fix(resources): remove unsupported executable resource type

The executable resource type was not defined in the specification (docs/specification.md lines 10567-10575). The supported resource types are: git-checkout, git, fs-mount, fs-directory, fs-file, container-instance, and devcontainer-instance. Additionally, the agents resource list output included a Location column that is not part of the spec-defined columns (Name, ID, Type, Phys/Virt, Children, Projects per line 11051). Changes: - Remove executable type registration from _resource_registry_lsp.py - Remove executable from BUILTIN_TYPE_NAMES in _resource_type_validation.py - Update resource list table columns to match spec (Name, ID, Type, Phys/Virt, Children, Projects) removing Location and Description - Remove all executable-related test scenarios from features/resource_type_lsp.feature - Update robot/helper_resource_type_lsp.py to reflect 3 LSP types (not 4) and add assertions that executable is absent ISSUES CLOSED: #3077
2026-05-07 10:53:53 +00:00
34 changed files with 259 additions and 1629 deletions
@@ -5,17 +5,6 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).

 ## [Unreleased]

-### Added
-
- **CLI — `config get` spec-compliant output** (#3423): The `agents config get` command now
-  renders the full specification-required output including three Rich panels (*Config*, *Origin*,
-  *Resolution Chain*) with Overridden indicator, Winner line, and Origin details (File, Line,
-  Default). JSON output is wrapped in a standard envelope with `command`, `status`, `exit_code`,
-  `data` (key, value, source short-name, overridden, type spec-string, origin, resolution_chain,
-  winner), `timing` (started ISO-8601 + duration_ms), and messages. Added module-level helpers in
-  `_config_display.py` and `_config_helpers.py`, plus 18 new BDD scenarios in
-  `config_get_spec_output.feature`.
-
 ### Fixed

 - **Cross-actor subgraph cycle detection reads actor_ref field** (#1431): Fixed
@@ -38,14 +27,6 @@ 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):
@@ -58,8 +39,6 @@ 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:
@@ -132,15 +111,6 @@ 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
@@ -24,13 +24,10 @@ 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.
@@ -45871,8 +45871,6 @@ 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
@@ -9,7 +9,6 @@ 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 |
@@ -117,7 +116,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 ACMS entry count should be 10
+    Then the count should be 10

  Scenario: Remove entry from index
    Given I have an index with 3 entries
@@ -132,7 +131,6 @@ 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 |
@@ -99,23 +99,3 @@ 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"
@@ -78,14 +78,14 @@ Feature: Config CLI commands
  # --- RESOLUTION CHAIN ---
  Scenario: Resolution chain shows default source
    When I run config get key "core.log.level" formatted as "json"
-    Then the resolution chain should include "Default" source
+    Then the resolution chain should include "default" source

  # --- SET/GET ROUNDTRIP ---
  Scenario: Set and get roundtrip
    Given I have set config "core.log.file-enabled" to "true"
    When I run config get key "core.log.file-enabled" formatted as "json"
    Then the config get should succeed
-    And the get value source should be "config"
+    And the get value source should be "global"

  # --- FORMAT ---
  Scenario: Config list with JSON format
@@ -197,8 +197,8 @@ Feature: Config CLI safety-net coverage
  Scenario: safety-net config get rich format displays panel and chain
    Given a safety-net isolated temp config directory
    When the safety-net CLI gets key "core.log.level" with format "rich" and verbose
-    Then the safety-net get rich output should contain "Config"
-    And the safety-net get rich output should contain "Resolution Chain"
+    Then the safety-net get rich output should contain "Configuration Value"
+    And the safety-net get rich output should contain "Resolution chain"

  @tdd_issue @tdd_issue_4238 @tdd_expected_fail
  Scenario: safety-net config get yaml format produces valid YAML
@@ -285,4 +285,4 @@ Feature: Config CLI safety-net coverage
    Given a safety-net isolated temp config directory
    And a safety-net patched console for capturing rich output
    When the safety-net CLI gets key "core.log.level" with format "rich" and verbose
-    Then the safety-net captured console output should contain "Winner"
+    Then the safety-net captured console output should contain "active"
@@ -1,120 +0,0 @@
-Feature: agents config get spec-compliant output
-  As a developer
-  I want `agents config get` to render all spec-required panels and JSON fields
-  So that the output matches the specification exactly
-
-  Background:
-    Given a clean config get spec test environment
-
-  # --- Rich output: three panels ---
-
-  Scenario: Rich output shows Config panel with correct title
-    When I run spec config get "core.log.level"
-    Then the spec config get should succeed
-    And the spec config get rich output should contain panel "Config"
-
-  Scenario: Rich output shows Overridden field in Config panel
-    When I run spec config get "core.log.level"
-    Then the spec config get should succeed
-    And the spec config get rich output should contain "Overridden"
-
-  Scenario: Rich output shows Origin panel
-    When I run spec config get "core.log.level"
-    Then the spec config get should succeed
-    And the spec config get rich output should contain panel "Origin"
-
-  Scenario: Rich output shows Resolution Chain panel
-    When I run spec config get "core.log.level"
-    Then the spec config get should succeed
-    And the spec config get rich output should contain panel "Resolution Chain"
-
-  Scenario: Rich output shows Winner in Resolution Chain panel
-    When I run spec config get "core.log.level"
-    Then the spec config get should succeed
-    And the spec config get rich output should contain "Winner"
-
-  Scenario: Rich output uses spec type string not Python type name
-    When I run spec config get "core.log.level"
-    Then the spec config get should succeed
-    And the spec config get rich output should contain "string"
-    And the spec config get rich output should not contain "str"
-
-  Scenario: Rich output shows OK confirmation message
-    When I run spec config get "core.log.level"
-    Then the spec config get should succeed
-    And the spec config get rich output should contain "OK"
-
-  # --- JSON output: standard envelope ---
-
-  Scenario: JSON output is wrapped in standard envelope
-    When I run spec config get "core.log.level" with format "json"
-    Then the spec config get should succeed
-    And the spec config get JSON envelope should have field "command" equal to "config get"
-    And the spec config get JSON envelope should have field "status" equal to "ok"
-    And the spec config get JSON envelope should have field "exit_code" equal to 0
-    And the spec config get JSON envelope should have field "data"
-    And the spec config get JSON envelope should have field "timing"
-    And the spec config get JSON envelope should have field "messages"
-
-  Scenario: JSON data contains all required fields
-    When I run spec config get "core.log.level" with format "json"
-    Then the spec config get should succeed
-    And the spec config get JSON data should have field "key"
-    And the spec config get JSON data should have field "value"
-    And the spec config get JSON data should have field "source"
-    And the spec config get JSON data should have field "overridden"
-    And the spec config get JSON data should have field "type"
-    And the spec config get JSON data should have field "origin"
-    And the spec config get JSON data should have field "resolution_chain"
-    And the spec config get JSON data should have field "winner"
-
-  Scenario: JSON data type field uses spec type string
-    When I run spec config get "core.log.level" with format "json"
-    Then the spec config get should succeed
-    And the spec config get JSON data field "type" should equal "string"
-
-  Scenario: JSON data overridden field is false for default value
-    When I run spec config get "core.log.level" with format "json"
-    Then the spec config get should succeed
-    And the spec config get JSON data field "overridden" should be false
-
-  Scenario: JSON data overridden field is true when value is set
-    Given I have set spec config "core.log.level" to "DEBUG"
-    When I run spec config get "core.log.level" with format "json"
-    Then the spec config get should succeed
-    And the spec config get JSON data field "overridden" should be true
-
-  Scenario: JSON data origin contains default field
-    When I run spec config get "core.log.level" with format "json"
-    Then the spec config get should succeed
-    And the spec config get JSON origin should have field "default"
-
-  Scenario: JSON data resolution chain uses human-readable source names
-    When I run spec config get "core.log.level" with format "json"
-    Then the spec config get should succeed
-    And the spec config get JSON resolution chain should use human-readable source names
-
-  Scenario: JSON data winner contains source and level
-    When I run spec config get "core.log.level" with format "json"
-    Then the spec config get should succeed
-    And the spec config get JSON winner should have field "source"
-    And the spec config get JSON winner should have field "level"
-
-  Scenario: JSON timing contains started and duration_ms
-    When I run spec config get "core.log.level" with format "json"
-    Then the spec config get should succeed
-    And the spec config get JSON timing should have field "started"
-    And the spec config get JSON timing should have field "duration_ms"
-
-  # --- Source name mapping ---
-
-  Scenario: JSON source uses short name "config" when value from global config
-    Given I have set spec config "core.log.level" to "WARNING"
-    When I run spec config get "core.log.level" with format "json"
-    Then the spec config get should succeed
-    And the spec config get JSON data field "source" should equal "config"
-
-  Scenario: JSON source uses short name "default" when value from default
-    When I run spec config get "core.log.level" with format "json"
-    Then the spec config get should succeed
-    And the spec config get JSON data field "source" should equal "default"
@@ -16,18 +16,12 @@ Feature: LSP Resource Types

    Examples:
      | type_name     |
-      | executable    |
      | lsp-server    |
      | lsp-workspace |
      | lsp-document  |

  # ── User-addable flags ────────────────────────────────────────

-  Scenario: executable is user-addable for lsp_rt
-    Given the built-in executable YAML file for lsp_rt
-    When I load the YAML via schema for lsp_rt
-    Then the lsp_rt schema user_addable should be true
-
  Scenario: lsp-server is user-addable for lsp_rt
    Given the built-in lsp-server YAML file for lsp_rt
    When I load the YAML via schema for lsp_rt
@@ -51,12 +45,6 @@ Feature: LSP Resource Types
    Then the lsp_rt capability read should be true
    And the lsp_rt capability write should be true

-  Scenario: executable has read-only capabilities for lsp_rt
-    Given the built-in executable YAML file for lsp_rt
-    When I load the YAML via schema for lsp_rt
-    Then the lsp_rt capability read should be true
-    And the lsp_rt capability write should be false
-
  # ── Parent/child hierarchy ────────────────────────────────────

  Scenario: lsp-server has lsp-workspace as child type for lsp_rt
@@ -81,12 +69,6 @@ Feature: LSP Resource Types

  # ── Auto-discovery ────────────────────────────────────────────

-  Scenario: executable has auto-discovery from container-exec-env for lsp_rt
-    Given the built-in executable YAML file for lsp_rt
-    When I load the YAML via schema for lsp_rt
-    Then the lsp_rt auto_discovery should be present
-    And the lsp_rt auto_discovery trigger_types should contain "container-exec-env"
-
  Scenario: lsp-workspace has auto-discovery from lsp-server for lsp_rt
    Given the built-in lsp-workspace YAML file for lsp_rt
    When I load the YAML via schema for lsp_rt
@@ -97,8 +79,7 @@ Feature: LSP Resource Types

  Scenario: BUILTIN_NAMES includes all LSP types for lsp_rt
    Given the ResourceTypeSpec BUILTIN_NAMES set for lsp_rt
-    Then BUILTIN_NAMES should contain "executable" for lsp_rt
-    And BUILTIN_NAMES should contain "lsp-server" for lsp_rt
+    Then BUILTIN_NAMES should contain "lsp-server" for lsp_rt
    And BUILTIN_NAMES should contain "lsp-workspace" for lsp_rt
    And BUILTIN_NAMES should contain "lsp-document" for lsp_rt

@@ -106,9 +87,6 @@ Feature: LSP Resource Types

  Scenario: LSP types survive DB roundtrip after bootstrap for lsp_rt
    Given a lsp_rt fresh in-memory resource registry with bootstrap
-    When I query the lsp_rt registry for type "executable"
-    Then the lsp_rt db type "executable" should exist
-    And the lsp_rt db type "executable" should have kind "physical"
    When I query the lsp_rt registry for type "lsp-server"
    Then the lsp_rt db type "lsp-server" should exist
    When I query the lsp_rt registry for type "lsp-workspace"
@@ -361,7 +361,7 @@ def step_get_entry_count(context):
    context.entry_count = context.index.get_entry_count()


-@then("the ACMS entry count should be {count:d}")
+@then("the 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
@@ -1,8 +1,7 @@
 """Step definitions for behave_parallel_log_filtering.feature.

 Tests the conditional log replay and worker exception handling in
-``scripts/run_behave_parallel.py``, as well as the PassSuppressFormatter
-per-scenario output buffering behaviour.
+``scripts/run_behave_parallel.py``.
 """

 from __future__ import annotations
@@ -13,11 +12,8 @@ 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]


@@ -56,7 +52,6 @@ _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:
@@ -362,132 +357,3 @@ 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}"
-    )
@@ -27,8 +27,7 @@ def _restore_cwd(context):
        os.environ.pop("CLEVERAGENTS_HOME", None)
    else:
        os.environ["CLEVERAGENTS_HOME"] = context._init_original_home
