Persistent agent session UI with orchestration graph visualization

factory/agents/runner.py

@@ -163,6 +163,12 @@ async def invoke_agent(
 
     sid = _begin_span_safe(project_path, role, model=model, task=task)
 
+    parent_session_id = os.environ.get("FACTORY_SESSION_ID")
+    sess_id = _begin_session_safe(
+        project_path, role,
+        parent_id=parent_session_id, model=model, title=task[:200] if task else None,
+    )
+
     runner = get_runner(runner_name, project_path=project_path)
 
     agent_session_name = session_name or f"factory: {project_path.resolve().name}/{role}"
@@ -183,8 +189,11 @@ async def invoke_agent(
     )
 
     old_parent_span = os.environ.get("FACTORY_PARENT_SPAN_ID")
+    old_session_id = os.environ.get("FACTORY_SESSION_ID")
     if sid:
         os.environ["FACTORY_PARENT_SPAN_ID"] = sid
+    if sess_id:
+        os.environ["FACTORY_SESSION_ID"] = sess_id
     try:
         try:
             result = await runner.headless(request)
@@ -195,6 +204,7 @@ async def invoke_agent(
             logger.error("%s agent failed: %s", role, e)
             _emit_safe(project_path, "agent.failed", agent=role, data={"error": str(e)[:200]})
             _complete_span_safe(project_path, sid, status="failed")
+            _complete_session_safe(project_path, sess_id, status="failed")
             if _track_failures:
                 _consecutive_failures += 1
                 _check_failure_threshold(project_path, role)
@@ -210,6 +220,10 @@ async def invoke_agent(
                 project_path, sid, status="failed",
                 usage=usage, metadata=result.metadata, output=stdout,
             )
+            _complete_session_safe(
+                project_path, sess_id, status="failed",
+                usage=usage, metadata=result.metadata, output=stdout,
+            )
             if _track_failures:
                 _consecutive_failures += 1
                 _check_failure_threshold(project_path, role)
@@ -238,6 +252,10 @@ async def invoke_agent(
                 project_path, sid, status="completed",
                 usage=usage, metadata=result.metadata, output=stdout,
             )
+            _complete_session_safe(
+                project_path, sess_id, status="completed",
+                usage=usage, metadata=result.metadata, output=stdout,
+            )
             if _track_failures:
                 _consecutive_failures = 0
 
@@ -249,6 +267,10 @@ async def invoke_agent(
             os.environ["FACTORY_PARENT_SPAN_ID"] = old_parent_span
         elif sid:
             os.environ.pop("FACTORY_PARENT_SPAN_ID", None)
+        if old_session_id is not None:
+            os.environ["FACTORY_SESSION_ID"] = old_session_id
+        elif sess_id:
+            os.environ.pop("FACTORY_SESSION_ID", None)
 
 
 def _check_failure_threshold(project_path: Path, last_agent: str) -> None:
@@ -358,6 +380,52 @@ def _complete_span_safe(
         logger.debug("Failed to complete span %s", span_id, exc_info=True)
 
 
+def _begin_session_safe(
+    project_path: Path,
+    role: str,
+    *,
+    parent_id: str | None = None,
+    model: str | None = None,
+    title: str | None = None,
+    claude_session_id: str | None = None,
+) -> str | None:
+    """Begin a SQLite session, swallowing errors so agent invocation is never blocked."""
+    try:
+        from factory.sessions import begin_session
+
+        return begin_session(
+            project_path, role,
+            parent_id=parent_id, model=model, title=title,
+            claude_session_id=claude_session_id,
+        )
+    except Exception:
+        logger.debug("Failed to begin session for %s", role, exc_info=True)
+        return None
+
+
+def _complete_session_safe(
+    project_path: Path,
+    session_id: str | None,
+    *,
+    status: str = "completed",
+    usage: object | None = None,
+    metadata: dict[str, object] | None = None,
+    output: str | None = None,
+) -> None:
+    """Complete a SQLite session, swallowing errors so agent invocation is never blocked."""
+    if session_id is None:
+        return
+    try:
+        from factory.sessions import complete_session
+
+        complete_session(
+            project_path, session_id,
+            status=status, usage=usage, metadata=metadata, output=output,
+        )
+    except Exception:
+        logger.debug("Failed to complete session %s", session_id, exc_info=True)
+
+
 def _save_review(
     project_path: Path, role: str, output: str, return_code: int,
     review_tag: str | None = None,
@@ -396,10 +464,23 @@ def begin_cycle_session(
 ) -> str | None:
     """Create a root Langfuse trace for a factory cycle.
 
+    Also creates a root SQLite session so child agents can link via
+    FACTORY_SESSION_ID.
+
     Sets FACTORY_TRACE_ID and FACTORY_PARENT_SPAN_ID env vars so child
     agents link to this trace. Returns the span_id, or None if Langfuse
     is not configured.
     """
+    claude_session_id = os.environ.get("CLAUDE_CODE_SESSION_ID")
+
+    cycle_sess_id = _begin_session_safe(
+        project_path, "ceo", model=model,
+        title=f"cycle-{cycle_id}" if cycle_id else None,
+        claude_session_id=claude_session_id,
+    )
+    if cycle_sess_id:
+        os.environ["FACTORY_SESSION_ID"] = cycle_sess_id
+
     try:
         from factory.telemetry import begin_trace, is_enabled
 
@@ -425,7 +506,20 @@ def complete_cycle_session(
     project_path: Path,
     span_id: str | None,
 ) -> None:
-    """Mark a root Langfuse trace as finished and flush."""
+    """Mark a root Langfuse trace as finished and flush.
+
+    Also completes the root SQLite session if one was started.
+    """
+    cycle_sess_id = os.environ.pop("FACTORY_SESSION_ID", None)
+    claude_session_id = os.environ.get("CLAUDE_CODE_SESSION_ID")
+    metadata: dict[str, object] | None = None
+    if claude_session_id:
+        metadata = {"session_id": claude_session_id}
+    _complete_session_safe(
+        project_path, cycle_sess_id, status="completed",
+        metadata=metadata,
+    )
+
     if span_id is None:
         return
     try:

factory/cli.py

@@ -1885,6 +1885,46 @@ def cmd_backfill_archive(args: argparse.Namespace) -> int:
     return 0
 
 
+def cmd_sessions(args: argparse.Namespace) -> int:
+    """List agent sessions from the SQLite session database."""
+    from factory.sessions import get_cycles, get_sessions
+
+    project_path = Path(args.path).resolve()
+    cycle_filter = getattr(args, "cycle", None)
+    role_filter = getattr(args, "role", None)
+
+    if cycle_filter:
+        rows = get_sessions(project_path, cycle_id=cycle_filter, role=role_filter)
+    elif role_filter:
+        rows = get_sessions(project_path, role=role_filter)
+    else:
+        rows = get_cycles(project_path)
+
+    if not rows:
+        print("No sessions found.")
+        return 0
+
+    header = f"{'ID':<14} {'Role':<14} {'Status':<12} {'Duration':>10} {'Cost':>10}"
+    print(header)
+    print("-" * len(header))
+
+    for r in rows:
+        sid = r.get("id", "")[:13]
+        role = (r.get("agent_role") or "unknown")[:13]
+        status = (r.get("status") or "unknown")[:11]
+        duration_ms = r.get("duration_ms") or r.get("total_duration") or 0.0
+        duration_s = duration_ms / 1000.0
+        if duration_s >= 60:
+            dur_str = f"{duration_s / 60:.1f}m"
+        else:
+            dur_str = f"{duration_s:.1f}s"
+        cost = r.get("total_cost_usd") or r.get("total_cost") or 0.0
+        cost_str = f"${cost:.4f}" if cost else "-"
+        print(f"{sid:<14} {role:<14} {status:<12} {dur_str:>10} {cost_str:>10}")
+
+    return 0
+
+
 def cmd_diff(args: argparse.Namespace) -> int:
     """Compare two experiments side-by-side."""
     from factory.analysis import compare_experiments, format_comparison
@@ -2310,7 +2350,7 @@ def cmd_ceo(args: argparse.Namespace) -> int:
     With --headless: pipe mode via claude -p (for scripting, cron, etc.).
     With --mode design: brainstorm an idea via research + Strategist before building.
     """
-    from factory.agents.runner import resolve_prompt
+    from factory.agents.runner import begin_cycle_session, complete_cycle_session, resolve_prompt
     from factory.runners import get_runner
     from factory.user_config import load_config
 
@@ -2618,6 +2658,7 @@ def cmd_ceo(args: argparse.Namespace) -> int:
         # Uses completion guard to auto-resume on premature exit
         from factory.ceo_completion import run_ceo_with_completion_guard
 
+        span_id = begin_cycle_session(wt_path, cycle_id=session_name, model=model)
         try:
             result, code = _run(run_ceo_with_completion_guard(
                 wt_path,
@@ -2646,12 +2687,14 @@ def cmd_ceo(args: argparse.Namespace) -> int:
                 background=background,
             )
         finally:
+            complete_cycle_session(wt_path, span_id)
             remove_worktree(project_path, wt_path, wt_branch)
             if needs_materialize and _is_scaffold_only(project_path):
                 import shutil
                 shutil.rmtree(project_path, ignore_errors=True)
 
     # Interactive foreground mode: use subprocess.run so we can clean up the worktree.
+    span_id = begin_cycle_session(wt_path, cycle_id=session_name, model=model)
     try:
         if pending_ids:
             print(
@@ -2669,6 +2712,7 @@ def cmd_ceo(args: argparse.Namespace) -> int:
             session_name=session_name,
         ))
     finally:
+        complete_cycle_session(wt_path, span_id)
         remove_worktree(project_path, wt_path, wt_branch)
         if needs_materialize and _is_scaffold_only(project_path):
             import shutil
@@ -3871,6 +3915,12 @@ def build_parser() -> argparse.ArgumentParser:
     p = sub.add_parser("research", help="Print research citation index for experiments")
     p.add_argument("path", help="Path to the project")
 
+    # sessions
+    p = sub.add_parser("sessions", help="List agent sessions from the local SQLite database")
+    p.add_argument("path", help="Path to the project")
+    p.add_argument("--cycle", default=None, help="Filter by cycle (root session) ID")
+    p.add_argument("--role", default=None, help="Filter by agent role")
+
     # diff
     p = sub.add_parser("diff", help="Compare two experiments side-by-side")
     p.add_argument("path", help="Path to the project")
@@ -4320,6 +4370,7 @@ def main(argv: list[str] | None = None) -> int:
         "research": cmd_research,
         "backfill-citations": cmd_backfill_citations,
         "backfill-archive": cmd_backfill_archive,
+        "sessions": cmd_sessions,
         "diff": cmd_diff,
         "explain": cmd_explain,
         "export": cmd_export,