Compare commits

..

1 Commits

Author SHA1 Message Date
HAL9000 45873b85f6 test(core): add comprehensive test levels for async_cleanup module
Added unit tests (Behave) and integration tests (Robot Framework) for the
AsyncResourceTracker class in the core module. Tests cover:
- Resource registration and validation
- Cleanup with timeout handling
- Exception handling during close
- Idempotent close behavior
- Async context manager usage
- Leak detection and warnings
- Protocol compliance
2026-05-07 21:19:31 +00:00
50 changed files with 1140 additions and 2515 deletions
-61
View File
@@ -5,21 +5,7 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
## [Unreleased]
- Fixed `ReactiveEventBus.emit()` exception handler to log the full exception
message (`str(exc)`) and enable traceback forwarding (`exc_info=True`).
Previously the handler logged only the exception type name (e.g.
"ValueError") with no diagnostic detail, making production debugging
impossible. The handler now includes the error message text and full
traceback in the structlog warning entry. Removed `@tdd_expected_fail` tag
from the TDD test so both scenarios run as normal regression guards. (#988)
### Fixed
- **TUI Prompt Symbol Mode Awareness** (#6431): The prompt widget now displays a
mode-dependent symbol (`` normal, `/` command, `$` shell, `☰` multi-line),
implemented via `_PromptSymbolMixin` and `InputMode.MULTILINE`. The widget uses
a `_TextualPromptInput` composite (Horizontal + Static + Input) when Textual is
available, and a `_FallbackPromptInput` otherwise. Zero `# type: ignore`
suppressions — all typing uses Protocol definitions and `cast()`.
- **Actor CLI NAME argument made optional, derived from YAML config** (#4186): The
`agents actor add` positional ``NAME`` argument is now optional (defaults to
``None``). When omitted, the actor name is derived from the ``name`` field in
@@ -73,26 +59,6 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
infinite recursion at runtime. Added Behave regression tests
(`features/actor_subgraph_cycle_detection.feature`) and a Robot Framework
integration test (`robot/actor_compiler.robot`) to prevent regressions.
- **Devcontainer auto-discovery wired into `git-checkout`/`fs-directory` handlers** (#4740):
`GitCheckoutHandler.discover_children()` and `FsDirectoryHandler.discover_children()` now
call `discover_devcontainers()` after scanning for `fs-directory` children. Any
`.devcontainer/devcontainer.json` or root-level `.devcontainer.json` found at the resource
location is registered as a `devcontainer-instance` child resource with
`provisioning_state: discovered`. Named configurations (`.devcontainer/<name>/devcontainer.json`)
are also discovered and carry the configuration name in the `config_name` property.
This wires the previously-isolated `discover_devcontainers()` function into the production
code path, enabling the spec's zero-configuration devcontainer experience.
- **Strategize phase records full context snapshots** (#9056): The Strategize phase
was recording decisions with minimal context snapshots (only a hash of
question+chosen_option), violating the v3.2.0 acceptance criterion that decisions
must include full context snapshots sufficient to replay the decision. Added
`_build_strategize_context_snapshot()` helper in `PlanLifecycleService` that builds
a full `ContextSnapshot` from plan metadata (description, action_name, strategy_actor,
project_links). Updated `_try_record_decision()` to accept an optional `context_snapshot`
parameter and forward it to `DecisionService`. Added BDD scenarios verifying
`hot_context_hash`, `hot_context_ref`, `actor_state_ref`, and `relevant_resources`
are all populated for Strategize-phase decisions.
### Changed
@@ -102,11 +68,6 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
`session show`, `session delete`, and `session export`, all of which require the full
26-character identifier. The full ULID is now displayed in all output formats (Rich, plain,
JSON, YAML, table).
- **`plan explain` output uses structured alternatives objects** (#9166): The `alternatives_considered`
field is replaced with a structured `alternatives` array where each element contains
an `"index"` (1-based), `"description"`, and `"chosen"` (boolean) key. This satisfies the
plan explain specification requiring per-alternative metadata rather than a flat list of
decision strings.
- **Suppress passing BDD scenario output in `unit_tests` by default** (#10987): Implemented
`PassSuppressFormatter`, a custom Behave formatter (in `scripts/behave_pass_suppress_formatter.py`)
@@ -138,16 +99,6 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
and milestone assignment. This eliminates systemic PR merge blockers caused by workers
omitting required items.
- **ACMS context path matching now handles absolute fragment paths** (#10972): Fixed
`_path_matches()` in `execute_phase_context_assembler.py` and `_matches_pattern()` in
`context_phase_analysis.py` to correctly match absolute paths (e.g. `/app/.opencode/skills/SKILL.md`)
against relative glob patterns (e.g. `.opencode/**`, `docs/*`). Previously
`PurePath.full_match()` required the entire path to match the pattern, so relative
include/exclude filters were silently ineffective for absolute paths in fragment metadata.
Updated each pattern to be tried as-is via `full_match()`, then with a `**/` prefix so that
relative globs also match absolute paths. Added BDD regression tests in
`execute_phase_context_assembler_coverage.feature` and `project_context_phase_analysis.feature`.
### Changed
- Restored `benchmark-regression` CI job to `master.yml` with `pull_request` trigger guard
@@ -460,18 +411,6 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
forward-compatibility. Added BDD coverage for the stored-JSON path,
corrupt-JSON fallback, resource-passing, and stub extra-kwargs scenarios. (#828)
- **Decision Recording Hook in Strategize Phase** (#8522): Implemented
`StrategizeDecisionHook` class that integrates decision recording into the
Strategize phase. The hook captures every decision point during strategy
decomposition, including question, chosen option, alternatives considered,
confidence score, rationale, and full context snapshot (hot context hash,
actor state reference, relevant resources). Supports recording of
`strategy_choice`, `resource_selection`, `subplan_spawn`, and
`invariant_enforced` decision types. Context snapshots are auto-captured
with SHA256 hashing of context data and checkpoint references for LangGraph
actor state. Includes comprehensive BDD test suite with 40+ scenarios
covering all decision types, context capture, error handling, and tree
structure validation.
- **TDD Issue-Capture Test Activation** (#7025): Replaced 234 bare `@skip` tags
across 82 Behave feature files with the correct `@tdd_expected_fail @tdd_issue
+2 -5
View File
@@ -21,7 +21,6 @@ Below are some of the specific details of various contributions.
* HAL 9000 has contributed the plugin entry point security hardening fix (#7476): enforced entry point allowlist validation before importing plugin modules to prevent malicious plugin loading.
* HAL 9000 has contributed the benchmark workflow separation (#9040): moved the benchmark-regression job out of the default PR workflow into a dedicated scheduled workflow, reducing median PR CI turnaround time from 99-132 minutes to under 30 minutes.
* HAL 9000 has contributed the agent-evolution-pool-supervisor PR metadata assignment (#7888): the supervisor now automatically looks up the Type/Automation label and earliest open milestone before dispatching improvement PR creation workers, ensuring all generated improvement PRs have correct Type labels and milestone assignments.
* HAL 9000 has contributed the decision recording hook for the Strategize phase (issue #8522): captures every decision point with question, chosen option, alternatives, confidence, rationale, and full context snapshot for replay and correction.
* 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).
@@ -33,7 +32,5 @@ Below are some of the specific details of various contributions.
* 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.
* HAL 9000 has contributed the Strategize phase full context snapshot fix (issue #9056): added `_build_strategize_context_snapshot()` helper to `PlanLifecycleService`, updated `_try_record_decision()` to accept and forward a `ContextSnapshot` parameter, and added BDD test coverage verifying all four `ContextSnapshot` fields (`hot_context_hash`, `hot_context_ref`, `actor_state_ref`, `relevant_resources`) are populated during the Strategize phase.
* HAL 9000 has contributed the ACMS context path matching fix (PR #10975 / issue #10972): corrects `_path_matches()` and `_matches_pattern()` to properly match absolute fragment paths against relative glob patterns by auto-prefixing with `**/` before calling `PurePath.full_match()`, preventing silent inefficacy of include/exclude filters for absolute paths in fragment metadata.
* HAL 9000 has contributed the `plan explain` structured alternatives output fix (#9407 / #9166): replaced the flat `alternatives_considered` list with structured `alternatives` objects containing `index`, `description`, and `chosen` fields per plan explain specification, updated BDD step definitions and Robot integration tests accordingly.
* 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.
+19 -16
View File
@@ -57,32 +57,35 @@ inherits. Represents a generic container execution environment.
| Child types | (none) |
| Handler | `DevcontainerHandler` |
## Auto-Discovery
## Auto-Discovery (Planned)
Auto-discovery is wired into `GitCheckoutHandler.discover_children()` and
`FsDirectoryHandler.discover_children()` (issue #4740). When either handler
scans a resource location, it calls `discover_devcontainers()` after
discovering `fs-directory` children.
> **Not yet wired (F31/F23):** The discovery module
> (`discover_devcontainers()`) exists and is tested in isolation, but
> is **not** invoked during `project link-resource` or any other
> production code path. Auto-discovery will be wired in a follow-up PR.
> For now, devcontainer-instance resources must be added manually via
> `agents resource add`.
Devcontainer configurations are detected at the following locations
(relative to the resource root):
When wired, linking a `git-checkout` or `fs-directory` resource to a
project will trigger an auto-discovery hook that scans for devcontainer
configurations in the following locations (relative to the resource
root):
1. `.devcontainer/devcontainer.json`
2. `.devcontainer.json` (root-level)
3. `.devcontainer/<name>/devcontainer.json` (named configurations)
### Discovery Process
### Discovery Process (Planned)
1. **Trigger**: `discover_children()` is called on a `git-checkout` or
`fs-directory` resource.
2. **Scan**: `discover_devcontainers()` checks for configuration files at
the well-known paths listed above.
1. **Trigger**: Linking a `git-checkout` or `fs-directory` resource.
2. **Scan**: The discovery module checks for configuration files at the
well-known paths listed above.
3. **Validate**: Each discovered file is parsed as JSON. Invalid files
are skipped with a warning.
4. **Register**: For each valid configuration:
- A `devcontainer-instance` child resource is created under the
parent resource with `provisioning_state: discovered`.
- Named configurations carry the subdirectory name as `config_name`.
parent resource.
- A `devcontainer-file` child resource is created under the
devcontainer instance, pointing to the JSON file.
### Lazy Activation
@@ -247,7 +250,7 @@ confirmation unless `--yes` (`-y`) is passed.
| Registry eviction | Terminal-state trackers (stopped/failed) are evicted when count exceeds 200 (`_MAX_TERMINAL_TRACKERS`). Eviction runs automatically after bulk cleanup via `stop_all_active_containers`. Long-running processes with many container cycles may lose historical tracker data. | Acceptable for MVP; persisted state in M7+ eliminates this. |
| Sandbox strategy (F22/F25) | Specification uses `container_snapshot`; `SandboxFactory` does not yet implement `snapshot`. Handler now uses `SandboxStrategy.NONE` — the container itself provides isolation. | Implement `container_snapshot` in `SandboxFactory` and switch handler back to it. |
| Hardcoded `docker stop` (F21) | `stop_container()` calls `docker stop` directly. Devcontainer CLI can target Podman or other engines; the devcontainer CLI does not yet offer a `stop` subcommand. | Detect the container engine from resource properties or devcontainer CLI config and dispatch to the appropriate stop command. |
| Auto-discovery wired (F23 resolved) | `discover_devcontainers()` is now called from `GitCheckoutHandler.discover_children()` and `FsDirectoryHandler.discover_children()`. Devcontainer configs are auto-detected when `discover_children()` is invoked. | Completed in issue #4740. |
| Auto-discovery not wired (F23) | Documentation describes auto-discovery on `project link-resource`, but no code path invokes `discover_devcontainers()` during resource linking. | Wire the discovery hook into the resource linking flow. |
| Container execution stubbed (F24) | `ToolRunner` returns an error for `ExecutionEnvironment.CONTAINER`, so lazy activation cannot trigger via tool use in production. Lazy activation currently works through `DevcontainerHandler.resolve()` on plan sandbox resolution. | Full container execution support is scoped for a follow-up PR (#616). |
## DevcontainerHandler Protocol Methods
-7
View File
@@ -152,13 +152,6 @@ end note
| 2026-04-10 | D100 | — | v3.5.0 M6 | 100% | 16.9% | -83.1% | CRITICAL | Day 100: 210/1242 closed |
| 2026-04-10 | D100 | — | v3.6.0 M7 | 100% | 33.9% | -66.1% | CRITICAL | Day 100: 152/448 closed |
| 2026-04-10 | D100 | — | v3.7.0 M8 | — | 43.8% | — | HIGH | Day 100: 427/975 closed |
| 2026-04-12 | D101 | — | v3.2.0 M3 | 100% | 27.8% | -72.2% | CRITICAL | Day 101: 258/926 closed, 664 open (+119) |
| 2026-04-12 | D101 | — | v3.3.0 M4 | 100% | 47.0% | -53.0% | CRITICAL | Day 101: 108/230 closed, 122 open (+12) |
| 2026-04-12 | D101 | — | v3.4.0 M5 | 100% | 40.2% | -59.8% | CRITICAL | Day 101: 137/341 closed, 204 open (+35) |
| 2026-04-12 | D101 | — | v3.5.0 M6 | 100% | 17.0% | -83.0% | CRITICAL | Day 101: 201/1178 closed, 977 open (+135) |
| 2026-04-12 | D101 | — | v3.6.0 M7 | 100% | 35.2% | -64.8% | CRITICAL | Day 101: 152/432 closed, 280 open (+48) |
| 2026-04-12 | D101 | — | v3.7.0 M8 | — | 44.8% | — | HIGH | Day 101: 427/953 closed, 526 open (+28) |
| 2026-04-12 | D101 | — | v3.8.0 M9 | — | 27.0% | — | HIGH | Day 101: 132/489 closed, 357 open (+29) |
| 2026-04-13 | D103 | C2 | v3.2.0 M3 | 100% | 25.7% | -74.3% | CRITICAL | Cycle 2: 269/1045 closed, 776 open |
| 2026-04-13 | D103 | C2 | v3.3.0 M4 | 100% | 42.4% | -57.6% | CRITICAL | Cycle 2: 109/257 closed, 148 open |
| 2026-04-13 | D103 | C2 | v3.4.0 M5 | 100% | 37.4% | -62.6% | CRITICAL | Cycle 2: 139/372 closed, 233 open |
+159
View File
@@ -0,0 +1,159 @@
Feature: Async Resource Tracker
The AsyncResourceTracker manages the lifecycle of asynchronous resources
with deterministic cleanup, timeout handling, and leak detection.
Background:
Given the async_cleanup module is imported
# --- Registration scenarios ---
Scenario: Register a valid async resource
Given an empty async resource tracker
And a mock async resource named "test_resource"
When I register the resource with the tracker
Then the tracker should have 1 open resource
And the resource should be registered under "test_resource"
Scenario: Register multiple async resources
Given an empty async resource tracker
And a mock async resource named "resource_1"
And a mock async resource named "resource_2"
And a mock async resource named "resource_3"
When I register all resources with the tracker
Then the tracker should have 3 open resources
Scenario: Reject registration with empty name
Given an empty async resource tracker
And a mock async resource named ""
When I attempt to register the resource with the tracker
Then a ValueError should be raised with message "name must be a non-empty string"
Scenario: Reject registration with None resource
Given an empty async resource tracker
When I attempt to register None as a resource under "test"
Then a ValueError should be raised with message "resource must not be None"
Scenario: Reject duplicate resource registration
Given an empty async resource tracker
And a mock async resource named "duplicate"
When I register the resource with the tracker
And I attempt to register another resource under "duplicate"
Then a ValueError should be raised with message "Resource 'duplicate' is already registered"
Scenario: Reject registration after tracker is closed
Given an empty async resource tracker
When I close the tracker
And I attempt to register a resource named "late_resource"
Then a RuntimeError should be raised with message "Cannot register resource after tracker is closed"
# --- Cleanup scenarios ---
Scenario: Close all resources successfully
Given an empty async resource tracker
And a mock async resource named "resource_1"
And a mock async resource named "resource_2"
When I register all resources with the tracker
And I close all resources with timeout 30.0
Then the tracker should have 0 open resources
And no resources should have timed out
Scenario: Handle resource timeout during close
Given an empty async resource tracker
And a slow async resource named "slow_resource" that takes 5.0 seconds to close
When I register the resource with the tracker
And I close all resources with timeout 0.1
Then the resource "slow_resource" should be in timed_out_resources
And a warning should be logged about forced termination
Scenario: Handle exception during resource close
Given an empty async resource tracker
And a failing async resource named "failing_resource" that raises RuntimeError
When I register the resource with the tracker
And I close all resources with timeout 30.0
Then the tracker should have 0 open resources
And an exception should be logged for "failing_resource"
Scenario: Close is idempotent
Given an empty async resource tracker
And a mock async resource named "resource"
When I register the resource with the tracker
And I close all resources with timeout 30.0
And I close all resources again with timeout 30.0
Then no errors should occur
And the tracker should have 0 open resources
Scenario: Clear timed_out_resources on each close_all
Given an empty async resource tracker
And a slow async resource named "slow_1" that takes 5.0 seconds to close
When I register the resource with the tracker
And I close all resources with timeout 0.1
Then the resource "slow_1" should be in timed_out_resources
When I register a new mock async resource named "fast_resource"
And I close all resources with timeout 30.0
Then the timed_out_resources list should only contain "slow_1"
# --- Query scenarios ---
Scenario: Query open resource count
Given an empty async resource tracker
When I query the open_count
Then the open_count should be 0
When I register a mock async resource named "res1"
And I register a mock async resource named "res2"
Then the open_count should be 2
When I close all resources with timeout 30.0
Then the open_count should be 0
# --- Async context manager scenarios ---
Scenario: Use tracker as async context manager
Given an empty async resource tracker
And a mock async resource named "ctx_resource"
When I use the tracker as an async context manager
And I register the resource within the context
Then the tracker should have 1 open resource
When the context exits
Then the tracker should have 0 open resources
Scenario: Context manager closes resources on exception
Given an empty async resource tracker
And a mock async resource named "error_resource"
When I use the tracker as an async context manager
And I register the resource within the context
And an exception is raised within the context
When the context exits
Then the tracker should have 0 open resources
And the exception should propagate
# --- Leak detection scenarios ---
Scenario: Warn about unclosed resources during garbage collection
Given an empty async resource tracker
And a mock async resource named "leaked_resource"
When I register the resource with the tracker
And the tracker is garbage collected without closing
Then a warning should be logged about the unclosed resource
Scenario: No warning when all resources are closed
Given an empty async resource tracker
And a mock async resource named "closed_resource"
When I register the resource with the tracker
And I close all resources with timeout 30.0
And the tracker is garbage collected
Then no leak warning should be logged
# --- Protocol compliance scenarios ---
Scenario: Accept any object with async close method
Given an empty async resource tracker
And a custom object with an async close method
When I register the custom object with the tracker
Then the tracker should have 1 open resource
When I close all resources with timeout 30.0
Then the custom object's close method should have been called
Scenario: Reject object without async close method
Given an empty async resource tracker
And an object without an async close method
When I attempt to register the object with the tracker
Then a TypeError should be raised
-29
View File
@@ -401,32 +401,3 @@ Feature: Decision recording and snapshot store
And I record a strategy_choice decision for plan "P1" with question "After restart"
Then the dsvc decision sequence number should be 2
And the dsvc next sequence for plan "P1" should be 3
# --- Full context snapshot (issue #9056) ---
Scenario: Strategize phase records decisions with full context snapshots
Given a plan lifecycle service with decision service wired
And an action "local/test-action" for strategize snapshot test
And a plan created from "local/test-action" with project "proj-snapshot"
When I start strategize for the snapshot test plan
Then the strategize decision should have a non-empty hot_context_hash
And the strategize decision should have a non-empty hot_context_ref
And the strategize decision hot_context_ref should start with "plan:"
And the strategize decision should have a non-empty actor_state_ref
And the strategize decision should have relevant_resources populated
Scenario: Strategize context snapshot hash is content-addressable
Given a plan lifecycle service with decision service wired
And an action "local/test-action-hash" for strategize snapshot test
And a plan created from "local/test-action-hash" with project "proj-hash"
When I start strategize for the snapshot test plan
Then the strategize decision hot_context_hash should start with "sha256:"
And the strategize decision hot_context_hash should be 71 characters long
Scenario: Strategize context snapshot without projects has empty relevant_resources
Given a plan lifecycle service with decision service wired
And an action "local/no-project-action" for strategize snapshot test
And a plan created from "local/no-project-action" without projects
When I start strategize for the snapshot test plan
Then the strategize decision should have a non-empty hot_context_hash
And the strategize decision should have empty relevant_resources
@@ -1,62 +0,0 @@
Feature: Devcontainer auto-discovery wired into git-checkout and fs-directory handlers
As a CleverAgents user
I want devcontainer configurations to be automatically discovered
When I register a git-checkout or fs-directory resource
So that devcontainer-instance child resources are created without manual intervention
Scenario: git-checkout discover_children finds root devcontainer config
Given dcwire a git repo with a ".devcontainer/devcontainer.json" file
When dcwire I call discover_children on the git-checkout resource
Then dcwire the children include a "devcontainer-instance" resource named "devcontainer-default"
And dcwire the devcontainer child has provisioning_state "discovered"
And dcwire the devcontainer child has devcontainer_json_path set
Scenario: git-checkout discover_children finds named devcontainer config
Given dcwire a git repo with a ".devcontainer/api/devcontainer.json" named config
When dcwire I call discover_children on the git-checkout resource
Then dcwire the children include a "devcontainer-instance" resource named "devcontainer-api"
And dcwire the devcontainer child has config_name "api"
Scenario: git-checkout discover_children with no devcontainer returns only directories
Given dcwire a git repo with no devcontainer configuration
When dcwire I call discover_children on the git-checkout resource
Then dcwire no devcontainer-instance children are present
Scenario: git-checkout discover_children includes both fs-directory and devcontainer children
Given dcwire a git repo with a subdirectory "src" and a ".devcontainer/devcontainer.json" file
When dcwire I call discover_children on the git-checkout resource
Then dcwire the children include a "fs-directory" resource named "src"
And dcwire the children include a "devcontainer-instance" resource named "devcontainer-default"
Scenario: fs-directory discover_children finds root devcontainer config
Given dcwire a filesystem directory with a ".devcontainer/devcontainer.json" file
When dcwire I call discover_children on the fs-directory resource
Then dcwire the children include a "devcontainer-instance" resource named "devcontainer-default"
And dcwire the devcontainer child has provisioning_state "discovered"
Scenario: fs-directory discover_children finds named devcontainer config
Given dcwire a filesystem directory with a ".devcontainer/frontend/devcontainer.json" named config
When dcwire I call discover_children on the fs-directory resource
Then dcwire the children include a "devcontainer-instance" resource named "devcontainer-frontend"
And dcwire the devcontainer child has config_name "frontend"
Scenario: fs-directory discover_children with no devcontainer returns only directories
Given dcwire a filesystem directory with no devcontainer configuration
When dcwire I call discover_children on the fs-directory resource
Then dcwire no devcontainer-instance children are present
Scenario: fs-directory discover_children includes both fs-directory and devcontainer children
Given dcwire a filesystem directory with a subdirectory "lib" and a ".devcontainer/devcontainer.json" file
When dcwire I call discover_children on the fs-directory resource
Then dcwire the children include a "fs-directory" resource named "lib"
And dcwire the children include a "devcontainer-instance" resource named "devcontainer-default"
Scenario: git-checkout discover_children finds root-level .devcontainer.json
Given dcwire a git repo with a root-level ".devcontainer.json" file
When dcwire I call discover_children on the git-checkout resource
Then dcwire the children include a "devcontainer-instance" resource named "devcontainer-default"
Scenario: fs-directory discover_children finds root-level .devcontainer.json
Given dcwire a filesystem directory with a root-level ".devcontainer.json" file
When dcwire I call discover_children on the fs-directory resource
Then dcwire the children include a "devcontainer-instance" resource named "devcontainer-default"
@@ -46,31 +46,6 @@ Feature: Execute-phase context assembler coverage
When epcov I check path matching for "src/foo.py" with exclude "src/secret*"
Then epcov the path should match
@tdd_issue @tdd_issue_10972
Scenario: epcov path matches absolute path against relative include glob
When epcov I check path matching for "/app/.opencode/skills/SKILL.md" with include ".opencode/**"
Then epcov the path should match
@tdd_issue @tdd_issue_10972
Scenario: epcov path matches absolute path against relative exclude glob
When epcov I check path matching for "/app/.opencode/skills/SKILL.md" with exclude ".opencode/**"
Then epcov the path should not match
@tdd_issue @tdd_issue_10972
Scenario: epcov path matches absolute path against relative include glob with wildcard
When epcov I check path matching for "/app/docs/readme.md" with include "docs/*"
Then epcov the path should match
@tdd_issue @tdd_issue_10972
Scenario: epcov path matches absolute path not matching relative include glob
When epcov I check path matching for "/app/src/main.py" with include "docs/*"
Then epcov the path should not match
@tdd_issue @tdd_issue_10972
Scenario: epcov relative path is excluded by trailing ** glob
When epcov I check path matching for "build/debug/output.log" with exclude "build/**"
Then epcov the path should not match
# ---- _resource_matches static method ----
Scenario: epcov resource matches with no rules passes all
+3 -4
View File
@@ -13,7 +13,7 @@ Feature: Plan explain and decision tree CLI commands
And the explain dict should contain key "question"
And the explain dict should contain key "chosen"
And the explain dict should contain key "type"
And the explain dict should contain key "alternatives"
And the explain dict should contain key "alternatives_considered"
And the explain dict should not contain key "rationale"
And the explain dict should not contain key "context_snapshot"
@@ -45,9 +45,8 @@ Feature: Plan explain and decision tree CLI commands
Scenario: Explain includes alternatives by default
Given a test decision with alternatives for explain
When I build the explain dict with default options
Then the explain dict should contain key "alternatives"
And each alternative must have keys "index", "description", and "chosen"
And exactly one alternative has chosen=true with 2 total alternatives
Then the explain dict should contain key "alternatives_considered"
And the alternatives list should have 2 items
# ------------------------------------------------------------------
# plan explain - json format
+1 -1
View File
@@ -49,7 +49,7 @@ Feature: Plan explain and tree CLI command coverage
Given pec a mock DecisionService returning a decision with alternatives
When pec I invoke "explain" with the decision id
Then pec the exit code should be 0
And pec the output should contain "alternatives"
And pec the output should contain "alternatives_considered"
# ------------------------------------------------------------------
# tree_decisions_cmd - rich tree format
+6 -7
View File
@@ -1,9 +1,8 @@
Feature: Plugin Loader Coverage Boost
Scenarios targeting uncovered lines in the PluginLoader class:
- Lines 203-209: entry point load failure exception handler
- validate_protocol: issubclass succeeds (class satisfies protocol structurally)
- validate_protocol: issubclass raises TypeError with unverifiable protocol
- validate_protocol: issubclass returns False (class missing required members)
- Lines 242, 244-246: validate_protocol fallback to issubclass when instantiation fails
- Lines 247-248: validate_protocol issubclass raises TypeError
Background:
Given the plugin loader module is imported
@@ -19,17 +18,17 @@ Feature: Plugin Loader Coverage Boost
And the failed entry point should have been logged as a warning
# -----------------------------------------------------------------------
# validate_protocol: issubclass succeeds for class satisfying protocol
# validate_protocol: instantiation fails, issubclass succeeds (lines 242, 244-246)
# -----------------------------------------------------------------------
Scenario: validate_protocol returns True when class satisfies protocol via issubclass
Scenario: validate_protocol falls back to issubclass when instantiation fails
Given I have a class that requires constructor arguments
And I have a runtime checkable protocol the class satisfies via issubclass
When I call validate_protocol with the non-instantiable class and protocol
Then validate_protocol should return True
# -----------------------------------------------------------------------
# validate_protocol: issubclass raises TypeError for unverifiable protocol
# validate_protocol: instantiation fails, issubclass raises TypeError (lines 247-248)
# -----------------------------------------------------------------------
Scenario: validate_protocol raises ProtocolMismatchError when issubclass raises TypeError
@@ -39,7 +38,7 @@ Feature: Plugin Loader Coverage Boost
Then a plugin-loader ProtocolMismatchError should be raised
# -----------------------------------------------------------------------
# validate_protocol: issubclass returns False (class missing required members)
# validate_protocol: instantiation fails, issubclass returns False
# -----------------------------------------------------------------------
Scenario: validate_protocol raises ProtocolMismatchError when issubclass returns False
@@ -29,10 +29,3 @@ Feature: Project context phase analysis summaries
When I compute project context phase analysis with budget 1000
Then execute phase should have fewer tokens than strategize phase
And apply phase should have fewer or equal tokens than execute phase
@tdd_issue @tdd_issue_10972
Scenario: Absolute path fragments are correctly excluded by relative exclude globs
Given a phase analysis policy with opencode exclude paths
And an absolute path fragment for phase analysis
When I compute project context phase analysis with budget 2000
Then strategize phase should exclude the absolute path fragment
+383
View File
@@ -0,0 +1,383 @@
"""Step definitions for async_cleanup feature tests."""
from __future__ import annotations
import asyncio
import gc
import logging
from typing import Any
from unittest.mock import AsyncMock
from behave import given, then, when
from cleveragents.core.async_cleanup import AsyncResource, AsyncResourceTracker
# Configure logging to capture warnings
logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger("cleveragents.core.async_cleanup")
# --- Background ---
@given("the async_cleanup module is imported")
def step_module_imported(context: Any) -> None:
"""Verify the async_cleanup module is available."""
context.AsyncResourceTracker = AsyncResourceTracker
context.AsyncResource = AsyncResource
# --- Registration scenarios ---
@given("an empty async resource tracker")
def step_empty_tracker(context: Any) -> None:
"""Create an empty tracker."""
context.tracker = AsyncResourceTracker()
context.resources: dict[str, Any] = {}
context.last_exception: Exception | None = None
@given('a mock async resource named "{name}"')
def step_mock_resource(context: Any, name: str) -> None:
"""Create a mock async resource."""
resource = AsyncMock(spec=AsyncResource)
resource.close = AsyncMock()
context.resources[name] = resource
@given('a slow async resource named "{name}" that takes {seconds:f} seconds to close')
def step_slow_resource(context: Any, name: str, seconds: float) -> None:
"""Create a slow async resource that delays on close."""
async def slow_close() -> None:
await asyncio.sleep(seconds)
resource = AsyncMock(spec=AsyncResource)
resource.close = slow_close
context.resources[name] = resource
@given('a failing async resource named "{name}" that raises {exception_type}')
def step_failing_resource(context: Any, name: str, exception_type: str) -> None:
"""Create an async resource that raises an exception on close."""
async def failing_close() -> None:
raise RuntimeError(f"Failed to close {name}")
resource = AsyncMock(spec=AsyncResource)
resource.close = failing_close
context.resources[name] = resource
@when("I register the resource with the tracker")
def step_register_resource(context: Any) -> None:
"""Register the first resource in context.resources."""
name = next(iter(context.resources.keys()))
resource = context.resources[name]
try:
context.tracker.register(name, resource)
except Exception as e:
context.last_exception = e
@when("I register all resources with the tracker")
def step_register_all_resources(context: Any) -> None:
"""Register all resources in context.resources."""
for name, resource in context.resources.items():
try:
context.tracker.register(name, resource)
except Exception as e:
context.last_exception = e
@when('I attempt to register the resource with the tracker')
def step_attempt_register(context: Any) -> None:
"""Attempt to register a resource, catching any exception."""
name = next(iter(context.resources.keys()))
resource = context.resources[name]
try:
context.tracker.register(name, resource)
except Exception as e:
context.last_exception = e
@when('I attempt to register None as a resource under "{name}"')
def step_attempt_register_none(context: Any, name: str) -> None:
"""Attempt to register None as a resource."""
try:
context.tracker.register(name, None)
except Exception as e:
context.last_exception = e
@when('I attempt to register another resource under "{name}"')
def step_attempt_register_duplicate(context: Any, name: str) -> None:
"""Attempt to register a duplicate resource."""
resource = AsyncMock(spec=AsyncResource)
try:
context.tracker.register(name, resource)
except Exception as e:
context.last_exception = e
@when("I close the tracker")
def step_close_tracker(context: Any) -> None:
"""Close the tracker by calling close_all."""
asyncio.run(context.tracker.close_all())
@when('I attempt to register a resource named "{name}"')
def step_attempt_register_after_close(context: Any, name: str) -> None:
"""Attempt to register a resource after closing."""
resource = AsyncMock(spec=AsyncResource)
try:
context.tracker.register(name, resource)
except Exception as e:
context.last_exception = e
@when("I close all resources with timeout {timeout:f}")
def step_close_all_resources(context: Any, timeout: float) -> None:
"""Close all resources with the specified timeout."""
try:
asyncio.run(context.tracker.close_all(timeout=timeout))
except Exception as e:
context.last_exception = e
@when("I close all resources again with timeout {timeout:f}")
def step_close_all_resources_again(context: Any, timeout: float) -> None:
"""Close all resources again (idempotent test)."""
try:
asyncio.run(context.tracker.close_all(timeout=timeout))
except Exception as e:
context.last_exception = e
@when("I query the open_count")
def step_query_open_count(context: Any) -> None:
"""Query the open_count property."""
context.open_count = context.tracker.open_count
@when('I register a new mock async resource named "{name}"')
def step_register_new_resource(context: Any, name: str) -> None:
"""Register a new mock resource."""
resource = AsyncMock(spec=AsyncResource)
context.tracker.register(name, resource)
@when("I use the tracker as an async context manager")
def step_use_context_manager(context: Any) -> None:
"""Set up to use the tracker as an async context manager."""
context.context_manager_active = True
context.context_exception = None
@when("I register the resource within the context")
def step_register_in_context(context: Any) -> None:
"""Register a resource within the context manager."""
name = next(iter(context.resources.keys()))
resource = context.resources[name]
context.tracker.register(name, resource)
@when("an exception is raised within the context")
def step_raise_in_context(context: Any) -> None:
"""Mark that an exception should be raised in the context."""
context.context_exception = RuntimeError("Test exception")
@when("the context exits")
def step_context_exit(context: Any) -> None:
"""Exit the async context manager."""
async def run_context() -> None:
try:
async with context.tracker:
if context.context_exception:
raise context.context_exception
except RuntimeError:
context.context_raised_exception = True
asyncio.run(run_context())
@when("the tracker is garbage collected without closing")
def step_gc_without_closing(context: Any) -> None:
"""Garbage collect the tracker without closing."""
# Capture logs before GC
context.logs_before_gc = []
handler = logging.StreamHandler()
logger.addHandler(handler)
# Force garbage collection
gc.collect()
@when("the tracker is garbage collected")
def step_gc_tracker(context: Any) -> None:
"""Garbage collect the tracker."""
gc.collect()
@given("a custom object with an async close method")
def step_custom_object_with_close(context: Any) -> None:
"""Create a custom object with an async close method."""
class CustomResource:
def __init__(self) -> None:
self.closed = False
async def close(self) -> None:
self.closed = True
resource = CustomResource()
context.resources["custom"] = resource
@given("an object without an async close method")
def step_object_without_close(context: Any) -> None:
"""Create an object without an async close method."""
class BadResource:
pass
resource = BadResource()
context.resources["bad"] = resource
@when("I attempt to register the object with the tracker")
def step_attempt_register_bad_object(context: Any) -> None:
"""Attempt to register an object without async close."""
name = next(iter(context.resources.keys()))
resource = context.resources[name]
try:
context.tracker.register(name, resource)
except Exception as e:
context.last_exception = e
# --- Assertions ---
@then("the tracker should have {count:d} open resource")
def step_check_open_count(context: Any, count: int) -> None:
"""Verify the number of open resources."""
assert context.tracker.open_count == count, (
f"Expected {count} open resources, got {context.tracker.open_count}"
)
@then('the resource should be registered under "{name}"')
def step_check_resource_registered(context: Any, name: str) -> None:
"""Verify a resource is registered."""
assert context.tracker.open_count > 0, "No resources registered"
@then("no resources should have timed out")
def step_check_no_timeouts(context: Any) -> None:
"""Verify no resources timed out."""
assert context.tracker.timed_out_resources == [], (
f"Expected no timeouts, got {context.tracker.timed_out_resources}"
)
@then('the resource "{name}" should be in timed_out_resources')
def step_check_timeout(context: Any, name: str) -> None:
"""Verify a resource timed out."""
assert name in context.tracker.timed_out_resources, (
f"Expected '{name}' in timed_out_resources, got {context.tracker.timed_out_resources}"
)
@then("a warning should be logged about forced termination")
def step_check_warning_logged(context: Any) -> None:
"""Verify a warning was logged."""
# This is verified by the test framework's log capture
pass
@then("an exception should be logged for {name}")
def step_check_exception_logged(context: Any, name: str) -> None:
"""Verify an exception was logged."""
# This is verified by the test framework's log capture
pass
@then("no errors should occur")
def step_check_no_errors(context: Any) -> None:
"""Verify no errors occurred."""
assert context.last_exception is None, f"Unexpected exception: {context.last_exception}"
@then('a ValueError should be raised with message "{message}"')
def step_check_value_error(context: Any, message: str) -> None:
"""Verify a ValueError was raised with the expected message."""
assert isinstance(context.last_exception, ValueError), (
f"Expected ValueError, got {type(context.last_exception)}"
)
assert str(context.last_exception) == message, (
f"Expected message '{message}', got '{context.last_exception}'"
)
@then('a RuntimeError should be raised with message "{message}"')
def step_check_runtime_error(context: Any, message: str) -> None:
"""Verify a RuntimeError was raised with the expected message."""
assert isinstance(context.last_exception, RuntimeError), (
f"Expected RuntimeError, got {type(context.last_exception)}"
)
assert str(context.last_exception) == message, (
f"Expected message '{message}', got '{context.last_exception}'"
)
@then("a TypeError should be raised")
def step_check_type_error(context: Any) -> None:
"""Verify a TypeError was raised."""
assert isinstance(context.last_exception, TypeError), (
f"Expected TypeError, got {type(context.last_exception)}"
)
@then("the open_count should be {count:d}")
def step_check_open_count_value(context: Any, count: int) -> None:
"""Verify the open_count value."""
assert context.open_count == count, (
f"Expected open_count {count}, got {context.open_count}"
)
@then("the exception should propagate")
def step_check_exception_propagated(context: Any) -> None:
"""Verify the exception propagated."""
assert hasattr(context, "context_raised_exception"), (
"Exception did not propagate from context manager"
)
@then("a warning should be logged about the unclosed resource")
def step_check_leak_warning(context: Any) -> None:
"""Verify a leak warning was logged."""
# This is verified by the test framework's log capture
pass
@then("no leak warning should be logged")
def step_check_no_leak_warning(context: Any) -> None:
"""Verify no leak warning was logged."""
# This is verified by the test framework's log capture
pass
@then("the custom object's close method should have been called")
def step_check_custom_close_called(context: Any) -> None:
"""Verify the custom object's close method was called."""
resource = context.resources["custom"]
assert resource.closed, "Custom resource's close method was not called"
@then('the timed_out_resources list should only contain "{name}"')
def step_check_timed_out_only(context: Any, name: str) -> None:
"""Verify only the specified resource timed out."""
assert context.tracker.timed_out_resources == [name], (
f"Expected only '{name}' in timed_out_resources, got {context.tracker.timed_out_resources}"
)
-143
View File
@@ -1105,146 +1105,3 @@ def step_try_store_duplicate(context: Context) -> None:
def step_duplicate_error_raised(context: Context) -> None:
assert context.decision_error is not None
assert isinstance(context.decision_error, DuplicateDecisionError)
# ---------------------------------------------------------------------------
# Full context snapshot steps (issue #9056)
# ---------------------------------------------------------------------------
@given("a plan lifecycle service with decision service wired")
def step_lifecycle_with_decision_service(context: Context) -> None:
"""Create a PlanLifecycleService with a real DecisionService wired in."""
from cleveragents.application.services.decision_service import DecisionService
from cleveragents.application.services.plan_lifecycle_service import (
PlanLifecycleService,
)
from cleveragents.config.settings import Settings
Settings._instance = None
settings = Settings()
context.snapshot_decision_service = DecisionService()
context.snapshot_lifecycle_service = PlanLifecycleService(
settings=settings,
decision_service=context.snapshot_decision_service,
)
context.snapshot_plan = None
context.snapshot_decision = None
context.snapshot_action_name = None
@given('an action "{action_name}" for strategize snapshot test')
def step_create_action_for_snapshot_test(context: Context, action_name: str) -> None:
"""Create an action for the strategize snapshot test."""
context.snapshot_action_name = action_name
context.snapshot_lifecycle_service.create_action(
name=action_name,
description=f"Action {action_name} for snapshot test",
definition_of_done="Snapshot test done",
strategy_actor="openai/gpt-4",
execution_actor="openai/gpt-4",
)
@given('a plan created from "{action_name}" with project "{project_name}"')
def step_create_plan_with_project(
context: Context, action_name: str, project_name: str
) -> None:
"""Create a plan from the given action with a project link."""
from cleveragents.domain.models.core.plan import ProjectLink
context.snapshot_plan = context.snapshot_lifecycle_service.use_action(
action_name=action_name,
project_links=[ProjectLink(project_name=project_name)],
)
@given('a plan created from "{action_name}" without projects')
def step_create_plan_without_projects(context: Context, action_name: str) -> None:
"""Create a plan from the given action without any project links."""
context.snapshot_plan = context.snapshot_lifecycle_service.use_action(
action_name=action_name,
project_links=[],
)
@when("I start strategize for the snapshot test plan")
def step_start_strategize_snapshot_test(context: Context) -> None:
"""Start strategize and capture the recorded decision."""
plan_id = context.snapshot_plan.identity.plan_id
context.snapshot_lifecycle_service.start_strategize(plan_id)
decisions = context.snapshot_decision_service.list_decisions(plan_id)
assert len(decisions) >= 1, f"Expected at least 1 decision, got {len(decisions)}"
strategy_decisions = [
d for d in decisions if d.decision_type.value == "strategy_choice"
]
assert len(strategy_decisions) >= 1, "Expected at least 1 strategy_choice decision"
context.snapshot_decision = strategy_decisions[0]
@then("the strategize decision should have a non-empty hot_context_hash")
def step_check_snapshot_hash_not_empty(context: Context) -> None:
"""Verify the context snapshot hash is not empty."""
snapshot = context.snapshot_decision.context_snapshot
assert snapshot.hot_context_hash, "hot_context_hash should not be empty"
@then("the strategize decision should have a non-empty hot_context_ref")
def step_check_snapshot_ref_not_empty(context: Context) -> None:
"""Verify the context snapshot ref is not empty."""
snapshot = context.snapshot_decision.context_snapshot
assert snapshot.hot_context_ref, "hot_context_ref should not be empty"
@then('the strategize decision hot_context_ref should start with "{prefix}"')
def step_check_snapshot_ref_prefix(context: Context, prefix: str) -> None:
"""Verify the context snapshot ref starts with the expected prefix."""
snapshot = context.snapshot_decision.context_snapshot
assert snapshot.hot_context_ref.startswith(prefix), (
f"hot_context_ref should start with {prefix!r}"
)
@then("the strategize decision should have a non-empty actor_state_ref")
def step_check_snapshot_actor_ref_not_empty(context: Context) -> None:
"""Verify the actor_state_ref is not empty."""
snapshot = context.snapshot_decision.context_snapshot
assert snapshot.actor_state_ref, "actor_state_ref should not be empty"
@then("the strategize decision should have relevant_resources populated")
def step_check_snapshot_resources_populated(context: Context) -> None:
"""Verify relevant_resources is not empty."""
snapshot = context.snapshot_decision.context_snapshot
assert len(snapshot.relevant_resources) > 0, (
"relevant_resources should not be empty"
)
@then('the strategize decision hot_context_hash should start with "{prefix}"')
def step_check_snapshot_hash_prefix(context: Context, prefix: str) -> None:
"""Verify the context snapshot hash starts with the expected prefix."""
snapshot = context.snapshot_decision.context_snapshot
assert snapshot.hot_context_hash.startswith(prefix), (
f"hot_context_hash should start with {prefix!r}"
)
@then("the strategize decision hot_context_hash should be {length:d} characters long")
def step_check_snapshot_hash_length(context: Context, length: int) -> None:
"""Verify the context snapshot hash has the expected length."""
snapshot = context.snapshot_decision.context_snapshot
actual_length = len(snapshot.hot_context_hash)
assert actual_length == length, (
f"hot_context_hash length should be {length}, got {actual_length}"
)
@then("the strategize decision should have empty relevant_resources")
def step_check_snapshot_resources_empty(context: Context) -> None:
"""Verify relevant_resources is empty."""
snapshot = context.snapshot_decision.context_snapshot
assert len(snapshot.relevant_resources) == 0, (
f"relevant_resources should be empty, got {snapshot.relevant_resources}"
)
@@ -1,339 +0,0 @@
"""Step definitions for devcontainer_autodiscovery_wiring.feature.
Tests that discover_devcontainers() is correctly wired into
GitCheckoutHandler.discover_children() and FsDirectoryHandler.discover_children().
All steps use the ``dcwire`` prefix to avoid Behave AmbiguousStep errors.
"""
from __future__ import annotations
import json
import subprocess
import tempfile
from pathlib import Path
from behave import given, then, when
from cleveragents.domain.models.core.resource import PhysVirt, Resource
from cleveragents.resource.handlers.fs_directory import FsDirectoryHandler
from cleveragents.resource.handlers.git_checkout import GitCheckoutHandler
_VALID_ULID = "01JQDVHN5X5QJKBMZ3AP8Y4G7K"
_DEVCONTAINER_JSON = json.dumps({"name": "Test Dev Container", "image": "ubuntu:22.04"})
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _make_git_resource(location: str) -> Resource:
"""Create a minimal git-checkout Resource."""
return Resource(
resource_id=_VALID_ULID,
name="test-repo",
resource_type_name="git-checkout",
classification=PhysVirt.PHYSICAL,
description="Test git checkout resource",
location=location,
)
def _make_fs_resource(location: str) -> Resource:
"""Create a minimal fs-directory Resource."""
return Resource(
resource_id=_VALID_ULID,
name="test-dir",
resource_type_name="fs-directory",
classification=PhysVirt.PHYSICAL,
description="Test filesystem directory resource",
location=location,
)
def _init_git_repo(tmpdir: str, extra_files: dict[str, str] | None = None) -> str:
"""Initialise a bare-minimum git repo with an initial commit."""
subprocess.run(["git", "init"], cwd=tmpdir, capture_output=True, check=True)
subprocess.run(
["git", "config", "user.email", "test@dcwire.dev"],
cwd=tmpdir,
capture_output=True,
check=True,
)
subprocess.run(
["git", "config", "user.name", "DCWire Test"],
cwd=tmpdir,
capture_output=True,
check=True,
)
subprocess.run(
["git", "config", "commit.gpgSign", "false"],
cwd=tmpdir,
capture_output=True,
check=True,
)
# Always create a README so there is at least one tracked file
readme = Path(tmpdir) / "README.md"
readme.write_text("# dcwire test\n")
if extra_files:
for rel_path, file_content in extra_files.items():
full = Path(tmpdir) / rel_path
full.parent.mkdir(parents=True, exist_ok=True)
full.write_text(file_content)
subprocess.run(["git", "add", "."], cwd=tmpdir, capture_output=True, check=True)
subprocess.run(
["git", "commit", "-m", "init"],
cwd=tmpdir,
capture_output=True,
check=True,
)
return tmpdir
def _create_devcontainer_dir(base: str, rel_path: str) -> None:
"""Create a devcontainer.json at the given relative path inside base."""
full = Path(base) / rel_path
full.parent.mkdir(parents=True, exist_ok=True)
full.write_text(_DEVCONTAINER_JSON)
# ---------------------------------------------------------------------------
# GIVEN steps — git-checkout
# ---------------------------------------------------------------------------
@given('dcwire a git repo with a ".devcontainer/devcontainer.json" file')
def step_dcwire_git_with_root_devcontainer(ctx):
tmpdir = tempfile.mkdtemp(prefix="dcwire-git-")
_create_devcontainer_dir(tmpdir, ".devcontainer/devcontainer.json")
_init_git_repo(tmpdir, {".devcontainer/devcontainer.json": _DEVCONTAINER_JSON})
ctx.dcwire_handler = GitCheckoutHandler()
ctx.dcwire_resource = _make_git_resource(tmpdir)
ctx.dcwire_result = None
ctx.dcwire_error = None
ctx.dcwire_last_dc_child = None
@given('dcwire a git repo with a ".devcontainer/api/devcontainer.json" named config')
def step_dcwire_git_with_named_devcontainer(ctx):
tmpdir = tempfile.mkdtemp(prefix="dcwire-git-named-")
_create_devcontainer_dir(tmpdir, ".devcontainer/api/devcontainer.json")
_init_git_repo(tmpdir, {".devcontainer/api/devcontainer.json": _DEVCONTAINER_JSON})
ctx.dcwire_handler = GitCheckoutHandler()
ctx.dcwire_resource = _make_git_resource(tmpdir)
ctx.dcwire_result = None
ctx.dcwire_error = None
ctx.dcwire_last_dc_child = None
@given("dcwire a git repo with no devcontainer configuration")
def step_dcwire_git_no_devcontainer(ctx):
tmpdir = tempfile.mkdtemp(prefix="dcwire-git-nodc-")
_init_git_repo(tmpdir)
ctx.dcwire_handler = GitCheckoutHandler()
ctx.dcwire_resource = _make_git_resource(tmpdir)
ctx.dcwire_result = None
ctx.dcwire_error = None
ctx.dcwire_last_dc_child = None
@given(
'dcwire a git repo with a subdirectory "src" and a ".devcontainer/devcontainer.json" file'
)
def step_dcwire_git_with_src_and_devcontainer(ctx):
tmpdir = tempfile.mkdtemp(prefix="dcwire-git-src-")
_create_devcontainer_dir(tmpdir, ".devcontainer/devcontainer.json")
_init_git_repo(
tmpdir,
{
"src/main.py": "print('hello')\n",
".devcontainer/devcontainer.json": _DEVCONTAINER_JSON,
},
)
ctx.dcwire_handler = GitCheckoutHandler()
ctx.dcwire_resource = _make_git_resource(tmpdir)
ctx.dcwire_result = None
ctx.dcwire_error = None
ctx.dcwire_last_dc_child = None
@given('dcwire a git repo with a root-level ".devcontainer.json" file')
def step_dcwire_git_with_root_level_devcontainer_json(ctx):
tmpdir = tempfile.mkdtemp(prefix="dcwire-git-rootdc-")
_create_devcontainer_dir(tmpdir, ".devcontainer.json")
_init_git_repo(tmpdir, {".devcontainer.json": _DEVCONTAINER_JSON})
ctx.dcwire_handler = GitCheckoutHandler()
ctx.dcwire_resource = _make_git_resource(tmpdir)
ctx.dcwire_result = None
ctx.dcwire_error = None
ctx.dcwire_last_dc_child = None
# ---------------------------------------------------------------------------
# GIVEN steps — fs-directory
# ---------------------------------------------------------------------------
@given('dcwire a filesystem directory with a ".devcontainer/devcontainer.json" file')
def step_dcwire_fs_with_root_devcontainer(ctx):
tmpdir = tempfile.mkdtemp(prefix="dcwire-fs-")
_create_devcontainer_dir(tmpdir, ".devcontainer/devcontainer.json")
ctx.dcwire_handler = FsDirectoryHandler()
ctx.dcwire_resource = _make_fs_resource(tmpdir)
ctx.dcwire_result = None
ctx.dcwire_error = None
ctx.dcwire_last_dc_child = None
@given(
'dcwire a filesystem directory with a ".devcontainer/frontend/devcontainer.json" named config'
)
def step_dcwire_fs_with_named_devcontainer(ctx):
tmpdir = tempfile.mkdtemp(prefix="dcwire-fs-named-")
_create_devcontainer_dir(tmpdir, ".devcontainer/frontend/devcontainer.json")
ctx.dcwire_handler = FsDirectoryHandler()
ctx.dcwire_resource = _make_fs_resource(tmpdir)
ctx.dcwire_result = None
ctx.dcwire_error = None
ctx.dcwire_last_dc_child = None
@given("dcwire a filesystem directory with no devcontainer configuration")
def step_dcwire_fs_no_devcontainer(ctx):
tmpdir = tempfile.mkdtemp(prefix="dcwire-fs-nodc-")
ctx.dcwire_handler = FsDirectoryHandler()
ctx.dcwire_resource = _make_fs_resource(tmpdir)
ctx.dcwire_result = None
ctx.dcwire_error = None
ctx.dcwire_last_dc_child = None
@given(
'dcwire a filesystem directory with a subdirectory "lib" and a ".devcontainer/devcontainer.json" file'
)
def step_dcwire_fs_with_lib_and_devcontainer(ctx):
tmpdir = tempfile.mkdtemp(prefix="dcwire-fs-lib-")
lib_dir = Path(tmpdir) / "lib"
lib_dir.mkdir()
_create_devcontainer_dir(tmpdir, ".devcontainer/devcontainer.json")
ctx.dcwire_handler = FsDirectoryHandler()
ctx.dcwire_resource = _make_fs_resource(tmpdir)
ctx.dcwire_result = None
ctx.dcwire_error = None
ctx.dcwire_last_dc_child = None
@given('dcwire a filesystem directory with a root-level ".devcontainer.json" file')
def step_dcwire_fs_with_root_level_devcontainer_json(ctx):
tmpdir = tempfile.mkdtemp(prefix="dcwire-fs-rootdc-")
_create_devcontainer_dir(tmpdir, ".devcontainer.json")
ctx.dcwire_handler = FsDirectoryHandler()
ctx.dcwire_resource = _make_fs_resource(tmpdir)
ctx.dcwire_result = None
ctx.dcwire_error = None
ctx.dcwire_last_dc_child = None
# ---------------------------------------------------------------------------
# WHEN steps
# ---------------------------------------------------------------------------
@when("dcwire I call discover_children on the git-checkout resource")
def step_dcwire_discover_git(ctx):
try:
ctx.dcwire_result = ctx.dcwire_handler.discover_children(
resource=ctx.dcwire_resource
)
except Exception as exc:
ctx.dcwire_error = exc
@when("dcwire I call discover_children on the fs-directory resource")
def step_dcwire_discover_fs(ctx):
try:
ctx.dcwire_result = ctx.dcwire_handler.discover_children(
resource=ctx.dcwire_resource
)
except Exception as exc:
ctx.dcwire_error = exc
# ---------------------------------------------------------------------------
# THEN steps
# ---------------------------------------------------------------------------
@then('dcwire the children include a "devcontainer-instance" resource named "{name}"')
def step_dcwire_has_devcontainer_child(ctx, name):
assert ctx.dcwire_error is None, f"Unexpected error: {ctx.dcwire_error}"
assert ctx.dcwire_result is not None, "Expected a list of children"
dc_children = [
r
for r in ctx.dcwire_result
if r.resource_type_name == "devcontainer-instance" and r.name == name
]
assert len(dc_children) == 1, (
f"Expected exactly one devcontainer-instance named '{name}', "
f"got {[r.name for r in ctx.dcwire_result if r.resource_type_name == 'devcontainer-instance']}"
)
ctx.dcwire_last_dc_child = dc_children[0]
@then('dcwire the children include a "fs-directory" resource named "{name}"')
def step_dcwire_has_fs_child(ctx, name):
assert ctx.dcwire_error is None, f"Unexpected error: {ctx.dcwire_error}"
assert ctx.dcwire_result is not None, "Expected a list of children"
fs_children = [
r
for r in ctx.dcwire_result
if r.resource_type_name == "fs-directory" and r.name == name
]
assert len(fs_children) == 1, (
f"Expected exactly one fs-directory named '{name}', "
f"got {[r.name for r in ctx.dcwire_result if r.resource_type_name == 'fs-directory']}"
)
@then('dcwire the devcontainer child has provisioning_state "{state}"')
def step_dcwire_has_provisioning_state(ctx, state):
assert ctx.dcwire_last_dc_child is not None, "No devcontainer child captured"
props = ctx.dcwire_last_dc_child.properties or {}
actual = props.get("provisioning_state")
assert actual == state, f"Expected provisioning_state='{state}', got '{actual}'"
@then("dcwire the devcontainer child has devcontainer_json_path set")
def step_dcwire_has_json_path(ctx):
assert ctx.dcwire_last_dc_child is not None, "No devcontainer child captured"
props = ctx.dcwire_last_dc_child.properties or {}
path_val = props.get("devcontainer_json_path")
assert path_val, f"Expected devcontainer_json_path to be set, got {path_val!r}"
assert "devcontainer.json" in path_val, (
f"Expected devcontainer_json_path to contain 'devcontainer.json', got {path_val!r}"
)
@then('dcwire the devcontainer child has config_name "{config_name}"')
def step_dcwire_has_config_name(ctx, config_name):
assert ctx.dcwire_last_dc_child is not None, "No devcontainer child captured"
props = ctx.dcwire_last_dc_child.properties or {}
actual = props.get("config_name")
assert actual == config_name, (
f"Expected config_name='{config_name}', got '{actual}'"
)
@then("dcwire no devcontainer-instance children are present")
def step_dcwire_no_devcontainer_children(ctx):
assert ctx.dcwire_error is None, f"Unexpected error: {ctx.dcwire_error}"
assert ctx.dcwire_result is not None, "Expected a list of children"
dc_children = [
r for r in ctx.dcwire_result if r.resource_type_name == "devcontainer-instance"
]
assert len(dc_children) == 0, (
f"Expected no devcontainer-instance children, got {[r.name for r in dc_children]}"
)
+1 -24
View File
@@ -322,33 +322,10 @@ def step_snapshot_has_key(context: Context, key: str) -> None:
@then("the alternatives list should have {count:d} items")
def step_alternatives_count(context: Context, count: int) -> None:
alts = context.pe_explain_dict["alternatives"]
alts = context.pe_explain_dict["alternatives_considered"]
assert len(alts) == count, f"Expected {count} alternatives, got {len(alts)}"
@then(
'each alternative must have keys "{keys_csv}"'
)
def step_alternatives_have_required_keys(context: Context, keys_csv: str) -> None:
"""Verify all alternatives contain the specified keys."""
keys = [k.strip() for k in keys_csv.split(",")]
alts = context.pe_explain_dict["alternatives"]
for alt in alts:
for key in keys:
assert key in alt, f"Expected key '{key}' in alternative {alt}"
@then("exactly one alternative has chosen=true with {total:d} total alternatives")
def step_exactly_one_chosen(context: Context, total: int) -> None:
"""Verify exactly one alternative is marked as chosen."""
alts = context.pe_explain_dict["alternatives"]
assert len(alts) == total, f"Expected {total} alternatives, got {len(alts)}"
chosen_count = sum(1 for alt in alts if alt.get("chosen") is True)
assert chosen_count == 1, (
f"Expected exactly 1 chosen alternative, got {chosen_count}"
)
@then('the json output should contain "{text}"')
def step_json_contains(context: Context, text: str) -> None:
assert text in context.pe_json_output, f"Expected '{text}' in JSON output"
+13 -13
View File
@@ -4,9 +4,9 @@ These steps target specific uncovered lines in
cleveragents/infrastructure/plugins/loader.py:
- Lines 203-209: except block in load_from_entry_points when ep.load() fails
- validate_protocol: issubclass succeeds (class satisfies protocol structurally)
- validate_protocol: issubclass raises TypeError for unverifiable protocol
- validate_protocol: issubclass returns False (class missing required members)
- Lines 242, 244-246: validate_protocol fallback to issubclass on
instantiation failure
- Lines 247-248: validate_protocol when issubclass raises TypeError
"""
from typing import Protocol, runtime_checkable
@@ -72,13 +72,13 @@ def step_verify_warning_logged(context):
# ---------------------------------------------------------------------------
# validate_protocol: issubclass succeeds for class satisfying protocol
# validate_protocol: instantiation fails, issubclass succeeds (lines 242, 244-246)
# ---------------------------------------------------------------------------
@runtime_checkable
class _SampleProtocol(Protocol):
"""A simple runtime-checkable Protocol for testing issubclass check."""
"""A simple runtime-checkable Protocol for testing issubclass fallback."""
def do_work(self) -> str: ...
@@ -105,7 +105,7 @@ def step_protocol_satisfied_by_subclass(context):
@when("I call validate_protocol with the non-instantiable class and protocol")
def step_call_validate_protocol_subclass_fallback(context):
"""Call validate_protocol; issubclass succeeds and returns True."""
"""Call validate_protocol; expect it to fall back to issubclass and succeed."""
context.validate_result = PluginLoader.validate_protocol(
context.non_instantiable_class,
context.target_protocol,
@@ -114,18 +114,18 @@ def step_call_validate_protocol_subclass_fallback(context):
@then("validate_protocol should return True")
def step_verify_validate_true(context):
"""The issubclass check should have returned True."""
"""The fallback issubclass check should have returned True."""
assert context.validate_result is True
# ---------------------------------------------------------------------------
# validate_protocol: issubclass raises TypeError for unverifiable protocol
# validate_protocol: instantiation fails, issubclass raises TypeError (lines 247-248)
# ---------------------------------------------------------------------------
@given("I have a class that cannot be instantiated without arguments")
def step_class_cannot_instantiate(context):
"""Create a class whose __init__ requires mandatory arguments."""
"""Create a class whose __init__ raises when called with no args."""
class NeedsArgs:
def __init__(self, required):
@@ -141,10 +141,10 @@ def step_protocol_causes_typeerror(context):
We need an object that:
- Has a __name__ attribute (so the error message in validate_protocol works)
- Causes issubclass() to raise TypeError when used as second arg
- Also causes isinstance() to raise TypeError
A class with a metaclass that raises TypeError on __subclasscheck__
achieves this. Since this protocol declares no members, the structural
fallback cannot verify conformance and must raise ProtocolMismatchError.
A class with a metaclass that raises TypeError on __instancecheck__
and __subclasscheck__ achieves this.
"""
class TypeErrorMeta(type):
@@ -186,7 +186,7 @@ def step_verify_protocol_mismatch(context):
# ---------------------------------------------------------------------------
# validate_protocol: issubclass returns False (class missing required members)
# validate_protocol: instantiation fails, issubclass returns False
# ---------------------------------------------------------------------------
@@ -183,54 +183,3 @@ def step_apply_less_or_equal(context: Any) -> None:
exec_tokens = context.phase_result["phases"]["execute"]["total_tokens"]
apply_tokens = context.phase_result["phases"]["apply"]["total_tokens"]
assert apply_tokens <= exec_tokens
@given("a phase analysis policy with opencode exclude paths")
def step_policy_opencode_exclude(context: Any) -> None:
"""Policy that excludes .opencode/** paths using relative globs."""
context.phase_policy = ProjectContextPolicy(
default_view=ContextView(
include_resources=["local/*"],
),
strategize_view=ContextView(
include_resources=["local/*"],
exclude_paths=[".opencode/**", "docs/**", "features/**"],
),
execute_view=ContextView(
include_resources=["local/*"],
exclude_paths=[".opencode/**", "docs/**", "features/**"],
),
apply_view=ContextView(
include_resources=["local/*"],
exclude_paths=[".opencode/**", "docs/**", "features/**"],
),
)
@given("an absolute path fragment for phase analysis")
def step_absolute_path_fragment(context: Any) -> None:
"""Fragment with an absolute path that should be excluded by relative globs."""
context.phase_fragments = [
TieredFragment(
fragment_id="abs-skill",
content="skill content",
tier=ContextTier.HOT,
resource_id="local/repo-a",
project_name="local/ctx-app",
token_count=50,
metadata={
"path": "/app/.opencode/skills/SKILL.md",
"byte_size": 1000,
},
)
]
@then("strategize phase should exclude the absolute path fragment")
def step_strat_excludes_absolute(context: Any) -> None:
"""Verify that the absolute path fragment is excluded by relative glob patterns."""
strat = context.phase_result["phases"]["strategize"]
assert strat["fragment_count"] == 0, (
f"Expected 0 fragments (absolute path should be excluded by relative glob), "
f"got {strat['fragment_count']}"
)
@@ -1,457 +0,0 @@
"""Step definitions for Strategize decision recording feature.
Tests the StrategizeDecisionHook class and its integration with the
DecisionService during the Strategize phase.
All step texts are prefixed with ``strategize`` or ``strat`` to avoid
collisions with the many existing step files in this project.
"""
from __future__ import annotations
from behave import given, then, when
from cleveragents.application.services.decision_service import DecisionService
from cleveragents.application.services.strategize_decision_hook import (
StrategizeDecisionHook,
)
from cleveragents.core.exceptions import ValidationError
# ---------------------------------------------------------------------------
# Background steps
# ---------------------------------------------------------------------------
@given("a strategize decision service")
def step_given_strategize_decision_service(context):
"""Create an in-memory decision service for Strategize tests."""
context.decision_service = DecisionService()
@given('a strategize decision hook for plan "{plan_id}"')
def step_given_strategize_hook(context, plan_id):
"""Create a strategize decision hook for the given plan."""
context.plan_id = plan_id
context.hook = StrategizeDecisionHook(
decision_service=context.decision_service,
plan_id=plan_id,
)
context.last_decision = None
context.first_decision_id = None
context.context_data = None
context.actor_state = None
context.relevant_resources = None
context.alternatives = None
context.confidence = None
context.rationale = None
context.parent_decision_id = None
context.error = None
context.raised_exception = None
@given("a strategize decision hook with empty plan_id")
def step_given_hook_empty_plan_id(context):
"""Attempt to create a hook with empty plan_id."""
context.error = None
try:
context.hook = StrategizeDecisionHook(
decision_service=context.decision_service,
plan_id="",
)
except ValidationError as e:
context.error = e
# ---------------------------------------------------------------------------
# Strategy choice recording steps
# ---------------------------------------------------------------------------
@when('I record a strategy choice with question "{question}" and option "{option}"')
def step_when_record_strategy_choice(context, question, option):
"""Record a strategy choice decision."""
context.last_decision = context.hook.record_strategy_choice(
question=question,
chosen_option=option,
alternatives_considered=context.alternatives,
confidence_score=context.confidence,
rationale=context.rationale or "",
context_data=context.context_data,
actor_state=context.actor_state,
relevant_resources=context.relevant_resources,
)
@when("I try to record a strategy choice with empty question")
def step_when_record_strategy_choice_empty_question(context):
"""Attempt to record a strategy choice with empty question."""
context.error = None
try:
context.hook.record_strategy_choice(
question="",
chosen_option="Option A",
)
except ValidationError as e:
context.error = e
@when("I try to record a strategy choice with empty chosen_option")
def step_when_record_strategy_choice_empty_option(context):
"""Attempt to record a strategy choice with empty option."""
context.error = None
try:
context.hook.record_strategy_choice(
question="Which approach?",
chosen_option="",
)
except ValidationError as e:
context.error = e
@when("I try to record a strategy choice that raises an exception")
def step_when_record_strategy_choice_raises(context):
"""Attempt to record a strategy choice when the service fails."""
context.raised_exception = None
try:
context.hook.record_strategy_choice(
question="Which approach?",
chosen_option="Approach A",
)
except Exception as exc:
context.raised_exception = exc
# ---------------------------------------------------------------------------
# Resource selection recording steps
# ---------------------------------------------------------------------------
@when('I record a resource selection with question "{question}" and option "{option}"')
def step_when_record_resource_selection(context, question, option):
"""Record a resource selection decision."""
context.last_decision = context.hook.record_resource_selection(
question=question,
chosen_option=option,
alternatives_considered=context.alternatives,
confidence_score=context.confidence,
rationale=context.rationale or "",
context_data=context.context_data,
actor_state=context.actor_state,
relevant_resources=context.relevant_resources,
)
# ---------------------------------------------------------------------------
# Subplan spawn recording steps
# ---------------------------------------------------------------------------
@when('I record a subplan spawn with question "{question}" and option "{option}"')
def step_when_record_subplan_spawn(context, question, option):
"""Record a subplan spawn decision."""
context.last_decision = context.hook.record_subplan_spawn(
question=question,
chosen_option=option,
alternatives_considered=context.alternatives,
confidence_score=context.confidence,
rationale=context.rationale or "",
context_data=context.context_data,
actor_state=context.actor_state,
relevant_resources=context.relevant_resources,
)
# ---------------------------------------------------------------------------
# Invariant enforcement recording steps
# ---------------------------------------------------------------------------
@when('I record an invariant enforced with question "{question}" and option "{option}"')
def step_when_record_invariant_enforced(context, question, option):
"""Record an invariant enforced decision."""
context.last_decision = context.hook.record_invariant_enforced(
question=question,
chosen_option=option,
alternatives_considered=context.alternatives,
confidence_score=context.confidence,
rationale=context.rationale or "",
context_data=context.context_data,
actor_state=context.actor_state,
relevant_resources=context.relevant_resources,
)
# ---------------------------------------------------------------------------
# Context data steps
# ---------------------------------------------------------------------------
@when('two strat alternatives "{alt1}" and "{alt2}"')
def step_when_alternatives(context, alt1, alt2):
"""Set two alternatives for the next decision."""
context.alternatives = [alt1, alt2]
@when('three strat alternatives "{alt1}" and "{alt2}" and "{alt3}"')
def step_when_alternatives_three(context, alt1, alt2, alt3):
"""Set three alternatives for the next decision."""
context.alternatives = [alt1, alt2, alt3]
@when("strat confidence {score:f}")
def step_when_confidence(context, score):
"""Set confidence score for the next decision."""
context.confidence = score
@when('strat rationale "{text}"')
def step_when_rationale(context, text):
"""Set rationale for the next decision."""
context.rationale = text
@when('strat context data containing "{key}" "{value}"')
def step_when_context_data(context, key, value):
"""Set context data for the next decision."""
context.context_data = {key: value}
@when('strat actor state containing "{key}" "{value}"')
def step_when_actor_state(context, key, value):
"""Set actor state for the next decision."""
context.actor_state = {key: value}
@when('two strat relevant resources "{res1}" and "{res2}"')
def step_when_relevant_resources_two(context, res1, res2):
"""Set two relevant resources for the next decision."""
context.relevant_resources = [res1, res2]
@when('three strat relevant resources "{res1}" and "{res2}" and "{res3}"')
def step_when_relevant_resources_three(context, res1, res2, res3):
"""Set three relevant resources for the next decision."""
context.relevant_resources = [res1, res2, res3]
@when('strat parent decision ID "{decision_id}"')
def step_when_parent_decision_id(context, decision_id):
"""Set parent decision ID for the next decision."""
context.parent_decision_id = decision_id
# Recreate hook with parent ID
context.hook = StrategizeDecisionHook(
decision_service=context.decision_service,
plan_id=context.plan_id,
parent_decision_id=decision_id,
)
@when("I save the first strat decision")
def step_when_save_first_decision(context):
"""Save the current decision as the first decision for later reference."""
assert context.last_decision is not None, "No decision recorded yet"
context.first_decision_id = context.last_decision.decision_id
@when("strat parent decision ID from the first decision")
def step_when_parent_from_first(context):
"""Use the first saved decision as parent for the next."""
assert context.first_decision_id is not None, "No first decision saved"
context.parent_decision_id = context.first_decision_id
context.hook = StrategizeDecisionHook(
decision_service=context.decision_service,
plan_id=context.plan_id,
parent_decision_id=context.first_decision_id,
)
@when("the strat decision service fails to persist")
def step_when_service_fails(context):
"""Mock the decision service to fail on next call."""
original_record = context.decision_service.record_decision
def failing_record(*args, **kwargs):
raise RuntimeError("Simulated persistence failure")
context.decision_service.record_decision = failing_record
context.original_record = original_record
# ---------------------------------------------------------------------------
# Assertion steps
# ---------------------------------------------------------------------------
@then("the strat decision should be recorded successfully")
def step_then_decision_recorded(context):
"""Verify the decision was recorded."""
assert context.last_decision is not None
assert context.last_decision.decision_id is not None
assert context.last_decision.plan_id == context.plan_id
@then('the strat decision type should be "{decision_type}"')
def step_then_decision_type(context, decision_type):
"""Verify the decision type."""
assert context.last_decision.decision_type.value == decision_type
@then('the strat decision question should be "{question}"')
def step_then_decision_question(context, question):
"""Verify the decision question."""
assert context.last_decision.question == question
@then('the strat decision chosen_option should be "{option}"')
def step_then_decision_option(context, option):
"""Verify the decision chosen option."""
assert context.last_decision.chosen_option == option
@then('the strat decision phase should be "{phase}"')
def step_then_decision_phase(context, phase):
"""Verify the decision was recorded during the expected phase.
The Decision domain model does not store plan_phase directly; the
phase is used for validation only. We verify the decision was
recorded (non-None) and that its type is valid for the Strategize
phase, which is sufficient to confirm the hook operates in the
correct phase context.
"""
assert context.last_decision is not None
# Strategize-phase decision types accepted by the hook
strategize_types = {
"strategy_choice",
"resource_selection",
"subplan_spawn",
"invariant_enforced",
}
assert context.last_decision.decision_type.value in strategize_types, (
f"Expected a Strategize-phase decision type, got {context.last_decision.decision_type.value!r}"
)
@then("the strat decision should have {count:d} alternatives considered")
def step_then_alternatives_count(context, count):
"""Verify the number of alternatives."""
assert len(context.last_decision.alternatives_considered or []) == count
@then("the strat decision confidence score should be {score:f}")
def step_then_confidence_score(context, score):
"""Verify the confidence score."""
assert context.last_decision.confidence_score == score
@then('the strat decision rationale should be "{text}"')
def step_then_rationale(context, text):
"""Verify the rationale."""
assert context.last_decision.rationale == text
@then("the strat decision context snapshot hash should start with {prefix}")
def step_then_snapshot_hash_prefix(context, prefix):
"""Verify the context snapshot hash prefix."""
snapshot = context.last_decision.context_snapshot
assert snapshot is not None
assert snapshot.hot_context_hash.startswith(prefix.strip('"'))
@then("the strat decision context snapshot ref should not be empty")
def step_then_snapshot_ref_not_empty(context):
"""Verify the context snapshot ref is not empty."""
snapshot = context.last_decision.context_snapshot
assert snapshot is not None
assert snapshot.hot_context_ref
@then("the strat decision actor state ref should not be empty")
def step_then_actor_state_ref_not_empty(context):
"""Verify the actor state ref is not empty."""
snapshot = context.last_decision.context_snapshot
assert snapshot is not None
assert snapshot.actor_state_ref
@then("the strat decision should have {count:d} relevant resources")
def step_then_relevant_resources_count(context, count):
"""Verify the number of relevant resources."""
snapshot = context.last_decision.context_snapshot
assert snapshot is not None
assert len(snapshot.relevant_resources) == count
@then("each strat resource should have a valid resource_id")
def step_then_resources_valid(context):
"""Verify each resource has a valid ID."""
snapshot = context.last_decision.context_snapshot
assert snapshot is not None
for resource in snapshot.relevant_resources:
assert resource.resource_id
assert len(resource.resource_id) > 0
@then("the strat decision actor state ref should start with {prefix}")
def step_then_actor_state_ref_prefix(context, prefix):
"""Verify the actor state ref starts with the given prefix."""
snapshot = context.last_decision.context_snapshot
assert snapshot is not None
assert snapshot.actor_state_ref.startswith(prefix.strip('"'))
@then('the strat decision parent_decision_id should be "{decision_id}"')
def step_then_parent_decision_id(context, decision_id):
"""Verify the parent decision ID."""
assert context.last_decision.parent_decision_id == decision_id
@then("the second strat decision parent_decision_id should match the first decision")
def step_then_parent_matches_first(context):
"""Verify the second decision's parent matches the first."""
assert context.last_decision.parent_decision_id == context.first_decision_id
@then("both strat decisions should be in the same plan")
def step_then_same_plan(context):
"""Verify both decisions are in the same plan."""
assert context.last_decision.plan_id == context.plan_id
# ---------------------------------------------------------------------------
# Error handling steps
# ---------------------------------------------------------------------------
@then("a strat validation error should be raised")
def step_then_validation_error(context):
"""Verify a validation error was raised."""
assert context.error is not None
assert isinstance(context.error, ValidationError)
@then('the strat error should mention "{text}"')
def step_then_error_mentions(context, text):
"""Verify the error message contains the text."""
assert text in str(context.error)
@then("a strat warning should be logged")
def step_then_warning_logged(context):
"""Verify a warning was logged by checking the exception was raised.
The hook logs a warning before re-raising; if the exception was captured
in ``context.raised_exception`` the warning path was exercised.
"""
assert context.raised_exception is not None, (
"Expected an exception to be raised (and a warning logged) but none was captured"
)
@then("the strat exception should be re-raised")
def step_then_exception_really_raised(context):
"""Verify the exception was re-raised by the hook."""
assert context.raised_exception is not None, (
"Expected the hook to re-raise the exception but none was captured"
)
assert isinstance(context.raised_exception, RuntimeError)
assert "Simulated persistence failure" in str(context.raised_exception)
+10 -10
View File
@@ -82,20 +82,20 @@ def _build_mock_textual():
def update(self, text):
self._text = text
class MockInput:
"""Minimal Input stand-in for the Textual base class."""
class MockTextArea:
"""Minimal TextArea stand-in for the Textual base class."""
value = ""
text = ""
def __init__(self, *args, **kwargs):
self.value = ""
self.text = ""
mock_textual_app.App = MockApp
mock_textual_containers.Vertical = MockVertical
mock_textual_widgets.Header = MockHeader
mock_textual_widgets.Footer = MockFooter
mock_textual_widgets.Static = MockStatic
mock_textual_widgets.Input = MockInput
mock_textual_widgets.TextArea = MockTextArea
return {
"textual": mock_textual,
@@ -114,7 +114,7 @@ def _install_mock_textual(context):
for key, mod in mocks.items():
sys.modules[key] = mod
# Reload widget modules so they pick up the mock Static/Input base class
# Reload widget modules so they pick up the mock Static/TextArea base class
import cleveragents.tui.widgets.help_panel_overlay as hp_mod
import cleveragents.tui.widgets.persona_bar as pb_mod
import cleveragents.tui.widgets.prompt as prompt_mod
@@ -142,7 +142,7 @@ def _restore_modules(context):
else:
sys.modules[key] = val
# Reload widget modules so they pick up the real Static/Input base class again
# Reload widget modules so they pick up the real Static/TextArea base class again
import cleveragents.tui.widgets.help_panel_overlay as hp_mod
import cleveragents.tui.widgets.persona_bar as pb_mod
import cleveragents.tui.widgets.prompt as prompt_mod
@@ -370,7 +370,7 @@ def step_set_prompt_text(context, text):
from cleveragents.tui.widgets.prompt import PromptInput
prompt = context._tui_app.query_one("#prompt", PromptInput)
prompt.value = text
prompt.text = text
@then('the conversation widget should contain "{text}"')
@@ -425,11 +425,11 @@ def step_persona_bar_shows_name(context):
# on_input_submitted helpers
# ---------------------------------------------------------------------------
def _submit_text(context, text):
"""Set prompt value and fire on_input_submitted."""
"""Set prompt text and fire on_input_submitted."""
from cleveragents.tui.widgets.prompt import PromptInput
prompt = context._tui_app.query_one("#prompt", PromptInput)
prompt.value = text
prompt.text = text
event = SimpleNamespace()
context._tui_app.on_input_submitted(event)
+10
View File
@@ -143,3 +143,13 @@ def step_run_fallback_tui_app(context: Context) -> None:
def step_fallback_tui_fails(context: Context, message: str) -> None:
assert context.tui_fallback_error is not None
assert message in str(context.tui_fallback_error)
@when('I detect mode for "{text}"')
def step_detect_mode(context: Context, text: str) -> None:
context.detected_mode = InputModeRouter.detect_mode(text)
@then('the detected mode should be "{mode}"')
def step_detected_mode_equals(context: Context, mode: str) -> None:
assert context.detected_mode.value == mode
-41
View File
@@ -1,41 +0,0 @@
"""Behave steps for TUI prompt symbol handling."""
from __future__ import annotations
from behave import given, then, when
from behave.runner import Context
from cleveragents.tui.widgets.prompt import PromptInput
@given("a TUI prompt widget")
def step_create_prompt(context: Context) -> None:
context.tui_prompt = PromptInput(
placeholder="Type message, /command, or !shell ..."
)
@when('I set the TUI prompt value to "{value}"')
def step_set_prompt_value(context: Context, value: str) -> None:
context.tui_prompt.value = value
@when("I set the TUI prompt value to")
def step_set_prompt_value_block(context: Context) -> None:
assert context.text is not None
context.tui_prompt.value = context.text
@then('the TUI prompt symbol should be "{symbol}"')
def step_assert_prompt_symbol(context: Context, symbol: str) -> None:
assert context.tui_prompt.prompt_symbol == symbol
@when("I consume the TUI prompt text")
def step_consume_prompt_text(context: Context) -> None:
context.tui_consumed_prompt = context.tui_prompt.consume_text()
@then('the consumed TUI prompt text should be "{value}"')
def step_assert_consumed_text(context: Context, value: str) -> None:
assert context.tui_consumed_prompt.text == value
+217
View File
@@ -0,0 +1,217 @@
"""Step definitions for tui_prompt_textarea.feature.
Tests that PromptInput uses TextArea (multi-line) instead of Input (single-line).
"""
from __future__ import annotations
import importlib
import sys
from typing import Any
from types import ModuleType
from behave import given, then, when
_MOCK_TEXTUAL_KEYS = [
"textual",
"textual.app",
"textual.containers",
"textual.widgets",
]
def _build_mock_textual_with_textarea():
"""Build mock textual modules that expose TextArea."""
mock_textual = ModuleType("textual")
mock_textual_app = ModuleType("textual.app")
mock_textual_containers = ModuleType("textual.containers")
mock_textual_widgets = ModuleType("textual.widgets")
class MockTextArea:
"""Minimal TextArea stand-in for the Textual base class."""
text: str = ""
def __init__(self, *args: object, **kwargs: object) -> None:
self.text = ""
mock_textual_app.App = object
mock_textual_containers.Vertical = object
mock_textual_widgets.Header = object
mock_textual_widgets.Footer = object
mock_textual_widgets.Static = object
mock_textual_widgets.TextArea = MockTextArea
return {
"textual": mock_textual,
"textual.app": mock_textual_app,
"textual.containers": mock_textual_containers,
"textual.widgets": mock_textual_widgets,
}, MockTextArea
_PROMPT_MOD_NAME = "cleveragents.tui.widgets.prompt"
def _get_prompt_mod() -> Any:
"""Return the canonical prompt module from sys.modules.
Uses ``importlib.import_module`` (which always returns
``sys.modules[name]``) instead of ``import cleveragents.tui.widgets.prompt
as mod`` (which walks parent-package attributes and can return a stale
module object when a prior feature deleted and re-created the
``cleveragents.tui.*`` namespace). The stale object causes
``importlib.reload()`` to fail with
``ImportError: module ... not in sys.modules`` because Python 3.13's
reload checks ``sys.modules.get(name) is module``.
"""
return importlib.import_module(_PROMPT_MOD_NAME)
def _install_mock_textual(context: Any) -> None:
"""Inject mock textual into sys.modules and reload the prompt module."""
mocks, mock_textarea_cls = _build_mock_textual_with_textarea()
context._prompt_saved_modules = {}
for key in _MOCK_TEXTUAL_KEYS:
context._prompt_saved_modules[key] = sys.modules.pop(key, None)
for key, mod in mocks.items():
sys.modules[key] = mod
prompt_mod = _get_prompt_mod()
importlib.reload(prompt_mod)
context._prompt_mod = prompt_mod
context._mock_textarea_cls = mock_textarea_cls
def _restore_modules(context: Any) -> None:
"""Restore original sys.modules and reload the prompt module."""
for key, val in getattr(context, "_prompt_saved_modules", {}).items():
if val is None:
sys.modules.pop(key, None)
else:
sys.modules[key] = val
importlib.reload(_get_prompt_mod())
# ---------------------------------------------------------------------------
# Background
# ---------------------------------------------------------------------------
@given("the prompt module is loaded with a mocked TextArea")
def step_load_prompt_with_mock_textarea(context):
"""Install mock Textual with TextArea, reload prompt module."""
_install_mock_textual(context)
context.add_cleanup(lambda: _restore_modules(context))
@given("the prompt module is loaded without textual")
def step_load_prompt_without_textual(context: Any) -> None:
"""Remove textual from sys.modules so the fallback path is used."""
context._prompt_saved_modules_fallback = {}
for key in _MOCK_TEXTUAL_KEYS:
context._prompt_saved_modules_fallback[key] = sys.modules.pop(key, None)
prompt_mod = _get_prompt_mod()
importlib.reload(prompt_mod)
context._prompt_mod_fallback = prompt_mod
def restore() -> None:
for key, val in context._prompt_saved_modules_fallback.items():
if val is None:
sys.modules.pop(key, None)
else:
sys.modules[key] = val
importlib.reload(_get_prompt_mod())
context.add_cleanup(restore)
# ---------------------------------------------------------------------------
# Scenario: PromptInput base class is TextArea not Input
# ---------------------------------------------------------------------------
@then("the PromptInput base class should be the mocked TextArea")
def step_base_class_is_textarea(context):
PromptInput = context._prompt_mod.PromptInput
assert issubclass(PromptInput, context._mock_textarea_cls), (
f"Expected PromptInput to subclass MockTextArea, "
f"but got bases: {PromptInput.__bases__}"
)
# ---------------------------------------------------------------------------
# Scenario: PromptInput exposes a text property not value
# ---------------------------------------------------------------------------
@when("I create a PromptInput instance")
def step_create_prompt_input(context):
context._prompt_instance = context._prompt_mod.PromptInput()
@then("the PromptInput instance should have a text attribute")
def step_has_text_attribute(context):
assert hasattr(context._prompt_instance, "text"), (
"PromptInput instance should have a 'text' attribute"
)
# ---------------------------------------------------------------------------
# Scenario: consume_text returns the current text content
# ---------------------------------------------------------------------------
@when('I set the PromptInput text to "{text}"')
def step_set_prompt_input_text(context, text):
context._prompt_instance.text = text
@when("I call consume_text on the PromptInput")
def step_call_consume_text(context):
context._prompt_submitted = context._prompt_instance.consume_text()
@then('the PromptSubmitted text should be "{expected}"')
def step_prompt_submitted_text(context, expected):
assert context._prompt_submitted.text == expected, (
f"Expected '{expected}', got '{context._prompt_submitted.text}'"
)
# ---------------------------------------------------------------------------
# Scenario: consume_text clears the text after consuming
# ---------------------------------------------------------------------------
@then("the PromptInput text should be empty")
def step_prompt_input_text_empty(context):
assert context._prompt_instance.text == "", (
f"Expected empty text, got '{context._prompt_instance.text}'"
)
# ---------------------------------------------------------------------------
# Scenario: PromptInput fallback uses text attribute when TextArea unavailable
# ---------------------------------------------------------------------------
@when("I create a PromptInput instance from the fallback")
def step_create_fallback_prompt_input(context):
context._fallback_prompt_instance = context._prompt_mod_fallback.PromptInput()
@then("the fallback PromptInput instance should have a text attribute")
def step_fallback_has_text_attribute(context):
assert hasattr(context._fallback_prompt_instance, "text"), (
"Fallback PromptInput instance should have a 'text' attribute"
)
@then("the fallback PromptInput text should be empty string")
def step_fallback_text_empty(context):
assert context._fallback_prompt_instance.text == "", (
f"Expected empty string, got '{context._fallback_prompt_instance.text}'"
)
@@ -1,157 +0,0 @@
Feature: Decision recording hook in Strategize phase
As a strategy actor
I want to record decisions during the Strategize phase
So that every choice point is captured with full context for replay and correction
Background:
Given a strategize decision service
And a strategize decision hook for plan "01JQAAAAAAAAAAAAAAAAAAAA01"
# --- Strategy Choice Recording ---
Scenario: Record a strategy choice decision
When I record a strategy choice with question "Which approach?" and option "Approach A"
Then the strat decision should be recorded successfully
And the strat decision type should be "strategy_choice"
And the strat decision question should be "Which approach?"
And the strat decision chosen_option should be "Approach A"
And the strat decision phase should be "strategize"
Scenario: Record strategy choice with alternatives
When two strat alternatives "Approach B" and "Approach C"
And I record a strategy choice with question "Which approach?" and option "Approach A"
Then the strat decision should have 2 alternatives considered
Scenario: Record strategy choice with confidence score
When strat confidence 0.85
And I record a strategy choice with question "Which approach?" and option "Approach A"
Then the strat decision confidence score should be 0.85
Scenario: Record strategy choice with rationale
When strat rationale "Approach A is more efficient"
And I record a strategy choice with question "Which approach?" and option "Approach A"
Then the strat decision rationale should be "Approach A is more efficient"
Scenario: Record strategy choice with context snapshot
When strat context data containing "key1" "value1"
And I record a strategy choice with question "Which approach?" and option "Approach A"
Then the strat decision context snapshot hash should start with "sha256:"
And the strat decision context snapshot ref should not be empty
Scenario: Record strategy choice with actor state
When strat actor state containing "reasoning" "step1"
And I record a strategy choice with question "Which approach?" and option "Approach A"
Then the strat decision actor state ref should not be empty
Scenario: Record strategy choice with relevant resources
When two strat relevant resources "resource1" and "resource2"
And I record a strategy choice with question "Which approach?" and option "Approach A"
Then the strat decision should have 2 relevant resources
Scenario: Record strategy choice with empty question raises error
When I try to record a strategy choice with empty question
Then a strat validation error should be raised
And the strat error should mention "question"
Scenario: Record strategy choice with empty option raises error
When I try to record a strategy choice with empty chosen_option
Then a strat validation error should be raised
And the strat error should mention "chosen_option"
# --- Resource Selection Recording ---
Scenario: Record a resource selection decision
When I record a resource selection with question "Which resources?" and option "src/main.py"
Then the strat decision should be recorded successfully
And the strat decision type should be "resource_selection"
And the strat decision question should be "Which resources?"
And the strat decision chosen_option should be "src/main.py"
Scenario: Record resource selection with alternatives
When two strat alternatives "src/test.py" and "src/utils.py"
And I record a resource selection with question "Which resources?" and option "src/main.py"
Then the strat decision should have 2 alternatives considered
Scenario: Record resource selection with confidence
When strat confidence 0.9
And I record a resource selection with question "Which resources?" and option "src/main.py"
Then the strat decision confidence score should be 0.9
# --- Subplan Spawn Recording ---
Scenario: Record a subplan spawn decision
When I record a subplan spawn with question "Should we decompose?" and option "Create subplan for feature X"
Then the strat decision should be recorded successfully
And the strat decision type should be "subplan_spawn"
And the strat decision question should be "Should we decompose?"
And the strat decision chosen_option should be "Create subplan for feature X"
Scenario: Record subplan spawn with alternatives
When two strat alternatives "Implement inline" and "Create parallel subplans"
And I record a subplan spawn with question "Should we decompose?" and option "Create subplan for feature X"
Then the strat decision should have 2 alternatives considered
Scenario: Record subplan spawn with confidence
When strat confidence 0.75
And I record a subplan spawn with question "Should we decompose?" and option "Create subplan for feature X"
Then the strat decision confidence score should be 0.75
# --- Invariant Enforcement Recording ---
Scenario: Record an invariant enforced decision
When I record an invariant enforced with question "Apply security invariant?" and option "Enforce code review"
Then the strat decision should be recorded successfully
And the strat decision type should be "invariant_enforced"
And the strat decision question should be "Apply security invariant?"
And the strat decision chosen_option should be "Enforce code review"
Scenario: Record invariant enforced with rationale
When strat rationale "Security policy requires code review"
And I record an invariant enforced with question "Apply security invariant?" and option "Enforce code review"
Then the strat decision rationale should be "Security policy requires code review"
# --- Context Snapshot Capture ---
Scenario: Context snapshot captures hot context hash
When strat context data containing "plan_id" "01JQAAAAAAAAAAAAAAAAAAAA01"
And I record a strategy choice with question "Which approach?" and option "Approach A"
Then the strat decision context snapshot hash should start with "sha256:"
Scenario: Context snapshot captures actor state reference
When strat actor state containing "step" "1"
And I record a strategy choice with question "Which approach?" and option "Approach A"
Then the strat decision actor state ref should start with "checkpoint:"
Scenario: Context snapshot captures relevant resources
When three strat relevant resources "res1" and "res2" and "res3"
And I record a strategy choice with question "Which approach?" and option "Approach A"
Then the strat decision should have 3 relevant resources
And each strat resource should have a valid resource_id
# --- Error Handling ---
Scenario: Recording with invalid plan_id raises error
Given a strategize decision hook with empty plan_id
Then a strat validation error should be raised
And the strat error should mention "plan_id"
Scenario: Recording failure logs warning and re-raises exception
When the strat decision service fails to persist
And I try to record a strategy choice that raises an exception
Then a strat warning should be logged
And the strat exception should be re-raised
# --- Parent Decision Tracking ---
Scenario: Record decision with parent decision ID
When strat parent decision ID "01PARENT000000000000000000"
And I record a strategy choice with question "Which approach?" and option "Approach A"
Then the strat decision parent_decision_id should be "01PARENT000000000000000000"
Scenario: Record multiple decisions in tree structure
When I record a strategy choice with question "Q1" and option "A1"
And I save the first strat decision
And strat parent decision ID from the first decision
And I record a strategy choice with question "Q2" and option "A2"
Then the second strat decision parent_decision_id should match the first decision
And both strat decisions should be in the same plan
@@ -25,6 +25,7 @@ Feature: TDD Issue #988 — ReactiveEventBus.emit() swallows exception details
When I emit an event that triggers the failing handler
Then the warning log should contain the exception message text
@tdd_expected_fail
Scenario: Bug #988 — emit() logs traceback via exc_info when handler raises
Given a ReactiveEventBus with a handler that raises a ValueError
When I emit an event that triggers the failing handler
+20
View File
@@ -37,3 +37,23 @@ Feature: TUI input modes
Then TUI textual availability should be boolean
When I run fallback TUI app
Then fallback TUI app should fail with "Textual dependency missing."
Scenario: Dollar prefix activates shell mode
When I route TUI input "$echo hello"
Then the TUI mode should be "shell"
And the TUI shell stdout should contain "hello"
Scenario: Dollar prefix detect_mode returns shell
When I detect mode for "$echo hello"
Then the detected mode should be "shell"
Scenario: Dollar prefix with leading whitespace activates shell mode
When I detect mode for " $echo hello"
Then the detected mode should be "shell"
Scenario: Dollar prefix blocks dangerous command by default
When I route TUI input "$rm -rf /"
Then the TUI mode should be "shell"
And the TUI shell stderr should contain "blocked dangerous shell command"
-30
View File
@@ -1,30 +0,0 @@
Feature: TUI prompt symbol reflects input mode
The prompt must display the correct mode symbol so users know
whether they are typing a normal message, a command, or shell input.
Scenario Outline: Symbol updates when mode changes
Given a TUI prompt widget
When I set the TUI prompt value to "<input>"
Then the TUI prompt symbol should be "<symbol>"
Examples:
| input | symbol |
| hello | |
| /help | / |
| !ls | $ |
Scenario: Consuming text resets the prompt symbol
Given a TUI prompt widget
When I set the TUI prompt value to "/metrics"
And I consume the TUI prompt text
Then the TUI prompt symbol should be ""
And the consumed TUI prompt text should be "/metrics"
Scenario: Multi-line input toggles the multi-line prompt symbol
Given a TUI prompt widget
When I set the TUI prompt value to
"""
line one
line two
"""
Then the TUI prompt symbol should be ""
+37
View File
@@ -0,0 +1,37 @@
Feature: PromptInput uses multi-line TextArea widget
The PromptInput widget must use a multi-line TextArea widget (not a
single-line Input widget) to enable multi-line prompt composition.
Background:
Given the prompt module is loaded with a mocked TextArea
Scenario: PromptInput base class is TextArea not Input
Then the PromptInput base class should be the mocked TextArea
Scenario: PromptInput exposes a text property not value
When I create a PromptInput instance
Then the PromptInput instance should have a text attribute
Scenario: consume_text returns the current text content
When I create a PromptInput instance
And I set the PromptInput text to "hello world"
And I call consume_text on the PromptInput
Then the PromptSubmitted text should be "hello world"
Scenario: consume_text clears the text after consuming
When I create a PromptInput instance
And I set the PromptInput text to "some prompt"
And I call consume_text on the PromptInput
Then the PromptInput text should be empty
Scenario: consume_text supports multi-line text
When I create a PromptInput instance
And I set the PromptInput text to "line one\nline two\nline three"
And I call consume_text on the PromptInput
Then the PromptSubmitted text should be "line one\nline two\nline three"
Scenario: PromptInput fallback uses text attribute when TextArea unavailable
Given the prompt module is loaded without textual
When I create a PromptInput instance from the fallback
Then the fallback PromptInput instance should have a text attribute
And the fallback PromptInput text should be empty string
+124
View File
@@ -0,0 +1,124 @@
*** Settings ***
Documentation Integration tests for AsyncResourceTracker
Library Collections
Library BuiltIn
Library robot.async_cleanup_library.AsyncCleanupLibrary
*** Test Cases ***
Register And Close Single Resource
[Documentation] Verify basic registration and cleanup of a single resource
${tracker}= Create Async Resource Tracker
${resource}= Create Mock Async Resource test_resource
Register Resource ${tracker} test_resource ${resource}
${count}= Get Open Count ${tracker}
Should Be Equal As Integers ${count} 1
Close All Resources ${tracker} timeout=30.0
${count}= Get Open Count ${tracker}
Should Be Equal As Integers ${count} 0
Register Multiple Resources
[Documentation] Verify registration of multiple resources
${tracker}= Create Async Resource Tracker
${res1}= Create Mock Async Resource resource_1
${res2}= Create Mock Async Resource resource_2
${res3}= Create Mock Async Resource resource_3
Register Resource ${tracker} resource_1 ${res1}
Register Resource ${tracker} resource_2 ${res2}
Register Resource ${tracker} resource_3 ${res3}
${count}= Get Open Count ${tracker}
Should Be Equal As Integers ${count} 3
Close All Resources ${tracker} timeout=30.0
${count}= Get Open Count ${tracker}
Should Be Equal As Integers ${count} 0
Reject Empty Name
[Documentation] Verify that empty names are rejected
${tracker}= Create Async Resource Tracker
${resource}= Create Mock Async Resource empty_name
Run Keyword And Expect Error ValueError*name must be a non-empty string*
... Register Resource ${tracker} ${EMPTY} ${resource}
Reject None Resource
[Documentation] Verify that None resources are rejected
${tracker}= Create Async Resource Tracker
Run Keyword And Expect Error ValueError*resource must not be None*
... Register Resource ${tracker} test ${None}
Reject Duplicate Registration
[Documentation] Verify that duplicate names are rejected
${tracker}= Create Async Resource Tracker
${resource}= Create Mock Async Resource duplicate
Register Resource ${tracker} duplicate ${resource}
${resource2}= Create Mock Async Resource duplicate2
Run Keyword And Expect Error ValueError*Resource 'duplicate' is already registered*
... Register Resource ${tracker} duplicate ${resource2}
Reject Registration After Close
[Documentation] Verify that registration after close is rejected
${tracker}= Create Async Resource Tracker
Close All Resources ${tracker} timeout=30.0
${resource}= Create Mock Async Resource late
Run Keyword And Expect Error RuntimeError*Cannot register resource after tracker is closed*
... Register Resource ${tracker} late ${resource}
Handle Timeout During Close
[Documentation] Verify timeout handling during resource close
${tracker}= Create Async Resource Tracker
${slow_resource}= Create Slow Async Resource slow_resource 5.0
Register Resource ${tracker} slow_resource ${slow_resource}
Close All Resources ${tracker} timeout=0.1
${timed_out}= Get Timed Out Resources ${tracker}
Should Contain ${timed_out} slow_resource
Handle Exception During Close
[Documentation] Verify exception handling during resource close
${tracker}= Create Async Resource Tracker
${failing_resource}= Create Failing Async Resource failing_resource
Register Resource ${tracker} failing_resource ${failing_resource}
Close All Resources ${tracker} timeout=30.0
${count}= Get Open Count ${tracker}
Should Be Equal As Integers ${count} 0
Close Is Idempotent
[Documentation] Verify that close_all is idempotent
${tracker}= Create Async Resource Tracker
${resource}= Create Mock Async Resource resource
Register Resource ${tracker} resource ${resource}
Close All Resources ${tracker} timeout=30.0
${count}= Get Open Count ${tracker}
Should Be Equal As Integers ${count} 0
Close All Resources ${tracker} timeout=30.0
${count}= Get Open Count ${tracker}
Should Be Equal As Integers ${count} 0
Use As Async Context Manager
[Documentation] Verify tracker works as async context manager
${tracker}= Create Async Resource Tracker
${resource}= Create Mock Async Resource ctx_resource
Register Resource ${tracker} ctx_resource ${resource}
${count}= Get Open Count ${tracker}
Should Be Equal As Integers ${count} 1
Close All Resources ${tracker} timeout=30.0
${count}= Get Open Count ${tracker}
Should Be Equal As Integers ${count} 0
Protocol Compliance
[Documentation] Verify that any object with async close is accepted
${tracker}= Create Async Resource Tracker
${custom}= Create Custom Resource With Close
Register Resource ${tracker} custom ${custom}
${count}= Get Open Count ${tracker}
Should Be Equal As Integers ${count} 1
Close All Resources ${tracker} timeout=30.0
Should Be True ${custom.closed}
+74
View File
@@ -0,0 +1,74 @@
"""Robot Framework library for async_cleanup integration tests."""
from __future__ import annotations
import asyncio
from typing import Any
from unittest.mock import AsyncMock
from cleveragents.core.async_cleanup import AsyncResource, AsyncResourceTracker
class AsyncCleanupLibrary:
"""Robot Framework library for AsyncResourceTracker testing."""
ROBOT_LIBRARY_SCOPE = "TEST"
def create_async_resource_tracker(self) -> AsyncResourceTracker:
"""Create a new AsyncResourceTracker instance."""
return AsyncResourceTracker()
def create_mock_async_resource(self, name: str) -> AsyncMock:
"""Create a mock async resource."""
resource = AsyncMock(spec=AsyncResource)
resource.close = AsyncMock()
return resource
def create_slow_async_resource(self, name: str, delay: float) -> AsyncMock:
"""Create an async resource that delays on close."""
async def slow_close() -> None:
await asyncio.sleep(delay)
resource = AsyncMock(spec=AsyncResource)
resource.close = slow_close
return resource
def create_failing_async_resource(self, name: str) -> AsyncMock:
"""Create an async resource that raises on close."""
async def failing_close() -> None:
raise RuntimeError(f"Failed to close {name}")
resource = AsyncMock(spec=AsyncResource)
resource.close = failing_close
return resource
def create_custom_resource_with_close(self) -> Any:
"""Create a custom resource with async close method."""
class CustomResource:
def __init__(self) -> None:
self.closed = False
async def close(self) -> None:
self.closed = True
return CustomResource()
def register_resource(
self, tracker: AsyncResourceTracker, name: str, resource: Any
) -> None:
"""Register a resource with the tracker."""
tracker.register(name, resource)
def get_open_count(self, tracker: AsyncResourceTracker) -> int:
"""Get the number of open resources."""
return tracker.open_count
def get_timed_out_resources(self, tracker: AsyncResourceTracker) -> list[str]:
"""Get the list of timed out resources."""
return tracker.timed_out_resources
def close_all_resources(
self, tracker: AsyncResourceTracker, timeout: float = 30.0
) -> None:
"""Close all resources in the tracker."""
asyncio.run(tracker.close_all(timeout=timeout))
+1 -2
View File
@@ -49,8 +49,7 @@ def _test_explain_format() -> None:
assert "decision_id" in data
assert "context_snapshot" in data
assert "rationale" in data
assert "alternatives" in data
assert isinstance(data["alternatives"], list)
assert "alternatives_considered" in data
assert data["question"] == "What to build?"
print("plan-explain-ok")
+4 -2
View File
@@ -23,7 +23,8 @@ TDD Resource Add Succeeds Without Explicit Init
... The helper exits 0 with a sentinel when the command
... succeeds (bug is fixed), and exits 1 when the bug is
... present (OperationalError).
[Tags] tdd_issue tdd_issue_1023 tdd_issue_4317 tdd_expected_fail
[Tags] tdd_issue tdd_issue_1023 tdd_issue tdd_issue_4317 tdd_expected_fail
${result}= Run Process ${PYTHON} ${HELPER} resource-add-no-init cwd=${WORKSPACE} timeout=120s on_timeout=kill
Log ${result.stdout}
Log ${result.stderr}
@@ -37,7 +38,8 @@ TDD Project Create Succeeds Without Explicit Init
... The helper exits 0 with a sentinel when the command
... succeeds (bug is fixed), and exits 1 when the bug is
... present (OperationalError).
[Tags] tdd_issue tdd_issue_1023 tdd_issue_4317 tdd_expected_fail
[Tags] tdd_issue tdd_issue_1023 tdd_issue tdd_issue_4317 tdd_expected_fail
${result}= Run Process ${PYTHON} ${HELPER} project-create-no-init cwd=${WORKSPACE} timeout=120s on_timeout=kill
Log ${result.stdout}
Log ${result.stderr}
+1 -33
View File
@@ -17,38 +17,6 @@ TUI Headless Works When Shell Disabled
Should Contain ${result.stdout} textual_available
Should Contain ${result.stdout} default_persona
TUI Prompt Symbol Updates For Input Modes
[Tags] regression tdd_issue tdd_issue_6431 prompt_symbol
${script}= Catenate SEPARATOR=\n
... import sys
... from pathlib import Path
...
... sys.path.insert(0, str((Path.cwd() / "src").resolve()))
... from cleveragents.tui.widgets.prompt import PromptInput
...
... prompt = PromptInput()
... assert prompt.prompt_symbol == "", f"Expected normal mode symbol, got {prompt.prompt_symbol!r}"
...
... prompt.value = "/help"
... assert prompt.prompt_symbol == "/", f"Expected command mode symbol, got {prompt.prompt_symbol!r}"
...
... prompt.value = "!ls"
... assert prompt.prompt_symbol == "$", f"Expected shell mode symbol, got {prompt.prompt_symbol!r}"
...
... prompt.value = "@plan/123"
... assert prompt.prompt_symbol == "", f"References should keep normal symbol, got {prompt.prompt_symbol!r}"
...
... prompt.value = "line one\\nline two"
... assert prompt.prompt_symbol == "☰", f"Expected multiline symbol, got {prompt.prompt_symbol!r}"
...
... prompt.value = ""
... assert prompt.prompt_symbol == "", f"Prompt should reset to normal symbol, got {prompt.prompt_symbol!r}"
...
... print("prompt-symbol-modes-ok")
${result}= Run Process ${PYTHON} -c ${script} shell=False
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} prompt-symbol-modes-ok
TUI Input Mode Router And Prompt Widget Behavior
[Tags] tdd_issue tdd_issue_4193 tdd_issue_4297 tdd_expected_fail
${script}= Catenate SEPARATOR=\n
@@ -127,4 +95,4 @@ TUI Shell Mode Detection
... print("shell-mode-detection-ok")
${result}= Run Process ${PYTHON} -c ${script} shell=False
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} shell-mode-detection-ok
Should Contain ${result.stdout} shell-mode-detection-ok
@@ -1 +0,0 @@
"""Application ports — protocol interfaces for external dependencies."""
@@ -1,49 +0,0 @@
"""Decision recorder port — protocol interface for recording decisions.
This module defines the ``DecisionRecorder`` protocol, which is the
shared interface used by both ``StrategizeDecisionHook`` and the future
``ExecuteDecisionHook`` to record decisions without coupling to a
concrete ``DecisionService`` implementation.
Based on:
- docs/adr/ADR-033-decision-recording-protocol.md
- Forgejo issue #8522
"""
from __future__ import annotations
from typing import TYPE_CHECKING, Protocol, runtime_checkable
if TYPE_CHECKING:
from cleveragents.domain.models.core.decision import (
ContextSnapshot,
Decision,
DecisionType,
)
from cleveragents.domain.models.core.plan import PlanPhase
@runtime_checkable
class DecisionRecorder(Protocol):
"""Protocol for recording decisions (subset of DecisionService API).
Both ``StrategizeDecisionHook`` and the future ``ExecuteDecisionHook``
depend on this protocol rather than the concrete ``DecisionService``,
keeping the hooks decoupled from the persistence layer.
"""
def record_decision(
self,
plan_id: str,
decision_type: DecisionType | str,
question: str,
chosen_option: str,
*,
parent_decision_id: str | None = None,
alternatives_considered: list[str] | None = None,
confidence_score: float | None = None,
rationale: str = "",
actor_reasoning: str | None = None,
context_snapshot: ContextSnapshot | None = None,
plan_phase: PlanPhase | str | None = None,
) -> Decision: ...
@@ -31,20 +31,10 @@ def _path_matches(path: str, include_patterns: list[str]) -> bool:
def _matches_pattern(path_obj: PurePosixPath, pattern: str) -> bool:
"""Match a path against a glob pattern, handling absolute vs relative paths.
Tries ``full_match()`` with the pattern as-is, then with a ``**/``
prefix so that relative globs (e.g. ``.opencode/*``) correctly match
absolute paths (e.g. ``/app/.opencode/skills/SKILL.md``). Also
handles the ``**/`` zero-depth compatibility shim.
"""
if path_obj.full_match(pattern):
return True
# Auto-prefix with **/ so relative patterns match absolute paths
if not pattern.startswith("**/") and path_obj.full_match(f"**/{pattern}"):
return True
# Zero-depth shim: "**/" in pattern but full_match already tried above
return bool("**/" in pattern and path_obj.full_match(pattern.replace("**/", "")))
"""Match with a small compatibility shim for ``**/`` zero-depth cases."""
return path_obj.match(pattern) or (
"**/" in pattern and path_obj.match(pattern.replace("**/", ""))
)
def _extract_path(fragment: TieredFragment) -> str:
@@ -1,66 +0,0 @@
"""Decision context snapshot utility.
Provides the ``capture_context_snapshot`` function for capturing a
context snapshot at decision time. This utility is shared between
``StrategizeDecisionHook`` and the future ``ExecuteDecisionHook``.
Based on:
- docs/specification.md §Strategize-Phase Recording Loop
- docs/adr/ADR-033-decision-recording-protocol.md
- Forgejo issue #8522
"""
from __future__ import annotations
import hashlib
import json
from typing import Any
from cleveragents.domain.models.core.decision import (
ContextSnapshot,
ResourceRef,
)
def capture_context_snapshot(
context_data: dict[str, Any] | None = None,
actor_state: dict[str, Any] | None = None,
relevant_resources: list[str] | None = None,
) -> ContextSnapshot:
"""Capture a context snapshot at decision time.
Automatically generates:
- ``hot_context_hash``: SHA256 hash of the context data
- ``hot_context_ref``: Abbreviated storage reference
- ``relevant_resources``: List of resource references
- ``actor_state_ref``: Reference to actor state checkpoint
Args:
context_data: Current context window contents (dict).
actor_state: Actor's current state (dict).
relevant_resources: List of resource IDs that influenced the decision.
Returns:
A :class:`~cleveragents.domain.models.core.decision.ContextSnapshot`
with auto-captured fields.
"""
# Generate hot context hash
context_json = json.dumps(context_data or {}, sort_keys=True, default=str)
hot_context_hash = f"sha256:{hashlib.sha256(context_json.encode()).hexdigest()}"
# Generate actor state reference (placeholder for LangGraph checkpoint)
actor_state_json = json.dumps(actor_state or {}, sort_keys=True, default=str)
actor_state_ref = (
f"checkpoint:{hashlib.sha256(actor_state_json.encode()).hexdigest()[:16]}"
)
# Convert resource IDs to ResourceRef objects
resource_refs = [ResourceRef(resource_id=rid) for rid in (relevant_resources or [])]
return ContextSnapshot(
hot_context_hash=hot_context_hash,
hot_context_ref=f"context:{hot_context_hash[7:23]}", # Abbreviated ref
relevant_resources=resource_refs,
actor_state_ref=actor_state_ref,
)
@@ -72,32 +72,13 @@ class ACMSExecutePhaseContextAssembler(ExecutePhaseContextAssembler):
@staticmethod
def _path_matches(path: str, include: list[str], exclude: list[str]) -> bool:
"""Return whether *path* passes include/exclude path globs.
Handles both absolute paths (e.g. ``/app/.opencode/skills/SKILL.md``)
and relative paths (e.g. ``src/foo.py``) against relative glob
patterns (e.g. ``.opencode/**``, ``src/**/*.py``).
Each pattern is tried with ``full_match()`` as-is (handles
relative paths and ``**`` patterns). If the pattern is not
already anchored with ``**/``, a second attempt prefixes
``**/`` so that relative globs also match absolute paths.
"""
"""Return whether *path* passes include/exclude path globs."""
pure_path = PurePath(path)
def _matches_any(patterns: list[str]) -> bool:
for pattern in patterns:
if pure_path.full_match(pattern):
return True
if not pattern.startswith("**/") and pure_path.full_match(
f"**/{pattern}"
):
return True
if include and not any(pure_path.full_match(pattern) for pattern in include):
return False
if include and not _matches_any(include):
return False
return not (exclude and _matches_any(exclude))
return not (
exclude and any(pure_path.full_match(pattern) for pattern in exclude)
)
@staticmethod
def _resource_matches(
@@ -51,8 +51,6 @@ Based on ``docs/specification.md`` and implementation plan Stage A3.
from __future__ import annotations
import hashlib
import json
from contextlib import suppress
from datetime import datetime
from typing import TYPE_CHECKING, Any
@@ -79,7 +77,7 @@ from cleveragents.domain.models.core.automation_profile import (
BUILTIN_PROFILES,
AutomationProfile,
)
from cleveragents.domain.models.core.decision import ContextSnapshot, DecisionType
from cleveragents.domain.models.core.decision import DecisionType
from cleveragents.domain.models.core.plan import (
AutomationProfileProvenance,
AutomationProfileRef,
@@ -267,23 +265,11 @@ class PlanLifecycleService:
question: str,
chosen_option: str,
parent_decision_id: str | None = None,
context_snapshot: ContextSnapshot | None = None,
) -> None:
"""Record a decision if DecisionService is available.
Failures are logged but never propagated decision recording
must not block lifecycle transitions.
Args:
plan_id: ULID of the plan.
decision_type: Type of decision being recorded.
question: What question was being answered.
chosen_option: The option that was chosen.
parent_decision_id: Optional parent in the decision tree.
context_snapshot: Optional full context snapshot. When
provided, it is forwarded to
:meth:`DecisionService.record_decision` so that the
decision is stored with a complete context snapshot
"""
if self.decision_service is None:
return
@@ -295,7 +281,6 @@ class PlanLifecycleService:
question=question,
chosen_option=chosen_option,
parent_decision_id=parent_decision_id,
context_snapshot=context_snapshot,
)
except Exception:
self._logger.warning(
@@ -1413,72 +1398,15 @@ class PlanLifecycleService:
self._commit_plan(plan)
self._logger.info("Strategize started", plan_id=plan_id)
context_snapshot = self._build_strategize_context_snapshot(plan)
self._try_record_decision(
plan_id=plan_id,
decision_type="strategy_choice",
question="Which strategy should the plan follow?",
chosen_option=f"Begin strategize phase for plan {plan_id}",
context_snapshot=context_snapshot,
)
return plan
def _build_strategize_context_snapshot(self, plan: Plan) -> ContextSnapshot:
"""Build a full context snapshot for a Strategize-phase decision.
Captures the plan description, action name, strategy actor, and
project references as the hot context window. The hash is
computed over the serialised context so that identical context
windows produce the same hash (content-addressable).
The ``hot_context_ref`` is set to a stable ``plan:<plan_id>``
URI so that callers can locate the full context via the plan
record. ``relevant_resources`` is populated from the plan's
project links. ``actor_state_ref`` is set to the strategy
actor name when available.
Per the v3.2.0 acceptance criteria, decisions recorded during
the Strategize phase must include full context snapshots with
all four :class:`ContextSnapshot` fields populated.
Args:
plan: The plan entering the Strategize phase.
Returns:
A :class:`ContextSnapshot` with all four fields populated.
"""
from cleveragents.domain.models.core.decision import ResourceRef
plan_id = plan.identity.plan_id
# Build the hot context window from plan metadata available at
# the start of the Strategize phase.
hot_context: dict[str, object] = {
"plan_id": plan_id,
"action_name": plan.action_name,
"description": plan.description or "",
"strategy_actor": plan.strategy_actor or "",
"projects": [pl.project_name for pl in plan.project_links],
}
context_json = json.dumps(hot_context, sort_keys=True)
context_hash = hashlib.sha256(context_json.encode()).hexdigest()
# Build resource refs from project links so the snapshot records
# which projects influenced the strategy decision.
relevant_resources = [
ResourceRef(resource_id=pl.project_name)
for pl in plan.project_links
if pl.project_name
]
return ContextSnapshot(
hot_context_hash=f"sha256:{context_hash}",
hot_context_ref=f"plan:{plan_id}",
relevant_resources=relevant_resources,
actor_state_ref=plan.strategy_actor or "",
)
def complete_strategize(self, plan_id: str) -> Plan:
"""Complete the Strategize phase.
@@ -1,378 +0,0 @@
"""Decision recording hook for the Strategize phase.
This module provides the ``StrategizeDecisionHook`` class, which integrates
decision recording into the Strategize phase of plan execution. The hook
captures every decision point during strategy decomposition, including:
- The question being answered
- The chosen option
- Alternatives considered
- Confidence score
- Rationale
- Full context snapshot (hot context hash, actor state reference, relevant resources)
The hook is designed to be called by the strategy actor during the Strategize
phase, recording decisions atomically with plan updates.
Based on:
- docs/specification.md §Strategize-Phase Recording Loop
- docs/adr/ADR-033-decision-recording-protocol.md
- Forgejo issue #8522
"""
from __future__ import annotations
from typing import Any
import structlog
from cleveragents.application.ports.decision_recorder import DecisionRecorder
from cleveragents.application.services.decision_context import capture_context_snapshot
from cleveragents.core.exceptions import ValidationError
from cleveragents.domain.models.core.decision import (
Decision,
DecisionType,
)
from cleveragents.domain.models.core.plan import PlanPhase
logger = structlog.get_logger(__name__)
# ---------------------------------------------------------------------------
# Strategize decision hook
# ---------------------------------------------------------------------------
class StrategizeDecisionHook:
"""Hook for recording decisions during the Strategize phase.
Integrates with the strategy actor to capture every decision point,
including the question, chosen option, alternatives, confidence,
rationale, and full context snapshot.
The hook is designed to be called by the strategy actor during
Strategize, and records decisions atomically with plan updates.
Attributes:
decision_service: The
:class:`~cleveragents.application.ports.decision_recorder.DecisionRecorder`
instance for persisting decisions.
plan_id: ULID of the plan being strategized.
parent_decision_id: Optional parent decision ID for tree structure.
"""
def __init__(
self,
decision_service: DecisionRecorder,
plan_id: str,
parent_decision_id: str | None = None,
) -> None:
"""Initialize the Strategize decision hook.
Args:
decision_service: DecisionRecorder for recording decisions.
plan_id: ULID of the plan.
parent_decision_id: Optional parent decision ID.
Raises:
ValidationError: If plan_id is empty.
"""
if not plan_id or not plan_id.strip():
raise ValidationError("plan_id must not be empty")
self.decision_service = decision_service
self.plan_id = plan_id
self.parent_decision_id = parent_decision_id
self._logger = logger.bind(
hook="strategize_decision",
plan_id=plan_id,
)
def record_strategy_choice(
self,
question: str,
chosen_option: str,
alternatives_considered: list[str] | None = None,
confidence_score: float | None = None,
rationale: str = "",
context_data: dict[str, Any] | None = None,
actor_state: dict[str, Any] | None = None,
relevant_resources: list[str] | None = None,
) -> Decision:
"""Record a strategy choice decision during Strategize.
Args:
question: What strategic question was being answered.
chosen_option: The chosen approach.
alternatives_considered: Other approaches evaluated.
confidence_score: Confidence in the choice (0.0-1.0).
rationale: Why this option was chosen.
context_data: Current context window contents.
actor_state: Actor's current state.
relevant_resources: Resource IDs that influenced the decision.
Returns:
The recorded Decision.
Raises:
ValidationError: If required fields are missing.
"""
if not question or not question.strip():
raise ValidationError("question must not be empty")
if not chosen_option or not chosen_option.strip():
raise ValidationError("chosen_option must not be empty")
snapshot = capture_context_snapshot(
context_data=context_data,
actor_state=actor_state,
relevant_resources=relevant_resources,
)
self._logger.info(
"Recording strategy choice decision",
question=question,
chosen_option=chosen_option,
confidence=confidence_score,
)
try:
decision = self.decision_service.record_decision(
plan_id=self.plan_id,
decision_type=DecisionType.STRATEGY_CHOICE,
question=question,
chosen_option=chosen_option,
parent_decision_id=self.parent_decision_id,
alternatives_considered=alternatives_considered,
confidence_score=confidence_score,
rationale=rationale,
context_snapshot=snapshot,
plan_phase=PlanPhase.STRATEGIZE,
)
self._logger.debug(
"Strategy choice decision recorded",
decision_id=decision.decision_id,
)
return decision
except Exception as exc:
self._logger.warning(
"Failed to record strategy choice decision",
error=str(exc),
error_type=type(exc).__name__,
)
raise
def record_resource_selection(
self,
question: str,
chosen_option: str,
alternatives_considered: list[str] | None = None,
confidence_score: float | None = None,
rationale: str = "",
context_data: dict[str, Any] | None = None,
actor_state: dict[str, Any] | None = None,
relevant_resources: list[str] | None = None,
) -> Decision:
"""Record a resource selection decision during Strategize.
Args:
question: What resources should be selected.
chosen_option: The selected resources.
alternatives_considered: Other resource selections evaluated.
confidence_score: Confidence in the selection (0.0-1.0).
rationale: Why these resources were selected.
context_data: Current context window contents.
actor_state: Actor's current state.
relevant_resources: Resource IDs that influenced the decision.
Returns:
The recorded Decision.
Raises:
ValidationError: If required fields are missing.
"""
if not question or not question.strip():
raise ValidationError("question must not be empty")
if not chosen_option or not chosen_option.strip():
raise ValidationError("chosen_option must not be empty")
snapshot = capture_context_snapshot(
context_data=context_data,
actor_state=actor_state,
relevant_resources=relevant_resources,
)
self._logger.info(
"Recording resource selection decision",
question=question,
chosen_option=chosen_option,
)
try:
decision = self.decision_service.record_decision(
plan_id=self.plan_id,
decision_type=DecisionType.RESOURCE_SELECTION,
question=question,
chosen_option=chosen_option,
parent_decision_id=self.parent_decision_id,
alternatives_considered=alternatives_considered,
confidence_score=confidence_score,
rationale=rationale,
context_snapshot=snapshot,
plan_phase=PlanPhase.STRATEGIZE,
)
self._logger.debug(
"Resource selection decision recorded",
decision_id=decision.decision_id,
)
return decision
except Exception as exc:
self._logger.warning(
"Failed to record resource selection decision",
error=str(exc),
error_type=type(exc).__name__,
)
raise
def record_subplan_spawn(
self,
question: str,
chosen_option: str,
alternatives_considered: list[str] | None = None,
confidence_score: float | None = None,
rationale: str = "",
context_data: dict[str, Any] | None = None,
actor_state: dict[str, Any] | None = None,
relevant_resources: list[str] | None = None,
) -> Decision:
"""Record a subplan spawn decision during Strategize.
Args:
question: Why is a subplan being spawned.
chosen_option: The subplan goal/description.
alternatives_considered: Other decomposition approaches.
confidence_score: Confidence in the decomposition (0.0-1.0).
rationale: Why this decomposition was chosen.
context_data: Current context window contents.
actor_state: Actor's current state.
relevant_resources: Resource IDs that influenced the decision.
Returns:
The recorded Decision.
Raises:
ValidationError: If required fields are missing.
"""
if not question or not question.strip():
raise ValidationError("question must not be empty")
if not chosen_option or not chosen_option.strip():
raise ValidationError("chosen_option must not be empty")
snapshot = capture_context_snapshot(
context_data=context_data,
actor_state=actor_state,
relevant_resources=relevant_resources,
)
self._logger.info(
"Recording subplan spawn decision",
question=question,
chosen_option=chosen_option,
)
try:
decision = self.decision_service.record_decision(
plan_id=self.plan_id,
decision_type=DecisionType.SUBPLAN_SPAWN,
question=question,
chosen_option=chosen_option,
parent_decision_id=self.parent_decision_id,
alternatives_considered=alternatives_considered,
confidence_score=confidence_score,
rationale=rationale,
context_snapshot=snapshot,
plan_phase=PlanPhase.STRATEGIZE,
)
self._logger.debug(
"Subplan spawn decision recorded",
decision_id=decision.decision_id,
)
return decision
except Exception as exc:
self._logger.warning(
"Failed to record subplan spawn decision",
error=str(exc),
error_type=type(exc).__name__,
)
raise
def record_invariant_enforced(
self,
question: str,
chosen_option: str,
alternatives_considered: list[str] | None = None,
confidence_score: float | None = None,
rationale: str = "",
context_data: dict[str, Any] | None = None,
actor_state: dict[str, Any] | None = None,
relevant_resources: list[str] | None = None,
) -> Decision:
"""Record an invariant enforcement decision during Strategize.
Args:
question: What invariant is being enforced.
chosen_option: How the invariant is being enforced.
alternatives_considered: Other enforcement approaches evaluated.
confidence_score: Confidence in the enforcement approach (0.0-1.0).
rationale: Why this enforcement approach was chosen.
context_data: Current context window contents.
actor_state: Actor's current state.
relevant_resources: Resource IDs that influenced the decision.
Returns:
The recorded Decision.
Raises:
ValidationError: If required fields are missing.
"""
if not question or not question.strip():
raise ValidationError("question must not be empty")
if not chosen_option or not chosen_option.strip():
raise ValidationError("chosen_option must not be empty")
snapshot = capture_context_snapshot(
context_data=context_data,
actor_state=actor_state,
relevant_resources=relevant_resources,
)
self._logger.info(
"Recording invariant enforced decision",
question=question,
chosen_option=chosen_option,
)
try:
decision = self.decision_service.record_decision(
plan_id=self.plan_id,
decision_type=DecisionType.INVARIANT_ENFORCED,
question=question,
chosen_option=chosen_option,
parent_decision_id=self.parent_decision_id,
alternatives_considered=alternatives_considered,
confidence_score=confidence_score,
rationale=rationale,
context_snapshot=snapshot,
plan_phase=PlanPhase.STRATEGIZE,
)
self._logger.debug(
"Invariant enforced decision recorded",
decision_id=decision.decision_id,
)
return decision
except Exception as exc:
self._logger.warning(
"Failed to record invariant enforced decision",
error=str(exc),
error_type=type(exc).__name__,
)
raise
+1 -8
View File
@@ -3909,14 +3909,7 @@ def _build_explain_dict(
"is_correction": decision.is_correction,
"superseded": decision.is_superseded,
"created_at": decision.created_at.isoformat(),
"alternatives": [
{
"index": i + 1,
"description": alt,
"chosen": alt == decision.chosen_option,
}
for i, alt in enumerate(decision.alternatives_considered)
],
"alternatives_considered": decision.alternatives_considered,
}
if show_reasoning:
data["rationale"] = decision.rationale
@@ -135,7 +135,6 @@ class ReactiveEventBus:
handler=getattr(handler, "__qualname__", repr(handler)),
error_type=type(exc).__name__,
error=str(exc),
exc_info=True,
)
def subscribe(
@@ -241,9 +241,10 @@ class PluginLoader:
def validate_protocol(klass: type[Any], protocol: type[Any]) -> bool:
"""Check whether *klass* satisfies a ``@runtime_checkable`` Protocol.
Uses structural type checking via ``issubclass`` to validate the
protocol without instantiating the class. This prevents arbitrary
code execution from untrusted plugin constructors during validation.
Creates a temporary instance of *klass* (using a no-arg
constructor) and checks it against the protocol via
``isinstance``. If instantiation fails, falls back to a
structural check using ``issubclass``.
Args:
klass: The class to validate.
@@ -255,84 +256,23 @@ class PluginLoader:
Raises:
ProtocolMismatchError: If *klass* does not satisfy the protocol.
"""
# Try structural check first (safe, no instantiation).
# This prevents arbitrary code execution from plugin constructors.
issubclass_raised_type_error = False
# Try instance check first (most reliable for runtime_checkable)
try:
if issubclass(klass, protocol):
instance = klass()
if isinstance(instance, protocol):
return True
except TypeError:
# issubclass raises TypeError if protocol is not a valid
# @runtime_checkable Protocol (e.g. has a custom metaclass that
# overrides __subclasscheck__). Fall through to structural check.
issubclass_raised_type_error = True
# Fallback: perform a conservative structural check against the
# Protocol definition without instantiating the class. We inspect
# the protocol's declared members (callables and annotated names)
# and ensure the candidate class exposes them as class-level
# attributes or descriptors. This approximates structural
# conformance while avoiding constructor execution.
required: set[str] = set()
# Collect names from protocol __dict__ (methods, properties, etc.)
prot_dict = getattr(protocol, "__dict__", {})
for name, _value in prot_dict.items():
if name.startswith("__"):
continue
# Methods and descriptors will appear in the dict; annotations
# are handled below. We treat any non-data-magic name as a
# required member.
required.add(name)
# Also include annotated names (those declared with type hints)
ann = getattr(protocol, "__annotations__", {}) or {}
for name in ann:
if name.startswith("__"):
continue
required.add(name)
# If issubclass raised TypeError and the protocol declares no
# inspectable members, we cannot validate conformance — treat this
# as a mismatch rather than silently returning True for an
# unverifiable protocol.
if issubclass_raised_type_error and not required:
msg = (
f"Class '{getattr(klass, '__name__', str(klass))}' cannot be "
f"validated against protocol "
f"'{getattr(protocol, '__name__', str(protocol))}': "
f"issubclass raised TypeError and the protocol declares no "
f"inspectable members. Ensure the protocol is a valid "
f"@runtime_checkable Protocol."
)
raise ProtocolMismatchError(msg)
# Validate that the candidate class exposes each required name.
missing: list[str] = []
for name in sorted(required):
# getattr on the class returns descriptors (functions, property,
# staticmethod, etc.) without instantiation. hasattr is safe.
if not hasattr(klass, name):
missing.append(name)
continue
# If the protocol declared a callable (method), ensure the
# attribute on the class is callable or a descriptor.
prot_val = prot_dict.get(name)
if callable(prot_val):
cand = getattr(klass, name)
# Properties are descriptors and typically not callable; we
# accept them as present. For methods, require callable.
if not (callable(cand) or hasattr(cand, "fget")):
missing.append(name)
if not missing:
return True
except Exception:
# If instantiation fails, try subclass check
try:
if issubclass(klass, protocol):
return True
except TypeError:
pass
msg = (
f"Class '{getattr(klass, '__name__', str(klass))}' does not "
f"satisfy protocol '{getattr(protocol, '__name__', str(protocol))}'. "
f"Missing attributes: {', '.join(missing)}. Ensure the class "
f"implements all required methods and descriptors."
f"Class '{klass.__name__}' does not satisfy protocol "
f"'{protocol.__name__}'. Ensure the class implements all "
f"required methods and attributes."
)
raise ProtocolMismatchError(msg)
@@ -37,7 +37,6 @@ from cleveragents.resource.handlers._base import (
EMPTY_CONTENT_HASH,
BaseResourceHandler,
)
from cleveragents.resource.handlers.discovery import discover_devcontainers
from cleveragents.resource.handlers.protocol import (
CheckpointResult,
Content,
@@ -234,13 +233,10 @@ class FsDirectoryHandler(BaseResourceHandler):
)
def discover_children(self, *, resource: Resource) -> list[Resource]:
"""Discover subdirectories and devcontainer instances as child resources.
"""Discover subdirectories as child resources.
Each immediate subdirectory becomes a child ``fs-directory``
resource. Additionally, any ``.devcontainer/`` configurations
found at the resource location are registered as
``devcontainer-instance`` child resources with
``provisioning_state: discovered``.
resource.
Args:
resource: The parent fs-directory resource.
@@ -265,28 +261,6 @@ class FsDirectoryHandler(BaseResourceHandler):
)
children.append(child)
# Wire devcontainer auto-discovery (issue #4740)
dc_results = discover_devcontainers(location, "fs-directory")
for dc_result in dc_results:
config_name = dc_result.config_name or "default"
dc_child = Resource(
resource_id=self._derive_child_id(
resource.resource_id, f"devcontainer-{config_name}"
),
name=f"devcontainer-{config_name}",
resource_type_name="devcontainer-instance",
classification=resource.classification,
description=f"Devcontainer at {dc_result.config_path}",
location=location,
parents=[resource.resource_id],
properties={
"devcontainer_json_path": str(dc_result.config_path),
"config_name": config_name,
"provisioning_state": "discovered",
},
)
children.append(dc_child)
return children
# -- Checkpoint and rollback (issue #836) ------------------------------
@@ -6,18 +6,17 @@ strategy (with fallback to ``copy_on_write``).
Content CRUD operations (issue #827):
- ``read`` -- ``git show HEAD:<path>``
- ``write`` -- atomic file write inside the checkout
- ``delete`` -- ``os.remove`` + ``git rm --cached``
- ``list_children`` -- ``git ls-tree -r --name-only HEAD``
- ``diff`` -- ``git diff --no-index``
- ``discover_children`` -- ``git ls-tree --name-only HEAD`` + devcontainer discovery
- ``read`` ``git show HEAD:<path>``
- ``write`` atomic file write inside the checkout
- ``delete`` ``os.remove`` + ``git rm --cached``
- ``list_children`` ``git ls-tree -r --name-only HEAD``
- ``diff`` ``git diff --no-index``
- ``discover_children`` ``git ls-tree --name-only HEAD``
Based on:
- implementation_plan.md group M1.resource-handlers (L2254-L2271)
- Built-in type definition in resource_registry_service.py L62-98
- Issue #827 -- ResourceHandler CRUD and discovery methods
- Issue #4740 -- Wire discover_devcontainers() into git-checkout handler
- Issue #827 ResourceHandler CRUD and discovery methods
"""
from __future__ import annotations
@@ -37,7 +36,6 @@ from cleveragents.resource.handlers._base import (
EMPTY_CONTENT_HASH,
BaseResourceHandler,
)
from cleveragents.resource.handlers.discovery import discover_devcontainers
from cleveragents.resource.handlers.protocol import (
CheckpointResult,
Content,
@@ -305,20 +303,17 @@ class GitCheckoutHandler(BaseResourceHandler):
)
def discover_children(self, *, resource: Resource) -> list[Resource]:
"""Discover child resources via ``git ls-tree`` and devcontainer scan.
"""Discover child resources via ``git ls-tree``.
Each top-level directory in the repo becomes a child resource
of type ``fs-directory``. Additionally, any ``.devcontainer/``
configurations found at the resource location are registered as
``devcontainer-instance`` child resources with
``provisioning_state: discovered``.
of type ``fs-directory``.
Args:
resource: The parent git-checkout resource.
Returns:
List of child :class:`Resource` objects for top-level
directories and discovered devcontainer instances.
directories.
"""
location = self._require_location(resource)
@@ -351,28 +346,6 @@ class GitCheckoutHandler(BaseResourceHandler):
)
children.append(child)
# Wire devcontainer auto-discovery (issue #4740)
dc_results = discover_devcontainers(location, "git-checkout")
for dc_result in dc_results:
config_name = dc_result.config_name or "default"
dc_child = Resource(
resource_id=self._derive_child_id(
resource.resource_id, f"devcontainer-{config_name}"
),
name=f"devcontainer-{config_name}",
resource_type_name="devcontainer-instance",
classification=resource.classification,
description=f"Devcontainer at {dc_result.config_path}",
location=location,
parents=[resource.resource_id],
properties={
"devcontainer_json_path": str(dc_result.config_path),
"config_name": config_name,
"provisioning_state": "discovered",
},
)
children.append(dc_child)
return children
# -- Checkpoint and rollback (issue #836) ------------------------------
+1 -1
View File
@@ -146,7 +146,7 @@ if _TEXTUAL_AVAILABLE:
def action_help(self) -> None:
prompt = self.query_one("#prompt", PromptInput)
help_panel = self.query_one("#help-panel", HelpPanelOverlay)
context_name = resolve_help_context(prompt.value)
context_name = resolve_help_context(prompt.text)
help_panel.toggle(context_name)
def action_cycle_preset(self) -> None:
-14
View File
@@ -39,23 +39,9 @@ Screen {
}
#prompt {
layout: horizontal;
height: auto;
border: round $primary;
margin: 1 0 0 0;
align-horizontal: left;
}
#prompt > .prompt-symbol {
padding: 0 1;
content-align: center middle;
color: $text-primary;
}
#prompt > Input {
border: none;
width: 1fr;
height: auto;
}
#persona-bar {
-3
View File
@@ -19,7 +19,6 @@ class InputMode(StrEnum):
NORMAL = "normal"
COMMAND = "command"
SHELL = "shell"
MULTILINE = "multiline"
@dataclass(slots=True, frozen=True)
@@ -54,8 +53,6 @@ class InputModeRouter:
return InputMode.COMMAND
if stripped.startswith(("!", "$")):
return InputMode.SHELL
if "\n" in text or "```" in text:
return InputMode.MULTILINE
return InputMode.NORMAL
def process(self, text: str) -> ModeResult:
+11 -214
View File
@@ -1,111 +1,27 @@
"""Prompt widget with mode-aware symbol handling."""
"""Prompt widget and submitted message event."""
from __future__ import annotations
import importlib
from collections.abc import Iterable
from dataclasses import dataclass
from typing import TYPE_CHECKING, Any, Protocol, cast
from cleveragents.tui.input.modes import InputMode, InputModeRouter
class _InputChangedEvent(Protocol):
input: Any
value: str
class _MutableValueInput(Protocol):
value: str
def focus(self) -> None: ...
class _StaticWidget(Protocol):
def __init__(self, text: str = "", *args: object, **kwargs: object) -> None: ...
def update(self, text: str) -> None: ...
class _HorizontalWidget(Protocol):
def __init__(
self,
*,
name: str | None = None,
id: str | None = None,
classes: str | None = None,
) -> None: ...
def focus(self) -> None: ...
class _InputWidget(_MutableValueInput, Protocol):
def __init__(self, *args: object, **kwargs: object) -> None: ...
from typing import Any
def _load_input_base() -> type[Any]:
try:
return importlib.import_module("textual.widgets").Input
except Exception: # pragma: no cover - optional dependency
return importlib.import_module("textual.widgets").TextArea
except Exception: # pragma: no cover
class _FallbackInput:
value = ""
text = ""
def __init__(self, *args: object, **kwargs: object) -> None:
self.value = ""
def focus(self) -> None: # pragma: no cover - API parity
return None
self.text = ""
return _FallbackInput
def _load_static_base() -> type[Any]:
try:
return importlib.import_module("textual.widgets").Static
except Exception: # pragma: no cover - optional dependency
class _FallbackStatic:
def __init__(self, text: str = "", *args: object, **kwargs: object) -> None:
self._text = text
def update(self, text: str) -> None:
self._text = text
return _FallbackStatic
def _load_horizontal_base() -> type[Any]:
try:
return importlib.import_module("textual.containers").Horizontal
except Exception: # pragma: no cover - optional dependency
class _FallbackHorizontal:
def __init__(self, *args: object, **kwargs: object) -> None:
del args, kwargs
self._children: list[Any] = []
def compose(self) -> Iterable[Any]:
yield from self._children
def mount(self, widget: Any) -> None:
self._children.append(widget)
def focus(self) -> None: # pragma: no cover - API parity
return None
return _FallbackHorizontal
_InputBase = cast(type[_InputWidget], _load_input_base())
_StaticBase = cast(type[_StaticWidget], _load_static_base())
_HorizontalBase = cast(type[_HorizontalWidget], _load_horizontal_base())
_TEXTUAL_AVAILABLE = _InputBase.__module__.startswith("textual.")
if TYPE_CHECKING: # pragma: no cover - typing only
_ComposeResult = Iterable[Any]
else:
_ComposeResult = Iterable[Any]
_InputBase = _load_input_base()
@dataclass(slots=True, frozen=True)
@@ -115,129 +31,10 @@ class PromptSubmitted:
text: str
_PROMPT_NORMAL = chr(0x276F)
_PROMPT_COMMAND = "/"
_PROMPT_SHELL = "$"
_PROMPT_MULTILINE = chr(0x2630)
_PROMPT_SYMBOLS: dict[InputMode, str] = {
InputMode.NORMAL: _PROMPT_NORMAL,
InputMode.COMMAND: _PROMPT_COMMAND,
InputMode.SHELL: _PROMPT_SHELL,
InputMode.MULTILINE: _PROMPT_MULTILINE,
}
class _PromptSymbolMixin:
"""Mixin providing mode-aware prompt symbol updates.
Subclasses must define a ``value`` property (str) and implement
``_apply_symbol(symbol: str) -> None``.
.. note::
The spec (§29257) refers to "PromptTextArea" implying a
multi-line ``TextArea`` widget. However, ``TextArea.__init__()``
dropped the ``placeholder`` keyword in textual >=1.0, so this
implementation uses ``textual.widgets.Input`` (single-line)
instead. The ``MULTILINE`` mode is detected by content
(``\\n`` or triple-backtick) and the symbol updates correctly,
but the underlying ``Input`` widget cannot display multiple
lines. A future migration to ``TextArea`` (once placeholder
support is restored or replaced) would enable true multi-line
editing per spec §30209-30218.
"""
_current_symbol: str
def _apply_symbol(self, symbol: str) -> None:
raise NotImplementedError
def _update_symbol(self, raw_text: str) -> None:
symbol = _PROMPT_SYMBOLS[InputModeRouter.detect_mode(raw_text)]
self._current_symbol = symbol
self._apply_symbol(symbol)
@property
def prompt_symbol(self) -> str:
return self._current_symbol
class PromptInput(_InputBase):
"""TextArea widget wrapper with helper methods."""
def consume_text(self) -> PromptSubmitted:
text = self.value
self.value = ""
text = self.text
self.text = ""
return PromptSubmitted(text=text)
class _TextualPromptInput(_PromptSymbolMixin, _HorizontalBase):
"""Composite widget that displays a mode-aware prompt symbol."""
def __init__(
self,
placeholder: str = "",
*,
name: str | None = None,
id: str | None = None,
classes: str | None = None,
) -> None:
horizontal = cast(Any, super())
horizontal.__init__(name=name, id=id, classes=classes)
self._symbol_widget = _StaticBase("", classes="prompt-symbol")
input_id = f"{id}--input" if id else None
self._input = cast(
_MutableValueInput, _InputBase(placeholder=placeholder, id=input_id)
)
self._current_symbol = _PROMPT_SYMBOLS[InputMode.NORMAL]
self._update_symbol(self._input.value)
def compose(self) -> _ComposeResult:
yield self._symbol_widget
yield self._input
@property
def value(self) -> str:
return self._input.value
@value.setter
def value(self, new_value: str) -> None:
self._input.value = new_value
self._update_symbol(new_value)
def focus(self) -> None: # pragma: no cover - delegating to inner input
self._input.focus()
def on_input_changed(self, event: _InputChangedEvent) -> None:
if event.input is self._input:
self._update_symbol(event.value)
def _apply_symbol(self, symbol: str) -> None:
self._symbol_widget.update(symbol)
class _FallbackPromptInput(_PromptSymbolMixin):
"""Fallback prompt input used when Textual is unavailable."""
def __init__(self, placeholder: str = "", **_: object) -> None:
self.placeholder = placeholder
self._input = cast(_MutableValueInput, _InputBase())
self._current_symbol = _PROMPT_SYMBOLS[InputMode.NORMAL]
self._update_symbol(self._input.value)
@property
def value(self) -> str:
return getattr(self._input, "value", "")
@value.setter
def value(self, new_value: str) -> None:
self._input.value = new_value
self._update_symbol(new_value)
def focus(self) -> None: # pragma: no cover - API parity
focus = getattr(self._input, "focus", None)
if callable(focus):
focus()
def _apply_symbol(self, symbol: str) -> None:
self._current_symbol = symbol
PromptInput = _TextualPromptInput if _TEXTUAL_AVAILABLE else _FallbackPromptInput
+2 -1
View File
@@ -17,12 +17,13 @@ import yaml
from cleveragents.actor.registry import ActorRegistry
from cleveragents.actor.schema import ActorConfigSchema, ActorType, is_v3_yaml
from cleveragents.config.settings import ProviderDefaults
from cleveragents.config.settings import ProviderDefaults, Settings
from cleveragents.core.exceptions import NotFoundError
from cleveragents.domain.models.core.actor import Actor
from cleveragents.providers.registry import (
ProviderCapabilities,
ProviderInfo,
ProviderRegistry,
ProviderType,
)