orchestrata_ultimate.session.models

Reference › orchestrata_ultimate › session › models

The Pydantic data models of the session package: the documents that ContextFileManager serialises into context.json and renders into context.md, and that the controllers, managers and CLI commands pass around.

For a single project: WorkItem (a deferred item with a TaskState, priority and notes; mark_done, mark_todo and mark_deferred change its state), LogEntry (an entry in the session work log), FileArtifact (a file created or changed), TechChoice (a decision and its reason), HandoffData (decisions, blockers, active work, key files, commands and alerts for the next session), SessionRecord (one session, with session_id and header_line helpers), ProjectState (current state of the project) and ContextData, the full document with its session history, deferred work and state; it can return the latest session, the next session number and the pending work, and append a session.

For several projects: PortfolioProjectRef, ProjectSessionLink, PortfolioSessionRecord (which can link a project session), PortfolioState and PortfolioData, managed by the PortfolioController.

For goals (SPEC-023): GoalCriterion (an exit criterion with a status and an optional verify command), GoalTask, GoalProgress and Goal, used by GoalMixin. default_ai_model resolves the current AI-model label at call time.

At a glance

Class diagram

Dependency diagram

API

Classes

TaskState (enum)

Bases: StrEnum · Line 47

State of a work item.

WorkItem (pydantic)

Bases: BaseModel · Decorators: traces(spec='020', story='US-006', reqs=['REQ-020-174']) · Line 58

Single work item to track.

LogEntry (pydantic)

Bases: BaseModel · Decorators: traces(spec='006', story='FR-012', reqs=['REQ-006-FR12-002', 'REQ-006-FR12-005']) · Line 92

Entry in session work log.

FileArtifact (pydantic)

Bases: BaseModel · Line 100

File created during session.

TechChoice (pydantic)

Bases: BaseModel · Line 108

Technical decision made.

HandoffData (pydantic)

Bases: BaseModel · Line 116

Data for session handoff.

SessionRecord (pydantic)

