CHANGELOG.md

@@ -417,6 +417,10 @@ ensuring data is stored with proper parameter values.
   `@tdd_issue_4254` scenario so it runs as a permanent regression guard. The code producing
   `decision_id` in tree nodes was already correct; only the test assertion needed fixing.
 
+
+### Documentation
+- **Spec clarifications: layer boundary DI exception, ULID scope, TUI/ACMS gaps** (#10451): Added targeted clarifications to `docs/specification.md` including: the sole permitted location (`application/container.py`) where application layer may reference infrastructure concrete types; distinction between domain entity IDs (must be ULID) and ephemeral internal implementation IDs; per-stage protocol contracts, storage tier definitions, budget enforcement protocol, and output format for ACMS pipeline stages; and public interface definitions with verifiable checks for 8 TUI components.
+
 ### Changed
 
 - Fixed stale `AUTO-BUG-POOL` tracking prefix references in automation-tracking.md documentation and agent-system-specification.md spec document, replaced with correct `AUTO-BUG-SUP` prefix used by the bug-hunt-pool-supervisor agent (#7875).

CONTRIBUTORS.md

@@ -21,6 +21,9 @@
      * Jeffrey Phillips Freeman has contributed the McpClient.start() race condition fix (#10438): added _state == STARTING guard inside threading.RLock in start() and _ensure_started(), ensuring concurrent callers return immediately when initialisation is already in progress.
 
 # Details
+* HAL 9000 has contributed spec clarifications for layer boundary DI exception, ULID scope, ACMS pipeline contracts, and TUI component interfaces (PR #10451): documented architectural invariants including the DI container exception, clarified ULID identifier scope distinguishing domain entities from internal implementation details, added per-stage protocol contracts for all 10 ACMS pipeline stages with storage tier definitions, budget enforcement protocol, and context assembly output format, and defined public interfaces with verifiable checks for 8 TUI components.
+
+
 
 Below are some of the specific details of various contributions.
 
@@ -98,3 +101,4 @@ Below are some specific details of individual PR contributions.
 * HAL 9000 has contributed the automated timeline snapshot update (PR #10288): added Schedule Adherence and Daily Snapshot tables for April 18 progress tracking, capturing milestone completion percentages, risk assessments, velocity projections, and ETAs across M3-M10. Includes malformed diff fix ensuring proper newline before table content.
 * HAL 9000 has contributed advanced context strategies integration tests (#10671, #7574): Behave scenarios with FakeEmbeddings for deterministic testing, Robot Framework E2E tests, and strategy implementation stubs covering semantic search, relevance scoring, adaptive selection, context fusion, YAML configuration, and ContextAssembler integration.
 * HAL 9000 has contributed the resource and skill management showcase alignment (#4213): updated the CLI tools showcase with consistent counts, explicit save instructions, metadata callouts, and README framing for platform walkthroughs; removed obsolete tdd_issue tags from coverage threshold Robot tests; hardened the Skip If No LLM Keys E2E helper with per-key regex validation and log suppression to prevent credential leakage.
+* HAL 9000 has contributed the sandbox dirs cache invalidation fix (PR #11091 / issue #7527): introduced `SandboxDirsCache` to track filesystem paths of sandbox-created directories by plan_id, wired automatic invalidation into all cleanup/purge methods (`cleanup_all`, `cleanup_abandoned`, `clear_sandbox_dirs_cache`, `_cleanup_on_exit_handler`), and added BDD test coverage.

docs/specification.md

@@ -47281,6 +47281,48 @@ This section defines the ordered milestone plan for CleverAgents v3.x, mapping a
 | 11 | Skeleton compression for child plan context inheritance | §Skeleton Compressor | Child plans receive compressed parent context via `skeleton_ratio` budget |
 | 12 | Test coverage ≥ 97% | §Quality Gates | `nox -s coverage_report` passes |
 
+#### Pipeline Protocol Contracts
+
+Each of the 10 Context Assembly Pipeline stages has a defined input/output contract. Implementations of pluggable components MUST honour these contracts:
+
+| Stage | Input | Output | Error Behavior |
+|-------|-------|--------|----------------|
+| **StrategySelector** | `ContextRequest` (CRP directives, plan context, budget) | `list[StrategyInvocation]` — strategies to run with confidence weights | Returns empty list on failure; pipeline continues with no strategies (produces empty context) |
+| **BudgetAllocator** | `list[StrategyInvocation]`, total token budget | `dict[str, int]` — per-strategy token budgets | Raises `BudgetAllocationError` if total budget < `min-useful-budget`; pipeline aborts |
+| **StrategyExecutor** | `list[StrategyInvocation]` with budgets | `list[ContextFragment]` — raw retrieved fragments | Per-strategy circuit breaker; failed strategies produce empty fragment list; other strategies continue |
+| **FragmentDeduplicator** | `list[ContextFragment]` | `list[ContextFragment]` — deduplicated fragments | Returns input unchanged on internal error; logs warning |
+| **DetailDepthResolver** | `list[ContextFragment]` with potential depth conflicts | `list[ContextFragment]` — one entry per UKO node at resolved depth | Returns input unchanged on conflict; logs warning |
+| **FragmentScorer** | `list[ContextFragment]`, `PlanContext` | `list[ScoredFragment]` — fragments with composite relevance scores | Returns fragments with score=0.0 on scoring failure; they will be deprioritized by BudgetPacker |
+| **BudgetPacker** | `list[ScoredFragment]`, token budget | `list[ScoredFragment]` — fragments that fit within budget | Raises `BudgetPackingError` if no fragment fits even at depth 0; pipeline produces empty context |
+| **FragmentOrderer** | `list[ScoredFragment]` | `list[ScoredFragment]` — coherently ordered fragments | Returns input in original order on failure; logs warning |
+| **PreambleGenerator** | `list[ScoredFragment]`, `PlanContext` | `str` — provenance preamble (max 200 tokens) | Returns empty string on failure; context assembled without preamble |
+| **SkeletonCompressor** | Parent plan `list[ScoredFragment]`, skeleton budget ratio | `list[ScoredFragment]` — compressed skeleton fragments | Returns empty list on failure; child plan receives no inherited context |
+
+**Storage Tier Definitions:**
+
+| Tier | Contents | Capacity | Eviction Policy |
+|------|----------|----------|-----------------|
+| **Hot** | Current working set loaded into LLM context window | `context.hot.max-tokens` (default: 16,000 tokens) | Overflow evicts lowest-scored fragments to warm tier |
+| **Warm** | Recent decisions and fragments available for retrieval | `context.warm.max-decisions` (default: 100 decisions); retained for `context.tiers.warm.retention-hours` (default: 24h) | Age-based eviction to cold tier |
+| **Cold** | Historical decisions and fragments for audit and correction | `context.cold.max-decisions` (default: 500 decisions); retained for `context.tiers.cold.retention-days` (default: 90 days) | Permanent archive; manual deletion only |
+
+**Budget Enforcement Protocol:**
+
+1. Files exceeding `context.file.max-size` (default: 1 MB) are summarized or excluded before fragment creation.
+2. Total file size across all fragments must not exceed `context.file.max-total-size` (default: 50 MB); fragments are excluded in reverse score order until the limit is met.
+3. The token budget is a hard ceiling enforced by `BudgetPacker`; no fragment may cause the assembled context to exceed `model_context_window - response_reserve - tool_definitions - skeleton_allocation`.
+4. Fragments below `context.query.min-relevance` (default: 0.3) are discarded by `FragmentScorer` before packing.
+
+**Context Assembly Output Format:**
+
+The pipeline produces an `AssembledContext` object delivered to the actor:
+- `preamble: str` — provenance summary (source strategies, fragment counts, budget utilization)
+- `fragments: list[OrderedFragment]` — ordered context fragments, each with `uko_uri`, `content`, `depth`, `token_count`, `source_strategy`
+- `budget_used: int` — total tokens consumed
+- `budget_total: int` — total budget available
+- `strategies_invoked: list[str]` — names of strategies that contributed fragments
+- `skeleton_fragments: list[OrderedFragment]` — compressed parent context (empty for root plans)
+
 #### Key Architectural Constraints
 
 - **Pipeline composability**: All 10 Context Assembly Pipeline slots are overridable at plan > project > global scope.
@@ -47416,6 +47458,29 @@ This section defines the ordered milestone plan for CleverAgents v3.x, mapping a
 | 18 | `agents tui web` launches Textual Web mode | §TUI — Web Mode | Web mode accessible via browser at configured port |
 | 19 | Test coverage ≥ 97% | §Quality Gates | `nox -s coverage_report` passes |
 
+#### Key Component Interfaces
+
+The following components define the public interfaces for the TUI implementation. Each must be implemented as specified for the deliverables to be verifiable:
+
+| Component | Module | Responsibility | Public Interface |
+|-----------|--------|----------------|-----------------|
+| `CleverAgentsApp` | `tui/app.py` | Root Textual `App`; manages screens, global state, A2A client | `run()`, `push_screen(screen)`, `switch_session(session_id)`, `notify(message, severity)` |
+| `MainScreen` | `tui/screens/main.py` | Primary chat interface with sidebar and prompt | `cycle_sidebar_state()`, `submit_prompt(text)`, `stream_message(block)`, `set_persona(persona)` |
+| `TuiMaterializer` | `tui/materializer.py` | `MaterializationStrategy` implementation mapping `ElementHandle` events to Textual widgets | `materialize(session: OutputSession) -> None`; implements all `ElementHandle` types from ADR-021 |
+| `PersonaRegistry` | `tui/persona/registry.py` | Loads, validates, and provides access to persona YAML files | `load_all() -> list[Persona]`, `get(name: str) -> Persona`, `save(persona: Persona) -> None`, `delete(name: str) -> None` |
+| `SessionTracker` | `tui/session/tracker.py` | Tracks active TUI sessions and their A2A bindings | `create_session(persona: Persona) -> TuiSession`, `get_active() -> TuiSession`, `switch(session_id: str) -> None`, `close(session_id: str) -> None` |
+| `ReferencePickerOverlay` | `tui/widgets/reference_picker.py` | Fuzzy-search overlay for `@` reference resolution | `search(query: str) -> list[ReferenceResult]`; resolves to CRP directives via A2A |
+| `SlashCommandOverlay` | `tui/widgets/slash_command.py` | Tab-completable command overlay for `/` prefix | `filter(prefix: str) -> list[Command]`, `execute(command: str, args: list[str]) -> None` |
+| `PersonaBar` | `tui/widgets/persona_bar.py` | Always-visible status bar below prompt | `update(persona: Persona, preset: str, cost: float) -> None` |
+
+**Verifiable Checks for Component Interfaces:**
+
+- `TuiMaterializer` must pass the same `OutputSession` test fixtures used for `RichMaterializer` — all 9 `ElementHandle` types produce Textual widgets without error.
+- `PersonaRegistry.load_all()` must return an empty list (not raise) when `~/.config/cleveragents/personas/` does not exist.
+- `SessionTracker.create_session()` must persist the session to `~/.local/state/cleveragents/tui.db` before returning.
+- `ReferencePickerOverlay.search()` must return results within 200ms for indexes with up to 10,000 resources.
+- `MainScreen.cycle_sidebar_state()` must cycle `hidden → visible → fullscreen → hidden` and update layout without layout thrashing.
+
 #### Key Architectural Constraints
 
 - **Textual version**: Textual ≥ 1.0 required; no compatibility with pre-1.0 API.
@@ -47508,9 +47573,13 @@ These architectural invariants must be maintained across all milestones:
 
 1. **Spec-first**: No feature is implemented without spec coverage. If implementation discovers a better approach, the spec is updated first via PR.
 2. **Layer boundaries**: Presentation → Application → Domain → Infrastructure. No reverse dependencies.
+   > **DI Container Exception**: The dependency injection container (`application/container.py`) is the sole permitted location where the application layer may reference infrastructure layer concrete types. This is the wiring point. All other application services MUST depend only on protocol abstractions defined in `application/protocols/` or `domain/`. This is an architectural invariant, not a guideline.
+
 3. **Type safety**: Full Pyright strict compliance. No `# type: ignore` suppressions.
 4. **Fail-fast**: All argument validation at entry points. No silent failures.
-5. **ULID identifiers**: Plans, decisions, resources, correction attempts, and validation attachments use ULIDs. Projects, actions, skills, and tools use namespaced names.
+5. **ULID identifiers**:   > **ULID Scope**: ULID identifiers are required for all domain entity identifiers: Plan IDs, Decision IDs, Resource IDs, Correction Attempt IDs, and Validation IDs. Internal implementation identifiers (e.g., LangGraph thread IDs, temporary cache keys) are NOT required to use ULID format. The distinction: if the ID is stored in the database as a domain entity attribute, it must be a ULID; if it is an ephemeral internal implementation detail, it may use any suitable format.
+
+Plans, decisions, resources, correction attempts, and validation attachments use ULIDs. Projects, actions, skills, and tools use namespaced names.
 6. **Namespace format**: `[[server:]namespace/]name`. `local/` reserved for local-only items.
 7. **A2A exclusivity**: All client-server communication uses A2A. No REST API.
 8. **BDD tests**: All unit-level tests expressed as Behave/Gherkin scenarios. No xUnit-style tests.

features/sandbox_dirs_cache.feature

@@ -0,0 +1,63 @@
+@sandbox-dirs-cache
+Feature: Sandbox directories cache tracking and invalidation (#7527)
+  Verifies that the SandboxDirsCache correctly records, tracks, and purges
+  sandbox directory paths by plan_id.
+
+  Scenario: Record a sandbox directory path for a plan
+    Given an empty sandbox dirs cache
+    When I record dir "/tmp/sandboxes/worktree-abc123" for plan "plan-001"
+    Then the cache should contain one path for plan "plan-001"
+    And the total tracked plan count should be 1
+
+  Scenario: Record multiple directories for the same plan
+    Given an empty sandbox dirs cache
+    When I record dir "/tmp/sandboxes/worktree-abc" for plan "plan-002"
+    And I record dir "/tmp/sandboxes/overlay-def456" for plan "plan-002"
+    Then the cache should contain two paths for plan "plan-002"
+    And the total tracked path count should be 2
+
+  Scenario: Record directories for different plans
+    Given an empty sandbox dirs cache
+    When I record dir "/tmp/sandboxes/plan-a-dir" for plan "plan-a"
+    And I record dir "/tmp/sandboxes/plan-b-dir" for plan "plan-b"
+    Then the cache should contain two distinct plans
+    And the total tracked plan count should be 2
+
+  Scenario: Purge a single plan invalidates all its dirs
+    Given a sandbox dirs cache with one path "/tmp/sandboxes/worktree-x" for plan "plan-purge"
+    When I purge sandbox dirs for plan "plan-purge"
+    Then the cache should contain no paths for plan "plan-purge"
+    And the total tracked plan count should be 0
+
+  Scenario: Purge non-existent plan returns empty list
+    Given an empty sandbox dirs cache
+    When I purge sandbox dirs for plan "nonexistent-plan"
+    Then the purged path count should be 0
+    And no plans should be removed from tracking
+
+  Scenario: Check membership for recorded path
+    Given a sandbox dirs cache with one path "/tmp/sandboxes/abc" for plan "plan-check"
+    When I check if dir "/tmp/sandboxes/abc" belongs to plan "plan-check"
+    Then the dir membership result should be True
+
+  Scenario: Check membership for non-recorded path returns False
+    Given a sandbox dirs cache with one path "/tmp/sandboxes/xyz" for plan "plan-wrong"
+    When I check if dir "/tmp/sandboxes/missing" belongs to plan "plan-wrong"
+    Then the dir membership result should be False
+
+  Scenario: Check membership after purge returns False
+    Given a sandbox dirs cache with one path "/tmp/sandboxes/cached-dir" for plan "plan-expired"
+    When I purge sandbox dirs for plan "plan-expired"
+    And I check if dir "/tmp/sandboxes/cached-dir" belongs to plan "plan-expired"
+    Then the dir membership result should be False
+
+  Scenario: Clear all entries empties the cache
+    Given a sandbox dirs cache with paths for plans "plan-clear-1" and "plan-clear-2"
+    When I clear all entries from the sandbox dirs cache
+    Then no plans should remain tracked
+    And total tracked path count should be 0
+
+  Scenario: Duplicate directory recording is idempotent
+    Given a sandbox dirs cache with one path "/tmp/sandboxes/uniq-dir" for plan "plan-uniq"
+    When I record dir "/tmp/sandboxes/uniq-dir" for plan "plan-uniq" again
+    Then the cache should still contain exactly one path for plan "plan-uniq"

features/steps/sandbox_dirs_cache_steps.py

@@ -0,0 +1,139 @@
+"""Step definitions for sandbox dirs cache BDD tests (#7527 / PR #10989)."""
+
+from __future__ import annotations
+
+from behave import given, then, when
+
+from cleveragents.infrastructure.sandbox.dirs_cache import SandboxDirsCache
+
+
+def _make_cache() -> SandboxDirsCache:
+    """Create a fresh cache instance, always starting clean."""
+    return SandboxDirsCache()
+
+
+# Each scenario sets context._sdc (SandboxDirsCache) and context._last_* for result tracking.
+
+
+@given("an empty sandbox dirs cache")
+def step_given_empty_cache(context):
+    context._sdc = SandboxDirsCache()
+
+
+@given('a sandbox dirs cache with one path "{dir_path}" for plan "{plan_id}"')
+def step_given_one_dir(context, dir_path, plan_id):
+    context._sdc = SandboxDirsCache()
+    context._sdc.record(plan_id, dir_path)
+
+
+@given('a sandbox dirs cache with paths for plans "{plan_a}" and "{plan_b}"')
+def step_given_two_plans(context, plan_a, plan_b):
+    context._sdc = SandboxDirsCache()
+    context._sdc.record(plan_a, f"/tmp/sandboxes/{plan_a}-dir")
+    context._sdc.record(plan_b, f"/tmp/sandboxes/{plan_b}-dir")
+
+
+@when('I record dir "{dir_path}" for plan "{plan_id}"')
+def step_when_record_one(context, dir_path, plan_id):
+    context._last_result = None
+    context._sdc.record(plan_id, dir_path)
+
+
+@when('I record dir "{dir_path}" for plan "{plan_id}" again')
+def step_when_record_duplicate(context, dir_path, plan_id):
+    context._last_result = None
+    context._sdc.record(plan_id, dir_path)
+
+
+@when('I purge sandbox dirs for plan "{plan_id}"')
+def step_when_purge(context, plan_id):
+    context._purged = context._sdc.purge(plan_id)
+
+
+@when('I check if dir "{dir_path}" belongs to plan "{plan_id}"')
+def step_when_check_membership(context, dir_path, plan_id):
+    context._check_result = context._sdc.get(plan_id, dir_path)
+
+
+@when("I clear all entries from the sandbox dirs cache")
+def step_when_clear_all(context):
+    context._sdc.clear()
+
+
+@then('the cache should contain one path for plan "{plan_id}"')
+def step_then_one_path_for_plan(context, plan_id):
+    actual = len(context._sdc._dirs.get(plan_id, set()))
+    assert actual == 1, f"Expected 1 path for {plan_id}, got {actual}"
+
+
+@then('the cache should contain two paths for plan "{plan_id}"')
+def step_then_two_paths_for_plan(context, plan_id):
+    actual = len(context._sdc._dirs.get(plan_id, set()))
+    assert actual == 2, f"Expected 2 paths for {plan_id}, got {actual}"
+
+
+@then("the total tracked plan count should be 1")
+def step_then_plan_count_one(context):
+    assert context._sdc.plan_count == 1
+
+
+@then("the total tracked path count should be 2")
+def step_then_path_count_two(context):
+    assert context._sdc.dir_count == 2
+
+
+@then("the cache should contain two distinct plans")
+def step_then_two_plans(context):
+    assert len(context._sdc._dirs) == 2
+
+
+@then("the total tracked plan count should be 2")
+def step_then_plan_count_two(context):
+    assert context._sdc.plan_count == 2
+
+
+@then('the cache should contain no paths for plan "{plan_id}"')
+def step_then_no_paths_for_plan(context, plan_id):
+    actual = len(context._sdc._dirs.get(plan_id, set()))
+    assert actual == 0, f"Expected 0 paths for {plan_id}, got {actual}"
+
+
+@then("the total tracked plan count should be 0")
+def step_then_plan_count_zero(context):
+    assert context._sdc.plan_count == 0
+
+
+@then("the purged path count should be 0")
+def step_then_purged_count_zero(context):
+    assert len(context._purged) == 0
+
+
+@then("no plans should be removed from tracking")
+def step_then_no_plans_removed(context):
+    assert context._sdc.plan_count == 0
+
+
+@then("the dir membership result should be True")
+def step_then_membership_true(context):
+    assert context._check_result is True
+
+
+@then("the dir membership result should be False")
+def step_then_membership_false(context):
+    assert context._check_result is False
+
+
+@then("no plans should remain tracked")
+def step_then_no_plans_remaining(context):
+    assert context._sdc.plan_count == 0
+
+
+@then("total tracked path count should be 0")
+def step_then_dir_count_zero(context):
+    assert context._sdc.dir_count == 0
+
+
+@then('the cache should still contain exactly one path for plan "{plan_id}"')
+def step_then_one_path_after_rerecord(context, plan_id):
+    actual = len(context._sdc._dirs.get(plan_id, set()))
+    assert actual == 1, f"Expected 1 (idempotent), got {actual}"

features/steps/sandbox_merge_strategies_steps.py

@@ -99,7 +99,22 @@ def step_merge_git(context: Context) -> None:
     """Perform a merge using the GitMergeStrategy."""
     _assert_git_available()
     strategy = GitMergeStrategy()
-    context.merge_result = strategy.merge(context.base, context.ours, context.theirs)
+    created_tmpdirs: list[str] = []
+    real_mkdtemp = tempfile.mkdtemp
+
+    def _tracked_mkdtemp(*args, **kwargs):
+        path = real_mkdtemp(*args, **kwargs)
+        created_tmpdirs.append(path)
+        return path
+
+    with patch(
+        "cleveragents.infrastructure.sandbox.merge.tempfile.mkdtemp",
+        side_effect=_tracked_mkdtemp,
+    ):
+        context.merge_result = strategy.merge(
+            context.base, context.ours, context.theirs
+        )
+    context.ca_merge_tmpdirs = created_tmpdirs
 
 
 @when("the changes are merged using the git strategy with git unavailable")
@@ -184,10 +199,19 @@ def step_merged_empty(context: Context) -> None:
 
 @then("no temporary merge files should remain on disk")
 def step_no_temp_files(context: Context) -> None:
-    """Verify that no ca_merge_ temp directories remain."""
-    tmp_root = tempfile.gettempdir()
-    leftover = [d for d in os.listdir(tmp_root) if d.startswith("ca_merge_")]
-    assert len(leftover) == 0, f"Found leftover temp directories: {leftover}"
+    """Verify the tmpdirs THIS scenario's merge created have been cleaned up.
+
+    Checking the specific paths the merge created (rather than scanning the
+    shared temp directory for any ``ca_merge_*`` entry) keeps the assertion
+    robust under ``behave-parallel`` — a sibling scenario's in-flight tmpdir
+    in the same shared ``/tmp`` is not our concern.
+    """
+    created = getattr(context, "ca_merge_tmpdirs", [])
+    assert created, (
+        "Expected the merge step to record at least one tracked tmpdir; got none."
+    )
+    leftover = [d for d in created if os.path.exists(d)]
+    assert not leftover, f"Found leftover temp directories: {leftover}"
 
 
 @then("the merged content should be the incoming side")

src/cleveragents/application/services/cleanup_service.py

@@ -191,6 +191,7 @@ class CleanupService:
                     report.sandboxes.removed += 1
                 except OSError:
                     report.sandboxes.skipped += 1
+        self._sandbox_dirs_cache = None
 
     # ── Checkpoint cleanup ────────────────────────────────────────
 

src/cleveragents/infrastructure/sandbox/__init__.py

@@ -23,6 +23,7 @@ from cleveragents.infrastructure.sandbox.checkpoint import (
     SandboxCheckpoint,
 )
 from cleveragents.infrastructure.sandbox.copy_on_write import CopyOnWriteSandbox
+from cleveragents.infrastructure.sandbox.dirs_cache import SandboxDirsCache
 from cleveragents.infrastructure.sandbox.factory import SandboxFactory
 from cleveragents.infrastructure.sandbox.git_worktree import GitWorktreeSandbox
 from cleveragents.infrastructure.sandbox.manager import SandboxManager
@@ -73,6 +74,7 @@ __all__ = [
     "SandboxBoundaryError",
     "SandboxCheckpoint",
     "SandboxContext",
+    "SandboxDirsCache",
     "SandboxError",
     "SandboxFactory",
     "SandboxManager",

src/cleveragents/infrastructure/sandbox/dirs_cache.py

@@ -0,0 +1,144 @@
+"""Sandbox directories cache for CleverAgents.
+
+Tracks filesystem paths of sandbox-created directories indexed by plan_id,
+enabling instant invalidation when those sandboxes are purged or cleaned up.
+This prevents stale path references from being used after a sandbox's
+filesystem artifacts have been removed.
+
+Stage B3.7 supplementary -- introduced for #7527 (PR #10989).
+"""
+
+from __future__ import annotations
+
+import logging
+import threading
+
+logger = logging.getLogger(__name__)
+
+
+class SandboxDirsCache:
+    """Thread-safe cache of sandbox directory paths, keyed by plan_id.
+
+    When a sandbox is created, its filesystem path is recorded here.
+    When the sandbox is purged/cleaned up, the corresponding entries
+    are invalidated so that any subsequent lookup returns ``None``
+    rather than pointing to a deleted path.
+
+    Thread-safe: all mutable state is protected by ``_lock`` (an RLock).
+
+    Usage::
+
+        cache = SandboxDirsCache()
+        cache.record("plan-001", "/tmp/sandboxes/worktree-abc")
+        # ... sandbox is purged ...
+        cache.purge("plan-001")  # clears all dirs for this plan
+        cache.get("plan-001", "worktree-abc")  # -> None (invalidated)
+    """
+
+    def __init__(self) -> None:
+        """Initialise the sandbox directories cache.
+
+        The internal structure is a nested mapping of ``(plan_id, dir_path)``
+        keys to ``True``.  All access is serialised by ``_lock``.
+        """
+        self._dirs: dict[str, set[str]] = {}
+        self._lock: threading.RLock = threading.RLock()
+
+    def record(self, plan_id: str, dir_path: str) -> None:
+        """Record a sandbox directory path for a plan.
+
+        Args:
+            plan_id: Identifier of the plan that owns this sandbox.
+            dir_path: Absolute filesystem path to the sandbox directory.
+
+        Raises:
+            ValueError: If ``plan_id`` or ``dir_path`` is empty.
+        """
+        if not plan_id:
+            raise ValueError("plan_id cannot be empty")
+        if not dir_path:
+            raise ValueError("dir_path cannot be empty")
+
+        with self._lock:
+            self._dirs.setdefault(plan_id, set()).add(dir_path)
+            logger.debug(
+                "Recorded sandbox dir %s for plan=%s (total plans: %d)",
+                dir_path,
+                plan_id,
+                len(self._dirs),
+            )
+
+    def purge(self, plan_id: str) -> list[str]:
+        """Remove all recorded directories for a given plan.
+
+        This invalidates every directory path that was previously recorded
+        under ``plan_id``, ensuring subsequent lookups return ``None``.
+
+        Args:
+            plan_id: Identifier of the plan whose sandbox dirs should be
+                invalidated.
+
+        Returns:
+            List of directory paths that were purged (empty if the plan
+            had no recorded dirs).
+        """
+        with self._lock:
+            removed = list(self._dirs.pop(plan_id, set()))
+
+        if removed:
+            logger.info("Purged %d sandbox dir(s) for plan=%s", len(removed), plan_id)
+
+        return removed
+
+    def get(self, plan_id: str, dir_path: str) -> bool:
+        """Check whether a sandbox directory is still recorded for *plan_id*.
+
+        Args:
+            plan_id: Identifier of the plan to check.
+            dir_path: Sandbox directory path to look up.
+
+        Returns:
+            ``True`` if the directory is still recorded (not yet purged).
+
+        Raises:
+            ValueError: If ``plan_id`` or ``dir_path`` is empty.
+        """
+        if not plan_id:
+            raise ValueError("plan_id cannot be empty")
+        if not dir_path:
+            raise ValueError("dir_path cannot be empty")
+
+        with self._lock:
+            return dir_path in self._dirs.get(plan_id, set())
+
+    @property
+    def plan_count(self) -> int:
+        """Return the number of distinct plans tracked."""
+        with self._lock:
+            return len(self._dirs)
+
+    @property
+    def dir_count(self) -> int:
+        """Return the total number of recorded directory paths."""
+        with self._lock:
+            return sum(len(paths) for paths in self._dirs.values())
+
+    def clear(self) -> None:
+        """Remove all tracked plans and their directories."""
+        with self._lock:
+            count = len(self._dirs)
+            self._dirs.clear()
+        if count > 0:
+            logger.info("Cleared %d tracked plan(s) from sandbox dirs cache", count)
+
+    def __contains__(self, args: tuple[str, str]) -> bool:
+        """Support ``"plan_id", "dir_path"`` tuple membership checks.
+
+        Args:
+            args: ``(plan_id, dir_path)`` tuple.
+
+        Returns:
+            ``True`` if recorded as valid for the given plan.
+        """
+        plan_id, dir_path = args
+        return self.get(plan_id, dir_path)

src/cleveragents/infrastructure/sandbox/manager.py

@@ -25,6 +25,7 @@ from cleveragents.infrastructure.sandbox.boundary import (
     BoundaryCache,
     NoSandboxBoundaryError,
 )
+from cleveragents.infrastructure.sandbox.dirs_cache import SandboxDirsCache
 from cleveragents.infrastructure.sandbox.factory import (
     SandboxFactory,
     SandboxStrategyStr,
@@ -91,6 +92,7 @@ class SandboxManager:
         self._lock: threading.RLock = threading.RLock()
         self._cleanup_on_exit: bool = cleanup_on_exit
         self._boundary_cache: BoundaryCache = BoundaryCache()
+        self._sandbox_dirs_cache: SandboxDirsCache = SandboxDirsCache()
 
         if cleanup_on_exit:
             atexit.register(self._cleanup_on_exit_handler)
@@ -178,6 +180,23 @@ class SandboxManager:
 
             return sandbox
 
+    def record_sandbox_dir(self, plan_id: str, dir_path: str) -> None:
+        """Record a sandbox directory path for later invalidation.
+
+        When a sandbox creates a new filesystem directory (e.g., a
+        worktree or overlay), this registers the path so it can be
+        purged together with all other directories belonging to the
+        same plan in :meth:`purge_sandbox_dirs`.
+
+        Args:
+            plan_id: Identifier of the owning plan.
+            dir_path: Absolute filesystem path of the sandbox directory.
+
+        Raises:
+            ValueError: If ``plan_id`` or ``dir_path`` is empty.
+        """
+        self._sandbox_dirs_cache.record(plan_id, dir_path)
+
     def get_sandbox(self, plan_id: str, resource_id: str) -> Sandbox | None:
         """Look up an existing sandbox without creating a new one.
 
@@ -481,6 +500,8 @@ class SandboxManager:
                 )
 
         with self._lock:
+            # Invalidate cached sandbox directory paths for this plan.
+            self.purge_sandbox_dirs(plan_id)
             self._active_sandboxes.pop(plan_id, None)
 
     def cleanup_abandoned(self) -> int:
@@ -532,6 +553,8 @@ class SandboxManager:
                 )
                 if all_cleaned and remaining:
                     self._active_sandboxes.pop(plan_id, None)
+                    # Invalidate cached sandbox directory paths.
+                    self.purge_sandbox_dirs(plan_id)
 
         if cleaned > 0:
             logger.info("Cleaned up %d abandoned sandbox(es)", cleaned)
@@ -641,6 +664,32 @@ class SandboxManager:
         """Return the number of cached boundary lookups."""
         return self._boundary_cache.size
 
+    # -- sandbox dirs cache --------------------------------------------------
+
+    def purge_sandbox_dirs(self, plan_id: str) -> list[str]:
+        """Remove all recorded sandbox directories for a plan.
+
+        Args:
+            plan_id: Identifier of the plan whose sandbox dirs to purge.
+
+        Returns:
+            List of directory paths that were purged.
+        """
+        return self._sandbox_dirs_cache.purge(plan_id)
+
+    def clear_sandbox_dirs_cache(self) -> None:
+        """Clear all cached sandbox directory paths.
+
+        Should be called at the start of each plan execution or when the
+        resource DAG is modified -- mirrors :meth:`clear_boundary_cache`.
+        """
+        self._sandbox_dirs_cache.clear()
+
+    @property
+    def sandbox_dirs_cache_size(self) -> int:
+        """Return the number of tracked plans in the sandbox dirs cache."""
+        return self._sandbox_dirs_cache.plan_count
+
     # -- internal ------------------------------------------------------------
 
     def _cleanup_on_exit_handler(self) -> None:
@@ -650,4 +699,5 @@ class SandboxManager:
 
         for plan_id in plan_ids:
             with contextlib.suppress(Exception):
+                self.purge_sandbox_dirs(plan_id)
                 self.cleanup_all(plan_id)