feat(cli): implement context list and context add CLI commands for ACMS

Add new acms context list and acms context add CLI commands that
interact with the ACMS (Application Context Management System) tier-based
context pipeline. These commands manage context fragments across the hot,
warm, and cold storage tiers via the ContextTierService:

- acms context list: List all ACMS context fragments with tier placement,
  token counts, access frequency, and metrics. Supports --tier, --project,
  --project-filter, and format options (rich/json/yaml/table/plain).

- acms context add: Add resource files/directories to the ACMS tier store
  as TieredFragment entries under a target project scope and tier. Skips
  binary and oversized files automatically. Supports --tier, --recursive,
  --project, and format options. Also includes an reset subcommand to
  clear all fragments for a project.

.opencode/agents/implementation-supervisor.md

@@ -245,6 +245,16 @@ each work group's fetch algorithm:
 The prompt body to pass to workers you spawn:
   ```
     Implement or fix the indicated issue or pull request.
+
+    PR Compliance Checklist (MANDATORY — complete ALL items before creating a PR):
+    [ ] 1. CHANGELOG.md — add entry under [Unreleased] section
+    [ ] 2. CONTRIBUTORS.md — add or update contribution entry
+    [ ] 3. Commit footer — include `ISSUES CLOSED: #<issue-number>` in the commit message
+    [ ] 4. CI passes — all quality gates and tests green before requesting review
+    [ ] 5. BDD/Behave tests — added or updated for the changed behaviour
+    [ ] 6. Epic reference — PR description references the parent Epic issue number
+    [ ] 7. Labels — applied via forgejo-label-manager: State/In Review, Priority/<level>, MoSCoW/<level>, Type/<type>
+    [ ] 8. Milestone — PR assigned to the earliest open milestone matching the issue
   ```
 ```
 

CHANGELOG.md

@@ -5,6 +5,36 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
 
 ## [Unreleased]
 
