orchestrata_ultimate.session.portfolio

Reference › orchestrata_ultimate › session › portfolio

PortfolioController coordinates sessions and work across several projects. It owns a portfolio file with a PortfolioData record (see models). initialize creates a portfolio with a name and summary and load reads it; file_exists checks for it. Projects are managed with add_project, remove_project, list_projects and get_project. begin_session starts a portfolio-level session, and link_project_session links it to a numbered session of a project, with an optional focus. Portfolio work items are handled by add_work, pending_work and complete_work; get_state and update_state handle the portfolio state, get_session and get_latest_session read sessions, and report returns statistics. Usage is shown in examples/portfolio_usage.

At a glance

Class diagram

Dependency diagram

API

Classes

PortfolioController

Decorators: traces(spec='006', story='US-007', reqs=['REQ-006-US-007-001', 'REQ-006-US-007-002', 'REQ-006-US-007-003', 'REQ-006-US-… · Line 36

Controls portfolio operations.

Source

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

"""Fresh implementation of portfolio management.

Multi-project coordination across portfolios.
"""

import json
import logging
from datetime import datetime
from pathlib import Path

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.session.models import (
    PortfolioData,
    PortfolioProjectRef,
    PortfolioSessionRecord,
    PortfolioState,
    WorkItem,
    default_ai_model,
)

logger = logging.getLogger(__name__)


@traces(
    spec="006",
    story="US-007",
    reqs=[
        "REQ-006-US-007-001",
        "REQ-006-US-007-002",
        "REQ-006-US-007-003",
        "REQ-006-US-007-004",
        "REQ-006-US-007-005",
        "REQ-006-US-007-006",
    ],
)
class PortfolioController:
    """Controls portfolio operations."""

    FILE_NAME = "portfolio.json"

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

        Args:
            root: Portfolio root directory.
        """
        self.root = Path(root)
        self.file_path = self.root / self.FILE_NAME

    def file_exists(self) -> bool:
        """Check if portfolio file exists."""
        return self.file_path.is_file()

    def initialize(
        self,
        name: str,
        summary: str | None = None,
    ) -> PortfolioData:
        """Create new portfolio.

        Args:
            name: Portfolio name.
            summary: Optional description.

        Returns:
            Created portfolio.
        """
        portfolio = PortfolioData(name=name, summary=summary)
        self._save(portfolio)
        return portfolio

    def load(self) -> PortfolioData:
        """Load portfolio from file.

        Returns:
            Portfolio data.

        Raises:
            FileNotFoundError: If not found.
        """
        if not self.file_exists():
            raise FileNotFoundError(f"No portfolio at {self.file_path}")

        raw = self.file_path.read_text(encoding="utf-8")
        data = json.loads(raw)
        return PortfolioData.model_validate(data)

    def add_project(
        self,
        name: str,
        path: str,
        summary: str | None = None,
    ) -> PortfolioProjectRef:
        """Add project to portfolio.

        Args:
            name: Project name.
            path: Project path.
            summary: Optional description.

        Returns:
            Created reference.
        """
        portfolio = self.load()
        ref = portfolio.add_project(name, path, summary)
        self._save(portfolio)
        return ref

    def remove_project(self, name: str) -> bool:
        """Remove project from portfolio.

        Args:
            name: Project name.

        Returns:
            True if removed.
        """
        portfolio = self.load()
        removed = portfolio.drop_project(name)
        if removed:
            self._save(portfolio)
        return removed

    def list_projects(self) -> list[PortfolioProjectRef]:
        """Get all projects."""
        portfolio = self.load()
        return portfolio.projects

    def get_project(self, name: str) -> PortfolioProjectRef | None:
        """Get project by name."""
        portfolio = self.load()
        return portfolio.get_project(name)

    @traces(
        spec="006", story="US-001", reqs=["REQ-006-US-001-002", "REQ-006-US-001-003"]
    )
    def begin_session(
        self,
        purpose: str,
        objectives: list[str] | None = None,
        ai_model: str | None = None,
    ) -> PortfolioSessionRecord:
        """Start portfolio session.

        Args:
            purpose: Session purpose.
            objectives: Optional goals.
            ai_model: AI model name.

        Returns:
            Created session.
        """
        portfolio = self.load()
        seq = portfolio.next_session_num()

        session = PortfolioSessionRecord(
            seq=seq,
            purpose=purpose,
            objectives=objectives or [],
            ai_model=ai_model or default_ai_model(),
        )

        portfolio.add_session(session)
        self._save(portfolio)
        return session

    def link_project_session(
        self,
        portfolio_seq: int,
        project_name: str,
        project_seq: int,
        focus: str | None = None,
    ) -> bool:
        """Link project session to portfolio session.

        Args:
            portfolio_seq: Portfolio session number.
            project_name: Project name.
            project_seq: Project session number.
            focus: Optional focus.

        Returns:
            True if linked.
        """
        portfolio = self.load()

        for session in portfolio.sessions:
            if session.seq == portfolio_seq:
                session.link_session(project_name, project_seq, focus)
                self._save(portfolio)
                return True

        return False

    def get_session(self, seq: int) -> PortfolioSessionRecord | None:
        """Get session by number."""
        portfolio = self.load()
        for s in portfolio.sessions:
            if s.seq == seq:
                return s
        return None

    def get_latest_session(self) -> PortfolioSessionRecord | None:
        """Get most recent session."""
        portfolio = self.load()
        return portfolio.latest_session()

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

        Args:
            text: Work description.
            priority: Optional priority.

        Returns:
            Created work item.
        """
        portfolio = self.load()
        item = portfolio.add_work(text, priority)
        self._save(portfolio)
        return item

    def pending_work(self) -> list[WorkItem]:
        """Get pending work items."""
        portfolio = self.load()
        return portfolio.pending_work()

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

        Args:
            text: Work item text.

        Returns:
            True if marked.
        """
        portfolio = self.load()

        for item in portfolio.work_items:
            if item.text == text:
                item.mark_done()
                self._save(portfolio)
                return True

        return False

    def get_state(self) -> PortfolioState:
        """Get portfolio state."""
        portfolio = self.load()
        return portfolio.state

    def update_state(self, **updates: object) -> PortfolioState:
        """Update state fields.

        Args:
            **updates: Field updates.

        Returns:
            Updated state.
        """
        portfolio = self.load()

        for field, value in updates.items():
            if hasattr(portfolio.state, field):
                setattr(portfolio.state, field, value)

        portfolio.state.modified_at = datetime.now()
        self._save(portfolio)
        return portfolio.state

    def report(self) -> dict:
        """Generate portfolio statistics."""
        portfolio = self.load()

        return {
            "name": portfolio.name,
            "project_count": len(portfolio.projects),
            "active_projects": len([p for p in portfolio.projects if p.is_active]),
            "session_count": len(portfolio.sessions),
            "work_items": len(portfolio.work_items),
            "pending_work": len(portfolio.pending_work()),
            "created": portfolio.created,
            "modified": portfolio.state.modified_at,
        }

    def _save(self, portfolio: PortfolioData) -> None:
        """Save portfolio to file."""
        data = portfolio.model_dump(mode="json")
        content = json.dumps(data, indent=2, default=str)
        self.file_path.write_text(content, encoding="utf-8")