Bases: BaseModel · Decorators: traces(spec='006', story='US-001', reqs=['REQ-006-FR10-004', 'REQ-006-FR12-002', 'REQ-006-FR2-002', 'REQ-006-US-001-003… · Line 139

Single session record.

ProjectState (pydantic)

Bases: BaseModel · Decorators: traces(spec='020', story='US-006', reqs=['REQ-020-173']) · Line 169

Current project state.

ContextData (pydantic)

Bases: BaseModel · Decorators: traces(spec='006', story='FR-014', reqs=['REQ-006-FR14-006']) · Line 188

Full context document data.

PortfolioProjectRef (pydantic)

Bases: BaseModel · Line 221

Reference to project in portfolio.

ProjectSessionLink (pydantic)

Bases: BaseModel · Line 237

Link to project session from portfolio.

PortfolioSessionRecord (pydantic)

Bases: BaseModel · Line 245

Portfolio-level session.

PortfolioState (pydantic)

Bases: BaseModel · Line 274

Portfolio state.

PortfolioData (pydantic)

Bases: BaseModel · Line 282

Full portfolio data.

GoalCriterion (pydantic)

Bases: BaseModel · Line 355

Exit criterion for goal achievement.

GoalTask (pydantic)

Bases: BaseModel · Line 369

Task that contributes to goal achievement.

GoalProgress (pydantic)

Bases: BaseModel · Line 385

Logged progress entry for a goal.

Goal (pydantic)

Bases: BaseModel · Line 397

A persistent goal with measurable exit criteria.

Functions

Source

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

"""Fresh implementation of session models.

Pydantic models for session management system.
All models use strict validation.
"""

# Spec: 006 - Session Management

import os
from datetime import datetime
from enum import StrEnum
from pathlib import Path

from pydantic import BaseModel, Field

from orchestrata_ultimate.core.base.traces import traces

# Portfolio goal domain types now live in core (layer 1) so the agents layer
# (U:PM GoalRegistry) can import them without an upward cross-layer import.
# Re-exported here (redundant ``as`` alias = explicit re-export) for backward
# compatibility; PortfolioGoal is also used below by ProjectState.
from orchestrata_ultimate.core.domain.goal import (
    PortfolioGoal as PortfolioGoal,
)
from orchestrata_ultimate.core.domain.goal import (
    PortfolioGoalPriority as PortfolioGoalPriority,
)
from orchestrata_ultimate.core.domain.goal import (
    PortfolioGoalStatus as PortfolioGoalStatus,
)
from orchestrata_ultimate.core.domain.goal import (
    PortfolioSuccessCriteria as PortfolioSuccessCriteria,
)

_DEFAULT_AI_MODEL = "Claude Opus 4.8"


def default_ai_model() -> str:
    """Current AI-model label, resolved at call time.

    Reads ``ORCHESTRATA_AI_MODEL`` so the model recorded on session records isn't
    frozen to a stale hardcoded default (previously "Claude Opus 4.5" in 5 places).
    """
    return os.environ.get("ORCHESTRATA_AI_MODEL", _DEFAULT_AI_MODEL)


class TaskState(StrEnum):
    """State of a work item."""

    TODO = "todo"
    ACTIVE = "active"
    DONE = "done"
    BLOCKED = "blocked"
    DEFERRED = "deferred"


@traces(spec="020", story="US-006", reqs=["REQ-020-174"])
class WorkItem(BaseModel):
    """Single work item to track."""

    text: str
    state: TaskState = TaskState.TODO
    priority: str | None = None
    notes: str | None = None
    timestamp: datetime = Field(default_factory=datetime.now)
    done_at: datetime | None = None

    @traces(spec="020", story="US-006", reqs=["REQ-020-174"])
    def mark_done(self) -> None:
        """Mark item as done."""
        self.state = TaskState.DONE
        self.done_at = datetime.now()

    @traces(spec="020", story="US-006", reqs=["REQ-020-174"])
    def mark_todo(self) -> None:
        """Return item to TODO, clearing any completion stamp.

        The inverse of :meth:`mark_done`. It lives here rather than in the
        caller because reopening means clearing ``done_at`` as well as
        ``state`` -- a caller that reassigns only ``state`` leaves an item
        that is open but carries a completion timestamp.
        """
        self.state = TaskState.TODO
        self.done_at = None

    def mark_deferred(self) -> None:
        """Mark item as deferred."""
        self.state = TaskState.DEFERRED


@traces(spec="006", story="FR-012", reqs=["REQ-006-FR12-002", "REQ-006-FR12-005"])
class LogEntry(BaseModel):
    """Entry in session work log."""

    desc: str
    details: list[str] = Field(default_factory=list)
    at: datetime = Field(default_factory=datetime.now)


class FileArtifact(BaseModel):
    """File created during session."""

    filepath: str
    line_count: int | None = None
    summary: str | None = None


class TechChoice(BaseModel):
    """Technical decision made."""

    choice: str
    reason: str
    at: datetime = Field(default_factory=datetime.now)


class HandoffData(BaseModel):
    """Data for session handoff."""

    decisions: list[str] = Field(default_factory=list)
    blockers: list[str] = Field(default_factory=list)
    active_work: list[str] = Field(default_factory=list)
    key_files: list[str] = Field(default_factory=list)
    cmds: list[str] = Field(default_factory=list)
    alerts: list[str] = Field(default_factory=list)


@traces(
    spec="006",
    story="US-001",
    reqs=[
        "REQ-006-FR10-004",
        "REQ-006-FR12-002",
        "REQ-006-FR2-002",
        "REQ-006-US-001-003",
        "REQ-006-US-003-003",
        "REQ-006-US-005-002",
    ],
)
class SessionRecord(BaseModel):
    """Single session record."""

    seq: int
    when: datetime
    ai_model: str
    purpose: str
    length: str | None = None

    objectives: list[str] = Field(default_factory=list)
    log: list[LogEntry] = Field(default_factory=list)
    artifacts: list[FileArtifact] = Field(default_factory=list)
    choices: list[TechChoice] = Field(default_factory=list)
    achievements: list[str] = Field(default_factory=list)
    upcoming: list[str] = Field(default_factory=list)
    handoff: HandoffData | None = None

    @property
    def session_id(self) -> str:
        """Get formatted session ID."""
        return f"{self.seq:03d}"

    @property
    def header_line(self) -> str:
        """Get formatted header line."""
        time_str = self.when.strftime("%Y-%m-%d %H:%M")
        return f"Session {self.session_id} - {time_str} - {self.ai_model}"


@traces(spec="020", story="US-006", reqs=["REQ-020-173"])
class ProjectState(BaseModel):
    """Current project state.

    Includes SPEC-023 goal system for persistent, measurable goals.
    """

    active_session: int | None = None
    version: str = "0.1.0"
    git_branch: str = "master"
    test_state: str | None = None
    status_line: str = "Initial setup"
    # SPEC-023: Goal system persistence
    goal: Goal | None = None
    # SPEC-012: Portfolio goal system persistence (distinct from the
    # singular session goal above — see PortfolioGoal's docstring).
    portfolio_goals: list[PortfolioGoal] = Field(default_factory=list)


@traces(spec="006", story="FR-014", reqs=["REQ-006-FR14-006"])
class ContextData(BaseModel):
    """Full context document data."""

    state: ProjectState = Field(default_factory=ProjectState)
    deferred_work: list[WorkItem] = Field(default_factory=list)
    session_history: list[SessionRecord] = Field(default_factory=list)

    @traces(spec="006", story="FR-010", reqs=["REQ-006-FR10-004"])
    def latest_session(self) -> SessionRecord | None:
        """Get most recent session."""
        if not self.session_history:
            return None
        return max(self.session_history, key=lambda s: s.seq)

    @traces(spec="006", story="FR-014", reqs=["REQ-006-FR14-007"])
    def next_session_num(self) -> int:
        """Get next session number."""
        if not self.session_history:
            return 0
        return max(s.seq for s in self.session_history) + 1

    @traces(spec="020", story="US-006", reqs=["REQ-020-172"])
    def append_session(self, session: SessionRecord) -> None:
        """Add session to history."""
        self.session_history.append(session)
        self.state.active_session = session.seq

    @traces(spec="006", story="US-006", reqs=["REQ-006-US-006-001"])
    def pending_work(self) -> list[WorkItem]:
        """Get pending work items."""
        return [w for w in self.deferred_work if w.state != TaskState.DONE]


class PortfolioProjectRef(BaseModel):
    """Reference to project in portfolio."""

    name: str
    path: str
    summary: str | None = None
    is_active: bool = True

    def absolute_path(self, base: Path) -> Path:
        """Get absolute path."""
        p = Path(self.path)
        if p.is_absolute():
            return p
        return base / p


class ProjectSessionLink(BaseModel):
    """Link to project session from portfolio."""

    project: str
    session_num: int
    focus: str | None = None


class PortfolioSessionRecord(BaseModel):
    """Portfolio-level session."""

    seq: int
    when: datetime = Field(default_factory=datetime.now)
    ai_model: str = Field(default_factory=default_ai_model)
    purpose: str
    length: str | None = None
    objectives: list[str] = Field(default_factory=list)
    linked: list[ProjectSessionLink] = Field(default_factory=list)
    log: list[LogEntry] = Field(default_factory=list)
    achievements: list[str] = Field(default_factory=list)
    upcoming: list[str] = Field(default_factory=list)
    handoff: HandoffData | None = None

    @property
    def session_id(self) -> str:
        """Get formatted session ID."""
        return f"{self.seq:03d}"

    def link_session(
        self, project: str, session_num: int, focus: str | None = None
    ) -> None:
        """Link a project session."""
        self.linked.append(
            ProjectSessionLink(project=project, session_num=session_num, focus=focus)
        )


class PortfolioState(BaseModel):
    """Portfolio state."""

    active_session: int | None = None
    status: str | None = None
    modified_at: datetime = Field(default_factory=datetime.now)


class PortfolioData(BaseModel):
    """Full portfolio data."""

    name: str
    summary: str | None = None
    created: datetime = Field(default_factory=datetime.now)
    state: PortfolioState = Field(default_factory=PortfolioState)
    projects: list[PortfolioProjectRef] = Field(default_factory=list)
    sessions: list[PortfolioSessionRecord] = Field(default_factory=list)
    work_items: list[WorkItem] = Field(default_factory=list)

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

    def get_project(self, name: str) -> PortfolioProjectRef | None:
        """Get project by name."""
        for p in self.projects:
            if p.name == name:
                return p
        return None

    def drop_project(self, name: str) -> bool:
        """Remove project by name."""
        for i, p in enumerate(self.projects):
            if p.name == name:
                self.projects.pop(i)
                return True
        return False

    def add_session(self, session: PortfolioSessionRecord) -> None:
        """Add session to portfolio."""
        self.sessions.append(session)
        self.state.active_session = session.seq
        self.state.modified_at = datetime.now()

    def latest_session(self) -> PortfolioSessionRecord | None:
        """Get most recent session."""
        if not self.sessions:
            return None
        return max(self.sessions, key=lambda s: s.seq)

    @traces(spec="006", story="FR-014", reqs=["REQ-006-FR14-007"])
    def next_session_num(self) -> int:
        """Get next session number."""
        if not self.sessions:
            return 0
        return max(s.seq for s in self.sessions) + 1

    def pending_work(self) -> list[WorkItem]:
        """Get pending work items."""
        return [
            w
            for w in self.work_items
            if w.state in (TaskState.TODO, TaskState.DEFERRED)
        ]

    def add_work(self, text: str, priority: str | None = None) -> WorkItem:
        """Add work item."""
        item = WorkItem(text=text, priority=priority, state=TaskState.TODO)
        self.work_items.append(item)
        return item


# ============================================================================
# SPEC-023: Goal-Driven Session System Models
# ============================================================================


class GoalCriterion(BaseModel):
    """Exit criterion for goal achievement.

    A boolean condition that must be true for the goal to be achieved.
    """

    id: str
    description: str
    status: str = Field(default="pending", description="pending, met, or failed")
    verify_command: str | None = None
    verify_pattern: str | None = None
    met_at: str | None = None


class GoalTask(BaseModel):
    """Task that contributes to goal achievement.

    A unit of work that moves one or more exit criteria from pending to met.
    """

    id: str
    description: str
    criteria: list[str] = Field(default_factory=list)
    status: str = Field(
        default="pending", description="pending, in_progress, done, or failed"
    )
    created_at: str = Field(default_factory=lambda: datetime.now().isoformat())
    done_at: str | None = None


class GoalProgress(BaseModel):
    """Logged progress entry for a goal.

    Records what changed toward the goal at a specific time.
    """

    at: str = Field(default_factory=lambda: datetime.now().isoformat())
    action: str
    note: str
    criteria_changed: dict[str, str] = Field(default_factory=dict)


class Goal(BaseModel):
    """A persistent goal with measurable exit criteria.

    Persists in context.json under state.goal. The agent commits to a goal,
    decomposes it into tasks, evaluates progress against exit criteria, and
    continues until the goal is achieved or blocked.
    """

    id: str = Field(
        default_factory=lambda: f"goal-{datetime.now().strftime('%Y%m%d-%H%M%S')}"
    )
    statement: str
    status: str = Field(
        default="active", description="active, achieved, blocked, or abandoned"
    )
    set_at: str = Field(default_factory=lambda: datetime.now().isoformat())
    achieved_at: str | None = None
    abandoned_at: str | None = None
    criteria: list[GoalCriterion] = Field(default_factory=list)
    tasks: list[GoalTask] = Field(default_factory=list)
    current_step: int = Field(
        default=0,
        ge=0,
        description="Index of the current task in ``tasks`` for plan-window rendering.",
    )
    progress: list[GoalProgress] = Field(default_factory=list)
    blockers: list[str] = Field(default_factory=list)
    retry_counts: dict[str, int] = Field(default_factory=dict)


# ============================================================================
# SPEC-012: Portfolio Goal System Models
# ============================================================================
# The portfolio goal domain types (PortfolioGoalStatus / PortfolioGoalPriority /
# PortfolioSuccessCriteria / PortfolioGoal) moved to
# ``orchestrata_ultimate.core.domain.goal`` (layer 1) so the agents layer
# (U:PM ``GoalRegistry``) can import them WITHOUT an upward cross-layer import.
# They are re-exported at the top of this module for backward compatibility, and
# ``ProjectState.portfolio_goals`` (above) references the re-exported name.