+### Fixed
+
+- **Cross-actor subgraph cycle detection reads actor_ref field** (#1431): Fixed
+  `_detect_subgraph_cycles()`, `_map_node()`, and the `compile_actor()` main loop
+  in `src/cleveragents/actor/compiler.py` to read `actor_ref` from the top-level
+  `NodeDefinition.actor_ref` field instead of `node.config.get("actor_ref", "")`.
+  Because `actor_ref` is a typed, validated Pydantic field (not a key inside the
+  untyped `config` dict), the old code always returned an empty string, causing
+  cross-actor cycle detection to silently fail and leaving the system vulnerable to
+  infinite recursion at runtime. Added Behave regression tests
+  (`features/actor_subgraph_cycle_detection.feature`) and a Robot Framework
+  integration test (`robot/actor_compiler.robot`) to prevent regressions.
+
+### Changed
+
+- **`agents session list` now displays full 26-character session ULIDs** (#10970): The Rich table
+  and Summary panel ("Most Recent" / "Oldest") previously showed only the first 8 characters of
+  each session ULID. This made the output unusable for copy-paste into `session tell`,
+  `session show`, `session delete`, and `session export`, all of which require the full
+  26-character identifier. The full ULID is now displayed in all output formats (Rich, plain,
+  JSON, YAML, table).
+
+- **Suppress passing BDD scenario output in `unit_tests` by default** (#10987): Implemented
+  `PassSuppressFormatter`, a custom Behave formatter (in `scripts/behave_pass_suppress_formatter.py`)
+  that buffers all per-scenario output and only flushes it to stdout when a scenario fails or
+  errors. An all-passing `nox -s unit_tests` run now produces ≤ 10 lines (the summary block
+  only), eliminating ~100,000 lines of noise that previously made CI logs unreadable. The
+  formatter is embedded in the `behave-parallel` in-process runner; coverage mode
+  (`BEHAVE_PARALLEL_COVERAGE=1`) is unaffected.
+
 ### Security
 
 - **aiohttp upgraded to >=3.13.4 to remediate CVE-2026-34513 and CVE-2026-34515** (#1549, #1544):
@@ -15,6 +45,18 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
   versions (<3.13.4) cannot be installed even if upstream transitive dependencies have loose
   version constraints.
 
+### Fixed
+
+- **Error suppression removed from `reactive_registry_adapter.py`** (#9060): Removed two `try...except Exception:` blocks in `register_registry_agents()` that silently suppressed errors, violating the CONTRIBUTING.md fail-fast policy. Exceptions from `actor_registry.list_actors()` and the route bridge refresh now propagate to the caller instead of being swallowed. Added Behave scenarios verifying RuntimeError, AttributeError, and TypeError propagation.
+
+- **Implementation Supervisor PR Compliance Checklist** (#9824): Added a mandatory
+  8-item PR Compliance Checklist to the worker prompt body in `implementation-supervisor.md`
+  that every implementation worker must complete before creating a PR. Checklist covers:
+  CHANGELOG.md update, CONTRIBUTORS.md update, commit footer (`ISSUES CLOSED: #N`),
+  CI verification, BDD tests, Epic reference, label application via `forgejo-label-manager`,
+  and milestone assignment. This eliminates systemic PR merge blockers caused by workers
+  omitting required items.
+
 ### Changed
 
 - Restored `benchmark-regression` CI job to `master.yml` with `pull_request` trigger guard
@@ -79,6 +121,28 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
 
 ### Fixed
 
+- **fix(repositories): derive PlanResult.success from result_success column instead of error_message** (#7501):
+  Fixed a critical bug in `PlanRepository._to_domain` where `PlanResult.success` was incorrectly
+  derived from `error_message is None`. Because `error_message` is shared between the build phase
+  and the result phase, a plan with a historical build error would be marked as failed even after
+  successfully completing and being applied. The fix introduces a dedicated `result_success` boolean
+  column in the `plans` table (migration `m9_003_plan_result_success_column`) and updates the
+  repository read path to use it. For backward compatibility, when `result_success` is NULL
+  (pre-migration records), the legacy `error_message is None` heuristic is preserved.
+
+- **`LLMTraceRepository.save()` premature commit breaks UnitOfWork transactions** (#7505):
+  Replaced the unconditional `session.commit()` in `LLMTraceRepository.save()` with a
+  dual-path implementation that respects the UnitOfWork (UoW) pattern. When an external
+  session is provided (UoW mode), the method now calls only `session.flush()`, leaving
+  transaction control to the caller. When no session is provided (standalone mode), the
+  method creates its own session, flushes, commits, and closes it to ensure durable
+  persistence. This eliminates three data-integrity violations: premature commit of outer
+  UoW transactions, loss of rollback capability for subsequent failures, and a mismatch
+  between the class docstring ("Callers are responsible for commit") and the implementation.
+  Input validation for the `trace` argument was also added. Two new BDD scenarios verify
+  the session contract: `Repository save() calls flush not commit` and `LLM trace rolled
+  back when UnitOfWork transaction rolls back`.
+
 - **git_tools._get_base_env() TOCTOU Race Condition** (#7619): Fixed a
   Time-Of-Check-To-Time-Of-Use race condition in `git_tools._get_base_env()`
   where two concurrent threads could both observe `_BASE_ENV is None`, both
@@ -265,6 +329,14 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
 
 ### Added
 
+- **ACMS Index Data Model and File Traversal Engine** (#9579): Implements the
+  foundational ACMS index data model with structured fields for file metadata
+  (path, size, last modified, type), tag system, and hot/warm/cold/archive
+  storage tier assignment. Introduces a timeout-safe large-project file traversal
+  engine capable of handling 10,000+ files without memory exhaustion through
+  chunked processing. Provides a complete index entry pipeline for creation,
+  storage, and retrieval with full queryability by path, tag, type, and recency.
+
 - **ACMS Large-Project Indexing BDD Coverage** (#8726): Added 7 Behave scenarios
   covering walk-based indexing of 10,000+ files without timeout, binary-file
   skipping, oversized-file skipping, git-checkout indexing, fallback to walk when
@@ -273,6 +345,7 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
   syscalls). Added `timeout=120` to git subprocess calls to prevent CI hangs.
   Cached `get_scoped_view` results in `When` steps to avoid redundant re-queries
   in `Then` steps.
+
 - **Agent Evolution Pool Supervisor PR Metadata Assignment** (#7888): The
   agent-evolution-pool-supervisor now automatically looks up the Type/Automation
   label and the earliest open milestone from the repository before dispatching
@@ -579,6 +652,15 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
   response format from the OpenCode API `/session/status` endpoint instead of an array.
   Workers now dispatch and verify correctly, preventing incorrect session deletion.
 
+---
+### Fixed
+
+- **CLI (`agents actor remove`)** (#6491): Restores output parity with the
+  other actor commands by honoring `--format`/`-f` for JSON/YAML/plain/Rich
+  envelopes. Adds a Robot Framework regression test to assert the JSON
+  envelope structure and updates the CLI synopsis in `docs/specification.md`
+  to document the option.
+
 ---
 
 ## [3.8.0] -- 2026-04-05

CONTRIBUTORS.md

@@ -7,7 +7,6 @@
 * Jeffrey Phillips Freeman <jeffrey.freeman@syncleus.com>
 * Luis Mendes <luis.p.mendes@gmail.com>
 * Rui Hu <rui.hu@cleverthis.com>
-* HAL 9000 <hal9000@cleverthis.com>
 
 # Details
 
@@ -28,4 +27,10 @@ Below are some of the specific details of various contributions.
 * HAL 9000 has contributed the architecture-pool-supervisor milestone assignment feature (PR #8188 / issue #7521): added `forgejo_update_pull_request` permission and documented the PR workflow for major spec changes, enabling automatic milestone assignment for specification PRs.
 * HAL 9000 has contributed the git worktree TOCTOU race condition fix (PR #8178 / issue #7507): replaced the unsafe mkdtemp() + rmdir() pattern with a parent-directory approach to eliminate the race window in concurrent git worktree operations.
 * HAL 9000 has contributed the git_tools TOCTOU race condition fix (PR #8255 / issue #7619): eliminated the Time-Of-Check-To-Time-Of-Use race in `_get_base_env()` by adding double-checked locking with a module-level `threading.Lock`, preventing concurrent threads from writing conflicting environment snapshots.
+* HAL 9000 has contributed the mandatory PR compliance checklist to `implementation-supervisor.md` (#9824): added an 8-item checklist to the worker prompt body with concrete items covering CHANGELOG.md, CONTRIBUTORS.md, commit footer, CI verification, BDD tests, Epic reference, labels, and milestone assignment to eliminate systemic PR merge blockers.
+* HAL 9000 has contributed the PlanResult.success derivation fix (PR #8214 / issue #7501): replaced the incorrect `error_message is None` heuristic with a dedicated `result_success` column in the plans table, ensuring plans with historical build errors are not incorrectly marked as failed after a successful apply.
 * HAL 9000 has contributed comprehensive milestone documentation for v3.6.0 (Advanced Concepts & Deferred Features) and v3.7.0 (TUI Implementation) (PR #9903): split into sub-documents covering context strategies, LLM backends, resource types, A2A rename, container tool execution, scope chain resolution, cost/safety budgets, E2E workflow tests, code review examples, plugin architecture, TUI layout, persona system, reference/command input, session management, configuration, and TuiMaterializer integration.
+* HAL 9000 has contributed the LLMTraceRepository data-integrity fix (PR #8185 / issue #7505): replaced the unconditional `session.commit()` in `LLMTraceRepository.save()` with a dual-path implementation that respects the UnitOfWork pattern — flushing only when an external session is provided, and flushing + committing + closing when operating standalone. This eliminates premature transaction commits, loss of rollback capability, and a docstring/implementation mismatch.
+* HAL 9000 has contributed the ACMS Index Data Model and File Traversal Engine (PR #9664 / issue #9579): foundational data structures for indexed context entries with hot/warm/cold/archive storage tier classification, tag system, and a timeout-safe chunked file traversal engine for large projects with 10,000+ files.
+
+* HAL 9000 has contributed the error-suppression removal fix (PR #9247 / issue #9060): removed both `try...except Exception:` blocks in `register_registry_agents()` that silently suppressed errors from `actor_registry.list_actors()` and the route bridge refresh, enabling exceptions to propagate per CONTRIBUTING.md fail-fast policy. Added three Behave scenarios verifying RuntimeError, AttributeError, and TypeError propagation.

docs/reference/session_cli.md

@@ -72,8 +72,8 @@ The `rich` format renders a sessions table with columns: **ID**, **Name**, **Act
 | Field | Description |
 |-------|-------------|
 | Total | Number of sessions |
-| Most Recent | Name or truncated ID of the most recently updated session |
-| Oldest | Name or truncated ID of the oldest session |
+| Most Recent | Name or full ULID of the most recently updated session |
+| Oldest | Name or full ULID of the oldest session |
 | Total Messages | Sum of messages across all sessions |
 | Storage | Estimated storage used |
 
@@ -85,7 +85,7 @@ Followed by a `✓ OK N sessions listed` success message.
 {
   "sessions": [
     {
-      "id": "01HXYZ...",
+      "id": "01HXYZ4M1Q3F0R0E5HR8K5T8A",
       "name": "my-session",
       "actor": "openai/gpt-4",
       "messages": 5,

docs/showcase/cli-tools/project-init-and-context-management.md

@@ -296,13 +296,13 @@ $ python -m cleveragents session list
 ┏━━━━━━━━━━┳━━━━━━━━━━━┳━━━━━━━━┳━━━━━━━━━━┳━━━━━━━━━━━━━━━━━━┓
 ┃ ID       ┃ Name      ┃ Actor  ┃ Messages ┃ Updated          ┃
 ┡━━━━━━━━━━╇━━━━━━━━━━━╇━━━━━━━━╇━━━━━━━━━━╇━━━━━━━━━━━━━━━━━━┩
-│ 01KNKK4Q │ (unnamed) │ (none) │        0 │ 2026-04-07 09:07 │
+│ 01KNKK4Q9GZ0TRR5B0NEJYGMWH │ (unnamed) │ (none) │        0 │ 2026-04-07 09:07 │
 └──────────┴───────────┴────────┴──────────┴──────────────────┘
 
 ╭────────────────────────────────── Summary ───────────────────────────────────╮
 │ Total:          1                                                            │
-│ Most Recent:    01KNKK4Q                                                     │
-│ Oldest:         01KNKK4Q                                                     │
+│ Most Recent:    01KNKK4Q9GZ0TRR5B0NEJYGMWH                                                     │
+│ Oldest:         01KNKK4Q9GZ0TRR5B0NEJYGMWH                                                     │
 │ Total Messages: 0                                                            │
 │ Storage:        0 KB                                                         │
 ╰──────────────────────────────────────────────────────────────────────────────╯
@@ -311,7 +311,7 @@ $ python -m cleveragents session list
 ```
 
 **What's Happening:**
-The session list shows all sessions with their truncated ID, optional name, bound actor, message count, and last update time. The summary panel provides aggregate statistics across all sessions.
+The session list shows all sessions with their full ULID, optional name, bound actor, message count, and last update time. The summary panel provides aggregate statistics across all sessions.
 
 ---
 
@@ -339,8 +339,8 @@ $ python -m cleveragents session list --format json
     ],
     "summary": {
       "total": 1,
-      "most_recent": "01KNKK4Q",
-      "oldest": "01KNKK4Q",
+      "most_recent": "01KNKK4Q9GZ0TRR5B0NEJYGMWH",
+      "oldest": "01KNKK4Q9GZ0TRR5B0NEJYGMWH",
       "total_messages": 0,
       "storage": "0 KB"
     }
@@ -469,7 +469,7 @@ $ python -m cleveragents session list
 ┏━━━━━━━━━━┳━━━━━━━━━━━┳━━━━━━━━┳━━━━━━━━━━┳━━━━━━━━━━━━━━━━━━┓
 ┃ ID       ┃ Name      ┃ Actor  ┃ Messages ┃ Updated          ┃
 ┡━━━━━━━━━━╇━━━━━━━━━━━╇━━━━━━━━╇━━━━━━━━━━╇━━━━━━━━━━━━━━━━━━┩
-│ 01KNKK4Q │ (unnamed) │ (none) │        0 │ 2026-04-07 09:07 │
+│ 01KNKK4Q9GZ0TRR5B0NEJYGMWH │ (unnamed) │ (none) │        0 │ 2026-04-07 09:07 │
 └──────────┴───────────┴────────┴──────────┴──────────────────┘
 ✓ OK 1 sessions listed
 ```

docs/showcase/cli-tools/session-management-workflows.md

@@ -180,14 +180,14 @@ $ python -m cleveragents session list
 ┏━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┓
 ┃ ID       ┃ Name       ┃ Actor          ┃ Messages ┃ Updated          ┃
 ┡━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┩
-│ 01HXYZ4M │ (unnamed)  │ openai/gpt-4o  │        3 │ 2026-04-07 09:22 │
-│ 01HXYZ3K │ (unnamed)  │ (none)         │        0 │ 2026-04-07 09:15 │
+│ 01HXYZ4M1Q3F0R0E5HR8K5T8A │ (unnamed)  │ openai/gpt-4o  │        3 │ 2026-04-07 09:22 │
+│ 01HXYZ3K9P2E9Q9D4GQ7J4S7Z │ (unnamed)  │ (none)         │        0 │ 2026-04-07 09:15 │
 └──────────┴────────────┴────────────────┴──────────┴──────────────────┘
 
 ╭──────────────────────────────────────────── Summary ─────────────────────────────────────────────╮
 │ Total:          2                                                                                 │
-│ Most Recent:    01HXYZ4M                                                                          │
-│ Oldest:         01HXYZ3K                                                                          │
+│ Most Recent:    01HXYZ4M1Q3F0R0E5HR8K5T8A                                                                          │
+│ Oldest:         01HXYZ3K9P2E9Q9D4GQ7J4S7Z                                                                          │
 │ Total Messages: 3                                                                                 │
 │ Storage:        0 KB                                                                              │
 ╰──────────────────────────────────────────────────────────────────────────────────────────────────╯
@@ -196,8 +196,8 @@ $ python -m cleveragents session list
 ```
 
 **What's Happening:**
-The list command renders a Rich table with five columns: truncated ID (first
-8 characters for readability), optional name, bound actor, message count, and
+The list command renders a Rich table with five columns: the full session
+ULID (26 characters), optional name, bound actor, message count, and
 last update time. The **Summary** panel below shows aggregate statistics
 including total sessions, most recent, oldest, total message count, and
 storage used.
@@ -229,8 +229,8 @@ $ python -m cleveragents session list --format json
   ],
   "summary": {
     "total": 2,
-    "most_recent": "01HXYZ4M",
-    "oldest": "01HXYZ3K",
+    "most_recent": "01HXYZ4M1Q3F0R0E5HR8K5T8A",
+    "oldest": "01HXYZ3K9P2E9Q9D4GQ7J4S7Z",
     "total_messages": 3,
     "storage": "0 KB"
   }
@@ -773,7 +773,7 @@ $ python -m cleveragents session list
 ✓ OK 2 sessions listed
 
 $ python -m cleveragents session list --format json
-{"sessions": [...], "summary": {"total": 2, "most_recent": "01HXYZ4M", ...}}
+{"sessions": [...], "summary": {"total": 2, "most_recent": "01HXYZ4M1Q3F0R0E5HR8K5T8A", ...}}
 
 $ python -m cleveragents session tell --session 01HXYZ4M1Q3F0R0E5HR8K5T8A "What is the capital of France?"
 user: What is the capital of France?

docs/specification.md

@@ -277,7 +277,7 @@ The following standards are integrated into the architecture:
                  [(<span style="color: magenta;"><span style="color: cyan;">--temperature</span>|<span style="color: yellow;">-t</span></span>) <span style="color: #66cc66;">&lt;TEMP&gt;</span>] [<span style="color: cyan;">--allow-rxpy-in-run-mode</span>]
                  [<span style="color: cyan;">--skill</span> <span style="color: #66cc66;">&lt;SKILL&gt;</span>]... <span style="color: #66cc66;">&lt;NAME&gt;</span> <span style="color: #66cc66;">&lt;PROMPT&gt;</span>
 <span style="color: cyan; font-weight: 600;">agents</span> actor add <span style="color: cyan;">--config</span>|<span style="color: yellow;">-c</span> <span style="color: #66cc66;">&lt;FILE&gt;</span> [<span style="color: cyan;">--update</span>]
-<span style="color: cyan; font-weight: 600;">agents</span> actor remove <span style="color: #66cc66;">&lt;NAME&gt;</span>
+<span style="color: cyan; font-weight: 600;">agents</span> actor remove [<span style="color: cyan;">--format</span> <span style="color: #66cc66;">&lt;FORMAT&gt;</span>] <span style="color: #66cc66;">&lt;NAME&gt;</span>
 <span style="color: cyan; font-weight: 600;">agents</span> actor list
 <span style="color: cyan; font-weight: 600;">agents</span> actor show <span style="color: #66cc66;">&lt;NAME&gt;</span>
 <span style="color: cyan; font-weight: 600;">agents</span> actor context remove [<span style="color: cyan;">--yes</span>|<span style="color: yellow;">-y</span>] (<span style="color: magenta;"><span style="color: cyan;">--all</span>|<span style="color: yellow;">-a</span>|<span style="color: #66cc66;">&lt;NAME&gt;</span></span>)
@@ -1723,8 +1723,8 @@ None.
     ╭─ Sessions ───────────────────────────────────────────────────────────────────╮
     │ <span style="color: cyan; font-weight: 600;">ID</span>        <span style="color: cyan; font-weight: 600;">Name</span>             <span style="color: cyan; font-weight: 600;">Actor</span>               <span style="color: cyan; font-weight: 600;">Messages</span>  <span style="color: cyan; font-weight: 600;">Updated</span>             │
     │ <span style="opacity: 0.7;">────────  ───────────────  ──────────────────  ────────  ────────────────</span>    │
-    │ 01HXM2A6  weekly-planning  local/orchestrator  6         2026-02-08 12:44    │
-    │ 01HXM1F2  refactor-sprint  local/orchestrator  14        2026-02-07 18:11    │
+    │ 01HXM2A61MQHZ4MRBAY3MPNJTN  weekly-planning  local/orchestrator  6         2026-02-08 12:44    │
+    │ 01HXM1F21MQHZ4MRBAY3MPNJTN  refactor-sprint  local/orchestrator  14        2026-02-07 18:11    │
     ╰──────────────────────────────────────────────────────────────────────────────╯
 
     ╭─ Summary ────────────────────╮
@@ -1746,8 +1746,8 @@ None.
     Sessions
     ID        Name             Actor               Messages  Updated
     --------  ---------------  ------------------  --------  ----------------
-    01HXM2A6  weekly-planning  local/orchestrator  6         2026-02-08 12:44
-    01HXM1F2  refactor-sprint  local/orchestrator  14        2026-02-07 18:11
+    01HXM2A61MQHZ4MRBAY3MPNJTN  weekly-planning  local/orchestrator  6         2026-02-08 12:44
+    01HXM1F21MQHZ4MRBAY3MPNJTN  refactor-sprint  local/orchestrator  14        2026-02-07 18:11
 
     Summary
       Total: 2
@@ -1769,14 +1769,14 @@ None.
       "data": {
         "sessions": [
           {
-            "id": "01HXM2A6",
+            "id": "01HXM2A61MQHZ4MRBAY3MPNJTN",
             "name": "weekly-planning",
             "actor": "local/orchestrator",
             "messages": 6,
             "updated": "2026-02-08T12:44:00Z"
           },
           {
-            "id": "01HXM1F2",
+            "id": "01HXM1F21MQHZ4MRBAY3MPNJTN",
             "name": "refactor-sprint",
             "actor": "local/orchestrator",
             "messages": 14,
@@ -1804,12 +1804,12 @@ None.
     exit_code: 0
     data:
       sessions:
-        - id: 01HXM2A6
+        - id: 01HXM2A61MQHZ4MRBAY3MPNJTN
           name: weekly-planning
           actor: local/orchestrator
           messages: 6
           updated: "2026-02-08T12:44:00Z"
-        - id: 01HXM1F2
+        - id: 01HXM1F21MQHZ4MRBAY3MPNJTN
           name: refactor-sprint
           actor: local/orchestrator
           messages: 14
@@ -45871,6 +45871,8 @@ The relational database follows a normalized design with foreign key constraints
 
 4. **Optimistic concurrency control**: The `updated_at` timestamp column on mutable entities serves as a version check. Update operations include `WHERE updated_at = <previous_value>` to detect concurrent modifications. On conflict, the operation fails with an explicit error rather than silently overwriting.
 
+6. **PlanResult Domain Model — `result_success` column** (migration `m9_003_plan_result_success_column`): The legacy `plans` table includes a dedicated `result_success BOOLEAN NULLABLE` column that explicitly records the success status of the result (apply) phase. `PlanRepository._to_domain` derives `PlanResult.success` from this column rather than from `error_message is None`. The `error_message` column is shared between the build phase and the result phase; using it to infer result success would incorrectly mark plans as failed when a build-phase error was recorded but the apply phase later succeeded. When `result_success` is `NULL` (pre-migration records), the repository falls back to the legacy `error_message is None` heuristic for backward compatibility.
+
 5. **Datetime handling contract**: All timestamps stored in the database are UTC ISO 8601 strings (format: `YYYY-MM-DDTHH:MM:SS.ffffff`). All Python datetime comparisons involving stored timestamps MUST parse the stored string back to a timezone-aware `datetime` object before comparing — **never** compare ISO timestamp strings lexicographically. String comparison is incorrect because different timezone offset formats (`+00:00` vs `Z` vs `+05:30`) produce strings that are not lexicographically comparable even when representing the same moment. The canonical pattern is:
 
     ```python

features/acms/index_data_model_and_traversal.feature

@@ -0,0 +1,140 @@
+Feature: ACMS Index Data Model and File Traversal Engine
+  As a developer
+  I want to index large projects with 10,000+ files
+  So that I can efficiently query and manage context entries at scale
+
+  Background:
+    Given I have an ACMS index
+    And I have a file traversal engine with chunk size 100
+
+  Scenario: Create an index entry with file metadata
+    When I create an index entry with:
+      | key | value |
+      | path | /project/src/main.py |
+      | file_type | python |
+      | size_bytes | 1024 |
+    Then the index entry should have path "/project/src/main.py"
+    And the index entry should have file type "python"
+    And the index entry should have size 1024 bytes
+
+  Scenario: Add tags to an index entry
+    Given I have an index entry with path "/project/src/main.py"
+    When I add tag "core" to the entry
+    And I add tag "important" to the entry
+    Then the entry should have tag "core"
+    And the entry should have tag "important"
+    And the entry should have 2 tags
+
+  Scenario: Set tier level for an index entry
+    Given I have an index entry with path "/project/src/main.py"
+    When I set the tier level to "hot"
+    Then the entry should have tier level "hot"
+
+  Scenario: Add entry to index
+    Given I have an index entry with path "/project/src/main.py"
+    When I add the entry to the index
+    Then the index should contain 1 entry
+    And I should be able to retrieve the entry by path "/project/src/main.py"
+
+  Scenario: Query index by path pattern
+    Given I have an index with entries:
+      | path | file_type |
+      | /project/src/main.py | python |
+      | /project/src/utils.py | python |
+      | /project/tests/test_main.py | python |
+      | /project/docs/readme.md | markdown |
+    When I query the index by path pattern "src"
+    Then I should get 2 results
+    And the results should include "/project/src/main.py"
+    And the results should include "/project/src/utils.py"
+
+  Scenario: Query index by file type
+    Given I have an index with entries:
+      | path | file_type |
+      | /project/src/main.py | python |
+      | /project/src/utils.py | python |
+      | /project/src/app.js | javascript |
+      | /project/docs/readme.md | markdown |
+    When I query the index by file type "python"
+    Then I should get 2 results
+    And all results should have file type "python"
+
+  Scenario: Query index by tag
+    Given I have an index with entries:
+      | path | tags |
+      | /project/src/main.py | core,important |
+      | /project/src/utils.py | supporting |
+      | /project/tests/test_main.py | test,important |
+    When I query the index by tag "important"
+    Then I should get 2 results
+    And the results should include "/project/src/main.py"
+    And the results should include "/project/tests/test_main.py"
+
+  Scenario: Query index by tier level
+    Given I have an index with entries:
+      | path | tier |
+      | /project/src/main.py | hot |
+      | /project/src/utils.py | warm |
+      | /project/tests/test_main.py | cold |
+    When I query the index by tier level "hot"
+    Then I should get 1 result
+    And the result should have path "/project/src/main.py"
+
+  Scenario: Query index by recency
+    Given I have an index with entries from different dates
+    When I query the index for entries modified after "2026-04-01"
+    Then I should get entries modified after that date
+
+  Scenario: Traverse and index a directory with multiple files
+    Given I have a test directory with 50 files
+    When I traverse and index the directory
+    Then the index should contain 50 entries
+    And all entries should have valid file paths
+
+  Scenario: Handle large project traversal with chunked processing
+    Given I have a test directory with 1000 files
+    When I traverse and index the directory with chunk size 100
+    Then the index should contain 1000 entries
+    And the traversal should complete without timeout
+
+  Scenario: Exclude patterns during traversal
+    Given I have a test directory with files including:
+      | path |
+      | /project/src/main.py |
+      | /project/.git/config |
+      | /project/__pycache__/main.cpython-39.pyc |
+      | /project/src/utils.py |
+    When I traverse and index the directory excluding ".git" and "__pycache__"
+    Then the index should contain 2 entries
+    And the index should not contain ".git" paths
+    And the index should not contain "__pycache__" paths
+
+  Scenario: Get all entries from index
+    Given I have an index with 5 entries
+    When I get all entries from the index
+    Then I should get 5 results
+
+  Scenario: Get entry count from index
+    Given I have an index with 10 entries
+    When I get the entry count
+    Then the ACMS entry count should be 10
+
+  Scenario: Remove entry from index
+    Given I have an index with 3 entries
+    When I remove an entry by path
+    Then the index should contain 2 entries
+
+  Scenario: Combined query with multiple filters
+    Given I have an index with entries:
+      | path | file_type | tags | tier |
+      | /project/src/main.py | python | core,important | hot |
+      | /project/src/utils.py | python | supporting | warm |
+      | /project/tests/test_main.py | python | test | cold |
+      | /project/docs/readme.md | markdown | docs | cold |
+    When I query the index with filters:
+      | filter | value |
+      | path_pattern | src |
+      | file_type | python |
+      | tier | hot |
+    Then I should get 1 result
+    And the result should have path "/project/src/main.py"

features/actor_cli_yaml.feature

@@ -59,6 +59,11 @@ Feature: Actor CLI YAML-first alignment
     When I run actor remove with namespaced name
     Then the actor remove should succeed for namespaced name
 
+  Scenario: Actor remove outputs JSON format
+    Given an actor CLI runner
+    When I run actor remove with format json
+    Then the actor remove output should be valid JSON envelope
+
   Scenario: Actor update outputs JSON format
     Given an actor CLI runner
     When I run actor update with format json

features/actor_subgraph_cycle_detection.feature

@@ -0,0 +1,38 @@
+Feature: Cross-actor subgraph cycle detection reads actor_ref field
+  As a CleverAgents developer
+  I want the actor compiler to correctly detect cross-actor subgraph cycles
+  So that mutually-referencing actors raise SubgraphCycleError at compile time
+
+  Background:
+    Given the actor compiler is available
+
+  # ────────────────────────────────────────────────────────────
+  # Bug fix: actor_ref is a top-level NodeDefinition field
+  # ────────────────────────────────────────────────────────────
+
+  @tdd_issue @tdd_issue_1431
+  Scenario: Mutually-referencing actors raise SubgraphCycleError
+    Given actor "test/actor-a" has a subgraph node with actor_ref "test/actor-b"
+    And actor "test/actor-b" has a subgraph node with actor_ref "test/actor-a"
+    And a registry containing both actors
+    When I compile actor "test/actor-a" with the registry resolver
+    Then the compilation should raise SubgraphCycleError
+    And the cycle error message should mention "cycle"
+
+  @tdd_issue @tdd_issue_1431
+  Scenario: Non-cyclic subgraph reference compiles successfully
+    Given actor "test/actor-x" has a subgraph node with actor_ref "test/actor-y"
+    And actor "test/actor-y" has no subgraph nodes
+    And a registry containing both actors
+    When I compile actor "test/actor-x" with the registry resolver
+    Then the actor compilation should succeed
+    And the actor subgraph_refs should map "sub" to "test/actor-y"
+
+  @tdd_issue @tdd_issue_1431
+  Scenario: Subgraph node actor_ref is reflected in compiled metadata
+    Given actor "test/actor-p" has a subgraph node with actor_ref "test/actor-q"
+    And actor "test/actor-q" has no subgraph nodes
+    And a registry containing both actors
+    When I compile actor "test/actor-p" with the registry resolver
+    Then the actor compilation should succeed
+    And the compiled node "sub" should have subgraph set to "test/actor-q"

features/behave_parallel_log_filtering.feature

@@ -99,3 +99,23 @@ Feature: Behave-parallel conditional log replay
     Given behave_parallel worker results with one crashed chunk and one passing chunk
     When behave_parallel I aggregate the worker results
     Then behave_parallel the aggregated summary should have failures
+
+  # ---- PassSuppressFormatter: per-scenario output suppression ----
+
+  Scenario: PassSuppressFormatter suppresses output for a passing scenario
+    Given behave_parallel a PassSuppressFormatter backed by a captured stream
+    When behave_parallel I simulate a passing scenario through the formatter
+    Then behave_parallel the formatter real stream output should be empty
+
+  Scenario: PassSuppressFormatter emits full output for a failing scenario
+    Given behave_parallel a PassSuppressFormatter backed by a captured stream
+    When behave_parallel I simulate a failing scenario through the formatter
+    Then behave_parallel the formatter real stream output should contain "Failing scenario"
+    And behave_parallel the formatter real stream output should contain "I fail"
+    And behave_parallel the formatter real stream output should contain "AssertionError"
+
+  Scenario: PassSuppressFormatter only shows failed scenarios in a mixed run
+    Given behave_parallel a PassSuppressFormatter backed by a captured stream
+    When behave_parallel I simulate one passing then one failing scenario through the formatter
+    Then behave_parallel the formatter real stream output should not contain "Passing scenario"
+    And behave_parallel the formatter real stream output should contain "Failing scenario"

features/consolidated_misc.feature

@@ -1714,6 +1714,12 @@ Feature: Consolidated Misc
     And the temporary connection should be closed afterward
 
 
+  Scenario: get_current_revision uses check_same_thread=False for SQLite engines
+    Given a migration runner configured for "sqlite:///:memory:"
+    When I request the current revision from the database
+    Then the SQLite engine for get_current_revision should use check_same_thread=False
+
+
   Scenario: File-based SQLite database directory is created if missing
     Given a migration runner configured for "sqlite:///tmp/test-db/mydb.db"
     When I initialize or upgrade a file-based SQLite database

features/environment.py

@@ -686,6 +686,12 @@ def after_scenario(context, scenario):
             pass  # Ignore cleanup errors
         context.test_dir = None
 
+    # Clean up TemporaryDirectory objects created by ACMS index traversal tests
+    if hasattr(context, "temp_dir") and context.temp_dir is not None:
+        with contextlib.suppress(Exception):
+            context.temp_dir.cleanup()
+        context.temp_dir = None
+
     # Clean up environment variables set during tests
     if hasattr(context, "env_vars_to_clean"):
         for key in context.env_vars_to_clean:

features/pr_compliance_checklist.feature

@@ -0,0 +1,58 @@
+@mock_only
+Feature: PR Compliance Checklist in Implementation Supervisor
+
+  As an implementation supervisor
+  I want to pass a mandatory PR compliance checklist to every worker prompt
+  So that workers complete all required items before creating a PR and avoid systemic merge blockers
+
+  Background:
+    Given the implementation-supervisor.md agent definition exists
+
+  Scenario: Supervisor worker prompt includes the PR compliance checklist
+    When I read the implementation supervisor agent definition
+    Then the worker prompt body includes the PR compliance checklist section
+    And the checklist is marked as MANDATORY
+
+  Scenario: Checklist item 1 — CHANGELOG.md update required
+    When I read the implementation supervisor agent definition
+    Then the worker prompt body includes a CHANGELOG.md checklist item
+    And the item instructs workers to add an entry under the Unreleased section
+
+  Scenario: Checklist item 2 — CONTRIBUTORS.md update required
+    When I read the implementation supervisor agent definition
+    Then the worker prompt body includes a CONTRIBUTORS.md checklist item
+    And the item instructs workers to add or update their contribution entry
+
+  Scenario: Checklist item 3 — commit footer required
+    When I read the implementation supervisor agent definition
+    Then the worker prompt body includes a commit footer checklist item
+    And the item specifies the ISSUES CLOSED footer format
+
+  Scenario: Checklist item 4 — CI must pass before PR creation
+    When I read the implementation supervisor agent definition
+    Then the worker prompt body includes a CI passes checklist item
+    And the item instructs workers to verify all quality gates are green
+
+  Scenario: Checklist item 5 — BDD/Behave tests required
+    When I read the implementation supervisor agent definition
+    Then the worker prompt body includes a BDD tests checklist item
+    And the item instructs workers to add or update Behave feature files
+
+  Scenario: Checklist item 6 — Epic reference required in PR description
+    When I read the implementation supervisor agent definition
+    Then the worker prompt body includes an Epic reference checklist item
+    And the item instructs workers to reference the parent Epic issue number
+
+  Scenario: Checklist item 7 — Labels must be applied
+    When I read the implementation supervisor agent definition
+    Then the worker prompt body includes a labels checklist item
+    And the item instructs workers to apply labels via forgejo-label-manager
+
+  Scenario: Checklist item 8 — Milestone must be assigned
+    When I read the implementation supervisor agent definition
+    Then the worker prompt body includes a milestone checklist item
+    And the item instructs workers to assign the earliest open milestone
+
+  Scenario: All 8 checklist items are present in the worker prompt
+    When I read the implementation supervisor agent definition
+    Then the worker prompt body contains all 8 mandatory checklist items

features/session_cli.feature

@@ -44,6 +44,28 @@ Feature: Session CLI commands
     When I run session CLI list with --format json
     Then the session CLI JSON list entries should match the documented contract
 
+  Scenario: List sessions displays full 26-character ULIDs in Rich table
+    Given there are mocked existing sessions
+    When I run session CLI list
+    Then the session CLI rich table should display full session ULIDs
+
+  Scenario: List sessions summary panel shows full ULIDs for unnamed sessions
+    Given there are mocked existing sessions
+    When I run session CLI list
+    Then the session CLI summary panel should contain full session ULIDs
+
+  Scenario: List sessions summary panel shows session names for named sessions
+    Given there are mocked existing named sessions
+    When I run session CLI list
+    Then the session CLI summary panel should show session names
+
+  Scenario: Full session ID from list output works with session tell
+    Given there are mocked existing sessions
+    When I run session CLI list
+    And I capture the first session full ULID from the output
+    And I run session CLI tell with the full session ID and prompt "Hello from list"
+    Then the session CLI tell should succeed
+
   # Show command tests
   Scenario: Show session with valid ID
     Given there is a mocked session with messages

features/steps/acms_index_data_model_traversal_steps.py

@@ -0,0 +1,395 @@
+"""Step definitions for ACMS Index Data Model and File Traversal Engine tests."""
+
+from __future__ import annotations
+
+import tempfile
+from datetime import datetime, timedelta
+from pathlib import Path
+
+from behave import given, then, when
+
+from cleveragents.acms.index import (
+    ACMSIndex,
+    FileTraversalEngine,
+    FileType,
+    IndexEntry,
+    TierLevel,
+)
+
+
+@given("I have an ACMS index")
+def step_create_index(context):
+    """Create a new ACMS index."""
+    context.index = ACMSIndex()
+
+
+@given("I have a file traversal engine with chunk size {chunk_size:d}")
+def step_create_traversal_engine(context, chunk_size):
+    """Create a file traversal engine with specified chunk size."""
+    context.engine = FileTraversalEngine(chunk_size=chunk_size)
+
+
+@when("I create an index entry with:")
+def step_create_index_entry(context):
+    """Create an index entry from table data."""
+    data = {row["key"]: row["value"] for row in context.table}
+
+    file_type = FileType(data.get("file_type", "other"))
+    size_bytes = int(data.get("size_bytes", "0"))
+
+    context.entry = IndexEntry(
+        path=data["path"],
+        file_type=file_type,
+        size_bytes=size_bytes,
+        created_at=datetime.now(),
+        modified_at=datetime.now(),
+    )
+
+
+@then('the index entry should have path "{path}"')
+def step_check_entry_path(context, path):
+    """Verify the index entry has the expected path."""
+    assert context.entry.path == path
+
+
+@then('the index entry should have file type "{file_type}"')
+def step_check_entry_file_type(context, file_type):
+    """Verify the index entry has the expected file type."""
+    assert context.entry.file_type == FileType(file_type)
+
+
+@then("the index entry should have size {size:d} bytes")
+def step_check_entry_size(context, size):
+    """Verify the index entry has the expected size."""
+    assert context.entry.size_bytes == size
+
+
+@given('I have an index entry with path "{path}"')
+def step_create_entry_with_path(context, path):
+    """Create an index entry with a specific path."""
+    context.entry = IndexEntry(
+        path=path,
+        file_type=FileType.PYTHON,
+        size_bytes=1024,
+        created_at=datetime.now(),
+        modified_at=datetime.now(),
+    )
+
+
+@when('I add tag "{tag}" to the entry')
+def step_add_tag_to_entry(context, tag):
+    """Add a tag to the current entry."""
+    context.entry.add_tag(tag)
+
+
+@then('the entry should have tag "{tag}"')
+def step_check_entry_has_tag(context, tag):
+    """Verify the entry has a specific tag."""
+    assert context.entry.has_tag(tag)
+
+
+@then("the entry should have {count:d} tags")
+def step_check_entry_tag_count(context, count):
+    """Verify the entry has the expected number of tags."""
+    assert len(context.entry.tags) == count
+
+
+@when('I set the tier level to "{tier}"')
+def step_set_entry_tier(context, tier):
+    """Set the tier level for the entry."""
+    context.entry.set_tier(TierLevel(tier))
+
+
+@then('the entry should have tier level "{tier}"')
+def step_check_entry_tier(context, tier):
+    """Verify the entry has the expected tier level."""
+    assert context.entry.tier == TierLevel(tier)
+
+
+@when("I add the entry to the index")
+def step_add_entry_to_index(context):
+    """Add the current entry to the index."""
+    context.index.add_entry(context.entry)
+
+
+@then("the index should contain {count:d} entry")
+def step_check_index_entry_count_singular(context, count):
+    """Verify the index has the expected number of entries."""
+    assert context.index.get_entry_count() == count
+
+
+@then("the index should contain {count:d} entries")
+def step_check_index_entry_count(context, count):
+    """Verify the index has the expected number of entries."""
+    assert context.index.get_entry_count() == count
+
+
+@then('I should be able to retrieve the entry by path "{path}"')
+def step_retrieve_entry_by_path(context, path):
+    """Verify we can retrieve an entry by path."""
+    entry = context.index.get_entry(path)
+    assert entry is not None
+    assert entry.path == path
+
+
+@given("I have an index with entries:")
+def step_create_index_with_entries(context):
+    """Create an index with multiple entries from table data."""
+    context.index = ACMSIndex()
+
+    for row in context.table:
+        path = row["path"]
+        file_type = FileType(row.get("file_type", "other"))
+
+        entry = IndexEntry(
+            path=path,
+            file_type=file_type,
+            size_bytes=1024,
+            created_at=datetime.now(),
+            modified_at=datetime.now(),
+        )
+
+        # Add tags if present
+        if "tags" in row:
+            for tag in row["tags"].split(","):
+                entry.add_tag(tag.strip())
+
+        # Set tier if present
+        if "tier" in row:
+            entry.set_tier(TierLevel(row["tier"]))
+
+        context.index.add_entry(entry)
+
+
+@given("I have an index with {count:d} entries")
+def step_create_index_with_n_entries(context, count):
+    """Create an index with a specified number of entries."""
+    context.index = ACMSIndex()
+    for i in range(count):
+        entry = IndexEntry(
+            path=f"/project/file{i}.py",
+            file_type=FileType.PYTHON,
+            size_bytes=1024,
+            created_at=datetime.now(),
+            modified_at=datetime.now(),
+        )
+        context.index.add_entry(entry)
+
+
+@when('I query the index by path pattern "{pattern}"')
+def step_query_by_path_pattern(context, pattern):
+    """Query the index by path pattern."""
+    context.query_results = context.index.query_by_path(pattern)
+
+
+@then("I should get {count:d} results")
+def step_check_query_result_count(context, count):
+    """Verify the query returned the expected number of results."""
+    assert len(context.query_results) == count, (
+        f"Expected {count} results, got {len(context.query_results)}"
+    )
+
+
+@then("I should get {count:d} result")
+def step_check_query_result_count_singular(context, count):
+    """Verify the query returned the expected number of results (singular)."""
+    assert len(context.query_results) == count, (
+        f"Expected {count} result, got {len(context.query_results)}"
+    )
+
+
+@then('the results should include "{path}"')
+def step_check_result_includes_path(context, path):
+    """Verify the query results include a specific path."""
+    paths = [entry.path for entry in context.query_results]
+    assert path in paths
+
+
+@when('I query the index by file type "{file_type}"')
+def step_query_by_file_type(context, file_type):
+    """Query the index by file type."""
+    context.query_results = context.index.query_by_type(FileType(file_type))
+
+
+@then('all results should have file type "{file_type}"')
+def step_check_all_results_file_type(context, file_type):
+    """Verify all results have the expected file type."""
+    expected_type = FileType(file_type)
+    for entry in context.query_results:
+        assert entry.file_type == expected_type
+
+
+@when('I query the index by tag "{tag}"')
+def step_query_by_tag(context, tag):
+    """Query the index by tag."""
+    context.query_results = context.index.query_by_tag(tag)
+
+
+@when('I query the index by tier level "{tier}"')
+def step_query_by_tier(context, tier):
+    """Query the index by tier level."""
+    context.query_results = context.index.query_by_tier(TierLevel(tier))
+
+
+@then('the result should have path "{path}"')
+def step_check_single_result_path(context, path):
+    """Verify the single result has the expected path."""
+    assert len(context.query_results) == 1
+    assert context.query_results[0].path == path
+
+
+@given("I have an index with entries from different dates")
+def step_create_index_with_dated_entries(context):
+    """Create an index with entries from different dates."""
+    context.index = ACMSIndex()
+
+    now = datetime.now()
+    dates = [
+        now - timedelta(days=10),
+        now - timedelta(days=5),
+        now - timedelta(days=1),
+        now,
+    ]
+
+    for i, date in enumerate(dates):
+        entry = IndexEntry(
+            path=f"/project/file{i}.py",
+            file_type=FileType.PYTHON,
+            size_bytes=1024,
+            created_at=date,
+            modified_at=date,
+        )
+        context.index.add_entry(entry)
+
+
+@when('I query the index for entries modified after "{date_str}"')
+def step_query_by_recency(context, date_str):
+    """Query the index for entries modified after a specific date."""
+    # Parse date string (format: YYYY-MM-DD)
+    date = datetime.strptime(date_str, "%Y-%m-%d")
+    context.query_results = context.index.query_by_recency(date)
+
+
+@then("I should get entries modified after that date")
+def step_check_recency_results(context):
+    """Verify the recency query returned valid results."""
+    assert len(context.query_results) > 0
+
+
+@given("I have a test directory with {count:d} files")
+def step_create_test_directory(context, count):
+    """Create a temporary directory with test files."""
+    context.temp_dir = tempfile.TemporaryDirectory()
+    temp_path = Path(context.temp_dir.name)
+
+    # Create subdirectories and files
+    for i in range(count):
+        subdir = temp_path / f"subdir{i % 10}"
+        subdir.mkdir(exist_ok=True)
+
+        file_path = subdir / f"file{i}.py"
+        file_path.write_text(f"# Test file {i}\nprint('Hello {i}')\n")
+
+
+@when("I traverse and index the directory")
+def step_traverse_and_index(context):
+    """Traverse and index the test directory."""
+    context.engine.reset_index()
+    context.index = context.engine.traverse_and_index(context.temp_dir.name)
+
+
+@when("I traverse and index the directory with chunk size {chunk_size:d}")
+def step_traverse_and_index_with_chunk_size(context, chunk_size):
+    """Traverse and index the test directory with a specific chunk size."""
+    engine = FileTraversalEngine(chunk_size=chunk_size)
+    context.index = engine.traverse_and_index(context.temp_dir.name)
+
+
+@then("all entries should have valid file paths")
+def step_check_valid_file_paths(context):
+    """Verify all entries have valid file paths."""
+    for entry in context.index.get_all_entries():
+        assert entry.path
+        assert len(entry.path) > 0
+
+
+@then("the traversal should complete without timeout")
+def step_check_no_timeout(context):
+    """Verify the traversal completed without timeout."""
+    # This step passes if we got here without timing out
+    assert True
+
+
+@given("I have a test directory with files including:")
+def step_create_test_directory_with_specific_files(context):
+    """Create a test directory with specific files."""
+    context.temp_dir = tempfile.TemporaryDirectory()
+    temp_path = Path(context.temp_dir.name)
+
+    for row in context.table:
+        file_path = temp_path / row["path"].lstrip("/")
+        file_path.parent.mkdir(parents=True, exist_ok=True)
+        file_path.write_text("test content")
+
+
+@when('I traverse and index the directory excluding "{exclude1}" and "{exclude2}"')
+def step_traverse_with_exclusions(context, exclude1, exclude2):
+    """Traverse and index with exclusion patterns."""
+    context.engine.reset_index()
+    context.index = context.engine.traverse_and_index(
+        context.temp_dir.name,
+        exclude_patterns=[exclude1, exclude2],
+    )
+
+
+@then('the index should not contain "{pattern}" paths')
+def step_check_no_excluded_paths(context, pattern):
+    """Verify the index doesn't contain paths matching the pattern."""
+    for entry in context.index.get_all_entries():
+        assert pattern not in entry.path
+
+
+@when("I get all entries from the index")
+def step_get_all_entries(context):
+    """Get all entries from the index."""
+    context.query_results = context.index.get_all_entries()
+
+
+@when("I get the entry count")
+def step_get_entry_count(context):
+    """Get the entry count from the index."""
+    context.entry_count = context.index.get_entry_count()
+
+
+@then("the ACMS entry count should be {count:d}")
+def step_check_entry_count_value(context, count):
+    """Verify the entry count matches the expected value."""
+    assert context.entry_count == count
+
+
+@when("I remove an entry by path")
+def step_remove_entry(context):
+    """Remove an entry from the index."""
+    # Get the first entry's path
+    entries = context.index.get_all_entries()
+    if entries:
+        context.index.remove_entry(entries[0].path)
+
+
+@when("I query the index with filters:")
+def step_query_with_combined_filters(context):
+    """Query the index with multiple filters."""
+    filters = {row["filter"]: row["value"] for row in context.table}
+
+    path_pattern = filters.get("path_pattern")
+    file_type_str = filters.get("file_type")
+    tier_str = filters.get("tier")
+
+    file_type = FileType(file_type_str) if file_type_str else None
+    tier = TierLevel(tier_str) if tier_str else None
+
+    context.query_results = context.index.query_combined(
+        path_pattern=path_pattern,
+        file_type=file_type,
+        tier=tier,
+    )

features/steps/actor_cli_yaml_steps.py

@@ -337,6 +337,63 @@ def step_remove_namespaced_ok(context: Any) -> None:
     )
 
 
+@when("I run actor remove with format json")
+def step_remove_format_json(context: Any) -> None:
+    with (
+        patch("cleveragents.cli.commands.actor._get_services") as mock_svc,
+        patch("cleveragents.cli.commands.actor._compute_actor_impact") as mock_impact,
+    ):
+        mock_registry = MagicMock()
+        mock_service = MagicMock()
+        actor = _make_actor(
+            name="local/remove-json",
+            provider="json-provider",
+            model="gpt-json",
+        )
+        mock_registry.get_actor.return_value = actor
+        mock_impact.return_value = (2, 1, 3)
+        mock_svc.return_value = (mock_service, mock_registry)
+
+        context.result = context.runner.invoke(
+            actor_app,
+            ["remove", actor.name, "--format", "json"],
+        )
+
+    context.mock_actor_registry = mock_registry
+    context.actor = actor
+    context.impact_counts = (2, 1, 3)
+
+
+@then("the actor remove output should be valid JSON envelope")
+def step_remove_json_valid(context: Any) -> None:
+    assert context.result.exit_code == 0
+    parsed = json.loads(context.result.output.strip())
+    assert _ENVELOPE_KEYS.issubset(parsed.keys())
+    assert parsed["command"] == f"agents actor remove {context.actor.name}"
+    assert parsed["status"] == "ok"
+    assert parsed["exit_code"] == 0
+    data = _unwrap_envelope(parsed)
+    assert isinstance(data, dict)
+    actor_data = data.get("actor_removed", {})
+    assert actor_data.get("name") == context.actor.name
+    assert actor_data.get("provider") == context.actor.provider
+    assert actor_data.get("model") == context.actor.model
+    impact = data.get("impact", {})
+    expected_sessions, expected_plans, expected_actions = context.impact_counts
+    assert impact.get("sessions") == expected_sessions
+    assert impact.get("active_plans") == expected_plans
+    assert impact.get("actions_referencing") == expected_actions
+    cleanup = data.get("cleanup", {})
+    assert cleanup.get("config") == "kept on disk"
+    assert cleanup.get("contexts") == "0 orphaned"
+    messages = parsed.get("messages", [])
+    assert messages, "expected messages in envelope"
+    first_message = messages[0]
+    assert first_message.get("level") == "ok"
+    assert "Actor removed" in first_message.get("text", "")
+    context.mock_actor_registry.remove_actor.assert_called_once_with(context.actor.name)
+
+
 # ------------------------------------------------------------------
 # Update with --format
 # ------------------------------------------------------------------

features/steps/actor_compiler_coverage_steps.py

@@ -135,7 +135,8 @@ def _subgraph_node(node_id: str, actor_ref: str = "") -> NodeDefinition:
         type=NodeType.SUBGRAPH,
         name=node_id.title(),
         description=f"Subgraph {node_id}",
-        config={"actor_ref": actor_ref},
+        config={},
+        actor_ref=actor_ref if actor_ref else None,
     )
 
 

features/steps/actor_compiler_steps.py

@@ -186,7 +186,8 @@ def step_given_subgraph_ref(context: Context, ref_name: str) -> None:
             type=NodeType.SUBGRAPH,
             name="Sub",
             description="Subgraph",
-            config={"actor_ref": ref_name},
+            config={},
+            actor_ref=ref_name,
         ),
     ]
     edges = [EdgeDefinition(from_node="main", to_node="sub")]
@@ -241,7 +242,8 @@ def step_given_outer_referencing_inner(
             type=NodeType.SUBGRAPH,
             name="Sub",
             description="Subgraph",
-            config={"actor_ref": inner_name},
+            config={},
+            actor_ref=inner_name,
         ),
     ]
     edges = [EdgeDefinition(from_node="main", to_node="sub")]
@@ -260,7 +262,8 @@ def step_given_resolver_cycle(context: Context, inner_name: str, back_ref: str)
             type=NodeType.SUBGRAPH,
             name="Child",
             description="Back-ref",
-            config={"actor_ref": back_ref},
+            config={},
+            actor_ref=back_ref,
         ),
     ]
     inner = _build_graph_config(inner_name, inner_nodes, [], "child", ["child"])

features/steps/actor_subgraph_cycle_detection_steps.py

@@ -0,0 +1,201 @@
+"""Step definitions for cross-actor subgraph cycle detection tests.
+
+Tests for features/actor_subgraph_cycle_detection.feature — validates that
+the actor compiler correctly reads actor_ref from the top-level
+NodeDefinition field (not from node.config) when detecting cross-actor
+subgraph cycles.
+
+This is the regression test for issue #1431: _detect_subgraph_cycles()
+was reading node.config.get("actor_ref", "") instead of node.actor_ref,
+causing cycle detection to always return an empty string and never detect
+cross-actor cycles.
+"""
+
+from __future__ import annotations
+
+from behave import given, then, when
+from behave.runner import Context
+
+from cleveragents.actor.compiler import (
+    SubgraphCycleError,
+    compile_actor,
+)
+from cleveragents.actor.schema import (
+    ActorConfigSchema,
+    ActorType,
+    EdgeDefinition,
+    NodeDefinition,
+    NodeType,
+    RouteDefinition,
+)
+
+
+# ────────────────────────────────────────────────────────────
+# Helpers
+# ────────────────────────────────────────────────────────────
+
+
+def _make_graph_actor(
+    name: str,
+    nodes: list[NodeDefinition],
+    edges: list[EdgeDefinition],
+    entry_node: str,
+    exit_nodes: list[str],
+) -> ActorConfigSchema:
+    """Build a minimal valid GRAPH actor for testing."""
+    route = RouteDefinition(
+        nodes=nodes,
+        edges=edges,
+        entry_node=entry_node,
+        exit_nodes=exit_nodes,
+    )
+    return ActorConfigSchema(
+        name=name,
+        type=ActorType.GRAPH,
+        description=f"Test actor {name}",
+        provider="openai",
+        model="gpt-4",
+        route=route,
+    )
+
+
+def _make_agent_node(node_id: str) -> NodeDefinition:
+    """Build a minimal AGENT node."""
+    return NodeDefinition(
+        id=node_id,
+        type=NodeType.AGENT,
+        name=node_id.title(),
+        description=f"Agent {node_id}",
+        config={"agent": "default"},
+    )
+
+
+def _make_subgraph_node(node_id: str, actor_ref: str) -> NodeDefinition:
+    """Build a SUBGRAPH node using the top-level actor_ref field.
+
+    The actor_ref is set as a first-class field on NodeDefinition, NOT
+    inside config.  This is the correct way to reference a subgraph actor.
+    """
+    return NodeDefinition(
+        id=node_id,
+        type=NodeType.SUBGRAPH,
+        name=node_id.title(),
+        description=f"Subgraph {node_id}",
+        config={},
+        actor_ref=actor_ref,
+    )
+
+
+# ────────────────────────────────────────────────────────────
+# Given steps
+# ────────────────────────────────────────────────────────────
+
+
+@given("the actor compiler is available")
+def step_compiler_available(context: Context) -> None:
+    """Ensure the actor compiler module is importable."""
+    context.actor_registry: dict[str, ActorConfigSchema] = {}
+
+
+@given('actor "{name}" has a subgraph node with actor_ref "{ref}"')
+def step_actor_has_subgraph_ref(context: Context, name: str, ref: str) -> None:
+    """Create a GRAPH actor with one agent node and one subgraph node."""
+    agent = _make_agent_node("start")
+    sub = _make_subgraph_node("sub", actor_ref=ref)
+    actor = _make_graph_actor(
+        name,
+        nodes=[agent, sub],
+        edges=[EdgeDefinition(from_node="start", to_node="sub")],
+        entry_node="start",
+        exit_nodes=["sub"],
+    )
+    context.actor_registry[name] = actor
+    context.actor_to_compile = name
+
+
+@given('actor "{name}" has no subgraph nodes')
+def step_actor_has_no_subgraph(context: Context, name: str) -> None:
+    """Create a GRAPH actor with only an agent node (no subgraph references)."""
+    agent = _make_agent_node("leaf")
+    actor = _make_graph_actor(
+        name,
+        nodes=[agent],
+        edges=[],
+        entry_node="leaf",
+        exit_nodes=["leaf"],
+    )
+    context.actor_registry[name] = actor
+
+
+@given("a registry containing both actors")
+def step_registry_contains_both(context: Context) -> None:
+    """Set up the resolver from the accumulated registry."""
+    registry = context.actor_registry
+
+    def resolver(actor_name: str) -> ActorConfigSchema | None:
+        return registry.get(actor_name)
+
+    context.resolver = resolver
+
+
+# ────────────────────────────────────────────────────────────
+# When steps
+# ────────────────────────────────────────────────────────────
+
+
+@when('I compile actor "{name}" with the registry resolver')
+def step_compile_actor(context: Context, name: str) -> None:
+    """Compile the named actor using the registry resolver."""
+    context.compile_error = None
+    context.compiled = None
+    actor = context.actor_registry[name]
+    try:
+        context.compiled = compile_actor(actor, actor_resolver=context.resolver)
+    except Exception as exc:
+        context.compile_error = exc
+
+
+# ────────────────────────────────────────────────────────────
+# Then steps
+# ────────────────────────────────────────────────────────────
+
+
+@then("the compilation should raise SubgraphCycleError")
+def step_raises_cycle_error(context: Context) -> None:
+    """Assert that compilation raised SubgraphCycleError."""
+    assert context.compile_error is not None, (
+        "Expected SubgraphCycleError but compilation succeeded"
+    )
+    assert isinstance(context.compile_error, SubgraphCycleError), (
+        f"Expected SubgraphCycleError, got {type(context.compile_error).__name__}: "
+        f"{context.compile_error}"
+    )
+
+
+@then("the actor compilation should succeed")
+def step_actor_compilation_succeeds(context: Context) -> None:
+    """Assert that compilation succeeded without errors."""
+    assert context.compile_error is None, (
+        f"Expected compilation to succeed but got: {context.compile_error}"
+    )
+    assert context.compiled is not None
+
+
+@then('the cycle error message should mention "{text}"')
+def step_cycle_error_mentions(context: Context, text: str) -> None:
+    """Assert that the cycle error message contains the given text."""
+    assert context.compile_error is not None
+    msg = str(context.compile_error)
+    assert text.lower() in msg.lower(), (
+        f"Expected '{text}' in error message, got: {msg}"
+    )
+
+
+@then('the actor subgraph_refs should map "{node}" to "{ref}"')
+def step_actor_subgraph_refs_map(context: Context, node: str, ref: str) -> None:
+    """Assert that the compiled metadata maps the node to the expected actor ref."""
+    assert context.compiled is not None
+    refs = context.compiled.metadata.subgraph_refs
+    assert refs.get(node) == ref, (
+        f"Expected subgraph_refs[{node!r}] == {ref!r}, got {refs}"
+    )

features/steps/behave_parallel_log_filtering_steps.py

@@ -1,7 +1,8 @@
 """Step definitions for behave_parallel_log_filtering.feature.
 
 Tests the conditional log replay and worker exception handling in
-``scripts/run_behave_parallel.py``.
+``scripts/run_behave_parallel.py``, as well as the PassSuppressFormatter
+per-scenario output buffering behaviour.
 """
 
 from __future__ import annotations
@@ -12,8 +13,11 @@ import sys
 from contextlib import redirect_stderr, redirect_stdout
 from pathlib import Path
 from types import ModuleType
+from typing import Any
 
 from behave import given, then, when  # type: ignore[import-untyped]
+from behave.configuration import Configuration  # type: ignore[import-untyped]
+from behave.formatter.base import StreamOpener  # type: ignore[import-untyped]
 from behave.runner import Context  # type: ignore[import-untyped]
 
 
@@ -52,6 +56,7 @@ _empty_summary = _runner_mod._empty_summary
 _worker_run_features = _runner_mod._worker_run_features
 _aggregate_worker_results = _runner_mod._aggregate_worker_results
 _has_failures = _runner_mod._has_failures
+PassSuppressFormatter = _runner_mod.PassSuppressFormatter
 
 
 def _passing_summary(features: int = 1, scenarios: int = 1) -> Summary:
@@ -357,3 +362,132 @@ def step_worker_summary_features_errors_1(context: Context) -> None:
         f"Expected features.errors=1 so _has_failures() detects partial crash, "
         f"got: {errors}"
     )
+
+
+# ---- PassSuppressFormatter helpers ----
+
+
+class _MockStatus:
+    """Minimal status object mirroring behave's Status enum interface."""
+
+    def __init__(self, name: str) -> None:
+        self.name = name
+
+
+class _MockScenario:
+    """Minimal scenario object for PassSuppressFormatter unit tests."""
+
+    def __init__(self, name: str, status_name: str) -> None:
+        self.keyword = "Scenario"
+        self.name = name
+        self.status = _MockStatus(status_name)
+
+
+class _MockStep:
+    """Minimal step object for PassSuppressFormatter unit tests."""
+
+    def __init__(
+        self,
+        keyword: str,
+        name: str,
+        status_name: str,
+        error_message: str | None,
+    ) -> None:
+        self.keyword = keyword
+        self.name = name
+        self.status = _MockStatus(status_name)
+        self.error_message = error_message
+
+
+def _make_pass_suppress_formatter() -> tuple[Any, io.StringIO]:
+    """Create a PassSuppressFormatter writing to a fresh StringIO buffer.
+
+    Returns ``(formatter, buffer)`` so callers can inspect what was written
+    to the formatter's real output stream after simulating scenario events.
+    """
+    buf: io.StringIO = io.StringIO()
+    opener = StreamOpener(stream=buf)
+    config = Configuration(command_args=["-q"])
+    formatter: Any = PassSuppressFormatter(opener, config)
+    return formatter, buf
+
+
+# ---- PassSuppressFormatter: Given ----
+
+
+@given("behave_parallel a PassSuppressFormatter backed by a captured stream")
+def step_create_pass_suppress_formatter(context: Context) -> None:
+    formatter, buf = _make_pass_suppress_formatter()
+    context.bp_formatter = formatter
+    context.bp_formatter_stream = buf
+
+
+# ---- PassSuppressFormatter: When ----
+
+
+@when("behave_parallel I simulate a passing scenario through the formatter")
+def step_simulate_passing_scenario(context: Context) -> None:
+    formatter: Any = context.bp_formatter
+    scenario = _MockScenario("Passing scenario", "passed")
+    step = _MockStep("Given", "I pass", "passed", None)
+    formatter.scenario(scenario)
+    formatter.result(step)
+    formatter.eof()
+
+
+@when("behave_parallel I simulate a failing scenario through the formatter")
+def step_simulate_failing_scenario(context: Context) -> None:
+    formatter: Any = context.bp_formatter
+    scenario = _MockScenario("Failing scenario", "failed")
+    step = _MockStep(
+        "When",
+        "I fail",
+        "failed",
+        "AssertionError: expected True got False",
+    )
+    formatter.scenario(scenario)
+    formatter.result(step)
+    formatter.eof()
+
+
+@when(
+    "behave_parallel I simulate one passing then one failing scenario"
+    " through the formatter"
+)
+def step_simulate_mixed_scenarios(context: Context) -> None:
+    formatter: Any = context.bp_formatter
+    # First: a passing scenario.
+    passing_scenario = _MockScenario("Passing scenario", "passed")
+    step1 = _MockStep("Given", "I pass", "passed", None)
+    formatter.scenario(passing_scenario)
+    formatter.result(step1)
+    # Second: a failing scenario.  Calling formatter.scenario() here finalises
+    # the previous (passing) scenario, which should be discarded.
+    failing_scenario = _MockScenario("Failing scenario", "failed")
+    step2 = _MockStep("When", "I fail", "failed", "AssertionError: it broke")
+    formatter.scenario(failing_scenario)
+    formatter.result(step2)
+    formatter.eof()
+
+
+# ---- PassSuppressFormatter: Then ----
+
+
+@then("behave_parallel the formatter real stream output should be empty")
+def step_formatter_output_empty(context: Context) -> None:
+    output: str = context.bp_formatter_stream.getvalue()
+    assert output == "", f"Expected empty output, got: {output!r}"
+
+
+@then('behave_parallel the formatter real stream output should contain "{text}"')
+def step_formatter_output_contains(context: Context, text: str) -> None:
+    output: str = context.bp_formatter_stream.getvalue()
+    assert text in output, f"Expected {text!r} in formatter output, got: {output!r}"
+
+
+@then('behave_parallel the formatter real stream output should not contain "{text}"')
+def step_formatter_output_not_contains(context: Context, text: str) -> None:
+    output: str = context.bp_formatter_stream.getvalue()
+    assert text not in output, (
+        f"Expected {text!r} NOT in formatter output, got: {output!r}"
+    )

features/steps/cli_init_yes_flag_steps.py

@@ -27,7 +27,8 @@ def _restore_cwd(context):
         os.environ.pop("CLEVERAGENTS_HOME", None)
     else:
         os.environ["CLEVERAGENTS_HOME"] = context._init_original_home
-    shutil.rmtree(context.temp_dir, ignore_errors=True)
+    if getattr(context, "temp_dir", None) is not None:
+        shutil.rmtree(context.temp_dir, ignore_errors=True)
 
 
 def _create_init_mocks(context):

features/steps/llm_trace_steps.py

@@ -809,6 +809,9 @@ class _BrokenSession:
     def rollback(self) -> None:
         pass
 
+    def close(self) -> None:
+        pass
+
     def query(self, *_args: Any, **_kwargs: Any) -> Any:
         raise SQLAlchemyDatabaseError("mock", {}, Exception("broken"))
 
@@ -985,8 +988,10 @@ def step_save_with_spy(context: Context) -> None:
     object.__setattr__(real_session, "flush", spy_flush)
     object.__setattr__(real_session, "commit", spy_commit)
 
+    # Pass the session explicitly to test the UoW path: save() must flush
+    # but must NOT commit (the caller owns the transaction boundary).
     repo = LLMTraceRepository(session_factory=lambda: real_session)
-    repo.save(context.trace)
+    repo.save(context.trace, session=real_session)
     # Commit so the data is visible for subsequent queries
     object.__setattr__(real_session, "commit", original_commit)
     real_session.commit()
@@ -1030,7 +1035,9 @@ def step_save_in_uow_rollback(context: Context) -> None:
     session = context.uow_session_factory()
     repo = LLMTraceRepository(session_factory=lambda: session)
     try:
-        repo.save(context.trace)
+        # Pass the session explicitly to use UoW mode: save() flushes but
+        # does NOT commit, so the caller's rollback can undo the change.
+        repo.save(context.trace, session=session)
         # Simulate a subsequent failure that triggers rollback
         raise RuntimeError("Simulated failure after save")
     except RuntimeError:

features/steps/migration_runner_steps.py

@@ -316,6 +316,17 @@ def step_then_temp_connection_closed(context) -> None:
     assert context.current_rev_fake_engine.connections[0].exit_called is True
 
 
+@then("the SQLite engine for get_current_revision should use check_same_thread=False")
+def step_then_get_current_revision_check_same_thread(context) -> None:
+    _url, kwargs = context.current_rev_create_call
+    assert "connect_args" in kwargs, (
+        "Expected connect_args to be passed to create_engine for SQLite"
+    )
+    assert kwargs["connect_args"].get("check_same_thread") is False, (
+        "Expected check_same_thread=False in connect_args for SQLite engine"
+    )
+
+
 @when("I initialize or upgrade a file-based SQLite database")
 def step_when_init_file_based_sqlite(context) -> None:
     import shutil

features/steps/pr_compliance_checklist_steps.py

@@ -0,0 +1,185 @@
+"""Step definitions for PR compliance checklist in implementation supervisor."""
+
+from pathlib import Path
+from typing import Any
+
+from behave import given, then, when
+
+PROJECT_ROOT = Path(__file__).resolve().parents[2]
+AGENT_DEF_PATH = PROJECT_ROOT / ".opencode" / "agents" / "implementation-supervisor.md"
+
+
+@given("the implementation-supervisor.md agent definition exists")
+def step_agent_def_exists(context: Any) -> None:
+    """Verify the implementation supervisor agent definition file exists."""
+    assert AGENT_DEF_PATH.exists(), f"Agent definition not found at {AGENT_DEF_PATH}"
+    context.agent_def_path = AGENT_DEF_PATH
+
+
+@when("I read the implementation supervisor agent definition")
+def step_read_agent_def(context: Any) -> None:
+    """Read the implementation supervisor agent definition."""
+    context.agent_def_content = AGENT_DEF_PATH.read_text(encoding="utf-8")
+
+
+@then("the worker prompt body includes the PR compliance checklist section")
+def step_prompt_includes_checklist(context: Any) -> None:
+    """Verify the worker prompt body includes the PR compliance checklist."""
+    assert "PR Compliance Checklist" in context.agent_def_content, (
+        "Worker prompt body does not include 'PR Compliance Checklist'"
+    )
+
+
+@then("the checklist is marked as MANDATORY")
+def step_checklist_is_mandatory(context: Any) -> None:
+    """Verify the checklist is marked as MANDATORY."""
+    assert "MANDATORY" in context.agent_def_content, (
+        "PR Compliance Checklist is not marked as MANDATORY"
+    )
+
+
+@then("the worker prompt body includes a CHANGELOG.md checklist item")
+def step_prompt_includes_changelog_item(context: Any) -> None:
+    """Verify the worker prompt body includes a CHANGELOG.md checklist item."""
+    assert "CHANGELOG.md" in context.agent_def_content, (
+        "Worker prompt body does not include a CHANGELOG.md checklist item"
+    )
+
+
+@then("the item instructs workers to add an entry under the Unreleased section")
+def step_changelog_item_unreleased(context: Any) -> None:
+    """Verify the CHANGELOG.md item mentions the Unreleased section."""
+    assert "[Unreleased]" in context.agent_def_content, (
+        "CHANGELOG.md checklist item does not mention the [Unreleased] section"
+    )
+
+
+@then("the worker prompt body includes a CONTRIBUTORS.md checklist item")
+def step_prompt_includes_contributors_item(context: Any) -> None:
+    """Verify the worker prompt body includes a CONTRIBUTORS.md checklist item."""
+    assert "CONTRIBUTORS.md" in context.agent_def_content, (
+        "Worker prompt body does not include a CONTRIBUTORS.md checklist item"
+    )
+
+
+@then("the item instructs workers to add or update their contribution entry")
+def step_contributors_item_add_update(context: Any) -> None:
+    """Verify the CONTRIBUTORS.md item instructs workers to add or update."""
+    assert "add or update" in context.agent_def_content, (
+        "CONTRIBUTORS.md checklist item does not instruct workers to add or update"
+    )
+
+
+@then("the worker prompt body includes a commit footer checklist item")
+def step_prompt_includes_commit_footer_item(context: Any) -> None:
+    """Verify the worker prompt body includes a commit footer checklist item."""
+    assert "Commit footer" in context.agent_def_content, (
+        "Worker prompt body does not include a commit footer checklist item"
+    )
+
+
+@then("the item specifies the ISSUES CLOSED footer format")
+def step_commit_footer_issues_closed(context: Any) -> None:
+    """Verify the commit footer item specifies the ISSUES CLOSED format."""
+    assert "ISSUES CLOSED" in context.agent_def_content, (
+        "Commit footer checklist item does not specify the ISSUES CLOSED format"
+    )
+
+
+@then("the worker prompt body includes a CI passes checklist item")
+def step_prompt_includes_ci_item(context: Any) -> None:
+    """Verify the worker prompt body includes a CI passes checklist item."""
+    assert "CI passes" in context.agent_def_content, (
+        "Worker prompt body does not include a CI passes checklist item"
+    )
+
+
+@then("the item instructs workers to verify all quality gates are green")
+def step_ci_item_quality_gates(context: Any) -> None:
+    """Verify the CI item instructs workers to verify quality gates."""
+    assert "quality gates" in context.agent_def_content, (
+        "CI checklist item does not mention quality gates"
+    )
+
+
+@then("the worker prompt body includes a BDD tests checklist item")
+def step_prompt_includes_bdd_item(context: Any) -> None:
+    """Verify the worker prompt body includes a BDD/Behave tests checklist item."""
+    assert "BDD/Behave tests" in context.agent_def_content, (
+        "Worker prompt body does not include a BDD/Behave tests checklist item"
+    )
+
+
+@then("the item instructs workers to add or update Behave feature files")
+def step_bdd_item_feature_files(context: Any) -> None:
+    """Verify the BDD item instructs workers to add or update feature files."""
+    assert "added or updated" in context.agent_def_content, (
+        "BDD checklist item does not instruct workers to add or update feature files"
+    )
+
+
+@then("the worker prompt body includes an Epic reference checklist item")
+def step_prompt_includes_epic_item(context: Any) -> None:
+    """Verify the worker prompt body includes an Epic reference checklist item."""
+    assert "Epic reference" in context.agent_def_content, (
+        "Worker prompt body does not include an Epic reference checklist item"
+    )
+
+
+@then("the item instructs workers to reference the parent Epic issue number")
+def step_epic_item_parent_reference(context: Any) -> None:
+    """Verify the Epic item instructs workers to reference the parent Epic."""
+    assert "parent Epic" in context.agent_def_content, (
+        "Epic checklist item does not instruct workers to reference the parent Epic"
+    )
+
+
+@then("the worker prompt body includes a labels checklist item")
+def step_prompt_includes_labels_item(context: Any) -> None:
+    """Verify the worker prompt body includes a labels checklist item."""
+    assert "Labels" in context.agent_def_content, (
+        "Worker prompt body does not include a labels checklist item"
+    )
+
+
+@then("the item instructs workers to apply labels via forgejo-label-manager")
+def step_labels_item_forgejo_label_manager(context: Any) -> None:
+    """Verify the labels item instructs workers to use forgejo-label-manager."""
+    assert "forgejo-label-manager" in context.agent_def_content, (
+        "Labels checklist item does not mention forgejo-label-manager"
+    )
+
+
+@then("the worker prompt body includes a milestone checklist item")
+def step_prompt_includes_milestone_item(context: Any) -> None:
+    """Verify the worker prompt body includes a milestone checklist item."""
+    assert "Milestone" in context.agent_def_content, (
+        "Worker prompt body does not include a milestone checklist item"
+    )
+
+
+@then("the item instructs workers to assign the earliest open milestone")
+def step_milestone_item_earliest(context: Any) -> None:
+    """Verify the milestone item instructs workers to assign the earliest open milestone."""
+    assert "earliest open milestone" in context.agent_def_content, (
+        "Milestone checklist item does not mention the earliest open milestone"
+    )
+
+
+@then("the worker prompt body contains all 8 mandatory checklist items")
+def step_prompt_contains_all_8_items(context: Any) -> None:
+    """Verify the worker prompt body contains all 8 mandatory checklist items."""
+    required_items = [
+        "CHANGELOG.md",
+        "CONTRIBUTORS.md",
+        "ISSUES CLOSED",
+        "CI passes",
+        "BDD/Behave tests",
+        "Epic reference",
+        "forgejo-label-manager",
+        "earliest open milestone",
+    ]
+    missing = [item for item in required_items if item not in context.agent_def_content]
+    assert not missing, (
+        f"Worker prompt body is missing the following checklist items: {missing}"
+    )

features/steps/session_cli_steps.py

@@ -4,6 +4,7 @@ from __future__ import annotations
 
 import json
 import os
+import re
 import tempfile
 from datetime import datetime
 from typing import Any
@@ -160,14 +161,34 @@ def step_existing_sessions(context: Context) -> None:
     context.mock_service.list.return_value = sessions
 
 
+@given("there are mocked existing named sessions")
+def step_existing_named_sessions(context: Context) -> None:
+    """Set up mocked sessions with names for Summary panel name-display test."""
+    sessions = [
+        _make_session(
+            session_id=_SESSION_ID,
+            actor_name="openai/gpt-4",
+            messages=[_make_message(sequence=0)],
+        ),
+        _make_session(session_id=_SESSION_ID_2),
+    ]
+    sessions[0].name = "weekly-planning"
+    sessions[1].name = "refactor-sprint"
+    context.mock_service.list.return_value = sessions
+
+
 @when("I run session CLI list")
 def step_list(context: Context) -> None:
-    context.result = context.runner.invoke(session_app, ["list"])
+    context.result = context.runner.invoke(
+        session_app, ["list"], env={"COLUMNS": "200"}
+    )
 
 
 @when("I run session CLI list with --format json")
 def step_list_json(context: Context) -> None:
-    context.result = context.runner.invoke(session_app, ["list", "--format", "json"])
+    context.result = context.runner.invoke(
+        session_app, ["list", "--format", "json"], env={"COLUMNS": "200"}
+    )
 
 
 @then("the session CLI should show all sessions in a table")
@@ -176,6 +197,125 @@ def step_list_shows_table(context: Context) -> None:
     assert "Sessions" in context.result.output
 
 
+@then("the session CLI rich table should display full session ULIDs")
+def step_list_rich_table_full_ulids(context: Context) -> None:
+    """Verify the Rich table displays the full 26-character session ULIDs."""
+    assert context.result.exit_code == 0
+    output = context.result.output
+    # Restrict assertion to the table region (before the Summary panel) so
+    # that a regression back to 8-char truncation is not masked by the full
+    # ULID appearing elsewhere (e.g. in the Summary panel).
+    summary_idx = output.find("Summary")
+    table_output = output[:summary_idx] if summary_idx != -1 else output
+    assert _SESSION_ID in table_output, (
+        f"Full ULID {_SESSION_ID} not found in Rich table output:\n"
+        f"{context.result.output}"
+    )
+    assert _SESSION_ID_2 in table_output, (
+        f"Full ULID {_SESSION_ID_2} not found in Rich table output:\n"
+        f"{context.result.output}"
+    )
+    # Negative guard: ensure the table contains full 26-character ULIDs,
+    # not the old 8-character truncated form.  A standalone 8-char prefix
+    # (not as part of the full ULID) would indicate the [:8] slice was not
+    # removed.
+    table_without_full_ids = table_output.replace(_SESSION_ID, "").replace(
+        _SESSION_ID_2, ""
+    )
+    assert _SESSION_ID[:8] not in table_without_full_ids, (
+        f"Truncated 8-char ID {_SESSION_ID[:8]} found in Rich table output:\n"
+        f"{context.result.output}"
+    )
+    assert _SESSION_ID_2[:8] not in table_without_full_ids, (
+        f"Truncated 8-char ID {_SESSION_ID_2[:8]} found in Rich table output:\n"
+        f"{context.result.output}"
+    )
+
+
+@then("the session CLI summary panel should contain full session ULIDs")
+def step_list_summary_full_ulids(context: Context) -> None:
+    """Verify the Summary panel shows full ULIDs for unnamed sessions."""
+    assert context.result.exit_code == 0
+    output = context.result.output
+    assert "Summary" in output, f"Summary panel not found in output:\n{output}"
+    # Extract the Summary panel region to avoid a false pass from the Rich
+    # table also containing the same full ULIDs.
+    summary_idx = output.find("Summary")
+    summary_output = output[summary_idx:]
+    # Both Most Recent and Oldest entries should display full ULIDs when
+    # sessions are unnamed.  Asserting only one ID would allow a regression
+    # that re-introduced [:8] on one fallback path while keeping the other
+    # intact to pass the test undetected.
+    assert _SESSION_ID in summary_output, (
+        f"Full ULID {_SESSION_ID} not found in Summary panel:\n{output}"
+    )
+    assert _SESSION_ID_2 in summary_output, (
+        f"Full ULID {_SESSION_ID_2} not found in Summary panel:\n{output}"
+    )
+
+
+@then("the session CLI summary panel should show session names")
+def step_list_summary_shows_names(context: Context) -> None:
+    """Verify the Summary panel shows session names (not ULIDs) for named sessions."""
+    assert context.result.exit_code == 0
+    output = context.result.output
+    summary_idx = output.find("Summary")
+    assert summary_idx != -1, f"Summary panel not found in output:\n{output}"
+    summary_output = output[summary_idx:]
+    assert "weekly-planning" in summary_output, (
+        f"'weekly-planning' not found in Summary panel:\n{output}"
+    )
+    assert "refactor-sprint" in summary_output, (
+        f"'refactor-sprint' not found in Summary panel:\n{output}"
+    )
+    # The Summary should NOT show the raw ULID when session names are present
+    assert _SESSION_ID not in summary_output, (
+        f"Full ULID {_SESSION_ID} unexpectedly found in Summary panel:\n{output}"
+    )
+    assert _SESSION_ID_2 not in summary_output, (
+        f"Full ULID {_SESSION_ID_2} unexpectedly found in Summary panel:\n{output}"
+    )
+
+
+@when("I capture the first session full ULID from the output")
+def step_capture_first_ulid(context: Context) -> None:
+    """Parse the first session's full ULID from the output and store it
+    for subsequent steps (round-trip tell test)."""
+    assert context.result.exit_code == 0
+    output = context.result.output
+    # Restrict the search to the table region (before the Summary panel)
+    # so that a ULID appearing in the Summary panel is not accidentally
+    # captured as the "first" session ID.
+    summary_idx = output.find("Summary")
+    search_region = output[:summary_idx] if summary_idx != -1 else output
+    match = re.search(r"[0-9A-HJKMNP-TV-Z]{26}", search_region)
+    assert match is not None, (
+        f"No 26-character ULID found in table output:\n{search_region}"
+    )
+    context.full_session_id = match.group()
+    assert len(context.full_session_id) == 26, (
+        f"Parsed ULID '{context.full_session_id}' is not 26 characters"
+    )
+    # Sanity check: the captured ID should match one of the known fixture IDs
+    assert context.full_session_id in (_SESSION_ID, _SESSION_ID_2), (
+        f"Captured ULID '{context.full_session_id}' does not match any fixture ID"
+    )
+
+
+@when('I run session CLI tell with the full session ID and prompt "{prompt}"')
+def step_tell_with_stored_ulid(context: Context, prompt: str) -> None:
+    """Run session tell using the full ULID stored from session list."""
+    session_id = context.full_session_id
+    context.mock_service.append_message.side_effect = [
+        _make_message(MessageRole.USER, prompt, 0),
+        _make_message(MessageRole.ASSISTANT, f"Acknowledged: {prompt}", 1),
+    ]
+    context.result = context.runner.invoke(
+        session_app,
+        ["tell", "--session", session_id, prompt],
+    )
+
+
 # ---------------------------------------------------------------------------
 # Show
 # ---------------------------------------------------------------------------

features/steps/tdd_migration_runner_get_current_revision_threading_10507_steps.py

@@ -0,0 +1,85 @@
+"""Steps for TDD Issue #10507 — get_current_revision() SQLite threading fix."""
+
+from __future__ import annotations
+
+from typing import Any
+from unittest.mock import MagicMock, patch
+
+from behave import then, when
+
+
+def _run_get_current_revision_and_capture_kwargs(
+    context: Any,
+) -> None:
+    """Shared helper: call get_current_revision() and capture create_engine kwargs.
+
+    Mocks ``create_engine`` and ``MigrationContext.configure`` so the call
+    completes without a real database.  The keyword arguments passed to
+    ``create_engine`` are stored on ``context.engine_creation_kwargs`` for
+    subsequent assertion steps.
+    """
+    fake_engine = MagicMock()
+    fake_connection = MagicMock()
+    fake_connection.__enter__ = MagicMock(return_value=fake_connection)
+    fake_connection.__exit__ = MagicMock(return_value=False)
+    fake_engine.connect.return_value = fake_connection
+
+    migration_ctx = MagicMock()
+    migration_ctx.get_current_revision.return_value = None
+
+    captured_kwargs: list[dict[str, Any]] = []
+
+    def fake_create_engine(url: str, **kwargs: Any) -> MagicMock:
+        captured_kwargs.append(kwargs)
+        return fake_engine
+
+    with (
+        patch(
+            "cleveragents.infrastructure.database.migration_runner.create_engine",
+            side_effect=fake_create_engine,
+        ),
+        patch(
+            "cleveragents.infrastructure.database.migration_runner.MigrationContext.configure",
+            return_value=migration_ctx,
+        ),
+    ):
+        context.revision_result = context.runner.get_current_revision()
+
+    context.engine_creation_kwargs = captured_kwargs
+
+
+@when("I request the current revision and capture the engine creation args")
+def step_when_capture_engine_args(context: Any) -> None:
+    """Call get_current_revision and capture the create_engine call arguments."""
+    _run_get_current_revision_and_capture_kwargs(context)
+
+
+@then("the SQLite engine should be created with check_same_thread set to False")
+def step_then_sqlite_engine_has_check_same_thread(context: Any) -> None:
+    """Verify the SQLite engine was created with check_same_thread=False."""
+    assert len(context.engine_creation_kwargs) == 1, (
+        f"Expected exactly 1 create_engine call, got {len(context.engine_creation_kwargs)}"
+    )
+    kwargs = context.engine_creation_kwargs[0]
+    assert "connect_args" in kwargs, (
+        "Expected connect_args in create_engine kwargs for SQLite, "
+        f"but got kwargs: {kwargs}"
+    )
+    assert kwargs["connect_args"].get("check_same_thread") is False, (
+        "Expected check_same_thread=False in connect_args, "
+        f"but got: {kwargs['connect_args']}"
+    )
+
+
+@then("the non-SQLite engine should be created without check_same_thread")
+def step_then_non_sqlite_engine_no_check_same_thread(context: Any) -> None:
+    """Verify non-SQLite engines are not given check_same_thread."""
+    assert len(context.engine_creation_kwargs) == 1, (
+        f"Expected exactly 1 create_engine call, got {len(context.engine_creation_kwargs)}"
+    )
+    kwargs = context.engine_creation_kwargs[0]
+    connect_args = kwargs.get("connect_args", {})
+    assert "check_same_thread" not in connect_args, (
+        "Expected check_same_thread to be absent for non-SQLite engine, "
+        f"but got connect_args: {connect_args}"
+    )

features/steps/tdd_plan_result_success_7501_steps.py

@@ -0,0 +1,211 @@
+"""Step definitions for TDD issue #7501 — PlanResult.success derivation.
+
+Validates that PlanRepository._to_domain derives PlanResult.success from
+the dedicated result_success column rather than from error_message is None.
+
+Targets ``PlanRepository`` in
+``src/cleveragents/infrastructure/database/repositories.py``.
+"""
+
+from __future__ import annotations
+
+import warnings
+from datetime import datetime
+
+from behave import given, then, when
+from behave.runner import Context
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+
+from cleveragents.infrastructure.database.models import Base, PlanModel
+from cleveragents.infrastructure.database.repositories import PlanRepository
+
+
+def _make_project(session: object) -> int:
+    """Insert a minimal project row and return its id."""
+    from cleveragents.infrastructure.database.models import ProjectModel
+
+    project = ProjectModel(
+        name="test-project-7501",
+        path="/tmp/test-project-7501",
+        settings={},
+    )
+    session.add(project)  # type: ignore[attr-defined]
+    session.flush()  # type: ignore[attr-defined]
+    return int(project.id)  # type: ignore[attr-defined]
+
+
+# ---------------------------------------------------------------------------
+# Background
+# ---------------------------------------------------------------------------
+
+
+@given("a clean in-memory database for plan result success tests")
+def step_clean_db_result_success(context: Context) -> None:
+    """Create a fresh in-memory SQLite database with all tables."""
+    engine = create_engine("sqlite:///:memory:")
+    Base.metadata.create_all(engine)
+    context.rs_db_engine = engine
+    session = sessionmaker(bind=engine)()
+    context.rs_db_session = session
+
+
+@given("a legacy plan repository backed by the database")
+def step_legacy_plan_repo(context: Context) -> None:
+    """Instantiate a PlanRepository using the in-memory session."""
+    with warnings.catch_warnings():
+        warnings.simplefilter("ignore", DeprecationWarning)
+        context.rs_plan_repo = PlanRepository(session=context.rs_db_session)
+    context.rs_retrieved_plan = None
+    context.rs_project_id = _make_project(context.rs_db_session)
+    context.rs_db_session.commit()
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+
+def _insert_plan_model(
+    session: object,
+    project_id: int,
+    applied_at: datetime | None = None,
+    error_message: str | None = None,
+    result_success: bool | None = None,
+) -> int:
+    """Insert a PlanModel row directly and return its id."""
+    db_plan = PlanModel(
+        project_id=project_id,
+        name="test-plan-7501",
+        prompt="Test prompt for issue 7501",
+        status="pending",
+        current=False,
+        applied_at=applied_at,
+        error_message=error_message,
+        result_success=result_success,
+    )
+    session.add(db_plan)  # type: ignore[attr-defined]
+    session.flush()  # type: ignore[attr-defined]
+    session.commit()  # type: ignore[attr-defined]
+    return int(db_plan.id)  # type: ignore[attr-defined]
+
+
+# ---------------------------------------------------------------------------
+# Given steps
+# ---------------------------------------------------------------------------
+
+
+@given("a legacy plan with applied_at set and result_success column True")
+def step_plan_result_success_true(context: Context) -> None:
+    """Insert a plan row with result_success=True."""
+    context.rs_plan_id = _insert_plan_model(
+        session=context.rs_db_session,
+        project_id=context.rs_project_id,
+        applied_at=datetime.now(),
+        error_message=None,
+        result_success=True,
+    )
+
+
+@given("a legacy plan with applied_at set and result_success column False")
+def step_plan_result_success_false(context: Context) -> None:
+    """Insert a plan row with result_success=False."""
+    context.rs_plan_id = _insert_plan_model(
+        session=context.rs_db_session,
+        project_id=context.rs_project_id,
+        applied_at=datetime.now(),
+        error_message=None,
+        result_success=False,
+    )
+
+
+@given("a legacy plan with a build error_message and result_success column True")
+def step_plan_build_error_result_success_true(context: Context) -> None:
+    """Insert a plan row with a build error but result_success=True.
+
+    This is the core bug scenario: a plan that had a build error but later
+    succeeded in the apply phase.  Without the fix, success would be derived
+    as False (because error_message is not None).  With the fix, success is
+    correctly derived as True from result_success.
+    """
+    context.rs_plan_id = _insert_plan_model(
+        session=context.rs_db_session,
+        project_id=context.rs_project_id,
+        applied_at=datetime.now(),
+        error_message="Build phase error: compilation failed",
+        result_success=True,
+    )
+
+
+@given(
+    "a legacy plan with applied_at set and result_success column NULL and no error_message"
+)
+def step_plan_null_result_success_no_error(context: Context) -> None:
+    """Insert a plan row with result_success=NULL and no error_message.
+
+    Simulates a pre-migration record.  The legacy heuristic should mark it
+    as success=True because error_message is None.
+    """
+    context.rs_plan_id = _insert_plan_model(
+        session=context.rs_db_session,
+        project_id=context.rs_project_id,
+        applied_at=datetime.now(),
+        error_message=None,
+        result_success=None,
+    )
+
+
+@given(
+    "a legacy plan with applied_at set and result_success column NULL and an error_message"
+)
+def step_plan_null_result_success_with_error(context: Context) -> None:
+    """Insert a plan row with result_success=NULL and an error_message.
+
+    Simulates a pre-migration record that failed.  The legacy heuristic
+    should mark it as success=False because error_message is not None.
+    """
+    context.rs_plan_id = _insert_plan_model(
+        session=context.rs_db_session,
+        project_id=context.rs_project_id,
+        applied_at=datetime.now(),
+        error_message="Apply phase error: deployment failed",
+        result_success=None,
+    )
+
+
+# ---------------------------------------------------------------------------
+# When steps
+# ---------------------------------------------------------------------------
+
+
+@when("the legacy plan is retrieved from the repository")
+def step_retrieve_legacy_plan(context: Context) -> None:
+    """Retrieve the plan via PlanRepository.get_by_id."""
+    context.rs_retrieved_plan = context.rs_plan_repo.get_by_id(context.rs_plan_id)
+
+
+# ---------------------------------------------------------------------------
+# Then steps
+# ---------------------------------------------------------------------------
+
+
+@then("the retrieved plan result success should be True")
+def step_check_result_success_true(context: Context) -> None:
+    """Assert that the retrieved plan's result.success is True."""
+    plan = context.rs_retrieved_plan
+    assert plan is not None, "Expected a plan but got None"
+    assert plan.result is not None, "Expected plan.result to be set but it was None"
+    assert plan.result.success is True, (
+        f"Expected plan.result.success=True but got {plan.result.success!r}"
+    )
+
+
+@then("the retrieved plan result success should be False")
+def step_check_result_success_false(context: Context) -> None:
+    """Assert that the retrieved plan's result.success is False."""
+    plan = context.rs_retrieved_plan
+    assert plan is not None, "Expected a plan but got None"
+    assert plan.result is not None, "Expected plan.result to be set but it was None"
+    assert plan.result.success is False, (
+        f"Expected plan.result.success=False but got {plan.result.success!r}"
+    )

features/tdd_migration_runner_get_current_revision_threading_10507.feature

@@ -0,0 +1,31 @@
+@tdd_issue @tdd_issue_10507
+Feature: TDD Issue #10507 — get_current_revision() must pass check_same_thread=False for SQLite
+  As a developer using MigrationRunner in a multi-threaded application
+  I want get_current_revision() to work safely from background threads
+  So that async startup flows and background migration checks do not crash
+
+  The root cause is that MigrationRunner.get_current_revision() calls
+  create_engine(self.database_url) without connect_args={"check_same_thread": False}
+  for SQLite databases.  When called from a thread other than the one that
+  created the engine, SQLite raises:
+      ProgrammingError: SQLite objects created in a thread can only be
+      used in that same thread.
+
+  The sibling method init_or_upgrade() already passes check_same_thread=False
+  for SQLite, making this an inconsistency in the same class.  Because
+  get_pending_migrations() and check_migrations_needed() both delegate to
+  get_current_revision(), the threading bug propagates to all three methods.
+
+  The fix adds connect_args={"check_same_thread": False} to the create_engine()
+  call in get_current_revision() when the database URL starts with "sqlite",
+  consistent with the existing pattern in init_or_upgrade().
+
+  Scenario: get_current_revision passes check_same_thread=False for SQLite engine
+    Given a migration runner configured for "sqlite:///:memory:"
+    When I request the current revision and capture the engine creation args
+    Then the SQLite engine should be created with check_same_thread set to False
+
+  Scenario: get_current_revision does not pass check_same_thread for non-SQLite engine
+    Given a migration runner configured for "postgresql://user:pass@localhost/testdb"
+    When I request the current revision and capture the engine creation args
+    Then the non-SQLite engine should be created without check_same_thread

features/tdd_plan_result_success_7501.feature

@@ -0,0 +1,43 @@
+@tdd_issue @tdd_issue_7501
+Feature: TDD Issue #7501 — PlanResult.success derived from result_success column
+  As a system operator managing plan lifecycle
+  I want PlanResult.success to be derived from the dedicated result_success column
+  So that plans with historical build errors are not incorrectly marked as failed
+
+  The root cause is that PlanRepository._to_domain derived PlanResult.success
+  from `error_message is None`. Because error_message is shared between the
+  build phase and the result phase, a plan that had a build error but later
+  succeeded in the apply phase would be incorrectly reconstructed as failed.
+
+  The fix adds a dedicated result_success column to the plans table and updates
+  _to_domain to use it. For backward compatibility, when result_success is NULL
+  (pre-migration records), the legacy heuristic (error_message is None) is used.
+
+  Background:
+    Given a clean in-memory database for plan result success tests
+    And a legacy plan repository backed by the database
+
+  Scenario: Plan with result_success=True is reconstructed as success=True
+    Given a legacy plan with applied_at set and result_success column True
+    When the legacy plan is retrieved from the repository
+    Then the retrieved plan result success should be True
+
+  Scenario: Plan with result_success=False is reconstructed as success=False
+    Given a legacy plan with applied_at set and result_success column False
+    When the legacy plan is retrieved from the repository
+    Then the retrieved plan result success should be False
+
+  Scenario: Plan with build error but result_success=True is reconstructed as success=True
+    Given a legacy plan with a build error_message and result_success column True
+    When the legacy plan is retrieved from the repository
+    Then the retrieved plan result success should be True
+
+  Scenario: Plan with NULL result_success falls back to error_message heuristic when no error
+    Given a legacy plan with applied_at set and result_success column NULL and no error_message
+    When the legacy plan is retrieved from the repository
+    Then the retrieved plan result success should be True
+
+  Scenario: Plan with NULL result_success falls back to error_message heuristic when error present
+    Given a legacy plan with applied_at set and result_success column NULL and an error_message
+    When the legacy plan is retrieved from the repository
+    Then the retrieved plan result success should be False

noxfile.py

@@ -111,6 +111,13 @@ def _install_behave_parallel(session: nox.Session) -> None:
     runner_script = Path(__file__).parent / "scripts" / "run_behave_parallel.py"
     (pkg_dir / "cli.py").write_text(runner_script.read_text(encoding="utf-8"))
 
+    formatter_script = (
+        Path(__file__).parent / "scripts" / "behave_pass_suppress_formatter.py"
+    )
+    (pkg_dir / "behave_pass_suppress_formatter.py").write_text(
+        formatter_script.read_text(encoding="utf-8")
+    )
+
     setup_path = source_dir / "setup.py"
     setup_path.write_text(
         "from setuptools import find_packages, setup\n"

robot/actor_compiler.robot

@@ -42,3 +42,15 @@ Reject LLM Actor Compilation
     Should Be Equal As Integers    ${result.rc}    0
     Should Contain    ${result.stdout}    actor-compiler-expected-fail
     Should Contain    ${result.stdout}    GRAPH
+
+Detect Cross-Actor Subgraph Cycle Via actor_ref Field
+    [Documentation]    Verify that mutually-referencing actors raise SubgraphCycleError.
+    ...    This is the regression test for issue #1431: _detect_subgraph_cycles()
+    ...    was reading node.config.get("actor_ref") instead of node.actor_ref,
+    ...    causing cycle detection to silently fail.
+    [Tags]    slow
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    cycle-detect    dummy    cwd=${WORKSPACE}
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    actor-compiler-cycle-detected

robot/actor_remove_cli.robot

@@ -0,0 +1,18 @@
+*** Settings ***
+Documentation    Integration tests for actor remove CLI format output
+Resource         ${CURDIR}/common.resource
+Suite Setup      Setup Test Environment
+Suite Teardown   Cleanup Test Environment
+
+*** Variables ***
+${HELPER}    ${CURDIR}/helper_actor_remove_cli.py
+
+*** Test Cases ***
+Actor Remove Format JSON Emits Spec Envelope
+    [Documentation]    Verify that ``actor remove --format json`` emits a spec-compliant envelope
+    [Tags]    actor_remove_cli_format
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    remove-json    cwd=${WORKSPACE}
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    actor-remove-json-format-ok

robot/helper_actor_compiler.py

@@ -66,6 +66,82 @@ def main() -> int:
             print(f"actor-compiler-fail: {exc}")
             return 1
 
+    if command == "cycle-detect":
+        # Test cross-actor cycle detection using actor_ref top-level field.
+        # Creates two mutually-referencing actors and verifies SubgraphCycleError
+        # is raised, confirming the fix for issue #1431.
+        from cleveragents.actor.compiler import SubgraphCycleError
+        from cleveragents.actor.schema import (
+            ActorType,
+            EdgeDefinition,
+            NodeDefinition,
+            NodeType,
+            RouteDefinition,
+        )
+
+        def _make_actor(
+            name: str, subgraph_ref: str | None = None
+        ) -> ActorConfigSchema:
+            if subgraph_ref is not None:
+                nodes = [
+                    NodeDefinition(
+                        id="start",
+                        type=NodeType.AGENT,
+                        name="Start",
+                        description="Start",
+                        config={"agent": "default"},
+                    ),
+                    NodeDefinition(
+                        id="sub",
+                        type=NodeType.SUBGRAPH,
+                        name="Sub",
+                        description="Subgraph",
+                        config={},
+                        actor_ref=subgraph_ref,
+                    ),
+                ]
+                edges = [EdgeDefinition(from_node="start", to_node="sub")]
+                entry, exits = "start", ["sub"]
+            else:
+                nodes = [
+                    NodeDefinition(
+                        id="leaf",
+                        type=NodeType.AGENT,
+                        name="Leaf",
+                        description="Leaf",
+                        config={"agent": "default"},
+                    )
+                ]
+                edges = []
+                entry, exits = "leaf", ["leaf"]
+            route = RouteDefinition(
+                nodes=nodes, edges=edges, entry_node=entry, exit_nodes=exits
+            )
+            return ActorConfigSchema(
+                name=name,
+                type=ActorType.GRAPH,
+                description=f"Test actor {name}",
+                model="gpt-4",
+                route=route,
+            )
+
+        actor_a = _make_actor("test/actor-a", subgraph_ref="test/actor-b")
+        actor_b = _make_actor("test/actor-b", subgraph_ref="test/actor-a")
+        registry = {"test/actor-a": actor_a, "test/actor-b": actor_b}
+
+        try:
+            compile_actor(actor_a, actor_resolver=registry.get)
+            print(
+                "actor-compiler-cycle-not-detected: FAIL — expected SubgraphCycleError"
+            )
+            return 1
+        except SubgraphCycleError as exc:
+            print(f"actor-compiler-cycle-detected: {exc}")
+            return 0
+        except Exception as exc:
+            print(f"actor-compiler-unexpected-error: {exc}")
+            return 1
+
     print(f"Unknown command: {command}")
     return 1
 

robot/helper_actor_remove_cli.py

@@ -0,0 +1,164 @@
+"""Helper script for Robot integration tests covering ``actor remove --format`` output.
+
+Exercises the real ``agents`` CLI via subprocess — no mocking of any kind.
+A test actor is seeded via ``agents actor add``, then removed via
+``agents actor remove --format json``, and the resulting JSON envelope is
+validated against the spec.
+
+Usage::
+
+    python helper_actor_remove_cli.py <command>
+
+Where *command* is one of:
+
+- ``remove-json`` — seed an actor, remove it with ``--format json``, validate envelope
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import sys
+from pathlib import Path
+
+# Ensure src is importable when run from workspace root
+_SRC = str(Path(__file__).resolve().parents[1] / "src")
+if _SRC not in sys.path:
+    sys.path.insert(0, _SRC)
+
+# Ensure robot/ is on the import path for helper_e2e_common.
+_ROBOT = str(Path(__file__).resolve().parent)
+if _ROBOT not in sys.path:
+    sys.path.insert(0, _ROBOT)
+
+from helper_e2e_common import cleanup_workspace, run_cli, setup_workspace  # noqa: E402
+
+_ACTOR_NAME = "local/robot-remove-actor"
+
+_ACTOR_CONFIG: dict[str, object] = {
+    "name": _ACTOR_NAME,
+    "provider": "openai",
+    "model": "gpt-4",
+}
+
+
+def _write_actor_config(workspace: str) -> str:
+    """Write actor config JSON to a temp file and return its path."""
+    config_path = os.path.join(workspace, "robot_remove_actor.json")
+    with open(config_path, "w", encoding="utf-8") as fh:
+        json.dump(_ACTOR_CONFIG, fh)
+    return config_path
+
+
+def test_remove_format_json() -> None:
+    """Seed an actor via the real CLI, remove it with ``--format json``.
+
+    Validates the resulting JSON envelope against the spec.
+    """
+    workspace = setup_workspace(prefix="robot_actor_remove_")
+    try:
+        config_path = _write_actor_config(workspace)
+
+        # Step 1: Add the actor using the real CLI.
+        add_result = run_cli(
+            "actor",
+            "add",
+            _ACTOR_NAME,
+            "--config",
+            config_path,
+            workspace=workspace,
+        )
+        assert add_result.returncode == 0, (
+            f"actor add failed (rc={add_result.returncode}):\n"
+            f"stdout: {add_result.stdout}\nstderr: {add_result.stderr}"
+        )
+
+        # Step 2: Remove the actor with --format json using the real CLI.
+        remove_result = run_cli(
+            "actor",
+            "remove",
+            _ACTOR_NAME,
+            "--format",
+            "json",
+            workspace=workspace,
+        )
+        assert remove_result.returncode == 0, (
+            f"actor remove --format json failed (rc={remove_result.returncode}):\n"
+            f"stdout: {remove_result.stdout}\nstderr: {remove_result.stderr}"
+        )
+
+        # Step 3: Parse and validate the JSON envelope.
+        output = remove_result.stdout.strip()
+        assert output, (
+            f"actor remove --format json produced no output.\n"
+            f"stderr: {remove_result.stderr}"
+        )
+
+        payload = json.loads(output)
+
+        assert payload["command"] == f"agents actor remove {_ACTOR_NAME}", (
+            f"Unexpected command field: {payload.get('command')!r}"
+        )
+        assert payload["status"] == "ok", (
+            f"Unexpected status: {payload.get('status')!r}"
+        )
+        assert payload["exit_code"] == 0, (
+            f"Unexpected exit_code: {payload.get('exit_code')!r}"
+        )
+
+        data = payload["data"]
+
+        removed = data.get("actor_removed", {})
+        assert removed.get("name") == _ACTOR_NAME, (
+            f"actor_removed.name mismatch: {removed.get('name')!r}"
+        )
+        assert removed.get("provider") == _ACTOR_CONFIG["provider"], (
+            f"actor_removed.provider mismatch: {removed.get('provider')!r}"
+        )
+        assert removed.get("model") == _ACTOR_CONFIG["model"], (
+            f"actor_removed.model mismatch: {removed.get('model')!r}"
+        )
+
+        impact = data.get("impact", {})
+        assert "sessions" in impact, f"Missing 'sessions' in impact: {impact}"
+        assert "active_plans" in impact, f"Missing 'active_plans' in impact: {impact}"
+        assert "actions_referencing" in impact, (
+            f"Missing 'actions_referencing' in impact: {impact}"
+        )
+
+        cleanup = data.get("cleanup", {})
+        assert cleanup.get("config") == "kept on disk", (
+            f"cleanup.config mismatch: {cleanup.get('config')!r}"
+        )
+        assert "contexts" in cleanup, f"Missing 'contexts' in cleanup: {cleanup}"
+
+        messages = payload.get("messages", [])
+        assert messages, f"Expected non-empty messages list, got: {messages}"
+        assert messages[0].get("level") == "ok", (
+            f"Unexpected message level: {messages[0].get('level')!r}"
+        )
+        assert "Actor removed" in messages[0].get("text", ""), (
+            f"Expected 'Actor removed' in message text: {messages[0].get('text')!r}"
+        )
+
+        print("actor-remove-json-format-ok")
+
+    finally:
+        cleanup_workspace(workspace)
+
+
+def main() -> None:
+    command = sys.argv[1] if len(sys.argv) > 1 else "remove-json"
+    dispatch: dict[str, object] = {
+        "remove-json": test_remove_format_json,
+    }
+    handler = dispatch.get(command)
+    if handler is None:
+        print(f"Unknown command: {command}", file=sys.stderr)
+        sys.exit(1)
+    assert callable(handler)
+    handler()
+
+
+if __name__ == "__main__":
+    main()

robot/helper_behave_parallel_log_filtering.py

@@ -26,6 +26,15 @@ def _load_runner_module() -> ModuleType:
     first or modify the path to be anchored relative to ``__file__``.
     """
     script_path = Path("scripts") / "run_behave_parallel.py"
+    # Ensure the scripts/ directory is on sys.path so that
+    # ``run_behave_parallel.py`` can ``from behave_pass_suppress_formatter
+    # import PassSuppressFormatter``.  This is necessary when the helper is
+    # invoked outside of a nox session (e.g. integration tests), where the
+    # behave_parallel package created by noxfile.py for unit_tests is not
+    # available.
+    scripts_dir = str(script_path.parent.resolve())
+    if scripts_dir not in sys.path:
+        sys.path.insert(0, scripts_dir)
     spec = importlib.util.spec_from_file_location(
         "run_behave_parallel", str(script_path)
     )

scripts/behave_pass_suppress_formatter.py

@@ -0,0 +1,125 @@
+"""PassSuppressFormatter — a custom Behave formatter that suppresses output
+for passing and skipped scenarios, reducing CI log noise from ~100,000 lines
+to ≤ 10 lines for an all-passing suite.
+
+This module is embedded in the same directory as ``run_behave_parallel.py``
+so that ``_install_behave_parallel()`` in ``noxfile.py`` can copy both files
+into the temporary ``behave_parallel`` package.
+"""
+
+from __future__ import annotations
+
+import io
+from typing import Any
+
+from behave.formatter.base import (
+    Formatter as _BehaveFormatter,  # type: ignore[import-untyped]
+)
+
+# Sentinel set of statuses whose output should be suppressed.
+# Every status not in this set (e.g. "failed", "undefined") causes the
+# buffered scenario output to be flushed to the real output stream.
+_SUPPRESS_STATUSES: frozenset[str] = frozenset({"passed", "skipped"})
+
+
+class PassSuppressFormatter(_BehaveFormatter):
+    """Behave formatter that suppresses output for passing scenarios.
+
+    All per-scenario output is buffered in a :class:`io.StringIO`.  When a
+    scenario ends (signalled by the next :meth:`scenario` call or by
+    :meth:`eof`):
+
+    * **Failed / errored scenario** — the buffer is flushed to the real
+      output stream so developers and CI tools see the full step-level
+      details.
+    * **Passed / skipped scenario** — the buffer is silently discarded.
+
+    The result for an all-passing suite is zero formatter output, leaving
+    only the ``_print_overall_summary`` block emitted by the runner as
+    visible output (~5-10 lines regardless of suite size).
+
+    Coverage mode (``BEHAVE_PARALLEL_COVERAGE=1``) bypasses this formatter
+    entirely via ``_make_runner()``, which falls back to behave's default
+    format so that slipcover can instrument a single sequential process
+    without output interference.
+    """
+
+    name: str = "pass_suppress"
+    description: str = "Suppress passing scenario output; show failures fully"
+
+    def __init__(self, stream_opener: Any, config: Any) -> None:
+        super().__init__(stream_opener, config)
+        # Ensure the real output stream is open (opens it if stream_opener
+        # was constructed with only a filename, not a pre-opened stream).
+        self.stream = self.open()
+
+        # Per-scenario buffering state.
+        self._current_scenario: Any = None
+        self._scenario_buf: io.StringIO = io.StringIO()
+
+    # ------------------------------------------------------------------
+    # Internal helpers
+    # ------------------------------------------------------------------
+
+    def _finalize_previous_scenario(self) -> None:
+        """Flush the scenario buffer to the real stream if scenario failed.
+
+        Called at the start of each new scenario and at end-of-feature so
+        the previous scenario's buffered output is either committed or
+        discarded based on the scenario's final status.
+        """
+        if self._current_scenario is None:
+            return
+        status: Any = getattr(self._current_scenario, "status", None)
+        status_name: str = (
+            getattr(status, "name", str(status)) if status is not None else ""
+        )
+        if status_name not in _SUPPRESS_STATUSES:
+            output = self._scenario_buf.getvalue()
+            if output:
+                self.stream.write(output)
+                self.stream.flush()
+        # Reset the buffer regardless — the next scenario starts fresh.
+        self._scenario_buf = io.StringIO()
+
+    # ------------------------------------------------------------------
+    # Formatter interface (behave lifecycle hooks)
+    # ------------------------------------------------------------------
+
+    def uri(self, uri: str) -> None:
+        """Called before each feature file; no output needed."""
+
+    def feature(self, feature: Any) -> None:
+        """Called at feature start; no output needed."""
+
+    def background(self, background: Any) -> None:
+        """Called when a feature background is declared; no output needed."""
+
+    def scenario(self, scenario: Any) -> None:
+        """Start buffering output for *scenario*, finalising the previous one."""
+        self._finalize_previous_scenario()
+        self._current_scenario = scenario
+        # Write the scenario header into the new buffer.
+        self._scenario_buf.write(f"\n  {scenario.keyword}: {scenario.name}\n")
+
+    def step(self, step: Any) -> None:
+        """Step announced before execution; no output needed at this point."""
+
+    def match(self, match: Any) -> None:
+        """Step matched; no output needed."""
+
+    def result(self, step: Any) -> None:
+        """Record a step result in the per-scenario buffer."""
+        status: Any = getattr(step, "status", None)
+        status_name: str = (
+            getattr(status, "name", "unknown") if status is not None else "unknown"
+        )
+        self._scenario_buf.write(f"    {step.keyword} {step.name} ... {status_name}\n")
+        error_msg: str | None = getattr(step, "error_message", None)
+        if error_msg:
+            self._scenario_buf.write(f"{error_msg}\n")
+
+    def eof(self) -> None:
+        """End of feature file: finalise the current (last) scenario."""
+        self._finalize_previous_scenario()
+        self._current_scenario = None

scripts/run_behave_parallel.py

@@ -1,20 +1,16 @@
 """In-process parallel behave runner.
 
-Replaces the old subprocess-per-feature model with direct use of
-behave's ``Runner`` API.  Step definitions and environment hooks are
-loaded once per process; feature files are parsed and executed without
-Python interpreter startup overhead.
+Uses behave's ``Runner`` API directly: step definitions and environment
+hooks are loaded once per process; feature files are parsed and executed
+without Python interpreter startup overhead.
 
 Parallelism modes
 -----------------
 * **Sequential** (``--processes 1`` or ``BEHAVE_PARALLEL_COVERAGE=1``):
   All features run in a single ``Runner.run()`` call.
 * **Parallel** (``--processes N``, N > 1, no coverage):
-  Features are split into *N* equal-size chunks.  A
-  ``multiprocessing.Pool`` with the ``fork`` start method dispatches
-  each chunk to a worker that creates its own ``Runner`` (hooks and
-  step definitions are re-loaded cheaply because all heavy modules are
-  already in memory from the parent).
+  Features split into *N* equal-size chunks dispatched via
+  ``multiprocessing.Pool`` with the ``fork`` start method.
 """
 
 from __future__ import annotations
@@ -31,6 +27,13 @@ from contextlib import redirect_stderr, redirect_stdout, suppress
 from pathlib import Path
 from typing import Any
 
+try:
+    from behave_pass_suppress_formatter import PassSuppressFormatter
+except ImportError:
+    from behave_parallel.behave_pass_suppress_formatter import (  # type: ignore[import-untyped,unused-ignore]
+        PassSuppressFormatter,
+    )
+
 DEFAULT_FEATURE_ROOT = "features/"
 
 
@@ -79,6 +82,7 @@ def _is_btrfs_or_overlayfs() -> bool:
 
 # Type alias for summary dictionaries
 Summary = dict[str, Any]
+WorkerResult = tuple[bool, str, str, Summary]
 
 
 # ---------------------------------------------------------------------------
@@ -174,10 +178,7 @@ def _format_duration(seconds: float) -> str:
     return f"{remainder:.3f}s"
 
 
-def _print_overall_summary(
-    total: Summary,
-    wall_seconds: float | None = None,
-) -> None:
+def _print_overall_summary(total: Summary, wall_seconds: float | None = None) -> None:
     print("\nOverall summary:")
     for bucket in ("features", "scenarios", "steps"):
         b = total[bucket]
@@ -272,35 +273,50 @@ def _make_runner(feature_paths: list[str], behave_args: list[str]) -> Any:
     Mirrors the format-defaulting logic from ``behave.__main__.run_behave``
     so that ``-q`` and bare invocations get a sensible formatter instead
     of crashing on ``config.format is None``.
+
+    When no explicit ``--format``/``-f`` flag is present in *behave_args* and
+    ``BEHAVE_PARALLEL_COVERAGE`` is **not** set, the runner uses
+    :class:`PassSuppressFormatter` so that passing scenarios produce no
+    output.  Coverage mode falls back to ``config.default_format`` (normally
+    ``pretty``) so that slipcover can instrument a single sequential process
+    without output interference.
     """
     from behave.configuration import Configuration
+    from behave.formatter._registry import register_as
     from behave.runner import Runner
     from behave.runner_util import reset_runtime
 
     reset_runtime()
+    # Register our custom formatter so behave can resolve it by name when
+    # make_formatters() is called during Runner initialisation.
+    register_as(PassSuppressFormatter.name, PassSuppressFormatter)
+
     args = list(behave_args) + [str(p) for p in feature_paths]
     config = Configuration(command_args=args)
     if not config.format:
-        config.format = [config.default_format]
+        # No explicit format was requested by the caller.  Use pass_suppress
+        # unless coverage instrumentation is active (BEHAVE_PARALLEL_COVERAGE=1),
+        # where slipcover needs unmodified output from a single sequential process.
+        coverage_mode = bool(os.environ.get("BEHAVE_PARALLEL_COVERAGE"))
+        if coverage_mode:
+            config.format = [config.default_format]
+        else:
+            config.format = [PassSuppressFormatter.name]
     return Runner(config)
 
 
-def _run_features_inprocess(
-    feature_paths: list[str], behave_args: list[str]
-) -> tuple[bool, Summary]:
+def _run_features_inprocess(paths: list[str], args: list[str]) -> tuple[bool, Summary]:
     """Run *all* feature_paths in a single behave Runner invocation.
 
     Returns ``(failed: bool, summary: dict)``.
     """
-    runner = _make_runner(feature_paths, behave_args)
+    runner = _make_runner(paths, args)
     failed = runner.run()
     summary = _extract_summary(runner)
     return failed, summary
 
 
-def _worker_run_features(
-    payload: tuple[list[str], list[str]],
-) -> tuple[bool, str, str, Summary]:
+def _worker_run_features(payload: tuple[list[str], list[str]]) -> WorkerResult:
     """Entry point for multiprocessing workers.
 
     Runs a chunk of feature files in a forked child process.  Heavy
@@ -348,9 +364,7 @@ def _worker_run_features(
 # ---------------------------------------------------------------------------
 
 
-def _aggregate_worker_results(
-    results: list[tuple[bool, str, str, Summary]],
-) -> Summary:
+def _aggregate_worker_results(results: list[WorkerResult]) -> Summary:
     """Aggregate worker results, replaying logs only for failed chunks.
 
     Iterates over the list of ``(worker_failed, stdout, stderr, summary)``

src/cleveragents/acms/__init__.py

@@ -5,12 +5,22 @@ technology-specific vocabulary extensions, and the DetailLevelMap
 inheritance mechanism for resolving named detail levels across the
 ontology hierarchy (Layer 3 -> Layer 2 -> Layer 1 -> Layer 0).
 
+Also provides the ACMS index data model and file traversal engine for
+indexing large projects.
+
 Based on ``docs/specification.md`` ~lines 42333-42422, 44405-44420.
 """
 
 from __future__ import annotations
 
 from cleveragents.acms import uko as _uko
+from cleveragents.acms.index import (
+    ACMSIndex,
+    FileTraversalEngine,
+    FileType,
+    IndexEntry,
+    TierLevel,
+)
 from cleveragents.acms.uko import (
     CODE_DETAIL_LEVEL_MAP,
     FUNC_DETAIL_LEVEL_MAP,
@@ -62,6 +72,14 @@ from cleveragents.acms.uko import (
     resolve_detail_level,
 )
 
-# Re-export everything published by the ``uko`` sub-package so the two
-# ``__all__`` lists stay in sync automatically.
-__all__: list[str] = list(_uko.__all__)
+# Combine exports from both uko and index modules
+_uko_exports = list(_uko.__all__)
+_index_exports = [
+    "ACMSIndex",
+    "FileTraversalEngine",
+    "FileType",
+    "IndexEntry",
+    "TierLevel",
+]
+
+__all__: list[str] = _uko_exports + _index_exports

src/cleveragents/acms/index.py

@@ -0,0 +1,412 @@
+"""ACMS Index Data Model and File Traversal Engine.
+
+Provides the foundational data model for indexed context entries and a
+file traversal engine that can handle 10,000+ files without timeout using
+chunked processing.
+
+Based on issue #9579 and ``docs/specification.md`` ~lines 44405-44420.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Iterator
+from datetime import datetime
+from enum import StrEnum
+from pathlib import Path
+
+from pydantic import BaseModel, Field
+
+
+class FileType(StrEnum):
+    """File type enumeration for index entries."""
+
+    PYTHON = "python"
+    JAVASCRIPT = "javascript"
+    TYPESCRIPT = "typescript"
+    JAVA = "java"
+    RUST = "rust"
+    MARKDOWN = "markdown"
+    TEXT = "text"
+    JSON = "json"
+    YAML = "yaml"
+    XML = "xml"
+    OTHER = "other"
+
+
+class TierLevel(StrEnum):
+    """Tier assignment levels for context entries.
+
+    Aligns with the hot/warm/cold storage tier vocabulary used throughout
+    the ACMS specification and milestone v3.4.0 acceptance criteria.
+    """
+
+    HOT = "hot"  # Core/essential
+    WARM = "warm"  # Important
+    COLD = "cold"  # Supporting
+    ARCHIVE = "archive"  # Reference
+
+
+class IndexEntry(BaseModel):
+    """Represents a single indexed context entry.
+
+    Attributes:
+        path: Absolute or relative file path
+        file_type: Type of file (from FileType enum)
+        size_bytes: File size in bytes
+        created_at: File creation timestamp
+        modified_at: File modification timestamp
+        tags: Set of tags for categorization
+        tier: Tier assignment level
+        metadata: Additional metadata dictionary
+    """
+
+    path: str
+    file_type: FileType
+    size_bytes: int
+    created_at: datetime
+    modified_at: datetime
+    tags: set[str] = Field(default_factory=set)
+    tier: TierLevel = TierLevel.COLD
+    metadata: dict[str, str] = Field(default_factory=dict)
+
+    def add_tag(self, tag: str) -> None:
+        """Add a tag to this entry.
+
+        Args:
+            tag: Tag string to add. Must be non-empty.
+
+        Raises:
+            ValueError: If tag is empty or whitespace-only.
+        """
+        if not tag or not tag.strip():
+            raise ValueError("tag must be a non-empty string")
+        self.tags.add(tag)
+
+    def remove_tag(self, tag: str) -> None:
+        """Remove a tag from this entry.
+
+        Args:
+            tag: Tag string to remove. Must be non-empty.
+
+        Raises:
+            ValueError: If tag is empty or whitespace-only.
+        """
+        if not tag or not tag.strip():
+            raise ValueError("tag must be a non-empty string")
+        self.tags.discard(tag)
+
+    def has_tag(self, tag: str) -> bool:
+        """Check if entry has a specific tag."""
+        return tag in self.tags
+
+    def set_tier(self, tier: TierLevel) -> None:
+        """Set the tier level for this entry."""
+        self.tier = tier
+
+
+class ACMSIndex:
+    """ACMS Index for storing and querying indexed context entries.
+
+    Provides storage and query capabilities for indexed files with support
+    for filtering by path, tags, type, and recency.
+
+    Attributes:
+        entries: Dictionary mapping file paths to IndexEntry objects
+    """
+
+    def __init__(self) -> None:
+        self.entries: dict[str, IndexEntry] = {}
+
+    def add_entry(self, entry: IndexEntry) -> None:
+        """Add an index entry to the index.
+
+        Args:
+            entry: The IndexEntry to add. Must not be None.
+
+        Raises:
+            TypeError: If entry is not an IndexEntry instance.
+        """
+        if not isinstance(entry, IndexEntry):
+            raise TypeError(
+                f"entry must be an IndexEntry instance, got {type(entry).__name__}"
+            )
+        self.entries[entry.path] = entry
+
+    def remove_entry(self, path: str) -> bool:
+        """Remove an entry by path. Returns True if removed, False if not found."""
+        if path in self.entries:
+            del self.entries[path]
+            return True
+        return False
+
+    def get_entry(self, path: str) -> IndexEntry | None:
+        """Get an entry by path."""
+        return self.entries.get(path)
+
+    def query_by_path(self, path_pattern: str) -> list[IndexEntry]:
+        """Query entries by path pattern (substring match).
+
+        Args:
+            path_pattern: Substring pattern to match against entry paths.
+                Must be non-empty.
+
+        Raises:
+            ValueError: If path_pattern is empty or whitespace-only.
+        """
+        if not path_pattern or not path_pattern.strip():
+            raise ValueError("path_pattern must be a non-empty string")
+        return [entry for entry in self.entries.values() if path_pattern in entry.path]
+
+    def query_by_tag(self, tag: str) -> list[IndexEntry]:
+        """Query entries by tag."""
+        return [entry for entry in self.entries.values() if entry.has_tag(tag)]
+
+    def query_by_type(self, file_type: FileType) -> list[IndexEntry]:
+        """Query entries by file type."""
+        return [
+            entry for entry in self.entries.values() if entry.file_type == file_type
+        ]
+
+    def query_by_tier(self, tier: TierLevel) -> list[IndexEntry]:
+        """Query entries by tier level."""
+        return [entry for entry in self.entries.values() if entry.tier == tier]
+
+    def query_by_recency(
+        self, after: datetime, before: datetime | None = None
+    ) -> list[IndexEntry]:
+        """Query entries by modification recency.
+
+        Args:
+            after: Return entries modified after this datetime.
+            before: Return entries modified before this datetime (optional).
+                When both are provided, before must be >= after.
+
+        Returns:
+            List of entries matching the recency criteria.
+
+        Raises:
+            ValueError: If before is earlier than after when both are provided.
+        """
+        if before is not None and before < after:
+            raise ValueError(
+                f"before ({before}) must be >= after ({after}) when both are provided"
+            )
+        results = [
+            entry for entry in self.entries.values() if entry.modified_at >= after
+        ]
+        if before:
+            results = [entry for entry in results if entry.modified_at <= before]
+        return results
+
+    def query_combined(
+        self,
+        path_pattern: str | None = None,
+        tags: set[str] | None = None,
+        file_type: FileType | None = None,
+        tier: TierLevel | None = None,
+        after: datetime | None = None,
+    ) -> list[IndexEntry]:
+        """Query entries with multiple filters (AND logic).
+
+        Args:
+            path_pattern: Filter by path pattern
+            tags: Filter by any of these tags
+            file_type: Filter by file type
+            tier: Filter by tier level
+            after: Filter by modification date
+
+        Returns:
+            List of entries matching all specified criteria
+        """
+        results = list(self.entries.values())
+
+        if path_pattern:
+            results = [entry for entry in results if path_pattern in entry.path]
+
+        if tags:
+            results = [
+                entry for entry in results if any(tag in entry.tags for tag in tags)
+            ]
+
+        if file_type:
+            results = [entry for entry in results if entry.file_type == file_type]
+
+        if tier:
+            results = [entry for entry in results if entry.tier == tier]
+
+        if after:
+            results = [entry for entry in results if entry.modified_at >= after]
+
+        return results
+
+    def get_all_entries(self) -> list[IndexEntry]:
+        """Get all entries in the index."""
+        return list(self.entries.values())
+
+    def get_entry_count(self) -> int:
+        """Get the total number of entries in the index."""
+        return len(self.entries)
+
+
+class FileTraversalEngine:
+    """Engine for traversing and indexing files in large projects.
+
+    Handles 10,000+ files without timeout using chunked processing to
+    prevent memory exhaustion.
+
+    Attributes:
+        chunk_size: Number of files to process in each chunk
+        index: The ACMS index to populate
+    """
+
+    def __init__(self, chunk_size: int = 100, index: ACMSIndex | None = None) -> None:
+        """Initialize the traversal engine.
+
+        Args:
+            chunk_size: Number of files to process per chunk (default: 100).
+                Must be a positive integer.
+            index: Optional pre-populated ACMSIndex to use. If None, a new
+                empty index is created (supports Dependency Inversion).
+
+        Raises:
+            ValueError: If chunk_size is not a positive integer.
+        """
+        if chunk_size <= 0:
+            raise ValueError(f"chunk_size must be a positive integer, got {chunk_size}")
+        self.chunk_size = chunk_size
+        self.index = index if index is not None else ACMSIndex()
+
+    def _get_file_type(self, file_path: Path) -> FileType:
+        """Determine file type from extension."""
+        suffix = file_path.suffix.lower()
+        type_map = {
+            ".py": FileType.PYTHON,
+            ".js": FileType.JAVASCRIPT,
+            ".ts": FileType.TYPESCRIPT,
+            ".tsx": FileType.TYPESCRIPT,
+            ".java": FileType.JAVA,
+            ".rs": FileType.RUST,
+            ".md": FileType.MARKDOWN,
+            ".txt": FileType.TEXT,
+            ".json": FileType.JSON,
+            ".yaml": FileType.YAML,
+            ".yml": FileType.YAML,
+            ".xml": FileType.XML,
+        }
+        return type_map.get(suffix, FileType.OTHER)
+
+    def _create_index_entry(self, file_path: Path) -> IndexEntry | None:
+        """Create an index entry from a file path.
+
+        Args:
+            file_path: Path to the file
+
+        Returns:
+            IndexEntry if successful, None if file cannot be read
+        """
+        try:
+            stat = file_path.stat()
+            return IndexEntry(
+                path=str(file_path),
+                file_type=self._get_file_type(file_path),
+                size_bytes=stat.st_size,
+                created_at=datetime.fromtimestamp(stat.st_ctime),
+                modified_at=datetime.fromtimestamp(stat.st_mtime),
+            )
+        except (OSError, ValueError):
+            return None
+
+    def _traverse_directory(
+        self,
+        root_path: Path,
+    ) -> Iterator[Path]:
+        """Recursively traverse directory and yield file paths.
+
+        Args:
+            root_path: Root directory to traverse
+
+        Yields:
+            Path objects for each file found
+        """
+        try:
+            for item in root_path.iterdir():
+                if item.is_file():
+                    yield item
+                elif item.is_dir():
+                    # Recursively traverse subdirectories
+                    yield from self._traverse_directory(item)
+        except (OSError, PermissionError):
+            # Skip directories we can't read
+            pass
+
+    def traverse_and_index(
+        self,
+        root_path: str | Path,
+        exclude_patterns: list[str] | None = None,
+    ) -> ACMSIndex:
+        """Traverse a directory and index all files.
+
+        Uses chunked processing to handle large projects without timeout
+        or memory exhaustion.
+
+        Args:
+            root_path: Root directory to traverse
+            exclude_patterns: List of patterns to exclude
+                (e.g., ['.git', '__pycache__'])
+
+        Returns:
+            Populated ACMSIndex
+        """
+        root = Path(root_path)
+        if not root.exists():
+            raise ValueError(f"Path does not exist: {root_path}")
+
+        exclude_patterns = exclude_patterns or []
+        chunk: list[IndexEntry] = []
+
+        for file_path in self._traverse_directory(root):
+            # Check if file matches any exclude pattern
+            if any(pattern in str(file_path) for pattern in exclude_patterns):
+                continue
+
+            # Create index entry
+            entry = self._create_index_entry(file_path)
+            if entry:
+                chunk.append(entry)
+
+            # Process chunk when it reaches the size limit
+            if len(chunk) >= self.chunk_size:
+                self._process_chunk(chunk)
+                chunk = []
+
+        # Process remaining entries
+        if chunk:
+            self._process_chunk(chunk)
+
+        return self.index
+
+    def _process_chunk(self, chunk: list[IndexEntry]) -> None:
+        """Process a chunk of index entries.
+
+        Args:
+            chunk: List of IndexEntry objects to add to the index
+        """
+        for entry in chunk:
+            self.index.add_entry(entry)
+
+    def get_index(self) -> ACMSIndex:
+        """Get the current index."""
+        return self.index
+
+    def reset_index(self) -> None:
+        """Reset the index to empty state."""
+        self.index = ACMSIndex()
+
+
+__all__ = [
+    "ACMSIndex",
+    "FileTraversalEngine",
+    "FileType",
+    "IndexEntry",
+    "TierLevel",
+]

src/cleveragents/actor/compiler.py

@@ -137,7 +137,7 @@ def _map_node(node: NodeDefinition) -> lg_nodes.NodeConfig:
             config.get("function") if node.type == NodeType.CONDITIONAL else None
         ),
         tools=config.get("tools", []) if node.type == NodeType.TOOL else [],
-        subgraph=(config.get("actor_ref") if node.type == NodeType.SUBGRAPH else None),
+        subgraph=(node.actor_ref if node.type == NodeType.SUBGRAPH else None),
         metadata=dict(config),
     )
 
@@ -187,6 +187,11 @@ def _detect_subgraph_cycles(
     """Detect cycles in subgraph references across actors.
 
     Returns list of actor names forming a cycle, or empty list.
+
+    The actor reference is read from the top-level actor_ref field on
+    NodeDefinition, not from node.config.  Reading from config
+    would always return an empty string because actor_ref is stored as a
+    first-class field on the schema model.
     """
     if resolver is None:
         return []
@@ -194,7 +199,7 @@ def _detect_subgraph_cycles(
     for node in route_nodes:
         if node.type != NodeType.SUBGRAPH:
             continue
-        ref_name = node.config.get("actor_ref", "")
+        ref_name = node.actor_ref or ""
         if not ref_name:
             continue
         if ref_name in visited:
@@ -290,7 +295,7 @@ def compile_actor(
         all_lsp_bindings.extend(bindings)
 
         if node_def.type == NodeType.SUBGRAPH:
-            ref = node_def.config.get("actor_ref", "")
+            ref = node_def.actor_ref or ""
             if ref:
                 subgraph_refs[node_def.id] = ref
 

src/cleveragents/cli/commands/acms.py

@@ -0,0 +1,466 @@
+"""ACMS context CLI commands.
+
+Implements ``agents acms context list`` and ``agents acms context add`` to
+interact with the Application Context Management System (ACMS) tier-based
+context pipeline.
+
+Commands:
+- ``agents acms context list``  - List all ACMS context fragments across
+  hot/warm/cold tiers for a project, showing fragment metadata and tier metrics.
+- ``agents acms context add``   - Add resource files or directories to the
+  ACMS context tier store (hydrate the ContextTierService).
+
+Based on the ACMS architecture (context_tier_hydrator, ContextTierService)
+and the hot/warm/cold tier fragment lifecycle.
+"""
+
+from __future__ import annotations
+
+import os
+import subprocess
+from datetime import UTC
+from pathlib import Path
+from typing import Annotated, Any
+
+import typer
+from rich.console import Console
+from rich.panel import Panel
+from rich.table import Table
+
+from cleveragents.cli.formatting import OutputFormat, format_output
+from cleveragents.domain.models.acms.tiers import ContextTier, TieredFragment
+
+app = typer.Typer(help="Manage ACMS context (tier-based fragment storage)")
+console = Console()
+err_console = Console(stderr=True)
+
+_FORMAT_HELP = "Output format: json, yaml, plain, table, or rich (default: rich)"
+
+_MAX_FILE_BYTES = 256 * 1024
+_MAX_TOTAL_BYTES = 10 * 1024 * 1024
+
+_SKIP_DIRS = frozenset(
+    {".git", ".hg", ".svn", "__pycache__", "node_modules",
+     ".venv", "venv", ".nox", ".tox", ".mypy_cache",
+     ".pytest_cache", ".ruff_cache", "dist", "build", ".eggs", ".cleveragents"}
+)
+
+_BINARY_EXTS = frozenset(
+    {".pyc", ".pyo", ".so", ".o", ".a", ".dll", ".exe",
+     ".png", ".jpg", ".jpeg", ".gif", ".bmp", ".ico", ".pdf",
+     ".zip", ".tar", ".gz", ".bz2", ".xz", ".whl", ".egg",
+     ".db", ".sqlite", ".sqlite3"}
+)
+
+
+def _format_size(size_bytes: int) -> str:
+    if size_bytes < 1024:
+        return f"{size_bytes} B"
+    elif size_bytes < 1024 * 1024:
+        return f"{size_bytes / 1024:.1f} KB"
+    else:
+        return f"{size_bytes / (1024 * 1024):.1f} MB"
+
+
+def _resolve_project_name(project_arg: str | None) -> Any:
+    from cleveragents.application.container import get_container
+    from cleveragents.application.services.project_service import ProjectService
+
+    container = get_container()
+    project_service: ProjectService = container.project_service()
+
+    if project_arg:
+        ns_repo = container.namespaced_project_repo()
+        try:
+            project = ns_repo.get(project_arg)
+            return project
+        except Exception as exc:
+            err_console.print(f"[red]Project not found:[/red] {project_arg}")
+            raise typer.Exit(1) from exc
+
+    project = project_service.get_current_project()
+    if not project:
+        err_console.print("[red]Error:[/red] No active project. Run 'cleveragents init' first.")
+        raise typer.Exit(1)
+    return project
+
+
+@app.command("list")
+def acms_context_list(
+    project: Annotated[
+        str | None,
+        typer.Option("--project", "-p", help="Project namespaced name (defaults to current project)"),
+    ] = None,
+    tier: Annotated[
+        ContextTier | None,
+        typer.Option("--tier", "-t", help="Filter by specific tier: hot, warm, or cold (default: all)"),
+    ] = None,
+    project_filter: Annotated[
+        list[str] | None,
+        typer.Option("--project-filter", "focus_projects", help="Project names to scope fragments from (repeatable)"),
+    ] = None,
+    fmt: Annotated[str, typer.Option("--format", "-f", help=_FORMAT_HELP)] = "rich",
+) -> None:
+    """List ACMS context fragments across hot/warm/cold tiers.
+
+    Displays all managed context fragments with their tier placement,
+    token counts, access frequency, and source resource information.
+
+    Examples::
+
+        # List all fragments for the current project
+        agents acms context list
+
+        # List only hot-tier fragments
+        agents acms context list --tier hot
+
+        # List fragments for a specific project by name
+        agents acms context list --project local/my-project
+
+        # Export as JSON for scripting
+        agents acms context list --format json
+    """
+    from cleveragents.application.container import get_container
+
+    try:
+        container = get_container()
+        tier_service = container.context_tier_service()
+
+        if project_filter:
+            fragments = tier_service.get_scoped_view(project_filter)
+        else:
+            proj = _resolve_project_name(project)
+            project_name = getattr(proj, "namespaced_name", str(proj))
+            fragments = tier_service.get_scoped_view([project_name])
+
+        if tier is not None:
+            fragments = [f for f in fragments if f.tier == tier]
+
+        metrics = tier_service.get_metrics()
+        budget = tier_service.budget
+
+        target_project = (
+            str(project_filter) if project_filter else getattr(proj, "namespaced_name", str(proj))
+        )
+        result_data: dict[str, Any] = {
+            "project": target_project,
+            "tier_filter": tier.value if tier else "all",
+            "metrics": {
+                "hot_count": metrics.hot_count,
+                "warm_count": metrics.warm_count,
+                "cold_count": metrics.cold_count,
+                "total_fragments": metrics.total_fragments,
+                "hot_hit_rate": f"{metrics.hot_hit_rate:.2%}",
+                "budget": {
+                    "max_tokens_hot": budget.max_tokens_hot,
+                    "max_decisions_warm": budget.max_decisions_warm,
+                    "max_decisions_cold": budget.max_decisions_cold,
+                },
+            },
+            "fragments": [
+                {
+                    "fragment_id": f.fragment_id,
+                    "tier": f.tier.value,
+                    "resource_id": f.resource_id or "",
+                    "token_count": f.token_count,
+                    "access_count": f.access_count,
+                    "last_accessed": f.last_accessed.isoformat(),
+                    "metadata": f.metadata,
+                }
+                for f in fragments
+            ],
+        }
+
+        if fmt.lower() == OutputFormat.RICH:
+            _render_list_rich(fragments, metrics, budget, tier)
+        elif fmt.lower() == "json":
+            import json as _json_mod
+            console.print(_json_mod.dumps(result_data, indent=2))
+        else:
+            console.print(format_output(result_data, fmt))
+
+    except Exception as exc:
+        err_console.print(f"[red]Error listing ACMS context:[/red] {exc}")
+        raise typer.Exit(1) from exc
+
+
+def _render_list_rich(
+    fragments: list[TieredFragment],
+    metrics: "TierMetrics",  # noqa: F821
+    budget: "TierBudget",     # noqa: F821
+    tier_filter: ContextTier | None,
+) -> None:
+    """Render the ACMS context list using rich terminal output."""
+    tier_label = tier_filter.value if tier_filter else "all"
+
+    summary_lines: list[str] = []
+    summary_lines.append(f"[bold]Total fragments:[/bold] {metrics.total_fragments}")
+    summary_lines.append(
+        f"[bold]Hot:[/bold] {metrics.hot_count}  "
+        f"[bold]Warm:[/bold] {metrics.warm_count}  "
+        f"[bold]Cold:[/bold] {metrics.cold_count}"
+    )
+    summary_lines.append(f"[bold]Budget (hot tokens):[/bold] {budget.max_tokens_hot:,}")
+    summary_lines.append(
+        f"[bold]Hot hit rate:[/bold] "
+        f"{metrics.hot_hit_rate:.2%} ({metrics.hot_hit_count} hits / "
+        f"{metrics.hot_hit_count + metrics.hot_miss_count} accesses)"
+    )
+
+    total_tokens_in_fragments = sum(f.token_count for f in fragments)
+    pct_of_budget = (
+        (total_tokens_in_fragments / budget.max_tokens_hot * 100) if budget.max_tokens_hot > 0 else 0
+    )
+    summary_lines.append(f"[bold]Tokens in view:[/bold] {total_tokens_in_fragments:,} ({pct_of_budget:.1%} of hot budget)")
+
+    console.print(Panel("\n".join(summary_lines), title=f"ACMS Context Fragments ({tier_label} tier)", expand=False))
+
+    if not fragments:
+        console.print("[yellow]No context fragments found.[/yellow]")
+        console.print("Use 'agents acms context add <path>' to add files.")
+        return
+
+    table = Table(title=f"Fragments ({len(fragments)})")
+    table.add_column("#", style="dim", justify="right", width=4)
+    table.add_column("Tier", width=6)
+    table.add_column("Resource / ID", overflow="fold")
+    table.add_column("Tokens", justify="right", width=12)
+    table.add_column("Accesses", justify="right", width=8)
+    table.add_column("Last Accessed", width=20)
+
+    tier_order = {ContextTier.HOT: 0, ContextTier.WARM: 1, ContextTier.COLD: 2}
+    sorted_frags = sorted(fragments, key=lambda f: (tier_order.get(f.tier), -f.access_count))
+
+    for idx, frag in enumerate(sorted_frags, 1):
+        tier_style = {"hot": "green", "warm": "yellow", "cold": "blue"}[frag.tier.value]
+        short_id = (frag.fragment_id[:50] + "..." if len(frag.fragment_id) > 50 else frag.fragment_id)
+        table.add_row(
+            str(idx),
+            f"[{tier_style}]{frag.tier.value}[/{tier_style}]",
+            short_id,
+            f"{frag.token_count:,}",
+            str(frag.access_count),
+            frag.last_accessed.strftime("%Y-%m-%d %H:%M"),
+        )
+
+    console.print(table)
+
+
+@app.command("add")
+def acms_context_add(
+    paths: Annotated[list[str], typer.Argument(help="Paths (files or directories) to add to ACMS context")],
+    recursive: Annotated[bool, typer.Option("-r", "--recursive", help="Add directories recursively")] = True,
+    tier: Annotated[ContextTier, typer.Option("--tier", "-t", help="Target tier (default: hot)")] = ContextTier.HOT,
+    project: Annotated[str | None, typer.Option("--project", "-p", help="Project namespaced name for scoping")] = None,
+    fmt: Annotated[str, typer.Option("--format", "-f", help=_FORMAT_HELP)] = "rich",
+) -> None:
+    """Add resource files to the ACMS context tier store.
+
+    Reads files from disk and stores them as TieredFragment entries in
+    the ContextTierService under the specified project scope and target
+    tier. Binary and large files are automatically skipped.
+
+    Examples::
+
+        # Add current directory to hot tier (current project)
+        agents acms context add .
+
+        # Add specific files/dirs with recursive traversal
+        agents acms context add src/ docs/
+
+        # Add to warm tier for a named project
+        agents acms context add ./project-files --tier warm -p local/my-project
+
+        # Non-recursive (single file only)
+        agents acms context add README.md -r false
+    """
+    from cleveragents.application.container import get_container
+
+    try:
+        container = get_container()
+        tier_service = container.context_tier_service()
+
+        proj = _resolve_project_name(project)
+        project_name = getattr(proj, "namespaced_name", str(proj))
+
+        from cleveragents.application.services.resource_registry_service import ResourceRegistryService
+
+        resource_reg: ResourceRegistryService = container.resource_registry()
+        resources_for_project = []
+        try:
+            if hasattr(proj, "project_id"):
+                resources_for_project = resource_reg.list_resources(project=proj.project_id)
+        except Exception:
+            pass
+
+        if resources_for_project and len(resources_for_project) > 0:
+            res_obj = resources_for_project[0]
+            resource_id = str(res_obj.resource_id)
+            resource_location = getattr(res_obj, "location", None) or getattr(proj, "location", None) or "."
+        else:
+            resource_id = f"adhoc/{project_name}"
+            resource_location = str(Path.cwd())
+
+        added_frags: list[TieredFragment] = []
+        already_in_context: list[str] = []
+        skipped_files: list[str] = []
+        total_bytes = 0
+
+        for path_str in paths:
+            path = Path(path_str).resolve()
+
+            if not path.exists():
+                err_console.print(f"[red]Path does not exist:[/red] {path}")
+                continue
+
+            files_to_add: list[Path] = []
+            if path.is_file():
+                files_to_add = [path]
+            elif path.is_dir() and recursive:
+                for dirpath, dirnames, filenames in os.walk(path):
+                    dirnames[:] = [d for d in dirnames if d not in _SKIP_DIRS and not d.startswith(".")]
+                    for fname in filenames:
+                        if fname.startswith("."):
+                            continue
+                        ext = os.path.splitext(fname)[1].lower()
+                        if ext in _BINARY_EXTS:
+                            skipped_files.append(str(Path(dirpath) / fname))
+                            continue
+                        files_to_add.append(Path(dirpath) / fname)
+            elif path.is_dir() and not recursive:
+                raise typer.BadParameter(f"Directory requires --recursive flag. Use 'add <path>' for individual files only.")
+
+            for file_path in files_to_add:
+                try:
+                    size = file_path.stat().st_size
+                except OSError:
+                    continue
+
+                if size > _MAX_FILE_BYTES:
+                    skipped_files.append(str(file_path))
+                    continue
+                if total_bytes + size > _MAX_TOTAL_BYTES:
+                    break
+
+                content = ""
+                try:
+                    content = file_path.read_text(encoding="utf-8")
+                except (UnicodeDecodeError, OSError):
+                    skipped_files.append(str(file_path))
+                    continue
+
+                try:
+                    res_root = Path(resource_location) if isinstance(resource_location, str) else resource_location
+                    rel_path = str(file_path.relative_to(res_root))
+                except (ValueError, TypeError):
+                    rel_path = str(file_path)
+
+                fragment_id = f"{resource_id}:{rel_path}"
+
+                existing = tier_service._find_fragment(fragment_id)
+                if existing is not None:
+                    already_in_context.append(rel_path)
+                    continue
+
+                fragment = TieredFragment(
+                    fragment_id=fragment_id,
+                    content=content,
+                    tier=tier,
+                    resource_id=resource_id,
+                    project_name=project_name,
+                    token_count=len(content) // 4,
+                    metadata={
+                        "path": rel_path,
+                        "detail_depth": "1",
+                        "relevance_score": "0.5",
+                        "source": str(file_path),
+                    },
+                )
+
+                tier_service.store(fragment)
+                added_frags.append(fragment)
+                total_bytes += size
+
+        result_data: dict[str, Any] = {
+            "project": project_name,
+            "resource_id": resource_id,
+            "tier": tier.value,
+            "added_count": len(added_frags),
+            "already_exists_count": len(already_in_context),
+            "skipped_count": len(skipped_files),
+            "total_bytes": total_bytes,
+            "added_files": [f.fragment_id for f in added_frags],
+            "already_in_context": already_in_context,
+            "skipped_files": skipped_files,
+        }
+
+        if fmt.lower() == OutputFormat.RICH:
+            messages: list[tuple[str, str]] = []
+            if added_frags:
+                shown = min(len(added_frags), 10)
+                file_list = "\n".join(f"  {f.fragment_id}" for f in added_frags[:shown])
+                extra = "" if len(added_frags) <= 10 else f"\n  ... and {len(added_frags) - 10} more"
+                messages.append(("Added", f"[green]Added [bold]{len(added_frags)}[/bold] file(s) to [{tier.value}] tier ({_format_size(total_bytes)})\n{file_list}{extra}"))
+            if already_in_context:
+                shown = min(len(already_in_context), 10)
+                messages.append(("Already in context", f"[yellow]{len(already_in_context)}[/yellow] file(s) already stored:\n" + "\n".join(f"  - {f}" for f in already_in_context[:shown]) + (f"\n  ... and {len(already_in_context) - 10} more" if len(already_in_context) > 10 else "")))
+            if skipped_files:
+                messages.append(("Skipped", f"[dim]{len(skipped_files)} file(s) skipped (binary or oversized)[/dim]"))
+            for title, body in messages:
+                console.print(Panel(body, title=title, expand=False))
+        else:
+            console.print(format_output(result_data, fmt))
+
+    except Exception as exc:
+        err_console.print(f"[red]Error adding to ACMS context:[/red] {exc}")
+        raise typer.Exit(1) from exc
+
+
+@app.command("reset")
+def acms_context_reset(
+    project: Annotated[str | None, typer.Option("--project", "-p", help="Project namespaced name (defaults to current project)")] = None,
+    yes: Annotated[bool, typer.Option("--yes", "-y", help="Skip confirmation prompt")] = False,
+    fmt: Annotated[str, typer.Option("--format", "-f", help=_FORMAT_HELP)] = "rich",
+) -> None:
+    """Reset (clear) all ACMS context fragments for a project."""
+    from cleveragents.application.container import get_container
+    from cleveragents.domain.models.acms.tiers import ContextTier
+
+    try:
+        container = get_container()
+        tier_service = container.context_tier_service()
+
+        proj = _resolve_project_name(project)
+        project_name = getattr(proj, "namespaced_name", str(proj))
+
+        all_frags = tier_service.get_scoped_view([project_name])
+        fragment_ids_to_remove = [f.fragment_id for f in all_frags]
+
+        if not fragment_ids_to_remove:
+            console.print("[yellow]No fragments to reset.[/yellow]")
+            return
+
+        if not yes:
+            typer.echo(f"Found {len(fragment_ids_to_remove)} fragment(s) for project '{project_name}'.")
+            if not typer.confirm("Reset (clear) all fragments?"):
+                typer.echo("Reset cancelled.")
+                return
+
+        stores = {ContextTier.HOT: tier_service._hot, ContextTier.WARM: tier_service._warm, ContextTier.COLD: tier_service._cold}
+        removed_count = 0
+        for store in stores.values():
+            to_remove = [fid for fid, frag in store.items() if frag.project_name == project_name]
+            for fid in to_remove:
+                del store[fid]
+                removed_count += 1
+
+        console.print(Panel(
+            f"[green]Reset ACMS context for project '[bold]{project_name}[/bold]'\n"
+            f"[bold]Removed:[/bold] {removed_count} fragment(s)",
+            title="ACMS Context Reset",
+            expand=False,
+        ))
+
+    except Exception as exc:
+        err_console.print(f"[red]Error resetting ACMS context:[/red] {exc}")
+        raise typer.Exit(1) from exc

src/cleveragents/cli/commands/actor.py

@@ -816,12 +816,28 @@ def update(
 
 
 @app.command()
-def remove(name: Annotated[str, typer.Argument(help="Actor name to remove")]) -> None:
+def remove(
+    name: Annotated[str, typer.Argument(help="Actor name to remove")],
+    fmt: Annotated[
+        str,
+        typer.Option("--format", "-f", help=_FORMAT_HELP),
+    ] = OutputFormat.RICH.value,
+) -> None:
     """Remove a custom actor.
 
     Specify the namespaced name (e.g. ``local/my-actor``).
     """
 
+    # Validate --format argument first; fail fast before any side effects.
+    fmt_value = fmt.lower()
+    _valid_formats = {f.value for f in OutputFormat}
+    if fmt_value not in _valid_formats:
+        raise typer.BadParameter(
+            f"Invalid format {fmt!r}. "
+            f"Supported values: {', '.join(sorted(_valid_formats))}",
+            param_hint="'--format'",
+        )
+
     service, registry = _get_services()
     try:
         # Get actor details before removal for display
@@ -844,6 +860,40 @@ def remove(name: Annotated[str, typer.Argument(help="Actor name to remove")]) ->
         else:
             service.remove_actor(name)
 
+        command_name = f"agents actor remove {name}"
+        payload = {
+            "actor_removed": {
+                "name": name,
+                "provider": actor_provider,
+                "model": actor_model,
+            },
+            "impact": {
+                "sessions": session_count,
+                "active_plans": active_plan_count,
+                "actions_referencing": action_count,
+            },
+            "cleanup": {
+                "config": "kept on disk",
+                # NOTE: context-cleanup count is deferred; always 0 for now.
+                # Follow-up: implement dynamic orphaned-context detection.
+                "contexts": "0 orphaned",
+            },
+        }
+        messages = [{"level": "ok", "text": "Actor removed"}]
+
+        if fmt_value != OutputFormat.RICH.value:
+            rendered = format_output(
+                payload,
+                fmt_value,
+                command=command_name,
+                status="ok",
+                exit_code=0,
+                messages=messages,
+            )
+            if rendered:
+                console.print(rendered)
+            return
+
         # Display Actor Removed panel
         actor_info = (
             f"[cyan bold]Name:[/cyan bold] {name}\n"

src/cleveragents/cli/commands/session.py

@@ -151,8 +151,8 @@ def _session_list_dict(sessions: list[Session]) -> dict[str, Any]:
     # Find most recent and oldest sessions
     if sessions:
         sorted_sessions = sorted(sessions, key=lambda x: x.updated_at, reverse=True)
-        most_recent = sorted_sessions[0].name or sorted_sessions[0].session_id[:8]
-        oldest = sorted_sessions[-1].name or sorted_sessions[-1].session_id[:8]
+        most_recent = sorted_sessions[0].name or sorted_sessions[0].session_id
+        oldest = sorted_sessions[-1].name or sorted_sessions[-1].session_id
     else:
         most_recent = None
         oldest = None
@@ -347,7 +347,7 @@ def list_sessions(
 
     for s in sessions:
         table.add_row(
-            s.session_id[:8],  # Truncate ID for readability
+            s.session_id,  # Full ULID for copy-paste compatibility with session tell
             s.name or "(unnamed)",
             s.actor_name or "(none)",
             str(s.message_count),

src/cleveragents/cli/main.py

@@ -98,6 +98,7 @@ def _register_subcommands() -> None:
             tui,
             validation,
         )
+        from cleveragents.cli.commands.acms import app as acms_app
         from cleveragents.cli.commands.auto_debug import app as auto_debug_app
         from cleveragents.cli.commands.db import app as db_app
         from cleveragents.cli.commands.repl import _repl_app
@@ -228,6 +229,11 @@ def _register_subcommands() -> None:
         name="repo",
         help="Repository indexing management",
     )
+    app.add_typer(
+        acms_app,
+        name="acms",
+        help="Application Context Management System (tier-based context)",
+    )
 
     _subcommands_registered = True
 
@@ -684,6 +690,7 @@ def main(args: list[str] | None = None) -> int:
             "tui",  # Textual TUI
             "server",  # Server connection management
             "repo",  # Repository indexing management
+            "acms",  # ACMS - Application Context Management System
             "apply",  # Shortcut for plan apply
             "context-load",  # Shortcut for context add
             "context-add",  # Shortcut
@@ -716,6 +723,7 @@ def main(args: list[str] | None = None) -> int:
                 "apply",
                 "context-load",
                 "context-add",
+                "acms",
             }
         )
         if hasattr(app, "add_typer") and not (

src/cleveragents/infrastructure/database/llm_trace_repository.py

@@ -31,6 +31,17 @@ class LLMTraceRepository:
 
     Uses the session-factory pattern: each public method obtains a
     session from the factory.  Callers are responsible for commit.
+
+    When ``save()`` is called with an explicit ``session`` argument the
+    repository operates in *UnitOfWork mode*: it flushes the change into
+    the caller's transaction but does **not** commit or close the session.
+    The caller (or the enclosing ``UnitOfWork``) is responsible for the
+    final commit.
+
+    When ``save()`` is called without an explicit ``session`` argument the
+    repository operates in *standalone mode*: it creates its own session
+    from the factory, flushes, commits, and closes the session so that the
+    trace is durably persisted even outside a ``UnitOfWork``.
     """
 
     def __init__(
@@ -46,16 +57,26 @@ class LLMTraceRepository:
         return self._sf()
 
     @database_retry
-    def save(self, trace: LLMTrace) -> None:
+    def save(self, trace: LLMTrace, session: Session | None = None) -> None:
         """Persist a single ``LLMTrace`` row.
 
         Args:
-            trace: The trace to persist.
+            trace: The trace to persist.  Must not be ``None``.
+            session: Optional external SQLAlchemy session.  When provided
+                the repository flushes into the caller's transaction and
+                does **not** commit or close the session (UnitOfWork mode).
+                When omitted the repository creates its own session, commits,
+                and closes it (standalone mode).
 
         Raises:
+            ValueError: If ``trace`` is ``None``.
             DatabaseError: On unrecoverable persistence failure.
         """
-        session = self._session()
+        if trace is None:
+            raise ValueError("trace must not be None")
+
+        own_session = session is None
+        s: Session = self._session() if own_session else session
         try:
             model = LLMTraceModel(
                 trace_id=trace.trace_id,
@@ -77,11 +98,16 @@ class LLMTraceRepository:
                 error=trace.error,
                 timestamp=trace.timestamp.isoformat(),
             )
-            session.add(model)
-            session.flush()
+            s.add(model)
+            s.flush()
+            if own_session:
+                s.commit()
         except (SQLAlchemyDatabaseError, OperationalError) as exc:
-            session.rollback()
+            s.rollback()
             raise DatabaseError(f"Failed to save LLM trace: {exc}") from exc
+        finally:
+            if own_session:
+                s.close()
 
     @database_retry
     def get(self, trace_id: str) -> LLMTrace | None:

src/cleveragents/infrastructure/database/migration_runner.py

@@ -151,10 +151,22 @@ class MigrationRunner:
     def get_current_revision(self) -> str | None:
         """Get the current migration revision of the database.
 
+        For SQLite databases, the engine is created with
+        ``connect_args={"check_same_thread": False}`` so that this method
+        can be safely called from any thread — including background threads
+        used in async startup flows.  This is consistent with the pattern
+        used in :meth:`init_or_upgrade`.
+
         Returns:
             Current revision ID or None if no migrations have been applied
         """
-        engine = create_engine(self.database_url)
+        if self.database_url.startswith("sqlite"):
+            engine = create_engine(
+                self.database_url,
+                connect_args={"check_same_thread": False},
+            )
+        else:
+            engine = create_engine(self.database_url)
         with engine.connect() as connection:
             context = MigrationContext.configure(connection)
             return context.get_current_revision()

src/cleveragents/infrastructure/database/migrations/versions/m9_003_plan_result_success_column.py

@@ -0,0 +1,51 @@
+"""Add result_success column to plans table.
+
+This migration adds a dedicated ``result_success`` boolean column to the
+``plans`` table to accurately track the final result status of a plan's
+apply phase.
+
+Previously, ``PlanRepository._to_domain`` derived ``PlanResult.success``
+from ``error_message is None``.  Because ``error_message`` is shared
+between the build phase and the result phase, a plan that encountered a
+build error but later succeeded in the apply phase would be incorrectly
+reconstructed as failed.
+
+The new ``result_success`` column is nullable to preserve backward
+compatibility with existing database records.  When ``result_success``
+is NULL (pre-migration records), the repository falls back to the legacy
+``error_message is None`` heuristic.
+
+Revision ID: m9_003_plan_result_success_column
+Revises: m10_001_virtual_builtin_actors
+Create Date: 2026-05-05 00:00:00
+
+"""
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = "m9_003_plan_result_success_column"
+down_revision: str | Sequence[str] | None = "m10_001_virtual_builtin_actors"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    """Add ``result_success`` column to the ``plans`` table.
+
+    The column is nullable so that existing rows (which have no explicit
+    result-phase success signal) are not affected.  New rows written by
+    the updated repository will always populate this column.
+    """
+    op.add_column(
+        "plans",
+        sa.Column("result_success", sa.Boolean(), nullable=True),
+    )
+
+
+def downgrade() -> None:
+    """Remove ``result_success`` column from the ``plans`` table."""
+    op.drop_column("plans", "result_success")

src/cleveragents/infrastructure/database/models.py

@@ -142,6 +142,7 @@ class PlanModel(Base):
     files_created = Column(Integer, nullable=True, default=0)
     files_modified = Column(Integer, nullable=True, default=0)
     files_deleted = Column(Integer, nullable=True, default=0)
+    result_success = Column(Boolean, nullable=True)
 
     # Relationships
     project = relationship("ProjectModel", back_populates="plans")

src/cleveragents/infrastructure/database/repositories.py

@@ -294,6 +294,7 @@ class PlanRepository:
             files_modified=plan.files_modified,
             files_deleted=plan.files_deleted,
             error_message=plan.build.error_message if plan.build else None,
+            result_success=plan.result.success if plan.result else None,
         )
 
         self.session.add(db_plan)
@@ -372,6 +373,7 @@ class PlanRepository:
                 db_plan.files_created = plan.result.files_created  # type: ignore
                 db_plan.files_modified = plan.result.files_modified  # type: ignore
                 db_plan.files_deleted = plan.result.files_deleted  # type: ignore
+                db_plan.result_success = plan.result.success  # type: ignore
 
             self.session.flush()
 
@@ -420,8 +422,20 @@ class PlanRepository:
 
         result = None
         if db_plan.applied_at:  # type: ignore
+            # Derive success from the dedicated result_success column when
+            # available.  For legacy records where result_success is NULL
+            # (written before migration m9_003), fall back to the old
+            # heuristic of checking whether error_message is None.
+            result_success_col = getattr(db_plan, "result_success", None)
+            if result_success_col is True:
+                plan_success = True
+            elif result_success_col is False:
+                plan_success = False
+            else:
+                # NULL — pre-migration record; use legacy heuristic
+                plan_success = db_plan.error_message is None  # type: ignore
             result = PlanResult(
-                success=db_plan.error_message is None,  # type: ignore
+                success=plan_success,
                 files_created=db_plan.files_created or 0,  # type: ignore
                 files_modified=db_plan.files_modified or 0,  # type: ignore
                 files_deleted=db_plan.files_deleted or 0,  # type: ignore