-    if getattr(context, "temp_dir", None) is not None:
-        shutil.rmtree(context.temp_dir, ignore_errors=True)
+    shutil.rmtree(context.temp_dir, ignore_errors=True)


 def _create_init_mocks(context):
@@ -41,16 +41,12 @@ from cleveragents.application.services.config_service import (
    ResolvedValue,
 )
 from cleveragents.cli.commands import config as config_mod
-from cleveragents.cli.commands._config_helpers import (
-    _env_var_for_key,
-)
 from cleveragents.cli.commands.config import (
+    _env_var_for_key,
    _resolution_chain,
    _resolve_source,
 )
-from cleveragents.cli.commands.config import (
-    app as config_app,
-)
+from cleveragents.cli.commands.config import app as config_app

 _runner = CliRunner()

@@ -266,7 +262,6 @@ def step_cfg_boost_get_verbose_path(context: Context, key: str, fmt: str) -> Non
    chain_with_path = [
        {"source": "cli_flag", "value": None},
        {"source": "env_var", "value": None},
-        {"source": "local", "value": None},
        {"source": "project", "value": None},
        {"source": "global", "value": Path("/tmp/fake/path")},
        {"source": "default", "value": Path("/default/path")},
@@ -289,7 +284,7 @@ def step_cfg_boost_get_verbose_path(context: Context, key: str, fmt: str) -> Non

@then("the cfg-boost JSON resolution chain entries should have string values")
 def step_cfg_boost_json_chain_strings(context: Context) -> None:
-   parsed = json.loads(context.cfg_boost_result.output)
+    parsed = json.loads(context.cfg_boost_result.output)
    data = _unwrap_envelope(parsed)
    chain = data.get("resolution_chain", [])
    assert len(chain) > 0, "Expected non-empty resolution chain"
@@ -26,23 +26,19 @@ from rich.console import Console
 from typer.testing import CliRunner

 from cleveragents.cli.commands import config as config_mod
-from cleveragents.cli.commands._config_helpers import (
+from cleveragents.cli.commands.config import (
    _env_var_for_key,
    _is_secret_key,
    _mask_value,
    _normalize_key,
-    _validate_key,
-)
-from cleveragents.cli.commands.config import (
    _read_config_file,
    _resolution_chain,
    _resolve_source,
    _settings_fields,
+    _validate_key,
    _write_config_file,
 )
-from cleveragents.cli.commands.config import (
-    app as config_app,
-)
+from cleveragents.cli.commands.config import app as config_app

 _runner = CliRunner()

@@ -533,17 +529,16 @@ def step_sn_get_valid_yaml(context: Context) -> None:

@then('the safety-net get YAML should contain key "{key}"')
 def step_sn_get_yaml_key(context: Context, key: str) -> None:
-    # YAML output is now wrapped in a standard envelope; fields are in data
-    yaml_data = context._sn_get_yaml
-    inner = yaml_data.get("data", yaml_data)
-    assert key in inner, f"Key '{key}' not in YAML data: {list(inner.keys())}"
+    assert key in context._sn_get_yaml, (
+        f"Key '{key}' not in YAML: {list(context._sn_get_yaml.keys())}"
+    )


@then("the safety-net get output should be valid JSON with type field")
 def step_sn_get_json_type(context: Context) -> None:
    result = context._sn_result
    assert result.exit_code == 0, f"Exit {result.exit_code}: {result.output}"
-   parsed = json.loads(result.output.strip())
+    parsed = json.loads(result.output.strip())
    data = _unwrap_envelope(parsed)
    assert "type" in data, f"'type' not in JSON keys: {list(data.keys())}"
    assert "key" in data, f"'key' not in JSON keys: {list(data.keys())}"
@@ -143,11 +143,9 @@ def step_config_get_contains_key(context: Context, key: str) -> None:
@then("the config get output should show the resolution chain")
 def step_config_get_resolution_chain(context: Context) -> None:
    output = context.result.output
-    assert (
-        "Resolution Chain" in output
-        or "Resolution chain" in output
-        or "resolution_chain" in output
-    ), f"Expected resolution chain in output: {output}"
+    assert "Resolution chain" in output or "resolution_chain" in output, (
+        f"Expected resolution chain in output: {output}"
+    )


@then("the config get should fail with unknown key error")
@@ -33,12 +33,10 @@ from cleveragents.application.services.config_service import (
    ResolvedValue,
 )
 from cleveragents.cli.commands import config as config_mod
-from cleveragents.cli.commands._config_helpers import (
-    _settings_defaults,
-    _validate_key,
-)
 from cleveragents.cli.commands.config import (
    _resolve_source,
+    _settings_defaults,
+    _validate_key,
    _write_config_file,
 )
 from cleveragents.cli.commands.config import (
@@ -338,11 +336,9 @@ def step_get_verbose_succeed(context: Context) -> None:
 def step_get_verbose_chain(context: Context) -> None:
    result = context._cfg_branch_result
    output = result.output
-    assert (
-        "Resolution Chain" in output
-        or "Resolution chain" in output
-        or "resolution_chain" in output
-    ), f"Expected resolution chain in output: {output}"
+    assert "Resolution chain" in output or "resolution_chain" in output, (
+        f"Expected resolution chain in output: {output}"
+    )


 # ===================================================================
@@ -366,7 +362,6 @@ def step_mock_resolve_path(context: Context, key: str) -> None:
                "value": None,
                "env_name": f"CLEVERAGENTS_{key.upper().replace('.', '_').replace('-', '_')}",
            },
-            {"source": ConfigLevel.LOCAL.value, "value": None},
            {"source": ConfigLevel.PROJECT.value, "value": None},
            {
                "source": ConfigLevel.GLOBAL.value,
@@ -409,10 +404,8 @@ def step_get_result_json(context: Context) -> None:

@then("the config cli branch get JSON value should be a string not a Path")
 def step_get_json_value_string(context: Context) -> None:
-    envelope = context._cfg_branch_get_json
-    # JSON output is now wrapped in a standard envelope; value is in data
-    data = envelope.get("data", envelope)
-    # The "value" should be a string (serialised from Path)
+    data = context._cfg_branch_get_json
+    # The top-level "value" should be a string (serialised from Path)
    assert isinstance(data["value"], str), (
        f"Expected str, got {type(data['value'])}: {data['value']}"
    )
@@ -1,296 +0,0 @@
-"""Step definitions for config_get_spec_output.feature.
-
-Tests that `agents config get` output matches the specification:
- Rich output: three panels (Config, Origin, Resolution Chain)
- JSON output: standard envelope with all required fields
- Spec-required type strings and human-readable source names
-"""
-
-from __future__ import annotations
-
-import json
-import os
-import re
-import shutil
-import tempfile
-from pathlib import Path
-from typing import Any
-from unittest.mock import patch
-
-from behave import given, then, when
-from behave.runner import Context
-from typer.testing import CliRunner
-
-from cleveragents.cli.commands import config as config_mod
-from cleveragents.cli.commands.config import app as config_app
-
-_runner = CliRunner()
-
-# Human-readable source names per spec
-_SPEC_SOURCE_NAMES = {
-    "CLI flag",
-    "Env var",
-    "Local file",
-    "Project file",
-    "Config file",
-    "Default",
-}
-
-
-# ---------------------------------------------------------------------------
-# Helpers
-# ---------------------------------------------------------------------------
-
-
-def _setup_spec_temp(context: Context) -> None:
-    """Redirect config dir/path to a fresh temp directory."""
-    if not hasattr(context, "_spec_tmpdir"):
-        context._spec_tmpdir = tempfile.mkdtemp(prefix="cfg_spec_")
-    tmp = Path(context._spec_tmpdir)
-    context._spec_dir_patch = patch.object(config_mod, "_CONFIG_DIR", tmp)
-    context._spec_path_patch = patch.object(
-        config_mod, "_CONFIG_PATH", tmp / "config.toml"
-    )
-    context._spec_dir_patch.start()
-    context._spec_path_patch.start()
-
-    if not hasattr(context, "_cleanup_handlers"):
-        context._cleanup_handlers: list[Any] = []
-    context._cleanup_handlers.append(lambda: _teardown_spec_temp(context))
-
-
-def _teardown_spec_temp(context: Context) -> None:
-    for attr in ("_spec_dir_patch", "_spec_path_patch"):
-        patcher = getattr(context, attr, None)
-        if patcher is not None and getattr(patcher, "_active", True):
-            patcher.stop()
-    tmpdir = getattr(context, "_spec_tmpdir", None)
-    if tmpdir and os.path.isdir(tmpdir):
-        shutil.rmtree(tmpdir, ignore_errors=True)
-
-
-# ---------------------------------------------------------------------------
-# Background
-# ---------------------------------------------------------------------------
-
-
-@given("a clean config get spec test environment")
-def step_spec_clean_env(context: Context) -> None:
-    _setup_spec_temp(context)
-
-
-# ---------------------------------------------------------------------------
-# When
-# ---------------------------------------------------------------------------
-
-
-@when('I run spec config get "{key}"')
-def step_spec_run_get(context: Context, key: str) -> None:
-    context.spec_result = _runner.invoke(config_app, ["get", key])
-
-
-@when('I run spec config get "{key}" with format "{fmt}"')
-def step_spec_run_get_fmt(context: Context, key: str, fmt: str) -> None:
-    context.spec_result = _runner.invoke(config_app, ["get", key, "--format", fmt])
-
-
-@given('I have set spec config "{key}" to "{value}"')
-def step_spec_preset_config(context: Context, key: str, value: str) -> None:
-    result = _runner.invoke(config_app, ["set", key, value])
-    assert result.exit_code == 0, f"Pre-set failed: {result.output}"
-
-
-# ---------------------------------------------------------------------------
-# Then — basic
-# ---------------------------------------------------------------------------
-
-
-@then("the spec config get should succeed")
-def step_spec_get_succeed(context: Context) -> None:
-    assert context.spec_result.exit_code == 0, (
-        f"Expected exit 0, got {context.spec_result.exit_code}: "
-        f"{context.spec_result.output}"
-    )
-
-
-@then('the spec config get rich output should contain panel "{panel_title}"')
-def step_spec_rich_contains_panel(context: Context, panel_title: str) -> None:
-    output = context.spec_result.output
-    assert panel_title in output, (
-        f"Expected panel '{panel_title}' in Rich output: {output}"
-    )
-
-
-@then('the spec config get rich output should contain "{text}"')
-def step_spec_rich_contains(context: Context, text: str) -> None:
-    output = context.spec_result.output
-    assert text in output, f"Expected '{text}' in Rich output: {output}"
-
-
-@then('the spec config get rich output should not contain "{text}"')
-def step_spec_rich_not_contains(context: Context, text: str) -> None:
-    output = context.spec_result.output
-    # Check that the text doesn't appear as a standalone type name
-    # (e.g. "str" should not appear but "string" is fine)
-    # We check for exact word boundary to avoid false positives
-    # Look for the text as a standalone word (not part of "string", "integer", etc.)
-    pattern = rf"\b{re.escape(text)}\b"
-    # Exclude lines that contain "string", "integer", "boolean" etc.
-    lines = output.split("\n")
-    for line in lines:
-        if re.search(pattern, line):
-            # Check if this is a false positive (e.g. "str" inside "string")
-            # Only flag if the match is not part of a longer type word
-            if text == "str" and "string" in line:
-                continue
-            if text == "bool" and "boolean" in line:
-                continue
-            if text == "int" and "integer" in line:
-                continue
-            raise AssertionError(
-                f"Found '{text}' as standalone word in Rich output line: {line!r}"
-            )
-
-
-# ---------------------------------------------------------------------------
-# Then — JSON envelope
-# ---------------------------------------------------------------------------
-
-
-def _get_spec_json(context: Context) -> dict[str, Any]:
-    """Parse and cache the JSON output."""
-    if not hasattr(context, "_spec_json"):
-        context._spec_json = json.loads(context.spec_result.output.strip())
-    return context._spec_json
-
-
-@then(
-    'the spec config get JSON envelope should have field "{field}" equal to "{value}"'
-)
-def step_spec_json_envelope_field_str(context: Context, field: str, value: str) -> None:
-    envelope = _get_spec_json(context)
-    assert field in envelope, (
-        f"Field '{field}' not in envelope: {list(envelope.keys())}"
-    )
-    assert str(envelope[field]) == value, (
-        f"Expected envelope['{field}'] == '{value}', got {envelope[field]!r}"
-    )
-
-
-@then(
-    'the spec config get JSON envelope should have field "{field}" equal to {value:d}'
-)
-def step_spec_json_envelope_field_int(context: Context, field: str, value: int) -> None:
-    envelope = _get_spec_json(context)
-    assert field in envelope, (
-        f"Field '{field}' not in envelope: {list(envelope.keys())}"
-    )
-    assert envelope[field] == value, (
-        f"Expected envelope['{field}'] == {value}, got {envelope[field]!r}"
-    )
-
-
-@then('the spec config get JSON envelope should have field "{field}"')
-def step_spec_json_envelope_has_field(context: Context, field: str) -> None:
-    envelope = _get_spec_json(context)
-    assert field in envelope, (
-        f"Field '{field}' not in envelope: {list(envelope.keys())}"
-    )
-
-
-# ---------------------------------------------------------------------------
-# Then — JSON data
-# ---------------------------------------------------------------------------
-
-
-def _get_spec_json_data(context: Context) -> dict[str, Any]:
-    """Return the data object from the JSON envelope."""
-    envelope = _get_spec_json(context)
-    return envelope["data"]
-
-
-@then('the spec config get JSON data should have field "{field}"')
-def step_spec_json_data_has_field(context: Context, field: str) -> None:
-    data = _get_spec_json_data(context)
-    assert field in data, f"Field '{field}' not in JSON data: {list(data.keys())}"
-
-
-@then('the spec config get JSON data field "{field}" should equal "{value}"')
-def step_spec_json_data_field_str(context: Context, field: str, value: str) -> None:
-    data = _get_spec_json_data(context)
-    assert field in data, f"Field '{field}' not in JSON data: {list(data.keys())}"
-    assert data[field] == value, (
-        f"Expected data['{field}'] == '{value}', got {data[field]!r}"
-    )
-
-
-@then('the spec config get JSON data field "{field}" should be false')
-def step_spec_json_data_field_false(context: Context, field: str) -> None:
-    data = _get_spec_json_data(context)
-    assert field in data, f"Field '{field}' not in JSON data: {list(data.keys())}"
-    assert data[field] is False, (
-        f"Expected data['{field}'] == False, got {data[field]!r}"
-    )
-
-
-@then('the spec config get JSON data field "{field}" should be true')
-def step_spec_json_data_field_true(context: Context, field: str) -> None:
-    data = _get_spec_json_data(context)
-    assert field in data, f"Field '{field}' not in JSON data: {list(data.keys())}"
-    assert data[field] is True, f"Expected data['{field}'] == True, got {data[field]!r}"
-
-
-# ---------------------------------------------------------------------------
-# Then — JSON origin
-# ---------------------------------------------------------------------------
-
-
-@then('the spec config get JSON origin should have field "{field}"')
-def step_spec_json_origin_has_field(context: Context, field: str) -> None:
-    data = _get_spec_json_data(context)
-    origin = data.get("origin", {})
-    assert field in origin, f"Field '{field}' not in JSON origin: {list(origin.keys())}"
-
-
-# ---------------------------------------------------------------------------
-# Then — JSON resolution chain
-# ---------------------------------------------------------------------------
-
-
-@then(
-    "the spec config get JSON resolution chain should use human-readable source names"
-)
-def step_spec_json_chain_human_readable(context: Context) -> None:
-    data = _get_spec_json_data(context)
-    chain = data.get("resolution_chain", [])
-    assert len(chain) > 0, "Expected non-empty resolution chain"
-    for entry in chain:
-        src = entry.get("source", "")
-        assert src in _SPEC_SOURCE_NAMES, (
-            f"Source '{src}' is not a spec-required human-readable name. "
-            f"Expected one of: {_SPEC_SOURCE_NAMES}"
-        )
-
-
-# ---------------------------------------------------------------------------
-# Then — JSON winner
-# ---------------------------------------------------------------------------
-
-
-@then('the spec config get JSON winner should have field "{field}"')
-def step_spec_json_winner_has_field(context: Context, field: str) -> None:
-    data = _get_spec_json_data(context)
-    winner = data.get("winner", {})
-    assert field in winner, f"Field '{field}' not in JSON winner: {list(winner.keys())}"
-
-
-# ---------------------------------------------------------------------------
-# Then — JSON timing
-# ---------------------------------------------------------------------------
-
-
-@then('the spec config get JSON timing should have field "{field}"')
-def step_spec_json_timing_has_field(context: Context, field: str) -> None:
-    envelope = _get_spec_json(context)
-    timing = envelope.get("timing", {})
-    assert field in timing, f"Field '{field}' not in JSON timing: {list(timing.keys())}"
@@ -5,7 +5,7 @@ from typing import Any

 from behave import given, then, when

-PROJECT_ROOT = Path(__file__).resolve().parents[2]
+PROJECT_ROOT = Path(__file__).resolve().parents[3]
 AGENT_DEF_PATH = PROJECT_ROOT / ".opencode" / "agents" / "implementation-supervisor.md"


@@ -1,211 +0,0 @@
-"""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}"
-    )
@@ -1,43 +0,0 @@
-@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
@@ -111,13 +111,6 @@ 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"
@@ -26,15 +26,6 @@ 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)
    )
@@ -15,26 +15,30 @@ def _import_lsp_types() -> None:
        LSP_RESOURCE_TYPES,
    )

-    assert len(LSP_RESOURCE_TYPES) == 4, f"Expected 4, got {len(LSP_RESOURCE_TYPES)}"
+    assert len(LSP_RESOURCE_TYPES) == 3, f"Expected 3, got {len(LSP_RESOURCE_TYPES)}"
    names = [t["name"] for t in LSP_RESOURCE_TYPES]
-    for expected in ("executable", "lsp-server", "lsp-workspace", "lsp-document"):
+    for expected in ("lsp-server", "lsp-workspace", "lsp-document"):
        assert expected in names, f"{expected} not in LSP_RESOURCE_TYPES: {names}"

    print("import-lsp-types-ok")


 def _check_builtin_names() -> None:
-    """Verify all 4 LSP types are in BUILTIN_NAMES."""
+    """Verify all 3 LSP types are in BUILTIN_NAMES."""
    from cleveragents.domain.models.core.resource_type import ResourceTypeSpec

-    for name in ("executable", "lsp-server", "lsp-workspace", "lsp-document"):
+    for name in ("lsp-server", "lsp-workspace", "lsp-document"):
        assert name in ResourceTypeSpec.BUILTIN_NAMES, f"{name} not in BUILTIN_NAMES"

+    assert "executable" not in ResourceTypeSpec.BUILTIN_NAMES, (
+        "executable should not be in BUILTIN_NAMES (not a spec-defined type)"
+    )
+
    print("check-builtin-names-ok")


 def _db_roundtrip() -> None:
-    """Bootstrap in-memory DB and verify all 4 LSP types are retrievable."""
+    """Bootstrap in-memory DB and verify all 3 LSP types are retrievable."""
    from sqlalchemy import create_engine
    from sqlalchemy.orm import sessionmaker

@@ -48,7 +52,7 @@ def _db_roundtrip() -> None:
    factory = sessionmaker(bind=engine, expire_on_commit=False)
    svc = ResourceRegistryService(session_factory=factory)

-    for name in ("executable", "lsp-server", "lsp-workspace", "lsp-document"):
+    for name in ("lsp-server", "lsp-workspace", "lsp-document"):
        spec = svc.show_type(name)
        assert spec is not None, f"{name} not found after bootstrap"
        assert str(spec.resource_kind) == "physical", (
@@ -85,32 +89,27 @@ def _check_hierarchy() -> None:
    assert "lsp-workspace" in types["lsp-document"]["parent_types"], (
        "lsp-document missing lsp-workspace parent"
    )
-    # executable has no children
-    assert types["executable"]["child_types"] == [], (
-        "executable should have no children"
-    )

    print("check-hierarchy-ok")


 def _check_auto_discovery() -> None:
-    """Verify executable has auto_discovery with trigger_types."""
+    """Verify lsp-workspace has auto_discovery with trigger_types."""
    from cleveragents.application.services._resource_registry_lsp import (
        LSP_RESOURCE_TYPES,
    )

-    types = {t["name"]: t for t in LSP_RESOURCE_TYPES}
+    types = {t["name"] for t in LSP_RESOURCE_TYPES}

-    # executable has auto_discovery
-    ad = types["executable"].get("auto_discovery")
-    assert ad is not None, "executable missing auto_discovery"
-    assert "container-exec-env" in ad.get("trigger_types", []), (
-        "executable missing container-exec-env trigger"
+    # executable should not be present
+    assert "executable" not in types, (
+        "executable should not be in LSP_RESOURCE_TYPES (not a spec-defined type)"
    )
-    assert ad.get("lazy") is True, "executable auto_discovery should be lazy"
+
+    lsp_types = {t["name"]: t for t in LSP_RESOURCE_TYPES}

    # lsp-workspace has auto_discovery from lsp-server
-    ad_ws = types["lsp-workspace"].get("auto_discovery")
+    ad_ws = lsp_types["lsp-workspace"].get("auto_discovery")
    assert ad_ws is not None, "lsp-workspace missing auto_discovery"
    assert "lsp-server" in ad_ws.get("trigger_types", []), (
        "lsp-workspace missing lsp-server trigger"
@@ -1,125 +0,0 @@
-"""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
@@ -1,16 +1,20 @@
 """In-process parallel behave runner.

-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.
+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.

 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 split into *N* equal-size chunks dispatched via
-  ``multiprocessing.Pool`` with the ``fork`` start method.
+  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).
 """

 from __future__ import annotations
@@ -27,13 +31,6 @@ 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/"


@@ -82,7 +79,6 @@ def _is_btrfs_or_overlayfs() -> bool:

 # Type alias for summary dictionaries
 Summary = dict[str, Any]
-WorkerResult = tuple[bool, str, str, Summary]


 # ---------------------------------------------------------------------------
@@ -178,7 +174,10 @@ 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]
@@ -273,50 +272,35 @@ 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:
-        # 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]
+        config.format = [config.default_format]
    return Runner(config)


-def _run_features_inprocess(paths: list[str], args: list[str]) -> tuple[bool, Summary]:
+def _run_features_inprocess(
+    feature_paths: list[str], behave_args: list[str]
+) -> tuple[bool, Summary]:
    """Run *all* feature_paths in a single behave Runner invocation.

    Returns ``(failed: bool, summary: dict)``.
    """
-    runner = _make_runner(paths, args)
+    runner = _make_runner(feature_paths, behave_args)
    failed = runner.run()
    summary = _extract_summary(runner)
    return failed, summary


-def _worker_run_features(payload: tuple[list[str], list[str]]) -> WorkerResult:
+def _worker_run_features(
+    payload: tuple[list[str], list[str]],
+) -> tuple[bool, str, str, Summary]:
    """Entry point for multiprocessing workers.

    Runs a chunk of feature files in a forked child process.  Heavy
@@ -364,7 +348,9 @@ def _worker_run_features(payload: tuple[list[str], list[str]]) -> WorkerResult:
 # ---------------------------------------------------------------------------


-def _aggregate_worker_results(results: list[WorkerResult]) -> Summary:
+def _aggregate_worker_results(
+    results: list[tuple[bool, str, str, Summary]],
+) -> Summary:
    """Aggregate worker results, replaying logs only for failed chunks.

    Iterates over the list of ``(worker_failed, stdout, stderr, summary)``
@@ -10,12 +10,11 @@ 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."""
@@ -46,7 +45,8 @@ class TierLevel(StrEnum):
    ARCHIVE = "archive"  # Reference


-class IndexEntry(BaseModel):
+@dataclass
+class IndexEntry:
    """Represents a single indexed context entry.

    Attributes:
@@ -65,9 +65,9 @@ class IndexEntry(BaseModel):
    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,6 +104,7 @@ class IndexEntry(BaseModel):
        self.tier = tier


+@dataclass
 class ACMSIndex:
    """ACMS Index for storing and querying indexed context entries.

@@ -114,8 +115,7 @@ class ACMSIndex:
        entries: Dictionary mapping file paths to IndexEntry objects
    """

-    def __init__(self) -> None:
-        self.entries: dict[str, IndexEntry] = {}
+    entries: dict[str, IndexEntry] = field(default_factory=dict)

    def add_entry(self, entry: IndexEntry) -> None:
        """Add an index entry to the index.
@@ -1,15 +1,14 @@
 """LSP resource type definitions for the Resource Registry.

-Contains the built-in LSP-related type entries (``executable``,
-``lsp-server``, ``lsp-workspace``, ``lsp-document``) that are appended
-to ``BUILTIN_TYPES`` in ``_resource_registry_data``.
+Contains the built-in LSP-related type entries (``lsp-server``,
+``lsp-workspace``, ``lsp-document``) that are appended to
+``BUILTIN_TYPES`` in ``_resource_registry_data``.

 These types bridge the resource system with the LSP runtime, allowing
 actors to discover available language tooling and the LSP system to know
 which resources it operates on.

-Spec reference: docs/adr/ADR-039-container-resource-types.md (executable),
-                docs/adr/ADR-040-lsp-resource-types.md (lsp-server,
+Spec reference: docs/adr/ADR-040-lsp-resource-types.md (lsp-server,
                lsp-workspace, lsp-document)

 .. note::
@@ -28,52 +27,11 @@ __all__ = ["LSP_RESOURCE_TYPES"]
 # === LSP Resource Types (#832) ===
 #
 # These types model language tooling infrastructure in the resource DAG.
-# - executable: system binary / interpreter / LSP server binary
 # - lsp-server: a configured or running LSP server instance
 # - lsp-workspace: workspace root tracked by an LSP server
 # - lsp-document: text document tracked by an LSP server

 LSP_RESOURCE_TYPES: list[dict[str, Any]] = [
-    # ── executable ───────────────────────────────────────────────
-    {
-        "name": "executable",
-        "description": (
-            "System executable (language runtime, compiler, LSP server "
-            "binary). Discovered from container or host PATH."
-        ),
-        "resource_kind": "physical",
-        "sandbox_strategy": "none",
-        "user_addable": True,
-        "built_in": True,
-        "cli_args": [
-            {
-                "name": "path",
-                "type": "string",
-                "required": True,
-                "description": "Absolute path to the executable",
-            },
-            {
-                "name": "version",
-                "type": "string",
-                "required": False,
-                "description": "Executable version if detectable",
-            },
-        ],
-        "parent_types": ["container-exec-env", "fs-directory"],
-        "child_types": [],
-        "handler": "cleveragents.resource.handlers.executable:ExecutableHandler",
-        "auto_discovery": {
-            "trigger_types": ["container-exec-env", "fs-directory"],
-            "scan_paths": [],
-            "lazy": True,
-        },
-        "capabilities": {
-            "read": True,
-            "write": False,
-            "sandbox": False,
-            "checkpoint": False,
-        },
-    },
    # ── lsp-server ───────────────────────────────────────────────
    {
        "name": "lsp-server",
@@ -1,171 +0,0 @@
-"""Display helpers for the ``agents config get`` command.
-
-Provides spec-required formatting utilities for converting internal
-``ConfigService`` data structures to the output format defined in
-``docs/specification.md`` (section "agents config get").
-
-Separated from ``config.py`` to keep that module under the 500-line limit.
-"""
-
-from __future__ import annotations
-
-from typing import Any
-
-from cleveragents.application.services.config_service import ConfigLevel
-
-# ---------------------------------------------------------------------------
-# Source name mappings
-# ---------------------------------------------------------------------------
-
-# Mapping from internal ConfigLevel values to spec-required human-readable
-# names used in the resolution chain and winner objects.
-_SOURCE_DISPLAY_NAMES: dict[str, str] = {
-    ConfigLevel.CLI_FLAG.value: "CLI flag",
-    ConfigLevel.ENV_VAR.value: "Env var",
-    ConfigLevel.LOCAL.value: "Local file",
-    ConfigLevel.PROJECT.value: "Project file",
-    ConfigLevel.GLOBAL.value: "Config file",
-    ConfigLevel.DEFAULT.value: "Default",
-}
-
-# Short source names used in the top-level ``data.source`` field and the
-# Rich ``Source:`` line.  These match the spec examples exactly.
-_SOURCE_SHORT_NAMES: dict[str, str] = {
-    ConfigLevel.CLI_FLAG.value: "cli",
-    ConfigLevel.ENV_VAR.value: "env",
-    ConfigLevel.LOCAL.value: "local",
-    ConfigLevel.PROJECT.value: "project",
-    ConfigLevel.GLOBAL.value: "config",
-    ConfigLevel.DEFAULT.value: "default",
-}
-
-# Mapping from Python type names to spec-required type strings.
-_PYTHON_TYPE_TO_SPEC: dict[str, str] = {
-    "str": "string",
-    "bool": "boolean",
-    "int": "integer",
-    "float": "number",
-    "NoneType": "null",
-    "None": "null",
-}
-
-# ---------------------------------------------------------------------------
-# Public helpers
-# ---------------------------------------------------------------------------
-
-
-def spec_type_name(value: Any) -> str:
-    """Return the spec-required type string for *value*.
-
-    Maps Python type names to the spec-required strings:
-    ``str`` → ``string``, ``bool`` → ``boolean``, ``int`` → ``integer``,
-    ``float`` → ``number``, ``None`` → ``null``.
-    """
-    if value is None:
-        return "null"
-    python_name = type(value).__name__
-    return _PYTHON_TYPE_TO_SPEC.get(python_name, python_name)
-
-
-def spec_source_short(internal_source: str) -> str:
-    """Return the short source name for the ``data.source`` field.
-
-    Used in the top-level ``data.source`` JSON field and the Rich
-    ``Source:`` line.  Matches the spec examples (e.g. ``"config"``,
-    ``"default"``, ``"env"``).
-    """
-    return _SOURCE_SHORT_NAMES.get(internal_source, internal_source)
-
-
-def spec_source_display(internal_source: str) -> str:
-    """Return the human-readable source name for resolution chain entries.
-
-    Used inside ``resolution_chain[].source`` and ``winner.source``.
-    Returns names like ``"CLI flag"``, ``"Config file"``, ``"Default"``.
-    """
-    return _SOURCE_DISPLAY_NAMES.get(internal_source, internal_source)
-
-
-def build_spec_resolution_chain(
-    chain: list[dict[str, Any]],
-) -> list[dict[str, Any]]:
-    """Convert internal chain entries to spec-required format.
-
-    The spec requires:
-    ``[{"level": 1, "source": "CLI flag", "value": null}, ...]``
-
-    Human-readable source names are used inside the chain per spec.
-    """
-    result: list[dict[str, Any]] = []
-    for idx, entry in enumerate(chain, start=1):
-        result.append(
-            {
-                "level": idx,
-                "source": spec_source_display(entry.get("source", "")),
-                "value": entry.get("value"),
-            }
-        )
-    return result
-
-
-def find_winner_in_chain(
-    chain: list[dict[str, Any]],
-    winning_source: str,
-) -> dict[str, Any]:
-    """Return the winner dict ``{"source": ..., "level": ...}`` for the chain.
-
-    *winning_source* is the internal ``ConfigLevel.value`` string.
-    The spec shows ``winner.source`` in lowercase (e.g. ``"config file"``).
-    """
-    # Use lowercase human-readable name per spec example
-    display_name = spec_source_display(winning_source).lower()
-    for idx, entry in enumerate(chain, start=1):
-        if entry.get("source") == winning_source and entry.get("value") is not None:
-            return {"source": display_name, "level": idx}
-    # Fallback: last entry (default)
-    return {"source": display_name, "level": len(chain)}
-
-
-def build_origin(
-    chain: list[dict[str, Any]],
-    winning_source: str,
-    default_value: Any,
-) -> dict[str, Any]:
-    """Build the ``origin`` object for the JSON envelope.
-
-    The origin shows where the winning value came from (file path, line,
-    and the registered default).  Line numbers are not tracked by the
-    service, so ``line`` is always ``null``.
-
-    TODO: Track line numbers in ConfigService to populate ``line`` field.
-    """
-    file_path: str | None = None
-    line: int | None = None  # ConfigService does not track line numbers
-
-    # Find the winning chain entry to extract path info
-    for entry in chain:
-        if entry.get("source") == winning_source and entry.get("value") is not None:
-            raw_path = entry.get("path")
-            if raw_path:
-                file_path = str(raw_path)
-            break
-
-    # If no file path found, use a descriptive label based on source
-    if file_path is None:
-        if winning_source == ConfigLevel.ENV_VAR.value:
-            for entry in chain:
-                if entry.get("source") == winning_source:
-                    env_name = entry.get("env_name")
-                    if env_name:
-                        file_path = f"${env_name}"
-                    break
-        elif winning_source == ConfigLevel.CLI_FLAG.value:
-            file_path = "(CLI flag)"
-        elif winning_source == ConfigLevel.DEFAULT.value:
-            file_path = "(built-in default)"
-
-    return {
-        "file": file_path,
-        "line": line,
-        "default": default_value,
-    }
@@ -1,85 +0,0 @@
-"""Internal helpers for the ``agents config`` command group.
-
-Provides pure key-normalisation, validation, and secret-masking utilities
-used by ``config.py``.  Extracted to keep ``config.py`` under the
-500-line limit.  Functions here have no side effects and do not call
-``ConfigService``.
-"""
-
-from __future__ import annotations
-
-import re
-from typing import Any
-
-import typer
-
-from cleveragents.application.services.config_service import _REGISTRY
-
-# ---------------------------------------------------------------------------
-# Secret helpers
-# ---------------------------------------------------------------------------
-
-_SECRET_PATTERNS: re.Pattern[str] = re.compile(
-    r"(api[_\-]?key|token|secret|password)", re.IGNORECASE
-)
-
-
-def _is_secret_key(key: str) -> bool:
-    """Return ``True`` when *key* looks like it stores a secret."""
-    return _SECRET_PATTERNS.search(key) is not None
-
-
-def _mask_value(value: str) -> str:
-    """Replace a value with ``****``."""
-    return "****"
-
-
-# ---------------------------------------------------------------------------
-# Key helpers
-# ---------------------------------------------------------------------------
-
-
-def _normalize_key(key: str) -> str:
-    """Pass-through normalisation — keys are matched as-is against the registry.
-
-    Backward compatibility: underscores are converted to dots so that
-    ``log_level`` resolves to ``core.log.level`` if there is a single
-    match.  If the key already exists in the registry, it is returned
-    unchanged.
-    """
-    stripped = key.strip()
-    if stripped in _REGISTRY:
-        return stripped
-    # Try underscore-to-dot conversion
-    dotted = stripped.replace("_", ".")
-    if dotted in _REGISTRY:
-        return dotted
-    return stripped
-
-
-def _validate_key(key: str) -> str:
-    """Validate and normalise *key*, raising :class:`typer.BadParameter` on failure."""
-    normalized = _normalize_key(key)
-    if not normalized:
-        raise typer.BadParameter("Key must not be empty.")
-    if normalized not in _REGISTRY:
-        valid_sample = ", ".join(sorted(_REGISTRY)[:8])
-        raise typer.BadParameter(
-            f"Unknown configuration key: '{key}' "
-            f"(normalized: '{normalized}'). "
-            f"Valid keys include: {valid_sample} ..."
-        )
-    return normalized
-
-
-def _env_var_for_key(key: str) -> str:
-    """Return the environment-variable name for a registered key."""
-    entry = _REGISTRY.get(key)
-    if entry is not None:
-        return entry.env_var
-    return f"CLEVERAGENTS_{key.upper().replace('.', '_').replace('-', '_')}"
-
-
-def _settings_defaults() -> dict[str, Any]:
-    """Return all registered config keys mapped to their default values."""
-    return {key: entry.default for key, entry in _REGISTRY.items()}
@@ -1,9 +1,22 @@
 """Configuration management commands for CleverAgents CLI.

-Manages runtime configuration values persisted in
-``~/.cleveragents/config.toml``.  Commands: ``set``, ``get``, ``list``.
-Keys use hierarchical dot-path names (e.g. ``core.log.level``).
-Resolution is delegated to :class:`ConfigService`.
+The ``agents config`` command group manages runtime configuration values
+persisted in a TOML file at ``~/.cleveragents/config.toml``.
+
+## Commands
+
+| Command                         | Description                             |
+|---------------------------------|-----------------------------------------|
+| ``agents config set <K> <V>``   | Set a configuration value               |
+| ``agents config get <K>``       | Get a configuration value               |
+| ``agents config list [PATTERN]``| List configuration values               |
+
+## Key Format
+
+Keys use the hierarchical dot-path names defined in the specification
+(e.g. ``core.log.level``, ``plan.budget.per-plan``).  Resolution is
+delegated to :class:`ConfigService` which implements the five-level
+precedence chain.

 Based on implementation_plan.md task A8.cli.
 """
@@ -12,8 +25,6 @@ from __future__ import annotations

 import fnmatch
 import re
-import time
-from datetime import UTC, datetime
 from pathlib import Path
 from typing import Annotated, Any

@@ -27,30 +38,31 @@ from cleveragents.application.services.config_service import (
    ConfigScope,
    ConfigService,
 )
-from cleveragents.cli.commands._config_display import (
-    build_origin,
-    build_spec_resolution_chain,
-    find_winner_in_chain,
-    spec_source_short,
-    spec_type_name,
-)
-from cleveragents.cli.commands._config_helpers import (
-    _is_secret_key,
-    _mask_value,
-    _settings_defaults,
-    _validate_key,
-)
 from cleveragents.cli.formatting import OutputFormat, format_output
 from cleveragents.cli.renderers import _get_console

 app = typer.Typer(help="Manage configuration settings for CleverAgents.")
 console = _get_console()

+# ---------------------------------------------------------------------------
+# Constants
+# ---------------------------------------------------------------------------
+
 _CONFIG_DIR = Path.home() / ".cleveragents"
 _CONFIG_PATH = _CONFIG_DIR / "config.toml"
+
+_SECRET_PATTERNS: re.Pattern[str] = re.compile(
+    r"(api[_\-]?key|token|secret|password)", re.IGNORECASE
+)
+
 _FORMAT_HELP = "Output format: json, yaml, plain, table, or rich (default: rich)"


+# ---------------------------------------------------------------------------
+# Internal helpers
+# ---------------------------------------------------------------------------
+
+
 def _get_service() -> ConfigService:
    """Return a ``ConfigService`` wired to the standard config paths."""
    container = get_container()
@@ -61,6 +73,20 @@ def _get_service() -> ConfigService:
    )


+def _is_secret_key(key: str) -> bool:
+    """Return ``True`` when *key* looks like it stores a secret."""
+    return _SECRET_PATTERNS.search(key) is not None
+
+
+def _mask_value(value: str) -> str:
+    """Replace a value with ``****``."""
+    return "****"
+
+
+# Legacy helpers retained for backward compatibility with tests that
+# patch or call them directly.  They now delegate to ConfigService.
+
+
 def _settings_fields() -> dict[str, Any]:
    """Return all registered config keys mapped to their resolved values."""
    svc = _get_service()
@@ -71,6 +97,52 @@ def _settings_fields() -> dict[str, Any]:
    return results


+def _settings_defaults() -> dict[str, Any]:
+    """Return all registered config keys mapped to their default values."""
+    return {key: entry.default for key, entry in _REGISTRY.items()}
+
+
+def _normalize_key(key: str) -> str:
+    """Pass-through normalisation — keys are matched as-is against the registry.
+
+    Backward compatibility: underscores are converted to dots so that
+    ``log_level`` resolves to ``core.log.level`` if there is a single
+    match.  If the key already exists in the registry, it is returned
+    unchanged.
+    """
+    stripped = key.strip()
+    if stripped in _REGISTRY:
+        return stripped
+    # Try underscore-to-dot conversion
+    dotted = stripped.replace("_", ".")
+    if dotted in _REGISTRY:
+        return dotted
+    return stripped
+
+
+def _validate_key(key: str) -> str:
+    """Validate and normalise *key*, raising :class:`typer.BadParameter` on failure."""
+    normalized = _normalize_key(key)
+    if not normalized:
+        raise typer.BadParameter("Key must not be empty.")
+    if normalized not in _REGISTRY:
+        valid_sample = ", ".join(sorted(_REGISTRY)[:8])
+        raise typer.BadParameter(
+            f"Unknown configuration key: '{key}' "
+            f"(normalized: '{normalized}'). "
+            f"Valid keys include: {valid_sample} ..."
+        )
+    return normalized
+
+
+def _env_var_for_key(key: str) -> str:
+    """Return the environment-variable name for a registered key."""
+    entry = _REGISTRY.get(key)
+    if entry is not None:
+        return entry.env_var
+    return f"CLEVERAGENTS_{key.upper().replace('.', '_').replace('-', '_')}"
+
+
 def _resolve_source(key: str) -> str:
    """Determine which level supplied the winning value for *key*."""
    svc = _get_service()
@@ -103,6 +175,11 @@ def _write_config_file(data: dict[str, Any]) -> None:
    svc.write_config(data)


+# ---------------------------------------------------------------------------
+# Commands
+# ---------------------------------------------------------------------------
+
+
@app.command("set")
 def config_set(
    key: Annotated[str, typer.Argument(help="Configuration key (dot-path)")],
@@ -123,12 +200,16 @@ def config_set(
 ) -> None:
    """Set a configuration value.

-    Scopes: ``global``, ``project``, ``local``.
+    Writes the value to the config file for the specified scope.
+    Scopes: ``global`` (~/.cleveragents/config.toml),
+    ``project`` (<project-root>/config.toml),
+    ``local`` (<project-root>/config.local.toml).

    Examples::

        agents config set core.log.level DEBUG
        agents config set plan.concurrency 8 --scope local
+        agents config set core.automation-profile manual --project local/prod
    """
    normalized = _validate_key(key)
    svc = _get_service()
@@ -136,6 +217,8 @@ def config_set(

    # Coerce to the registered type
    coerced: Any = svc.validate_type(normalized, value)
+
+    # Resolve the target scope
    config_scope: ConfigScope | None = None
    if scope is not None:
        try:
@@ -144,8 +227,12 @@ def config_set(
            raise typer.BadParameter(
                f"Invalid scope: '{scope}'. Use 'global', 'project', or 'local'."
            ) from exc
+
+    # Read previous value
    config_data = svc.read_config()
+
    if project is not None:
+        # Legacy project-scoped set (--project flag)
        if not entry.project_scopable:
            raise typer.BadParameter(f"Key '{normalized}' is not project-scopable.")
        project_section = config_data.get("project", {})
@@ -161,6 +248,7 @@ def config_set(
        svc.write_config(config_data)
        scope_display = f"project:{project}"
    elif config_scope is not None:
+        # New three-scope set (--scope flag)
        if config_scope == ConfigScope.GLOBAL:
            previous = config_data.get(normalized)
        elif config_scope == ConfigScope.PROJECT:
@@ -204,128 +292,79 @@ def config_set(
@app.command("get")
 def config_get(
    key: Annotated[str, typer.Argument(help="Configuration key (dot-path)")],
+    verbose: Annotated[
+        bool,
+        typer.Option("--verbose", "-v", help="Show full resolution chain"),
+    ] = False,
    project: Annotated[
        str | None,
        typer.Option("--project", "-p", help="Project name for scoped resolution"),
    ] = None,
-    verbose: Annotated[
-        bool,
-        typer.Option(
-            "--verbose",
-            "-v",
-            hidden=True,
-            help="[DEPRECATED] Verbose output (no longer needed — chain always shown)",
-        ),
-    ] = False,
    fmt: Annotated[str, typer.Option("--format", "-f", help=_FORMAT_HELP)] = "rich",
 ) -> None:
    """Get a configuration value with its resolution chain.

-    Shows the Config, Origin, and Resolution Chain panels per spec.
-    The full chain and Winner indicator are always displayed.
+    Shows the effective value and which source it was resolved from
+    (CLI flag, environment variable, project config, global file, or
+    default).
+
+    Use ``--verbose`` to display the full five-level resolution chain.

    Examples::

        agents config get core.log.level
+        agents config get plan.budget.per-plan --verbose
        agents config get core.automation-profile --project local/prod
    """
-    started_at = datetime.now(tz=UTC)
-    t0 = time.monotonic()
    normalized = _validate_key(key)
    svc = _get_service()
-    entry = _REGISTRY[normalized]
-    # Always verbose — spec requires the full chain in all output formats.
-    resolved = svc.resolve(normalized, project_name=project, verbose=True)
-    elapsed_ms = round((time.monotonic() - t0) * 1000)
-    overridden = resolved.value != entry.default
-    spec_chain = build_spec_resolution_chain(resolved.chain)
-    winner = find_winner_in_chain(resolved.chain, resolved.source.value)
-    origin = build_origin(resolved.chain, resolved.source.value, entry.default)
-    spec_type = spec_type_name(resolved.value)
-    short_source = spec_source_short(resolved.source.value)
-    display_value: Any = resolved.value
-    if hasattr(display_value, "__fspath__"):
-        display_value = str(display_value)
+
+    resolved = svc.resolve(
+        normalized,
+        project_name=project,
+        verbose=verbose,
+    )
+
+    result: dict[str, Any] = {
+        "key": normalized,
+        "value": resolved.value,
+        "source": resolved.source.value,
+        "type": type(resolved.value).__name__ if resolved.value is not None else "None",
+    }
+    if verbose:
+        result["resolution_chain"] = resolved.chain

    if fmt != OutputFormat.RICH.value:
-        data: dict[str, Any] = {
-            "key": normalized,
-            "value": display_value,
-            "source": short_source,
-            "overridden": overridden,
-            "type": spec_type,
-            "origin": origin,
-            "resolution_chain": spec_chain,
-            "winner": winner,
-        }
-        # Serialise Path objects in chain values
-        for chain_entry in data["resolution_chain"]:
-            if hasattr(chain_entry.get("value"), "__fspath__"):
-                chain_entry["value"] = str(chain_entry["value"])
-
-        envelope: dict[str, Any] = {
-            "command": "config get",
-            "status": "ok",
-            "exit_code": 0,
-            "data": data,
-            "timing": {
-                "started": started_at.isoformat(),
-                "duration_ms": elapsed_ms,
-            },
-            "messages": ["Config read"],
-        }
-        typer.echo(format_output(envelope, fmt))
+        # Serialise Path objects to strings for JSON/YAML
+        if hasattr(result["value"], "__fspath__"):
+            result["value"] = str(result["value"])
+        for entry in result.get("resolution_chain", []):
+            if hasattr(entry.get("value"), "__fspath__"):
+                entry["value"] = str(entry["value"])
+        typer.echo(format_output(result, fmt))
        return

-    # Rich output: three panels per spec
-    overridden_display = "yes" if overridden else "no"
    console.print(
        Panel(
            f"[bold]Key:[/bold] {normalized}\n"
-            f"[bold]Value:[/bold] {display_value}\n"
-            f"[bold]Source:[/bold] {short_source}\n"
-            f"[bold]Overridden:[/bold] {overridden_display}\n"
-            f"[bold]Type:[/bold] {spec_type}",
-            title="Config",
-            expand=False,
-        )
-    )
-    origin_file = origin.get("file") or "(unknown)"
-    origin_line = origin.get("line")
-    origin_default = origin.get("default")
-    origin_line_display = str(origin_line) if origin_line is not None else "(unknown)"
-    origin_default_display = (
-        str(origin_default) if origin_default is not None else "(none)"
-    )
-    console.print(
-        Panel(
-            f"[bold]File:[/bold] {origin_file}\n"
-            f"[bold]Line:[/bold] {origin_line_display}\n"
-            f"[bold]Default:[/bold] {origin_default_display}",
-            title="Origin",
-            expand=False,
-        )
-    )
-    chain_lines: list[str] = []
-    for chain_entry in spec_chain:
-        val = chain_entry.get("value")
-        val_display = str(val) if val is not None else "(not set)"
-        chain_lines.append(
-            f"{chain_entry['level']}. {chain_entry['source']}: {val_display}"
-        )
-    winner_src = winner.get("source", "")
-    winner_level = winner.get("level", "")
-    chain_lines.append(f"[bold]Winner:[/bold] {winner_src} (level {winner_level})")
-
-    console.print(
-        Panel(
-            "\n".join(chain_lines),
-            title="Resolution Chain",
+            f"[bold]Value:[/bold] {resolved.value}\n"
+            f"[bold]Source:[/bold] {resolved.source.value}\n"
+            "[bold]Type:[/bold] "
+            + (type(resolved.value).__name__ if resolved.value is not None else "None"),
+            title="Configuration Value",
            expand=False,
        )
    )

-    console.print("[green]✓ OK[/green] Config read")
+    if verbose and resolved.chain:
+        console.print("\n[bold]Resolution chain[/bold] (highest → lowest priority):")
+        for chain_entry in resolved.chain:
+            src = chain_entry["source"]
+            val = chain_entry.get("value")
+            is_winner = src == resolved.source.value and val is not None
+            marker = " [green]◀ active[/green]" if is_winner else ""
+            display_val = str(val) if val is not None else "(not set)"
+            console.print(f"  {src:14s} → {display_val}{marker}")


@app.command("list")
@@ -350,13 +389,17 @@ def config_list(
 ) -> None:
    """List all configuration values.

-    Filter by key glob (e.g. ``plan.*``) and/or value regex.
-    Secrets are masked by default; use ``--show-secrets`` to reveal.
+    Optionally filter by key glob pattern (e.g. ``plan.*``, ``context.pipeline.*``)
+    and/or value regex.  Secret values (API keys, tokens, passwords) are
+    masked by default; use ``--show-secrets`` to reveal them.

    Examples::

+        agents config list
        agents config list "plan.*"
-        agents config list --filter-values "DEBUG" --show-secrets
+        agents config list "context.pipeline.*"
+        agents config list --filter-values "DEBUG"
+        agents config list --show-secrets
    """
    # Compile regex patterns (fail fast on invalid regex)
    key_re: re.Pattern[str] | None = None
@@ -377,6 +420,8 @@ def config_list(
            raise typer.Exit(code=1) from exc

    svc = _get_service()
+
+    # When --project is given, show only project-scoped overrides
    if project is not None:
        proj_overrides = svc.get_project_overrides(project)
        settings_list: list[dict[str, Any]] = []
@@ -384,14 +429,19 @@ def config_list(
            raw_value = proj_overrides[key]
            str_value = str(raw_value) if raw_value is not None else ""

+            # Key filter
            if (
                key_re is not None
                and not key_re.search(key)
                and (pattern is None or not fnmatch.fnmatch(key, pattern))
            ):
                continue
+
+            # Value filter
            if val_re is not None and not val_re.search(str_value):
                continue
+
+            # Mask secrets unless --show-secrets
            display_value: Any = raw_value
            if _is_secret_key(key) and not show_secrets and raw_value is not None:
                display_value = _mask_value(str(raw_value))
@@ -459,9 +509,12 @@ def config_list(
        if val_re is not None and not val_re.search(str_value):
            continue

+        # Modified flag: value differs from default
        default_val = defaults.get(key)
        modified = raw_value != default_val
-        display_value = raw_value
+
+        # Mask secrets unless --show-secrets
+        display_value: Any = raw_value
        if _is_secret_key(key) and not show_secrets and raw_value is not None:
            display_value = _mask_value(str(raw_value))

@@ -479,6 +532,7 @@ def config_list(
        return

    if fmt != OutputFormat.RICH.value:
+        # Serialise Path objects for JSON/YAML
        for entry in settings_list_all:
            if hasattr(entry["value"], "__fspath__"):
                entry["value"] = str(entry["value"])
@@ -125,7 +125,6 @@ BUILTIN_TYPE_NAMES: frozenset[str] = frozenset(
        "submodule",
        "symlink",
        # LSP resource types (#832)
-        "executable",
        "lsp-server",
        "lsp-workspace",
        "lsp-document",
@@ -1,51 +0,0 @@
-"""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")
@@ -142,7 +142,6 @@ 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")
@@ -294,7 +294,6 @@ 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)
@@ -373,7 +372,6 @@ 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()

@@ -422,20 +420,8 @@ 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=plan_success,
+                success=db_plan.error_message is None,  # type: ignore
                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