orchestrata_ultimate.session.carryover

Reference › orchestrata_ultimate › session › carryover

WorkCarryoverManager manages the deferred work items that carry from one session to the next. Items are WorkItem records from models, stored through the ContextFileManager. The CLI task commands in cli/tasks work on the same list.

The manager lists all or only pending items, filters by priority (high_priority_items returns P0 and P1), adds an item with an optional priority and notes, completes an item by text or by index, changes its priority, adds notes, removes it, and clears every done item, returning the count. summary returns counts, format_list renders a markdown list with or without done items, and import_from_session turns the upcoming items of a numbered session into work items and returns how many were imported. The interface that describes carryover in the dual-storage design is ICarryoverManager in interfaces.

At a glance

Class diagram

Dependency diagram

API

Classes

WorkCarryoverManager

Decorators: traces(spec='006', story='US-005', reqs=['REQ-006-US-005-001']) · Line 20

Manages work items that carry between sessions.

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/session/carryover.py

# mypy: ignore-errors
"""Fresh implementation of carryover management.

Manages work items that carry over between sessions.
"""

# Spec: 006 - Session Management

import logging
from pathlib import Path

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.session.context import ContextFileManager
from orchestrata_ultimate.session.models import SessionRecord, TaskState, WorkItem

logger = logging.getLogger(__name__)


@traces(spec="006", story="US-005", reqs=["REQ-006-US-005-001"])
class WorkCarryoverManager:
    """Manages work items that carry between sessions."""

    def __init__(self, root: str | Path) -> None:
        """Initialize manager.

        Args:
            root: Project root directory.
        """
        self.root = Path(root)
        self.ctx = ContextFileManager(root)

    def all_items(self) -> list[WorkItem]:
        """Get all work items."""
        return self.ctx.read_work_items()

    def pending_items(self) -> list[WorkItem]:
        """Get pending items (not done)."""
        items = self.ctx.read_work_items()
        return [i for i in items if i.state != TaskState.DONE]

    def items_by_priority(self, priority: str) -> list[WorkItem]:
        """Get items by priority.

        Args:
            priority: Priority level (P0, P1, etc).

        Returns:
            Matching items.
        """
        items = self.ctx.read_work_items()
        return [i for i in items if i.priority == priority]

    def high_priority_items(self) -> list[WorkItem]:
        """Get P0 and P1 items."""
        items = self.ctx.read_work_items()
        return [i for i in items if i.priority in ("P0", "P1")]

    def add_item(
        self,
        text: str,
        priority: str | None = None,
        notes: str | None = None,
    ) -> WorkItem:
        """Add new work item.

        Args:
            text: Item description.
            priority: Optional priority.
            notes: Optional notes.

        Returns:
            Created item.
        """
        item = WorkItem(
            text=text,
            priority=priority,
            notes=notes,
            state=TaskState.DEFERRED,
        )
        self.ctx.add_work_item(item)
        return item

    def complete_item(self, text: str) -> bool:
        """Mark item as done.

        Args:
            text: Item text to match.

        Returns:
            True if found and completed.
        """
        return self.ctx.complete_work_item(text)

    def complete_by_index(self, idx: int) -> bool:
        """Complete item by index.

        Args:
            idx: Index in list.

        Returns:
            True if completed.
        """
        if idx < 0:
            return False

        ctx = self.ctx.read()
        if idx >= len(ctx.deferred_work):
            return False

        ctx.deferred_work[idx].mark_done()
        self.ctx.write(ctx)
        return True

    def update_priority(self, text: str, priority: str) -> bool:
        """Update item priority.

        Args:
            text: Item text.
            priority: New priority.

        Returns:
            True if updated.
        """
        ctx = self.ctx.read()
        for item in ctx.deferred_work:
            if item.text == text:
                item.priority = priority
                self.ctx.write(ctx)
                return True
        return False

    def add_notes(self, text: str, notes: str) -> bool:
        """Add notes to item.

        Args:
            text: Item text.
            notes: Notes to add.

        Returns:
            True if updated.
        """
        ctx = self.ctx.read()
        for item in ctx.deferred_work:
            if item.text == text:
                if item.notes:
                    item.notes = f"{item.notes}; {notes}"
                else:
                    item.notes = notes
                self.ctx.write(ctx)
                return True
        return False

    def remove_item(self, text: str) -> bool:
        """Remove item.

        Args:
            text: Item text.

        Returns:
            True if removed.
        """
        ctx = self.ctx.read()
        for i, item in enumerate(ctx.deferred_work):
            if item.text == text:
                ctx.deferred_work.pop(i)
                self.ctx.write(ctx)
                return True
        return False

    def clear_done(self) -> int:
        """Remove all done items.

        Returns:
            Number removed.
        """
        ctx = self.ctx.read()
        original = len(ctx.deferred_work)
        ctx.deferred_work = [i for i in ctx.deferred_work if i.state != TaskState.DONE]
        removed = original - len(ctx.deferred_work)
        if removed > 0:
            self.ctx.write(ctx)
        return removed

    def summary(self) -> dict:
        """Get summary of work items.

        Returns:
            Summary dict.
        """
        items = self.ctx.read_work_items()
        by_priority = {}
        by_state = {}

        for item in items:
            p = item.priority or "none"
            by_priority[p] = by_priority.get(p, 0) + 1
            by_state[item.state.value] = by_state.get(item.state.value, 0) + 1

        pending = len([i for i in items if i.state != TaskState.DONE])

        return {
            "total": len(items),
            "pending": pending,
            "by_priority": by_priority,
            "by_state": by_state,
        }

    def format_list(self, include_done: bool = False) -> str:
        """Format work items as markdown.

        Args:
            include_done: Include done items.

        Returns:
            Formatted markdown.
        """
        items = self.ctx.read_work_items()

        if not include_done:
            items = [i for i in items if i.state != TaskState.DONE]

        if not items:
            return "## Carryover Tasks\n\nNo carryover tasks.\n"

        lines = ["## Carryover Tasks\n"]
        for item in items:
            mark = "x" if item.state == TaskState.DONE else " "
            priority = f"[{item.priority}] " if item.priority else ""
            lines.append(f"- [{mark}] {priority}{item.text}")
            if item.notes:
                lines.append(f"  - Notes: {item.notes}")

        return "\n".join(lines) + "\n"

    @traces(spec="001", story="US-004", reqs=["REQ-001-US-004-028"])
    def import_from_session(self, seq: int) -> int:
        """Import upcoming items from session as work items.

        Args:
            seq: Session number.

        Returns:
            Number imported.
        """
        ctx = self.ctx.read()
        session: SessionRecord | None = None

        for s in ctx.session_history:
            if s.seq == seq:
                session = s
                break

        if session is None:
            return 0

        existing = {i.text for i in ctx.deferred_work}
        imported = 0

        for step in session.upcoming:
            if step not in existing:
                ctx.deferred_work.append(WorkItem(text=step, state=TaskState.DEFERRED))
                imported += 1

        if imported > 0:
            self.ctx.write(ctx)

        return imported