fix(sessions): surface working_directory and agent_profile on list_sessions

CHANGELOG.md

@@ -9,6 +9,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ### Fixed
 
 - force fast-uri 3.1.5 in aidlc-portfolio examples (#551) (#552)
+- `list_sessions` ownership metadata now persists the effective canonical launch directory, stays stable after pane `cd`, and purges stale terminal rows before same-name session relaunches so reused sessions report the new directory/profile (#497)
 
 
 ### Other

src/cli_agent_orchestrator/clients/database.py

@@ -39,6 +39,7 @@ class TerminalModel(Base):
     tmux_window = Column(String, nullable=False)  # "window-name"
     provider = Column(String, nullable=False)  # "kiro_cli", "claude_code"
     agent_profile = Column(String)  # "developer", "reviewer" (optional)
+    working_directory = Column(String, nullable=True)  # launch-time cwd (optional)
     allowed_tools = Column(String, nullable=True)  # JSON-encoded list of CAO tool names
     shell_command = Column(String, nullable=True)  # shell process name captured before kiro launch
     caller_id = Column(String, nullable=True)  # terminal that created this one (callback target)
@@ -851,6 +852,10 @@ def _migrate_terminals_schema() -> None:
             conn.execute("ALTER TABLE terminals ADD COLUMN engine TEXT")
             conn.commit()
             logger.info("Migration: added engine column to terminals table")
+        if "working_directory" not in columns:
+            conn.execute("ALTER TABLE terminals ADD COLUMN working_directory TEXT")
+            conn.commit()
+            logger.info("Migration: added working_directory column to terminals table")
         conn.close()
     except Exception as e:
         logger.warning(f"Migration check for terminals schema failed: {e}")
@@ -866,6 +871,7 @@ def create_terminal(
     shell_command: Optional[str] = None,
     caller_id: Optional[str] = None,
     engine: Optional[str] = None,
+    working_directory: Optional[str] = None,
 ) -> Dict[str, Any]:
     """Create terminal metadata record."""
     import json as _json
@@ -877,6 +883,7 @@ def create_terminal(
             tmux_window=tmux_window,
             provider=provider,
             agent_profile=agent_profile,
+            working_directory=working_directory,
             allowed_tools=_json.dumps(allowed_tools) if allowed_tools else None,
             shell_command=shell_command,
             caller_id=caller_id,
@@ -890,6 +897,7 @@ def create_terminal(
             "tmux_window": terminal.tmux_window,
             "provider": terminal.provider,
             "agent_profile": terminal.agent_profile,
+            "working_directory": terminal.working_directory,
             "allowed_tools": allowed_tools,
             "shell_command": terminal.shell_command,
             "caller_id": terminal.caller_id,
@@ -916,6 +924,7 @@ def get_terminal_metadata(terminal_id: str) -> Optional[Dict[str, Any]]:
             "tmux_window": terminal.tmux_window,
             "provider": terminal.provider,
             "agent_profile": terminal.agent_profile,
+            "working_directory": terminal.working_directory,
             "allowed_tools": allowed_tools,
             "shell_command": terminal.shell_command,
             "caller_id": terminal.caller_id,
@@ -935,6 +944,7 @@ def list_terminals_by_session(tmux_session: str) -> List[Dict[str, Any]]:
                 "tmux_window": t.tmux_window,
                 "provider": t.provider,
                 "agent_profile": t.agent_profile,
+                "working_directory": t.working_directory,
                 "engine": t.engine or ("v2" if t.provider == "kiro_cli" else None),
                 "last_active": t.last_active,
             }
@@ -975,6 +985,7 @@ def list_all_terminals() -> List[Dict[str, Any]]:
                 "tmux_window": t.tmux_window,
                 "provider": t.provider,
                 "agent_profile": t.agent_profile,
+                "working_directory": t.working_directory,
                 "engine": t.engine or ("v2" if t.provider == "kiro_cli" else None),
                 "last_active": t.last_active,
             }

src/cli_agent_orchestrator/ops_mcp_server/models.py

@@ -2,7 +2,7 @@
 
 from typing import Any, Dict, List, Optional
 
-from pydantic import BaseModel, Field
+from pydantic import BaseModel, ConfigDict, Field
 
 from cli_agent_orchestrator.services.install_service import InstallResult
 
@@ -44,9 +44,25 @@ class SessionListResult(BaseModel):
         default=None,
         description="Error message when success is False",
     )
-    sessions: List[Dict[str, Any]] = Field(
+    sessions: List["SessionListEntry"] = Field(
         default_factory=list,
-        description="Active CAO sessions with terminal counts and statuses",
+        description="Active CAO sessions with ownership metadata and statuses",
+    )
+
+
+class SessionListEntry(BaseModel):
+    """A single active CAO session returned by list_sessions."""
+
+    model_config = ConfigDict(extra="allow")
+
+    id: Optional[str] = Field(default=None, description="Session identifier")
+    name: Optional[str] = Field(default=None, description="Session display name")
+    status: Optional[str] = Field(default=None, description="Backend session status")
+    working_directory: Optional[str] = Field(
+        default=None, description="Best-effort launch or pane working directory"
+    )
+    agent_profile: Optional[str] = Field(
+        default=None, description="Agent profile for the session's first known terminal"
     )
 
 
@@ -62,6 +78,7 @@ class SendMessageResult(BaseModel):
     "InstallResult",
     "LaunchResult",
     "ProfileListResult",
+    "SessionListEntry",
     "SendMessageResult",
     "SessionListResult",
 ]

src/cli_agent_orchestrator/services/session_service.py

@@ -20,8 +20,9 @@
 """
 
 import logging
-from typing import Dict, List
+from typing import Any, Dict, List
 
+from cli_agent_orchestrator.backends.base import TerminalBackend
 from cli_agent_orchestrator.backends.registry import get_backend
 from cli_agent_orchestrator.clients.database import list_terminals_by_session
 from cli_agent_orchestrator.constants import SESSION_PREFIX
@@ -103,11 +104,70 @@ async def create_session(
     return terminal
 
 
+def _enrich_session_ownership(
+    backend: TerminalBackend, session_data: Dict[str, Any]
+) -> Dict[str, Any]:
+    """Add best-effort ownership metadata from the session's first known terminal."""
+    enriched = dict(session_data)
+    enriched.setdefault("working_directory", None)
+    enriched.setdefault("agent_profile", None)
+
+    # `... or ""` (not `.get("id", "")`): an explicit id=None must collapse to
+    # "" too, matching the sibling guard in list_sessions. `.get("id", "")`
+    # would yield the truthy string "None" and try to enrich a bogus session.
+    session_name = enriched.get("id") or ""
+    if not session_name:
+        return enriched
+
+    try:
+        terminals = list_terminals_by_session(session_name)
+    except Exception as e:
+        logger.warning(f"Failed to load terminal metadata for {session_name}: {e}")
+        terminals = []
+
+    ownership_terminal: Dict[str, Any] = {}
+    for terminal in terminals:
+        if terminal.get("agent_profile") or terminal.get("working_directory"):
+            ownership_terminal = terminal
+            break
+
+    if not ownership_terminal:
+        for terminal in terminals:
+            if terminal.get("tmux_window"):
+                ownership_terminal = terminal
+                break
+
+    if ownership_terminal:
+        enriched["agent_profile"] = ownership_terminal.get("agent_profile")
+        persisted_working_directory = ownership_terminal.get("working_directory")
+        if persisted_working_directory:
+            enriched["working_directory"] = persisted_working_directory
+        elif ownership_terminal.get("tmux_window"):
+            try:
+                enriched["working_directory"] = backend.get_pane_working_directory(
+                    session_name, ownership_terminal["tmux_window"]
+                )
+            except Exception as e:
+                logger.warning(f"Failed to resolve working directory for {session_name}: {e}")
+
+    return enriched
+
+
 def list_sessions() -> List[Dict]:
     """List all sessions from tmux."""
     try:
-        tmux_sessions = get_backend().list_sessions()
-        return [s for s in tmux_sessions if s["id"].startswith(SESSION_PREFIX)]
+        backend = get_backend()
+        tmux_sessions = backend.list_sessions()
+        return [
+            _enrich_session_ownership(backend, s)
+            for s in tmux_sessions
+            # Use .get() rather than s["id"]: a backend that returns a session
+            # dict without an "id" key must not blank the entire list (KeyError
+            # in this comprehension is swallowed by the outer except and returns
+            # []). Shipped backends always populate "id"; this hardens against a
+            # future backend that does not.
+            if (s.get("id") or "").startswith(SESSION_PREFIX)
+        ]
     except Exception as e:
         logger.error(f"Failed to list sessions: {e}")
         return []

src/cli_agent_orchestrator/services/terminal_service.py

@@ -19,6 +19,7 @@
 
 import asyncio
 import logging
+import os
 import re
 import threading
 import time
@@ -33,6 +34,7 @@
 from cli_agent_orchestrator.clients.database import create_terminal as db_create_terminal
 from cli_agent_orchestrator.clients.database import delete_terminal as db_delete_terminal
 from cli_agent_orchestrator.clients.database import (
+    delete_terminals_by_session,
     get_terminal_metadata,
     update_last_active,
     update_terminal_shell_command,
@@ -72,6 +74,7 @@
 )
 from cli_agent_orchestrator.services.status_monitor import status_monitor
 from cli_agent_orchestrator.utils.agent_profiles import load_agent_profile
+from cli_agent_orchestrator.utils.path_validation import resolve_and_validate_path
 from cli_agent_orchestrator.utils.skills import build_skill_catalog
 from cli_agent_orchestrator.utils.terminal import (
     generate_session_name,
@@ -154,6 +157,16 @@ class OutputMode(str, Enum):
 }
 
 
+def _resolve_working_directory(working_directory: Optional[str]) -> str:
+    """Resolve launch cwd exactly as the tmux backend does before creation."""
+    return resolve_and_validate_path(
+        working_directory if working_directory is not None else os.getcwd(),
+        allow_create=False,
+        allow_file=False,
+        description="Working directory",
+    )
+
+
 async def create_terminal(
     provider: str,
     agent_profile: str,
@@ -289,6 +302,7 @@ async def create_terminal(
             session_name = generate_session_name()
 
         window_name = generate_window_name(agent_profile)
+        resolved_working_directory = _resolve_working_directory(working_directory)
 
         # Step 2: Create tmux session or window
         if new_session:
@@ -309,10 +323,11 @@ async def create_terminal(
                 session_name,
                 window_name,
                 terminal_id,
-                working_directory,
+                resolved_working_directory,
                 extra_env=env_vars,
             )
             session_created = True  # only set after successful creation
+            delete_terminals_by_session(session_name)
 
             # Persist forwarded env only after the tmux session actually
             # exists; the failure path below clears it if a later step
@@ -331,7 +346,7 @@ async def create_terminal(
                 session_name,
                 window_name,
                 terminal_id,
-                working_directory,
+                resolved_working_directory,
                 extra_env={**get_session_env(session_name), **(env_vars or {})},
             )
             window_created = True  # only set after successful creation
@@ -369,6 +384,7 @@ async def create_terminal(
             allowed_tools,
             caller_id=caller_id,
             engine=resolved_engine.value if resolved_engine is not None else None,
+            working_directory=resolved_working_directory,
         )
 
         # Step 4/5: Set up the FIFO event-driven output pipeline for pipe-pane