diff --git a/CHANGELOG.md b/CHANGELOG.md
index 368f48803..ddf8e99b5 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -11,6 +11,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 - tmux listing parse failures are retried once and reported as a distinct condition instead of surfacing as a bare `ValueError` that reads like "session not found" one layer up. libtmux 0.53.1+ zips `parse_output`'s fields with `strict=True`, so any short row (a pane or session vanishing mid-listing, or trailing fields tmux omits) raised `ValueError: zip() argument 2 is shorter than argument 1` — which propagated through `server.sessions`/`window.panes`, blocked launches outright, and left the pipe-liveness watchdog unable to tell a genuinely-gone session from a transient parse failure. Adds `TmuxLookupError` and routes the listing reads in `clients/tmux.py` through a single retry-and-classify wrapper; a failed `create_session` no longer leaves an orphaned tmux session that blocks relaunching the same name. Also caps `libtmux<0.53.1`, the last release that zips non-strict (caom-anv)
 - Codex handoff extraction now skips native TUI activity cells without relying on an English verb allowlist, including when the model's reply starts with prose (#545)
+- `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)
 
 ## [2.4.1] - 2026-08-04
 
diff --git a/src/cli_agent_orchestrator/clients/database.py b/src/cli_agent_orchestrator/clients/database.py
index ee18c94fc..ddec3adb8 100644
--- a/src/cli_agent_orchestrator/clients/database.py
+++ b/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)
@@ -1015,6 +1016,10 @@ def _migrate_terminals_schema() -> None:
             conn.execute('ALTER TABLE terminals ADD COLUMN "metadata" TEXT')
             conn.commit()
             logger.info("Migration: added metadata 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}")
@@ -1032,6 +1037,7 @@ def create_terminal(
     engine: Optional[str] = None,
     group: Optional[List[str]] = None,
     metadata: Optional[Dict[str, Any]] = None,
+    working_directory: Optional[str] = None,
 ) -> Dict[str, Any]:
     """Create terminal metadata record."""
     import json as _json
@@ -1043,6 +1049,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,
@@ -1058,6 +1065,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,
@@ -1094,6 +1102,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,
@@ -1262,6 +1271,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,
             }
@@ -1302,6 +1312,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,
             }
diff --git a/src/cli_agent_orchestrator/ops_mcp_server/models.py b/src/cli_agent_orchestrator/ops_mcp_server/models.py
index 2b12f732c..687c66b7c 100644
--- a/src/cli_agent_orchestrator/ops_mcp_server/models.py
+++ b/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",
 ]
diff --git a/src/cli_agent_orchestrator/services/session_service.py b/src/cli_agent_orchestrator/services/session_service.py
index dac57c3e3..1da51d9c8 100644
--- a/src/cli_agent_orchestrator/services/session_service.py
+++ b/src/cli_agent_orchestrator/services/session_service.py
@@ -22,6 +22,7 @@
 import logging
 from typing import Any, Dict, List, Optional
 
+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
@@ -112,11 +113,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 []
diff --git a/src/cli_agent_orchestrator/services/terminal_service.py b/src/cli_agent_orchestrator/services/terminal_service.py
index 841b72005..157b6d0e4 100644
--- a/src/cli_agent_orchestrator/services/terminal_service.py
+++ b/src/cli_agent_orchestrator/services/terminal_service.py
@@ -34,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,
     list_siblings_by_group_prefix,
     update_last_active,
@@ -82,6 +83,7 @@
 from cli_agent_orchestrator.services.status_monitor import status_monitor
 from cli_agent_orchestrator.services.step_output_store import _validate_key_part
 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,
@@ -169,6 +171,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,
@@ -347,6 +359,13 @@ async def create_terminal(
                 worktree_service.create_worktree, worktree_repo_root, terminal_id
             )
 
+        # Resolve AFTER the worktree block, not before: when `use_worktree` is set
+        # the block above REPLACES `working_directory` with the new worktree path,
+        # so resolving earlier would both launch tmux in the pre-worktree directory
+        # (defeating the isolation #100 provides) and persist that stale path as the
+        # terminal's working_directory. This is the effective launch cwd either way.
+        resolved_working_directory = _resolve_working_directory(working_directory)
+
         # Step 2: Create tmux session or window
         if new_session:
             # Ensure session name has the CAO prefix for identification
@@ -366,10 +385,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
@@ -388,7 +408,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
@@ -428,6 +448,7 @@ async def create_terminal(
             engine=resolved_engine.value if resolved_engine is not None else None,
             group=group,
             metadata=metadata,
+            working_directory=resolved_working_directory,
         )
 
         # Step 4/5: Set up the FIFO event-driven output pipeline for pipe-pane
diff --git a/test/clients/test_database.py b/test/clients/test_database.py
index a378ea969..2b623338b 100644
--- a/test/clients/test_database.py
+++ b/test/clients/test_database.py
@@ -81,6 +81,7 @@ def test_get_terminal_metadata_found(self, mock_session_class):
         mock_terminal.tmux_window = "window-0"
         mock_terminal.provider = "kiro_cli"
         mock_terminal.agent_profile = "developer"
+        mock_terminal.working_directory = "/workspace/project"
         mock_terminal.allowed_tools = None
         mock_terminal.group = None
         mock_terminal.metadata_json = None
@@ -97,6 +98,7 @@ def test_get_terminal_metadata_found(self, mock_session_class):
         assert result["id"] == "test123"
         assert result["group"] is None
         assert result["metadata"] is None
+        assert result["working_directory"] == "/workspace/project"
 
     @patch("cli_agent_orchestrator.clients.database.SessionLocal")
     def test_get_terminal_metadata_not_found(self, mock_session_class):
@@ -212,6 +214,7 @@ def test_list_terminals_by_session(self, mock_session_class):
         mock_terminal.tmux_window = "window-0"
         mock_terminal.provider = "kiro_cli"
         mock_terminal.agent_profile = "developer"
+        mock_terminal.working_directory = "/workspace/project"
         mock_terminal.last_active = datetime.now()
 
         mock_query = MagicMock()
@@ -223,6 +226,7 @@ def test_list_terminals_by_session(self, mock_session_class):
 
         assert len(result) == 1
         assert result[0]["id"] == "test123"
+        assert result[0]["working_directory"] == "/workspace/project"
 
     @patch("cli_agent_orchestrator.clients.database.SessionLocal")
     def test_list_pending_receiver_ids_by_provider(self, mock_session_class):
@@ -1415,10 +1419,10 @@ def test_init_db(self, mock_alias_migrate, mock_base):
 
 
 class TestTerminalsSchemaMigration:
-    """Tests for the terminals-table column-add migration (caller_id, issue #284)."""
+    """Tests for terminals-table additive column migrations."""
 
-    def test_caller_id_column_added_to_legacy_table(self, tmp_path, monkeypatch):
-        """A pre-#284 terminals table gains the caller_id column."""
+    def test_missing_terminal_columns_added_to_legacy_table(self, tmp_path, monkeypatch):
+        """A legacy terminals table gains nullable metadata columns."""
         import sqlite3
 
         from cli_agent_orchestrator.clients import database as db_mod
@@ -1447,9 +1451,10 @@ def test_caller_id_column_added_to_legacy_table(self, tmp_path, monkeypatch):
 
         with sqlite3.connect(str(db_file)) as conn:
             columns = {row[1] for row in conn.execute("PRAGMA table_info(terminals)")}
-            rows = conn.execute("SELECT id, caller_id FROM terminals").fetchall()
+            rows = conn.execute("SELECT id, caller_id, working_directory FROM terminals").fetchall()
         assert "caller_id" in columns
-        assert rows == [("abc12345", None)], "existing rows must get NULL caller_id"
+        assert "working_directory" in columns
+        assert rows == [("abc12345", None, None)], "existing rows must get NULL metadata values"
 
     def test_migration_is_idempotent(self, tmp_path, monkeypatch):
         """Running the migration twice must not fail or duplicate columns."""
@@ -1477,6 +1482,7 @@ def test_migration_is_idempotent(self, tmp_path, monkeypatch):
         with sqlite3.connect(str(db_file)) as conn:
             columns = [row[1] for row in conn.execute("PRAGMA table_info(terminals)")]
         assert columns.count("caller_id") == 1
+        assert columns.count("working_directory") == 1
         assert columns.count("allowed_tools") == 1
 
     def test_group_and_metadata_columns_added_to_legacy_table(self, tmp_path, monkeypatch):
@@ -1650,13 +1656,13 @@ def assert_seeded_rows_intact():
         assert columns.count("metadata") == 1
 
 
-class TestCallerIdRoundTrip:
-    """caller_id must round-trip create→read (issue #284): a write path that
-    persists it and a read path that drops it would silently break callback
-    routing for every worker."""
+class TestTerminalMetadataRoundTrip:
+    """Terminal metadata used by orchestration surfaces must round-trip."""
 
-    def test_caller_id_round_trips_through_real_db(self, tmp_path, monkeypatch):
-        """create_terminal persists caller_id; get_terminal_metadata returns it."""
+    def test_caller_id_and_working_directory_round_trip_through_real_db(
+        self, tmp_path, monkeypatch
+    ):
+        """create_terminal persists ownership metadata; reads return it."""
         from sqlalchemy import create_engine
         from sqlalchemy.orm import sessionmaker
 
@@ -1667,16 +1673,24 @@ def test_caller_id_round_trips_through_real_db(self, tmp_path, monkeypatch):
         monkeypatch.setattr(db_mod, "SessionLocal", sessionmaker(bind=engine))
 
         created = create_terminal(
-            "abc12345", "cao-s", "w-0", "kiro_cli", "developer", caller_id="def67890"
+            "abc12345",
+            "cao-s",
+            "w-0",
+            "kiro_cli",
+            "developer",
+            caller_id="def67890",
+            working_directory="/workspace/project",
         )
         assert created["caller_id"] == "def67890"
+        assert created["working_directory"] == "/workspace/project"
 
         fetched = get_terminal_metadata("abc12345")
         assert fetched is not None
         assert fetched["caller_id"] == "def67890"
+        assert fetched["working_directory"] == "/workspace/project"
 
-    def test_caller_id_defaults_to_none(self, tmp_path, monkeypatch):
-        """Operator-launched terminals (no caller) round-trip NULL."""
+    def test_nullable_metadata_defaults_to_none(self, tmp_path, monkeypatch):
+        """Operator-launched terminals without optional metadata round-trip NULL."""
         from sqlalchemy import create_engine
         from sqlalchemy.orm import sessionmaker
 
@@ -1688,10 +1702,12 @@ def test_caller_id_defaults_to_none(self, tmp_path, monkeypatch):
 
         created = create_terminal("abc12345", "cao-s", "w-0", "kiro_cli")
         assert created["caller_id"] is None
+        assert created["working_directory"] is None
 
         fetched = get_terminal_metadata("abc12345")
         assert fetched is not None
         assert fetched["caller_id"] is None
+        assert fetched["working_directory"] is None
 
 
 class TestProjectAliasMigration:
diff --git a/test/ops_mcp_server/test_server.py b/test/ops_mcp_server/test_server.py
index 411083cd6..c794d9155 100644
--- a/test/ops_mcp_server/test_server.py
+++ b/test/ops_mcp_server/test_server.py
@@ -508,6 +508,10 @@ async def test_list_sessions_returns_list(self) -> None:
             result = await list_sessions()
 
         assert result == SessionListResult(success=True, sessions=sessions)
+        dumped_session = result.model_dump()["sessions"][0]
+        assert dumped_session["session_name"] == "cao-123"
+        assert dumped_session["terminal_count"] == 2
+        assert "working_directory" in dumped_session
 
     async def test_list_sessions_returns_empty_list(self) -> None:
         """Empty session lists should still be a successful result."""
diff --git a/test/services/test_kiro_engine_phase0.py b/test/services/test_kiro_engine_phase0.py
index 63b5f4ac6..cf3df82ec 100644
--- a/test/services/test_kiro_engine_phase0.py
+++ b/test/services/test_kiro_engine_phase0.py
@@ -21,6 +21,9 @@
 _MODULE = "cli_agent_orchestrator.services.terminal_service"
 
 
+pytestmark = pytest.mark.usefixtures("isolated_memory_db")
+
+
 @pytest.mark.asyncio
 async def test_capability_probe_does_not_block_event_loop():
     """A synchronous capability probe runs in a worker while async work advances."""
@@ -161,6 +164,7 @@ async def test_omitted_engine_launches_as_explicitly_pinned_v2():
         ),
         patch(f"{_MODULE}.get_backend") as backend,
         patch(f"{_MODULE}.db_create_terminal") as db_create,
+        patch(f"{_MODULE}.delete_terminals_by_session"),
         patch(f"{_MODULE}.fifo_manager"),
         patch(f"{_MODULE}.provider_manager") as providers,
         patch(f"{_MODULE}.generate_terminal_id", return_value="test1234"),
@@ -209,6 +213,7 @@ async def test_explicit_model_override_is_probed_even_when_profile_has_none():
         ),
         patch(f"{_MODULE}.get_backend") as backend,
         patch(f"{_MODULE}.db_create_terminal"),
+        patch(f"{_MODULE}.delete_terminals_by_session"),
         patch(f"{_MODULE}.fifo_manager"),
         patch(f"{_MODULE}.provider_manager") as providers,
         patch(f"{_MODULE}.generate_terminal_id", return_value="test1234"),
diff --git a/test/services/test_plugin_event_emission.py b/test/services/test_plugin_event_emission.py
index 3bf7c75ff..6ac8f0631 100644
--- a/test/services/test_plugin_event_emission.py
+++ b/test/services/test_plugin_event_emission.py
@@ -23,6 +23,8 @@
     send_input,
 )
 
+pytestmark = pytest.mark.usefixtures("isolated_memory_db")
+
 
 def _registry_mock() -> MagicMock:
     """Build a registry double whose async dispatch can be asserted directly."""
@@ -141,6 +143,7 @@ class TestTerminalPluginEvents:
     """Verify terminal lifecycle events are emitted correctly."""
 
     @pytest.mark.asyncio
+    @patch("cli_agent_orchestrator.services.terminal_service.delete_terminals_by_session")
     @patch("cli_agent_orchestrator.services.terminal_service.TERMINAL_LOG_DIR")
     @patch("cli_agent_orchestrator.services.terminal_service.build_skill_catalog", return_value="")
     @patch("cli_agent_orchestrator.services.terminal_service.load_agent_profile")
@@ -165,6 +168,7 @@ async def test_create_terminal_dispatches_post_create_terminal_event_after_setup
         mock_load_agent_profile,
         mock_build_skill_catalog,
         mock_log_dir,
+        mock_delete_terminals_by_session,
     ):
         """Terminal creation should emit only after persistence and startup complete."""
         registry = _registry_mock()
@@ -225,6 +229,8 @@ async def record_dispatch(*_args):
         assert event.provider == "kiro_cli"
 
     @pytest.mark.asyncio
+    @patch("cli_agent_orchestrator.services.terminal_service.db_delete_terminal")
+    @patch("cli_agent_orchestrator.services.terminal_service.delete_terminals_by_session")
     @patch("cli_agent_orchestrator.services.terminal_service.TERMINAL_LOG_DIR")
     @patch("cli_agent_orchestrator.services.terminal_service.build_skill_catalog", return_value="")
     @patch("cli_agent_orchestrator.services.terminal_service.load_agent_profile")
@@ -249,6 +255,8 @@ async def test_create_terminal_does_not_dispatch_on_failure(
         mock_load_agent_profile,
         mock_build_skill_catalog,
         mock_log_dir,
+        mock_delete_terminals_by_session,
+        mock_db_delete_terminal,
     ):
         """Terminal creation failures must not emit post_create_terminal."""
         registry = _registry_mock()
@@ -336,6 +344,8 @@ class TestMessagePluginEvents:
     """Verify message delivery emits the correct event payloads."""
 
     @pytest.mark.parametrize("orchestration_type", ["send_message", "assign", "handoff"])
+    @patch("cli_agent_orchestrator.services.terminal_service.MemoryService")
+    @patch("cli_agent_orchestrator.services.terminal_service.status_monitor")
     @patch("cli_agent_orchestrator.services.terminal_service.update_last_active")
     @patch("cli_agent_orchestrator.backends.registry._backend")
     @patch("cli_agent_orchestrator.services.terminal_service.provider_manager")
@@ -346,10 +356,13 @@ def test_send_input_dispatches_post_send_message_event_for_each_orchestration_mo
         mock_provider_manager,
         mock_tmux,
         mock_update_last_active,
+        mock_status_monitor,
+        mock_memory_service,
         orchestration_type,
     ):
         """Every successful delivery should emit one post_send_message event."""
         registry = _registry_mock()
+        mock_memory_service.return_value.get_curated_memory_context.return_value = ""
         call_order: list[str] = []
 
         async def record_dispatch(*_args):
@@ -386,11 +399,18 @@ async def record_dispatch(*_args):
         assert event.message == "Hello from supervisor"
         assert event.orchestration_type == orchestration_type
 
+    @patch("cli_agent_orchestrator.services.terminal_service.MemoryService")
+    @patch("cli_agent_orchestrator.services.terminal_service.status_monitor")
     @patch("cli_agent_orchestrator.backends.registry._backend")
     @patch("cli_agent_orchestrator.services.terminal_service.provider_manager")
     @patch("cli_agent_orchestrator.services.terminal_service.get_terminal_metadata")
     def test_send_input_does_not_dispatch_on_failure(
-        self, mock_get_metadata, mock_provider_manager, mock_tmux
+        self,
+        mock_get_metadata,
+        mock_provider_manager,
+        mock_tmux,
+        mock_status_monitor,
+        mock_memory_service,
     ):
         """Message delivery failures must not emit post_send_message."""
         registry = _registry_mock()
diff --git a/test/services/test_session_service.py b/test/services/test_session_service.py
index 2d302334d..92d72aa89 100644
--- a/test/services/test_session_service.py
+++ b/test/services/test_session_service.py
@@ -1,16 +1,33 @@
 """Tests for the session service."""
 
+import asyncio
+import contextlib
+import os
+import shlex
+import uuid
+from pathlib import Path
 from unittest.mock import ANY, MagicMock, patch
 
 import pytest
-
+import pytest_asyncio
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+
+from cli_agent_orchestrator.backends import registry as backend_registry
+from cli_agent_orchestrator.backends.tmux_backend import TmuxBackend
+from cli_agent_orchestrator.clients import database as db_mod
+from cli_agent_orchestrator.clients.database import get_terminal_metadata
 from cli_agent_orchestrator.models.inbox import OrchestrationType
+from cli_agent_orchestrator.services import fifo_reader as fifo_reader_mod
+from cli_agent_orchestrator.services import terminal_service
+from cli_agent_orchestrator.services.event_bus import bus
 from cli_agent_orchestrator.services.session_service import (
     create_session,
     delete_session,
     get_session,
     list_sessions,
 )
+from cli_agent_orchestrator.services.status_monitor import status_monitor
 
 
 class TestCreateSession:
@@ -117,19 +134,39 @@ async def test_create_session_rejects_empty_initial_message(self, mock_create_te
 class TestListSessions:
     """Tests for list_sessions function."""
 
+    class _FakeTmuxClient:
+        def __init__(self, sessions, working_directories):
+            self._sessions = sessions
+            self._working_directories = working_directories
+            self.cwd_calls = []
+
+        def list_sessions(self):
+            return self._sessions
+
+        def get_pane_working_directory(self, session_name, window_name):
+            self.cwd_calls.append((session_name, window_name))
+            value = self._working_directories[(session_name, window_name)]
+            if isinstance(value, Exception):
+                raise value
+            return value
+
+    @patch("cli_agent_orchestrator.services.session_service.list_terminals_by_session")
     @patch("cli_agent_orchestrator.services.session_service.get_backend")
-    def test_list_sessions_success(self, mock_get_backend):
+    def test_list_sessions_success(self, mock_get_backend, mock_list_terminals):
         """Test listing sessions successfully."""
         mock_get_backend.return_value.list_sessions.return_value = [
             {"id": "cao-session1", "name": "Session 1"},
             {"id": "cao-session2", "name": "Session 2"},
             {"id": "other-session", "name": "Other"},
         ]
+        mock_list_terminals.return_value = []
 
         result = list_sessions()
 
         assert len(result) == 2
         assert all(s["id"].startswith("cao-") for s in result)
+        assert all("working_directory" in s for s in result)
+        assert all("agent_profile" in s for s in result)
 
     @patch("cli_agent_orchestrator.services.session_service.get_backend")
     def test_list_sessions_empty(self, mock_get_backend):
@@ -140,8 +177,9 @@ def test_list_sessions_empty(self, mock_get_backend):
 
         assert result == []
 
+    @patch("cli_agent_orchestrator.services.session_service.list_terminals_by_session")
     @patch("cli_agent_orchestrator.services.session_service.get_backend")
-    def test_list_sessions_no_cao_sessions(self, mock_get_backend):
+    def test_list_sessions_no_cao_sessions(self, mock_get_backend, mock_list_terminals):
         """Test listing sessions when no CAO sessions exist."""
         mock_get_backend.return_value.list_sessions.return_value = [
             {"id": "other-session1", "name": "Other 1"},
@@ -151,6 +189,7 @@ def test_list_sessions_no_cao_sessions(self, mock_get_backend):
         result = list_sessions()
 
         assert result == []
+        mock_list_terminals.assert_not_called()
 
     @patch("cli_agent_orchestrator.services.session_service.get_backend")
     def test_list_sessions_error(self, mock_get_backend):
@@ -161,6 +200,376 @@ def test_list_sessions_error(self, mock_get_backend):
 
         assert result == []
 
+    @patch("cli_agent_orchestrator.services.session_service.list_terminals_by_session")
+    @patch("cli_agent_orchestrator.services.session_service.get_backend")
+    def test_list_sessions_prefers_persisted_working_directory(
+        self, mock_get_backend, mock_list_terminals
+    ):
+        """Launch-time cwd from terminal metadata is the preferred ownership signal."""
+        fake_client = self._FakeTmuxClient(
+            [{"id": "cao-owned", "name": "cao-owned", "status": "detached"}],
+            {("cao-owned", "developer-abcd"): AssertionError("pane cwd should not be used")},
+        )
+        mock_get_backend.return_value = TmuxBackend(client=fake_client)
+        mock_list_terminals.return_value = [
+            {
+                "id": "term1",
+                "tmux_session": "cao-owned",
+                "tmux_window": "developer-abcd",
+                "agent_profile": "developer",
+                "working_directory": "/launch/project",
+            }
+        ]
+
+        result = list_sessions()
+
+        assert result == [
+            {
+                "id": "cao-owned",
+                "name": "cao-owned",
+                "status": "detached",
+                "agent_profile": "developer",
+                "working_directory": "/launch/project",
+            }
+        ]
+        assert fake_client.cwd_calls == []
+
+    @patch("cli_agent_orchestrator.services.session_service.list_terminals_by_session")
+    @patch("cli_agent_orchestrator.services.session_service.get_backend")
+    def test_list_sessions_falls_back_to_pane_working_directory(
+        self, mock_get_backend, mock_list_terminals
+    ):
+        """When no launch cwd is stored, list_sessions resolves the pane cwd."""
+        fake_client = self._FakeTmuxClient(
+            [{"id": "cao-owned", "name": "cao-owned", "status": "detached"}],
+            {("cao-owned", "developer-abcd"): "/pane/project"},
+        )
+        mock_get_backend.return_value = TmuxBackend(client=fake_client)
+        mock_list_terminals.return_value = [
+            {
+                "id": "term1",
+                "tmux_session": "cao-owned",
+                "tmux_window": "developer-abcd",
+                "agent_profile": "developer",
+                "working_directory": None,
+            }
+        ]
+
+        result = list_sessions()
+
+        assert result[0]["working_directory"] == "/pane/project"
+        assert result[0]["agent_profile"] == "developer"
+        assert fake_client.cwd_calls == [("cao-owned", "developer-abcd")]
+
+    @patch("cli_agent_orchestrator.services.session_service.list_terminals_by_session")
+    @patch("cli_agent_orchestrator.services.session_service.get_backend")
+    def test_list_sessions_keeps_session_when_working_directory_unresolvable(
+        self, mock_get_backend, mock_list_terminals
+    ):
+        """A cwd resolution failure affects only that field, not the session list."""
+        fake_client = self._FakeTmuxClient(
+            [{"id": "cao-owned", "name": "cao-owned", "status": "detached"}],
+            {("cao-owned", "developer-abcd"): RuntimeError("pane unavailable")},
+        )
+        mock_get_backend.return_value = TmuxBackend(client=fake_client)
+        mock_list_terminals.return_value = [
+            {
+                "id": "term1",
+                "tmux_session": "cao-owned",
+                "tmux_window": "developer-abcd",
+                "agent_profile": "developer",
+                "working_directory": None,
+            }
+        ]
+
+        result = list_sessions()
+
+        assert len(result) == 1
+        assert result[0]["id"] == "cao-owned"
+        assert result[0]["working_directory"] is None
+        assert result[0]["agent_profile"] == "developer"
+
+    @patch("cli_agent_orchestrator.services.session_service.list_terminals_by_session")
+    @patch("cli_agent_orchestrator.services.session_service.get_backend")
+    def test_list_sessions_handles_orphaned_tmux_session(
+        self, mock_get_backend, mock_list_terminals
+    ):
+        """A tmux session with no DB terminals still lists (null metadata)."""
+        fake_client = self._FakeTmuxClient(
+            [{"id": "cao-orphaned", "name": "Orphaned", "status": "active"}],
+            {},
+        )
+        mock_get_backend.return_value = TmuxBackend(client=fake_client)
+        mock_list_terminals.return_value = []
+
+        result = list_sessions()
+
+        assert len(result) == 1
+        assert result[0]["id"] == "cao-orphaned"
+        assert result[0]["working_directory"] is None
+        assert result[0]["agent_profile"] is None
+
+    @patch("cli_agent_orchestrator.services.session_service.list_terminals_by_session")
+    @patch("cli_agent_orchestrator.services.session_service.get_backend")
+    def test_list_sessions_handles_enrichment_exception_gracefully(
+        self, mock_get_backend, mock_list_terminals
+    ):
+        """One session's enrichment failure doesn't blank the entire list."""
+        fake_client = self._FakeTmuxClient(
+            [
+                {"id": "cao-good", "name": "Good", "status": "active"},
+                {"id": "cao-bad", "name": "Bad", "status": "active"},
+            ],
+            {("cao-good", "win-good"): "/home/user/project"},
+        )
+        mock_get_backend.return_value = TmuxBackend(client=fake_client)
+        mock_list_terminals.side_effect = [
+            [
+                {
+                    "id": "term-good",
+                    "tmux_session": "cao-good",
+                    "tmux_window": "win-good",
+                    "agent_profile": "developer",
+                    "working_directory": None,
+                }
+            ],
+            Exception("DB connection failed"),
+        ]
+
+        result = list_sessions()
+
+        assert len(result) == 2
+        assert result[0]["id"] == "cao-good"
+        assert result[0]["working_directory"] == "/home/user/project"
+        assert result[0]["agent_profile"] == "developer"
+        assert result[1]["id"] == "cao-bad"
+        assert result[1]["working_directory"] is None
+        assert result[1]["agent_profile"] is None
+
+    @patch("cli_agent_orchestrator.services.session_service.list_terminals_by_session")
+    @patch("cli_agent_orchestrator.services.session_service.get_backend")
+    def test_list_sessions_ignores_none_id_without_blanking_result(
+        self, mock_get_backend, mock_list_terminals
+    ):
+        """A backend row with id=None should be skipped without blanking valid rows."""
+        mock_get_backend.return_value.list_sessions.return_value = [
+            {"id": None, "name": "Bad"},
+            {"id": "cao-good", "name": "Good"},
+        ]
+        mock_list_terminals.return_value = []
+
+        result = list_sessions()
+
+        assert result == [
+            {
+                "id": "cao-good",
+                "name": "Good",
+                "working_directory": None,
+                "agent_profile": None,
+            }
+        ]
+
+    @patch("cli_agent_orchestrator.services.session_service.list_terminals_by_session")
+    @patch("cli_agent_orchestrator.services.session_service.get_backend")
+    def test_list_sessions_uses_one_terminal_for_profile_and_directory(
+        self, mock_get_backend, mock_list_terminals
+    ):
+        """Ownership metadata should not mix profile and cwd from different terminals."""
+        fake_client = self._FakeTmuxClient(
+            [{"id": "cao-owned", "name": "cao-owned", "status": "detached"}],
+            {("cao-owned", "developer-abcd"): "/pane/developer"},
+        )
+        mock_get_backend.return_value = TmuxBackend(client=fake_client)
+        mock_list_terminals.return_value = [
+            {
+                "id": "term1",
+                "tmux_session": "cao-owned",
+                "tmux_window": "developer-abcd",
+                "agent_profile": "developer",
+                "working_directory": None,
+            },
+            {
+                "id": "term2",
+                "tmux_session": "cao-owned",
+                "tmux_window": "reviewer-efgh",
+                "agent_profile": None,
+                "working_directory": "/launch/reviewer",
+            },
+        ]
+
+        result = list_sessions()
+
+        assert result[0]["agent_profile"] == "developer"
+        assert result[0]["working_directory"] == "/pane/developer"
+        assert fake_client.cwd_calls == [("cao-owned", "developer-abcd")]
+
+
+@pytest.fixture
+def real_session_db(tmp_path, monkeypatch):
+    """Route terminal metadata to a per-test real SQLite database."""
+    engine = create_engine(
+        f"sqlite:///{tmp_path / 'session-ownership.db'}",
+        connect_args={"check_same_thread": False},
+    )
+    db_mod.Base.metadata.create_all(bind=engine)
+    monkeypatch.setattr(
+        db_mod,
+        "SessionLocal",
+        sessionmaker(autocommit=False, autoflush=False, bind=engine),
+    )
+    try:
+        yield engine
+    finally:
+        engine.dispose()
+
+
+@pytest.fixture
+def real_tmux_backend(tmp_path, monkeypatch):
+    """Use a real tmux backend while keeping FIFO files in pytest's temp area."""
+    fifo_dir = Path(os.path.realpath(tmp_path / "fifos"))
+    fifo_dir.mkdir(parents=True, exist_ok=True)
+    monkeypatch.setattr(terminal_service, "FIFO_DIR", fifo_dir)
+    monkeypatch.setattr(fifo_reader_mod, "FIFO_DIR", fifo_dir)
+
+    backend = TmuxBackend()
+    monkeypatch.setattr(backend_registry, "_backend", backend)
+    return backend
+
+
+@pytest_asyncio.fixture
+async def running_status_monitor():
+    """Run the in-process status monitor used by mock_cli initialization."""
+    loop = asyncio.get_running_loop()
+    bus.set_loop(loop)
+    task = asyncio.create_task(status_monitor.run())
+    try:
+        yield
+    finally:
+        task.cancel()
+        with contextlib.suppress(asyncio.CancelledError):
+            await task
+        bus.set_loop(None)
+
+
+def _session_suffix() -> str:
+    return f"ownership-{uuid.uuid4().hex[:8]}"
+
+
+async def _wait_for_pane_directory(backend, session_name: str, window_name: str, expected: str):
+    deadline = asyncio.get_running_loop().time() + 8
+    while asyncio.get_running_loop().time() < deadline:
+        if backend.get_pane_working_directory(session_name, window_name) == expected:
+            return
+        await asyncio.sleep(0.2)
+    assert backend.get_pane_working_directory(session_name, window_name) == expected
+
+
+@pytest.mark.integration
+class TestSessionOwnershipIntegration:
+    """Regression tests for list_sessions ownership metadata persistence."""
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize(
+        "working_directory",
+        [None, "."],
+        ids=["omitted-working-directory", "relative-dot-working-directory"],
+    )
+    async def test_create_terminal_persists_effective_cwd_and_list_sessions_does_not_drift(
+        self,
+        working_directory,
+        tmp_path,
+        monkeypatch,
+        real_session_db,
+        real_tmux_backend,
+        running_status_monitor,
+    ):
+        project = tmp_path / "project"
+        drift = tmp_path / "drift"
+        project.mkdir()
+        drift.mkdir()
+        monkeypatch.chdir(project)
+        expected = os.path.realpath(project)
+        session_name = _session_suffix()
+        terminal = None
+
+        try:
+            terminal = await terminal_service.create_terminal(
+                provider="mock_cli",
+                agent_profile="developer",
+                session_name=session_name,
+                new_session=True,
+                working_directory=working_directory,
+            )
+
+            metadata = get_terminal_metadata(terminal.id)
+            assert metadata is not None
+            assert metadata["working_directory"] == expected
+
+            real_tmux_backend.send_keys(terminal.session_name, terminal.name, "/exit")
+            await asyncio.sleep(0.5)
+            real_tmux_backend.send_keys(
+                terminal.session_name,
+                terminal.name,
+                f"cd {shlex.quote(str(drift))}",
+            )
+            await _wait_for_pane_directory(
+                real_tmux_backend,
+                terminal.session_name,
+                terminal.name,
+                os.path.realpath(drift),
+            )
+
+            sessions = {s["id"]: s for s in list_sessions()}
+            assert sessions[terminal.session_name]["working_directory"] == expected
+            assert sessions[terminal.session_name]["agent_profile"] == "developer"
+        finally:
+            if terminal is not None:
+                with contextlib.suppress(Exception):
+                    delete_session(terminal.session_name)
+
+    @pytest.mark.asyncio
+    async def test_same_name_relaunch_purges_stale_terminal_metadata(
+        self,
+        tmp_path,
+        real_session_db,
+        real_tmux_backend,
+        running_status_monitor,
+    ):
+        old_project = tmp_path / "old-project"
+        new_project = tmp_path / "new-project"
+        old_project.mkdir()
+        new_project.mkdir()
+        session_name = _session_suffix()
+        live_session_name = f"cao-{session_name}"
+
+        first = await create_session(
+            provider="mock_cli",
+            agent_profile="developer",
+            session_name=session_name,
+            working_directory=str(old_project),
+        )
+        real_tmux_backend.kill_session(first.session_name)
+        terminal_service.fifo_manager.stop_reader(first.id)
+        terminal_service.status_monitor.clear_terminal(first.id)
+        terminal_service.provider_manager.cleanup_provider(first.id)
+
+        second = None
+        try:
+            second = await create_session(
+                provider="mock_cli",
+                agent_profile="reviewer",
+                session_name=session_name,
+                working_directory=str(new_project),
+            )
+
+            sessions = {s["id"]: s for s in list_sessions()}
+            assert sessions[live_session_name]["working_directory"] == os.path.realpath(new_project)
+            assert sessions[live_session_name]["agent_profile"] == "reviewer"
+        finally:
+            if second is not None:
+                with contextlib.suppress(Exception):
+                    delete_session(second.session_name)
+
 
 class TestGetSession:
     """Tests for get_session function."""
diff --git a/test/services/test_terminal_service_coverage.py b/test/services/test_terminal_service_coverage.py
index 1c31a626b..1fdbda835 100644
--- a/test/services/test_terminal_service_coverage.py
+++ b/test/services/test_terminal_service_coverage.py
@@ -10,11 +10,14 @@
 
 from cli_agent_orchestrator.models.agent_profile import AgentProfile
 
+pytestmark = pytest.mark.usefixtures("isolated_memory_db")
+
 
 class TestCreateTerminalCleanup:
     """Test error cleanup paths in create_terminal."""
 
     @pytest.mark.asyncio
+    @patch("cli_agent_orchestrator.services.terminal_service.delete_terminals_by_session")
     @patch("cli_agent_orchestrator.services.terminal_service.status_monitor")
     @patch("cli_agent_orchestrator.services.terminal_service.fifo_manager")
     @patch("cli_agent_orchestrator.services.terminal_service.TERMINAL_LOG_DIR")
@@ -42,6 +45,7 @@ async def test_cleanup_on_provider_init_failure(
         mock_log_dir,
         mock_fifo_manager,
         mock_status_monitor,
+        mock_delete_terminals_by_session,
     ):
         """When provider.initialize() fails, cleanup should kill session, cleanup
         provider, AND roll back the DB terminal row."""
@@ -235,6 +239,7 @@ async def test_cleanup_swallows_kill_window_errors(
         mock_tmux.kill_window.assert_called_once_with("cao-existing", "w1")
 
     @pytest.mark.asyncio
+    @patch("cli_agent_orchestrator.services.terminal_service.delete_terminals_by_session")
     @patch("cli_agent_orchestrator.services.terminal_service.status_monitor")
     @patch("cli_agent_orchestrator.services.terminal_service.fifo_manager")
     @patch("cli_agent_orchestrator.services.terminal_service.TERMINAL_LOG_DIR")
@@ -262,6 +267,7 @@ async def test_cleanup_ignores_cleanup_errors(
         mock_log_dir,
         mock_fifo_manager,
         mock_status_monitor,
+        mock_delete_terminals_by_session,
     ):
         """Cleanup errors should be swallowed, original error re-raised. The DB
         rollback still runs after cleanup_provider raises, and its own error is
@@ -293,6 +299,7 @@ async def test_cleanup_ignores_cleanup_errors(
         mock_db_delete.assert_called_once_with("tid1")
 
     @pytest.mark.asyncio
+    @patch("cli_agent_orchestrator.services.terminal_service.delete_terminals_by_session")
     @patch("cli_agent_orchestrator.services.terminal_service.status_monitor")
     @patch("cli_agent_orchestrator.services.terminal_service.fifo_manager")
     @patch("cli_agent_orchestrator.services.terminal_service.TERMINAL_LOG_DIR")
@@ -318,6 +325,7 @@ async def test_session_prefix_added_for_new_session(
         mock_log_dir,
         mock_fifo_manager,
         mock_status_monitor,
+        mock_delete_terminals_by_session,
     ):
         """New sessions without the prefix get it added automatically."""
         from cli_agent_orchestrator.services.terminal_service import create_terminal
@@ -357,6 +365,7 @@ class TestCreateTerminalSessionCleanupGuard:
     @patch("cli_agent_orchestrator.services.terminal_service.TERMINAL_LOG_DIR")
     @patch("cli_agent_orchestrator.backends.registry._backend")
     @patch("cli_agent_orchestrator.services.terminal_service.provider_manager")
+    @patch("cli_agent_orchestrator.services.terminal_service.db_delete_terminal")
     @patch("cli_agent_orchestrator.services.terminal_service.db_create_terminal")
     @patch(
         "cli_agent_orchestrator.services.terminal_service.generate_window_name", return_value="w1"
@@ -372,6 +381,7 @@ async def test_no_kill_session_when_session_already_exists(
         mock_tid,
         mock_wname,
         mock_db_create,
+        mock_db_delete,
         mock_pm,
         mock_tmux,
         mock_log_dir,
@@ -395,11 +405,13 @@ async def test_no_kill_session_when_session_already_exists(
         mock_tmux.kill_session.assert_not_called()
 
     @pytest.mark.asyncio
+    @patch("cli_agent_orchestrator.services.terminal_service.delete_terminals_by_session")
     @patch("cli_agent_orchestrator.services.terminal_service.status_monitor")
     @patch("cli_agent_orchestrator.services.terminal_service.fifo_manager")
     @patch("cli_agent_orchestrator.services.terminal_service.TERMINAL_LOG_DIR")
     @patch("cli_agent_orchestrator.backends.registry._backend")
     @patch("cli_agent_orchestrator.services.terminal_service.provider_manager")
+    @patch("cli_agent_orchestrator.services.terminal_service.db_delete_terminal")
     @patch("cli_agent_orchestrator.services.terminal_service.db_create_terminal")
     @patch(
         "cli_agent_orchestrator.services.terminal_service.generate_window_name", return_value="w1"
@@ -415,11 +427,13 @@ async def test_kill_session_when_we_created_it_and_later_step_fails(
         mock_tid,
         mock_wname,
         mock_db_create,
+        mock_db_delete,
         mock_pm,
         mock_tmux,
         mock_log_dir,
         mock_fifo_manager,
         mock_status_monitor,
+        mock_delete_terminals_by_session,
     ):
         """When we successfully created the session but a later step fails, cleanup SHOULD kill it."""
         from cli_agent_orchestrator.services.terminal_service import create_terminal
diff --git a/test/services/test_terminal_service_full.py b/test/services/test_terminal_service_full.py
index 1b6779d1a..d9c1ecf42 100644
--- a/test/services/test_terminal_service_full.py
+++ b/test/services/test_terminal_service_full.py
@@ -1,5 +1,6 @@
 """Full tests for terminal service."""
 
+import os
 from datetime import datetime
 from unittest.mock import AsyncMock, MagicMock, patch
 
@@ -19,11 +20,14 @@
     send_input,
 )
 
+pytestmark = pytest.mark.usefixtures("isolated_memory_db")
+
 
 class TestCreateTerminal:
     """Tests for create_terminal function."""
 
     @pytest.mark.asyncio
+    @patch("cli_agent_orchestrator.services.terminal_service.delete_terminals_by_session")
     @patch("cli_agent_orchestrator.services.terminal_service.status_monitor")
     @patch("cli_agent_orchestrator.services.terminal_service.fifo_manager")
     @patch("cli_agent_orchestrator.services.terminal_service.FIFO_DIR")
@@ -46,6 +50,7 @@ async def test_create_terminal_new_session(
         mock_fifo_dir,
         mock_fifo_manager,
         mock_status_monitor,
+        mock_delete_terminals_by_session,
     ):
         """Test creating terminal with new session."""
         mock_gen_id.return_value = "test1234"
@@ -65,6 +70,7 @@ async def test_create_terminal_new_session(
         mock_provider.initialize.assert_called_once()
 
     @pytest.mark.asyncio
+    @patch("cli_agent_orchestrator.services.terminal_service.delete_terminals_by_session")
     @patch("cli_agent_orchestrator.services.terminal_service._schedule_deferred_init")
     @patch("cli_agent_orchestrator.services.terminal_service.status_monitor")
     @patch("cli_agent_orchestrator.services.terminal_service.fifo_manager")
@@ -89,6 +95,7 @@ async def test_create_terminal_forwards_deferred_launch_payload(
         mock_fifo_manager,
         mock_status_monitor,
         mock_schedule_deferred_init,
+        mock_delete_terminals_by_session,
     ):
         """The real terminal layer sends the model to provider construction and
         the first task to the established deferred-init scheduler."""
@@ -127,6 +134,7 @@ async def test_create_terminal_forwards_deferred_launch_payload(
         )
 
     @pytest.mark.asyncio
+    @patch("cli_agent_orchestrator.services.terminal_service.delete_terminals_by_session")
     @patch("cli_agent_orchestrator.utils.tool_mapping.resolve_allowed_tools")
     @patch("cli_agent_orchestrator.services.terminal_service.status_monitor")
     @patch("cli_agent_orchestrator.services.terminal_service.fifo_manager")
@@ -151,6 +159,7 @@ async def test_create_terminal_persists_resolved_allowed_tools(
         mock_fifo_manager,
         mock_status_monitor,
         mock_resolve_allowed,
+        mock_delete_terminals_by_session,
     ):
         """Profile-derived restrictions should be persisted and used at launch."""
         mock_gen_id.return_value = "test1234"
@@ -182,10 +191,12 @@ async def test_create_terminal_persists_resolved_allowed_tools(
             engine="v2",
             group=None,
             metadata=None,
+            working_directory=os.path.realpath(os.getcwd()),
         )
         assert mock_provider_manager.create_provider.call_args.args[5] == ["fs_read"]
 
     @pytest.mark.asyncio
+    @patch("cli_agent_orchestrator.services.terminal_service.delete_terminals_by_session")
     @patch("cli_agent_orchestrator.services.terminal_service.status_monitor")
     @patch("cli_agent_orchestrator.services.terminal_service.fifo_manager")
     @patch("cli_agent_orchestrator.services.terminal_service.FIFO_DIR")
@@ -208,6 +219,7 @@ async def test_create_terminal_explicit_model_overrides_profile_model(
         mock_fifo_dir,
         mock_fifo_manager,
         mock_status_monitor,
+        mock_delete_terminals_by_session,
     ):
         """Regression: PR #501 review -- `model=model or (profile.model if
         profile else None)` in create_terminal is the line the entire
@@ -240,6 +252,7 @@ async def test_create_terminal_explicit_model_overrides_profile_model(
         )
 
     @pytest.mark.asyncio
+    @patch("cli_agent_orchestrator.services.terminal_service.delete_terminals_by_session")
     @patch("cli_agent_orchestrator.services.terminal_service.status_monitor")
     @patch("cli_agent_orchestrator.services.terminal_service.fifo_manager")
     @patch("cli_agent_orchestrator.services.terminal_service.FIFO_DIR")
@@ -262,6 +275,7 @@ async def test_create_terminal_falls_back_to_profile_model_when_no_override(
         mock_fifo_dir,
         mock_fifo_manager,
         mock_status_monitor,
+        mock_delete_terminals_by_session,
     ):
         """The other half of the same precedence line: with no explicit
         override, the profile's own model still reaches provider creation
@@ -286,6 +300,7 @@ async def test_create_terminal_falls_back_to_profile_model_when_no_override(
         )
 
     @pytest.mark.asyncio
+    @patch("cli_agent_orchestrator.services.terminal_service.delete_terminals_by_session")
     @patch("cli_agent_orchestrator.services.terminal_service.status_monitor")
     @patch("cli_agent_orchestrator.services.terminal_service.fifo_manager")
     @patch("cli_agent_orchestrator.services.terminal_service.FIFO_DIR")
@@ -308,6 +323,7 @@ async def test_create_terminal_persists_caller_id(
         mock_fifo_dir,
         mock_fifo_manager,
         mock_status_monitor,
+        mock_delete_terminals_by_session,
     ):
         """caller_id reaches the database row and the returned Terminal (issue #284)."""
         mock_gen_id.return_value = "test1234"
@@ -369,13 +385,20 @@ async def test_create_terminal_existing_session(
         mock_tmux.create_window.assert_called_once()
 
     @pytest.mark.asyncio
+    @patch("cli_agent_orchestrator.services.terminal_service.db_delete_terminal")
     @patch("cli_agent_orchestrator.backends.registry._backend")
     @patch("cli_agent_orchestrator.services.terminal_service.generate_window_name")
     @patch("cli_agent_orchestrator.services.terminal_service.generate_session_name")
     @patch("cli_agent_orchestrator.services.terminal_service.generate_terminal_id")
     @patch("cli_agent_orchestrator.services.terminal_service.load_agent_profile")
     async def test_create_terminal_session_not_found(
-        self, mock_load_profile, mock_gen_id, mock_gen_session, mock_gen_window, mock_tmux
+        self,
+        mock_load_profile,
+        mock_gen_id,
+        mock_gen_session,
+        mock_gen_window,
+        mock_tmux,
+        mock_db_delete,
     ):
         """Test creating terminal when session not found."""
         mock_gen_id.return_value = "test1234"
@@ -388,13 +411,20 @@ async def test_create_terminal_session_not_found(
             await create_terminal("kiro_cli", "developer", session_name="cao-nonexistent")
 
     @pytest.mark.asyncio
+    @patch("cli_agent_orchestrator.services.terminal_service.db_delete_terminal")
     @patch("cli_agent_orchestrator.backends.registry._backend")
     @patch("cli_agent_orchestrator.services.terminal_service.generate_window_name")
     @patch("cli_agent_orchestrator.services.terminal_service.generate_session_name")
     @patch("cli_agent_orchestrator.services.terminal_service.generate_terminal_id")
     @patch("cli_agent_orchestrator.services.terminal_service.load_agent_profile")
     async def test_create_terminal_session_already_exists(
-        self, mock_load_profile, mock_gen_id, mock_gen_session, mock_gen_window, mock_tmux
+        self,
+        mock_load_profile,
+        mock_gen_id,
+        mock_gen_session,
+        mock_gen_window,
+        mock_tmux,
+        mock_db_delete,
     ):
         """Test creating terminal when session already exists."""
         mock_gen_id.return_value = "test1234"
@@ -409,6 +439,7 @@ async def test_create_terminal_session_already_exists(
             )
 
     @pytest.mark.asyncio
+    @patch("cli_agent_orchestrator.services.terminal_service.delete_terminals_by_session")
     @patch("cli_agent_orchestrator.services.terminal_service.status_monitor")
     @patch("cli_agent_orchestrator.services.terminal_service.fifo_manager")
     @patch("cli_agent_orchestrator.services.terminal_service.FIFO_DIR")
@@ -435,6 +466,7 @@ async def test_create_terminal_appends_skill_catalog(
         mock_fifo_dir,
         mock_fifo_manager,
         mock_status_monitor,
+        mock_delete_terminals_by_session,
     ):
         """Providers that consume runtime prompts should receive the global skill catalog."""
         mock_gen_id.return_value = "test1234"
@@ -476,6 +508,7 @@ async def test_create_terminal_appends_skill_catalog(
         )
 
     @pytest.mark.asyncio
+    @patch("cli_agent_orchestrator.services.terminal_service.delete_terminals_by_session")
     @patch("cli_agent_orchestrator.services.terminal_service.status_monitor")
     @patch("cli_agent_orchestrator.services.terminal_service.fifo_manager")
     @patch("cli_agent_orchestrator.services.terminal_service.FIFO_DIR")
@@ -502,6 +535,7 @@ async def test_create_terminal_without_skills_is_unchanged(
         mock_fifo_dir,
         mock_fifo_manager,
         mock_status_monitor,
+        mock_delete_terminals_by_session,
     ):
         """Providers should receive an empty skill prompt when no skills are installed."""
         mock_gen_id.return_value = "test1234"
@@ -530,6 +564,7 @@ async def test_create_terminal_without_skills_is_unchanged(
 
     @pytest.mark.asyncio
     @pytest.mark.parametrize("provider_name", ["kiro_cli", "copilot_cli"])
+    @patch("cli_agent_orchestrator.services.terminal_service.delete_terminals_by_session")
     @patch("cli_agent_orchestrator.services.terminal_service.status_monitor")
     @patch("cli_agent_orchestrator.services.terminal_service.fifo_manager")
     @patch("cli_agent_orchestrator.services.terminal_service.FIFO_DIR")
@@ -556,6 +591,7 @@ async def test_create_terminal_does_not_pass_skill_prompt_to_non_runtime_provide
         mock_fifo_dir,
         mock_fifo_manager,
         mock_status_monitor,
+        mock_delete_terminals_by_session,
         provider_name,
     ):
         """Kiro, Q, and Copilot should receive skill_prompt=None."""
@@ -588,6 +624,7 @@ async def test_create_terminal_does_not_pass_skill_prompt_to_non_runtime_provide
         assert mock_provider_manager.create_provider.call_args.kwargs["skill_prompt"] is None
 
     @pytest.mark.asyncio
+    @patch("cli_agent_orchestrator.services.terminal_service.delete_terminals_by_session")
     @patch("cli_agent_orchestrator.services.terminal_service.status_monitor")
     @patch("cli_agent_orchestrator.services.terminal_service.fifo_manager")
     @patch("cli_agent_orchestrator.services.terminal_service.FIFO_DIR")
@@ -614,6 +651,7 @@ async def test_build_skill_catalog_called_for_runtime_prompt_provider(
         mock_fifo_dir,
         mock_fifo_manager,
         mock_status_monitor,
+        mock_delete_terminals_by_session,
     ):
         """build_skill_catalog() is called exactly once for runtime-prompt providers."""
         mock_gen_id.return_value = "test1234"
@@ -639,6 +677,7 @@ async def test_build_skill_catalog_called_for_runtime_prompt_provider(
         mock_build_skill_catalog.assert_called_once_with(["ads-*"])
 
     @pytest.mark.asyncio
+    @patch("cli_agent_orchestrator.services.terminal_service.delete_terminals_by_session")
     @patch("cli_agent_orchestrator.services.terminal_service.status_monitor")
     @patch("cli_agent_orchestrator.services.terminal_service.fifo_manager")
     @patch("cli_agent_orchestrator.services.terminal_service.FIFO_DIR")
@@ -665,6 +704,7 @@ async def test_build_skill_catalog_called_with_empty_filter_for_deny_all(
         mock_fifo_dir,
         mock_fifo_manager,
         mock_status_monitor,
+        mock_delete_terminals_by_session,
     ):
         """A `skills: []` deny-all profile threads the empty list through verbatim.
         It must NOT be coerced to None — that would leak the full catalog to an
@@ -692,6 +732,7 @@ async def test_build_skill_catalog_called_with_empty_filter_for_deny_all(
         mock_build_skill_catalog.assert_called_once_with([])
 
     @pytest.mark.asyncio
+    @patch("cli_agent_orchestrator.services.terminal_service.delete_terminals_by_session")
     @patch("cli_agent_orchestrator.services.terminal_service.status_monitor")
     @patch("cli_agent_orchestrator.services.terminal_service.fifo_manager")
     @patch("cli_agent_orchestrator.services.terminal_service.FIFO_DIR")
@@ -718,6 +759,7 @@ async def test_build_skill_catalog_called_with_none_for_missing_profile_runtime_
         mock_fifo_dir,
         mock_fifo_manager,
         mock_status_monitor,
+        mock_delete_terminals_by_session,
     ):
         """A runtime-prompt provider with no profile in the CAO store builds the
         catalog unfiltered (None). The `profile is None` guard must hold — no
@@ -741,6 +783,7 @@ async def test_build_skill_catalog_called_with_none_for_missing_profile_runtime_
 
     @pytest.mark.asyncio
     @pytest.mark.parametrize("provider_name", ["opencode_cli", "kiro_cli", "copilot_cli"])
+    @patch("cli_agent_orchestrator.services.terminal_service.delete_terminals_by_session")
     @patch("cli_agent_orchestrator.services.terminal_service.status_monitor")
     @patch("cli_agent_orchestrator.services.terminal_service.fifo_manager")
     @patch("cli_agent_orchestrator.services.terminal_service.FIFO_DIR")
@@ -767,6 +810,7 @@ async def test_build_skill_catalog_not_called_for_native_or_baked_provider(
         mock_fifo_dir,
         mock_fifo_manager,
         mock_status_monitor,
+        mock_delete_terminals_by_session,
         provider_name,
     ):
         """build_skill_catalog() is never called for providers that deliver skills natively or
@@ -789,6 +833,7 @@ async def test_build_skill_catalog_not_called_for_native_or_baked_provider(
         mock_build_skill_catalog.assert_not_called()
 
     @pytest.mark.asyncio
+    @patch("cli_agent_orchestrator.services.terminal_service.delete_terminals_by_session")
     @patch("cli_agent_orchestrator.services.terminal_service.status_monitor")
     @patch("cli_agent_orchestrator.services.terminal_service.fifo_manager")
     @patch("cli_agent_orchestrator.services.terminal_service.FIFO_DIR")
@@ -813,6 +858,7 @@ async def test_create_terminal_profile_not_found(
         mock_fifo_dir,
         mock_fifo_manager,
         mock_status_monitor,
+        mock_delete_terminals_by_session,
     ):
         """Terminal creation succeeds when agent profile is not in CAO store (e.g. JSON-only profiles)."""
         mock_gen_id.return_value = "test1234"
@@ -869,6 +915,7 @@ async def test_use_worktree_overrides_working_directory_for_the_new_window(
         mock_fifo_dir,
         mock_fifo_manager,
         mock_status_monitor,
+        tmp_path,
     ):
         mock_gen_id.return_value = "test1234"
         mock_gen_session.return_value = "cao-session"
@@ -880,24 +927,39 @@ async def test_use_worktree_overrides_working_directory_for_the_new_window(
         mock_provider.initialize.return_value = True
         mock_provider_manager.create_provider.return_value = mock_provider
         mock_fifo_dir.__truediv__ = MagicMock(return_value="fake.fifo")
+        # Both directories are real: create_terminal validates the EFFECTIVE launch
+        # cwd (post-worktree-override) before handing it to tmux. Using a real
+        # SOURCE dir too is deliberate -- it means the assertions below fail on the
+        # override semantics rather than on a synthetic path being rejected first,
+        # so this test still catches a regression that resolves the cwd before the
+        # worktree block instead of after it.
+        source_dir = tmp_path / "some" / "subdir"
+        source_dir.mkdir(parents=True)
+        worktree_dir = tmp_path / "worktrees" / "test1234"
+        worktree_dir.mkdir(parents=True)
         mock_worktree_service.find_repo_root.return_value = "/repo"
-        mock_worktree_service.create_worktree.return_value = "/repo/.cao/worktrees/test1234"
+        mock_worktree_service.create_worktree.return_value = str(worktree_dir)
 
         result = await create_terminal(
             "kiro_cli",
             "developer",
             session_name="cao-existing",
-            working_directory="/repo/some/subdir",
+            working_directory=str(source_dir),
             use_worktree=True,
         )
 
         assert result.id == "test1234"
-        mock_worktree_service.find_repo_root.assert_called_once_with("/repo/some/subdir")
+        mock_worktree_service.find_repo_root.assert_called_once_with(str(source_dir))
         mock_worktree_service.create_worktree.assert_called_once_with("/repo", "test1234")
         # The worktree path -- NOT the originally-given working_directory -- is
         # what actually reaches the tmux window (create_window's 4th positional
         # arg, per its own call site in terminal_service.py).
-        assert mock_tmux.create_window.call_args.args[3] == "/repo/.cao/worktrees/test1234"
+        assert mock_tmux.create_window.call_args.args[3] == os.path.realpath(worktree_dir)
+        # ...and is also what gets persisted as the terminal's working_directory,
+        # so list_sessions ownership metadata points at the isolated checkout.
+        assert mock_db_create.call_args.kwargs["working_directory"] == os.path.realpath(
+            worktree_dir
+        )
 
     @pytest.mark.asyncio
     @patch("cli_agent_orchestrator.services.terminal_service.status_monitor")
@@ -993,6 +1055,7 @@ async def test_use_worktree_rolls_back_the_worktree_on_a_later_failure(
         mock_fifo_dir,
         mock_fifo_manager,
         mock_status_monitor,
+        tmp_path,
     ):
         """The worktree WAS created before provider.initialize() failed later --
         the failure-cleanup path must roll it back too, or a provider-init
@@ -1008,8 +1071,12 @@ async def test_use_worktree_rolls_back_the_worktree_on_a_later_failure(
         mock_provider.initialize.side_effect = TimeoutError("provider init timed out")
         mock_provider_manager.create_provider.return_value = mock_provider
         mock_fifo_dir.__truediv__ = MagicMock(return_value="fake.fifo")
+        # Real directory so the effective-cwd validation passes and the failure
+        # under test is the provider-init timeout this test is actually about.
+        worktree_dir = tmp_path / "worktrees" / "test1234"
+        worktree_dir.mkdir(parents=True)
         mock_worktree_service.find_repo_root.return_value = "/repo"
-        mock_worktree_service.create_worktree.return_value = "/repo/.cao/worktrees/test1234"
+        mock_worktree_service.create_worktree.return_value = str(worktree_dir)
 
         with pytest.raises(TimeoutError):
             await create_terminal(
@@ -1208,6 +1275,7 @@ async def test_no_env_vars_existing_session_uses_session_env_only(
         assert extra_env == {"SESSION_VAR": "from-session"}
 
     @pytest.mark.asyncio
+    @patch("cli_agent_orchestrator.services.terminal_service.delete_terminals_by_session")
     @patch("cli_agent_orchestrator.services.terminal_service.set_session_env")
     @patch("cli_agent_orchestrator.services.terminal_service.status_monitor")
     @patch("cli_agent_orchestrator.services.terminal_service.fifo_manager")
@@ -1232,6 +1300,7 @@ async def test_new_session_true_path_unchanged(
         mock_fifo_manager,
         mock_status_monitor,
         mock_set_session_env,
+        mock_delete_terminals_by_session,
     ):
         """new_session=True is untouched by #408: env_vars go verbatim to
         create_session's extra_env and are persisted via set_session_env."""
@@ -1336,12 +1405,23 @@ def test_get_working_directory_not_found(self, mock_get_metadata):
 class TestSendInput:
     """Tests for send_input function."""
 
+    @patch("cli_agent_orchestrator.services.terminal_service.MemoryService")
+    @patch("cli_agent_orchestrator.services.terminal_service.status_monitor")
     @patch("cli_agent_orchestrator.services.terminal_service.update_last_active")
     @patch("cli_agent_orchestrator.services.terminal_service.provider_manager")
     @patch("cli_agent_orchestrator.backends.registry._backend")
     @patch("cli_agent_orchestrator.services.terminal_service.get_terminal_metadata")
-    def test_send_input_success(self, mock_get_metadata, mock_tmux, mock_pm, mock_update):
+    def test_send_input_success(
+        self,
+        mock_get_metadata,
+        mock_tmux,
+        mock_pm,
+        mock_update,
+        mock_status_monitor,
+        mock_memory_service,
+    ):
         """Test sending input successfully."""
+        mock_memory_service.return_value.get_curated_memory_context.return_value = ""
         mock_get_metadata.return_value = {
             "tmux_session": "cao-session",
             "tmux_window": "developer-abcd",
@@ -1363,13 +1443,20 @@ def test_send_input_success(self, mock_get_metadata, mock_tmux, mock_pm, mock_up
         )
         mock_update.assert_called_once_with("test1234")
 
+    @patch("cli_agent_orchestrator.services.terminal_service.MemoryService")
     @patch("cli_agent_orchestrator.services.terminal_service.status_monitor")
     @patch("cli_agent_orchestrator.services.terminal_service.update_last_active")
     @patch("cli_agent_orchestrator.services.terminal_service.provider_manager")
     @patch("cli_agent_orchestrator.backends.registry._backend")
     @patch("cli_agent_orchestrator.services.terminal_service.get_terminal_metadata")
     def test_send_input_clears_rolling_buffer_preserving_arm(
-        self, mock_get_metadata, mock_tmux, mock_pm, mock_update, mock_status_monitor
+        self,
+        mock_get_metadata,
+        mock_tmux,
+        mock_pm,
+        mock_update,
+        mock_status_monitor,
+        mock_memory_service,
     ):
         """send_input clears the byte buffer AFTER arming the sticky latch.
 
@@ -1385,6 +1472,7 @@ def test_send_input_clears_rolling_buffer_preserving_arm(
         placeholders from the pre-task buffer combining with input_received=
         True to trigger a false COMPLETED (the handoff-worker-killed-in-8s bug).
         """
+        mock_memory_service.return_value.get_curated_memory_context.return_value = ""
         mock_get_metadata.return_value = {
             "tmux_session": "cao-session",
             "tmux_window": "developer-abcd",
@@ -1470,15 +1558,23 @@ def test_send_input_blocked_message_uses_enum_value(
         mock_tmux.send_keys.assert_not_called()
         mock_update.assert_not_called()
 
+    @patch("cli_agent_orchestrator.services.terminal_service.MemoryService")
     @patch("cli_agent_orchestrator.services.terminal_service.status_monitor")
     @patch("cli_agent_orchestrator.services.terminal_service.update_last_active")
     @patch("cli_agent_orchestrator.services.terminal_service.provider_manager")
     @patch("cli_agent_orchestrator.backends.registry._backend")
     @patch("cli_agent_orchestrator.services.terminal_service.get_terminal_metadata")
     def test_send_input_allows_manual_answer_when_provider_waits_for_user_answer(
-        self, mock_get_metadata, mock_tmux, mock_pm, mock_update, mock_status_monitor
+        self,
+        mock_get_metadata,
+        mock_tmux,
+        mock_pm,
+        mock_update,
+        mock_status_monitor,
+        mock_memory_service,
     ):
         """Manual input can still answer clarify/approval prompts."""
+        mock_memory_service.return_value.get_curated_memory_context.return_value = ""
         mock_get_metadata.return_value = {
             "tmux_session": "cao-session",
             "tmux_window": "developer-abcd",