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.

CHANGELOG.md

@@ -27,6 +27,14 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
   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):
@@ -39,6 +47,8 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
 
 ### 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:
@@ -111,6 +121,15 @@ 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

CONTRIBUTORS.md

@@ -24,10 +24,13 @@ Below are some of the specific details of various contributions.
 * This project was made possible thanks to considerable donation of time, money, and resources by CleverThis, Inc.
 * HAL 9000 has contributed automated bug fixes, CLI output formatting improvements, and ongoing maintenance as part of the CleverAgents automation system.
 * HAL 9000 has contributed the file edit encoding parameter fix (PR #8258 / issue #7559).
-<<* 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 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/specification.md

@@ -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

@@ -9,6 +9,7 @@ Feature: ACMS Index Data Model and File Traversal Engine
 
   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 |
@@ -116,7 +117,7 @@ Feature: ACMS Index Data Model and File Traversal Engine
   Scenario: Get entry count from index
     Given I have an index with 10 entries
     When I get the entry count
-    Then the count should be 10
+    Then the ACMS entry count should be 10
 
   Scenario: Remove entry from index
     Given I have an index with 3 entries
@@ -131,6 +132,7 @@ Feature: ACMS Index Data Model and File Traversal Engine
       | /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 |

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/steps/acms_index_data_model_traversal_steps.py

@@ -361,7 +361,7 @@ def step_get_entry_count(context):
     context.entry_count = context.index.get_entry_count()
 
 
-@then("the count should be {count:d}")
+@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

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/pr_compliance_checklist_steps.py

@@ -5,7 +5,7 @@ from typing import Any
 
 from behave import given, then, when
 
-PROJECT_ROOT = Path(__file__).resolve().parents[3]
+PROJECT_ROOT = Path(__file__).resolve().parents[2]
 AGENT_DEF_PATH = PROJECT_ROOT / ".opencode" / "agents" / "implementation-supervisor.md"
 
 

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_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/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/index.py

@@ -10,11 +10,12 @@ Based on issue #9579 and ``docs/specification.md`` ~lines 44405-44420.
 from __future__ import annotations
 
 from collections.abc import Iterator
-from dataclasses import dataclass, field
 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."""
@@ -45,8 +46,7 @@ class TierLevel(StrEnum):
     ARCHIVE = "archive"  # Reference
 
 
-@dataclass
-class IndexEntry:
+class IndexEntry(BaseModel):
     """Represents a single indexed context entry.
 
     Attributes:
@@ -65,9 +65,9 @@ class IndexEntry:
     size_bytes: int
     created_at: datetime
     modified_at: datetime
-    tags: set[str] = field(default_factory=set)
+    tags: set[str] = Field(default_factory=set)
     tier: TierLevel = TierLevel.COLD
-    metadata: dict[str, str] = field(default_factory=dict)
+    metadata: dict[str, str] = Field(default_factory=dict)
 
     def add_tag(self, tag: str) -> None:
         """Add a tag to this entry.
@@ -104,7 +104,6 @@ class IndexEntry:
         self.tier = tier
 
 
-@dataclass
 class ACMSIndex:
     """ACMS Index for storing and querying indexed context entries.
 
@@ -115,7 +114,8 @@ class ACMSIndex:
         entries: Dictionary mapping file paths to IndexEntry objects
     """
 
-    entries: dict[str, IndexEntry] = field(default_factory=dict)
+    def __init__(self) -> None:
+        self.entries: dict[str, IndexEntry] = {}
 
     def add_entry(self, entry: IndexEntry) -> None:
         """Add an index entry to the index.

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/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/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