diff --git a/docs/behavior.md b/docs/behavior.md index 61b30fbb..69cc89aa 100644 --- a/docs/behavior.md +++ b/docs/behavior.md @@ -133,6 +133,24 @@ documented in `docs/beads-prefix-migration.md`. Repository-local Beads stores (for example `/.beads`) are not used for Atelier planning state. They are treated as external ticket sources. +## Refined Planning Readiness Contract + +- A changeset is refined only when metadata includes + `execution.strategy: refined`. +- Refined changesets must include a typed `planning.contract_json` payload. +- Planner guardrails fail closed when refined contract data is missing, + malformed, or conflicts with finalize semantics. +- Refined planning remains in review with `planning.stage: planning_in_review` + until operator approval. +- Promotion approval persists auditable metadata: `planning.stage=approved`, + `planning.approved_by`, `planning.approved_at`, and + `planning.approval_message_id`. +- Worker startup applies the same shared validator across next-changeset, + explicit review-feedback, merge-conflict, and global selector paths. +- Startup claim selection fails closed for refined changesets that are not + approved or have invalid contract evidence. +- Non-refined changesets keep existing startup/promotion behavior. + ## Command behavior (high level) - `atelier init` diff --git a/docs/plans/2026-03-26-trycycle-ready-planner-contract-test-plan.md b/docs/plans/2026-03-26-trycycle-ready-planner-contract-test-plan.md new file mode 100644 index 00000000..06718228 --- /dev/null +++ b/docs/plans/2026-03-26-trycycle-ready-planner-contract-test-plan.md @@ -0,0 +1,288 @@ +# Trycycle-Ready Planner Contract Test Plan + +## Strategy reconciliation + +The transcript goal and the implementation plan are aligned: ship a fail-closed +planner/worker contract for trycycle-targeted changesets and record auditable +approval evidence before work becomes runnable. + +No strategy changes requiring user approval were identified. + +Adjustments made during reconciliation: + +1. Expanded worker claim-gate coverage to include global review-feedback and + global merge-conflict startup paths, not only explicit-epic paths. Why: the + implementation plan requires all startup claim sources to share the same + fail-closed gate. +1. Added differential drift checks that compare planner and worker readiness + outcomes through the shared validator boundary. Why: the implementation plan + requires one reused validator and deterministic readiness semantics. + +## Harness requirements + +1. **Trycycle metadata fixture builder (new, low complexity)** + - What it does: builds issue payloads with description-field metadata for + `trycycle.targeted`, `trycycle.contract_json`, `trycycle.plan_stage`, and + approval evidence fields. + - Exposes: helper API for valid, invalid, missing-field, and unapproved + payload variants. + - Depends-on tests: 1-9, 17. +1. **Startup protocol fake-service extension (extend, medium complexity)** + - What it does: updates fake startup services to expose + `trycycle_claim_eligible(issue) -> (bool, reason)` and configurable + eligibility outcomes. + - Exposes: per-issue eligibility maps and captured rejection reasons for + assertions in claim-path tests. + - Depends-on tests: 10-16, 18. +1. **Promotion approval-audit harness (extend, medium complexity)** + - What it does: extends promotion script fakes to capture description-field + writes, audit message creation, and read-after-write verification paths. + - Exposes: recorded lifecycle transitions, metadata updates, and approval + message ids. + - Depends-on tests: 7-9. +1. **Projected skill bootstrap shim for new core import (extend, low + complexity)** + - What it does: updates projected-runtime fake module trees so planner skill + scripts importing the new core contract module still run in tests. + - Exposes: repo-src precedence assertions with the new import graph. + - Depends-on tests: 19. + +## Test plan + +1. **Name:** Targeted changeset without `trycycle.contract_json` is rejected by + readiness evaluation **Type:** regression **Disposition:** new **Harness:** + Trycycle metadata fixture builder **Preconditions:** Issue description + contains `trycycle.targeted: true` and omits `trycycle.contract_json`. + **Actions:** Call `evaluate_issue_trycycle_readiness(issue)` from + `src/atelier/trycycle_contract.py`. **Expected outcome:** Result is + fail-closed (`ok=False`, targeted true) with a deterministic error mentioning + missing `trycycle.contract_json`. Source: implementation plan, User-Visible + Behavior Contract items 1-3 and 6. **Interactions:** Description-field parser + and contract JSON edge parsing. + +1. **Name:** Valid contract in `planning_in_review` is planner-valid but not + worker-claim-eligible **Type:** invariant **Disposition:** new **Harness:** + Trycycle metadata fixture builder **Preconditions:** Targeted issue has valid + typed contract JSON and `trycycle.plan_stage: planning_in_review` with no + approval fields. **Actions:** Call readiness evaluator and claim-eligibility + helper used by worker startup. **Expected outcome:** Contract validation + passes, but claim eligibility is false until explicit approval metadata + exists. Source: implementation plan, User-Visible Behavior Contract items 3-5 + and Contracts and Invariants item 5. **Interactions:** Shared validator + output consumed by planner and worker. + +1. **Name:** Approval evidence requires all audit fields **Type:** boundary + **Disposition:** new **Harness:** Trycycle metadata fixture builder + **Preconditions:** Targeted issue has valid contract and stage `approved`, + but one of `approved_by`, `approved_at`, or `approval_message_id` is + missing/malformed. **Actions:** Evaluate readiness for each missing or + malformed field case. **Expected outcome:** Each case is non-eligible with + deterministic diagnostics naming the missing evidence field. Source: + implementation plan, User-Visible Behavior Contract item 4. **Interactions:** + Field normalization and timestamp parsing logic. + +1. **Name:** Completion-definition lifecycle conflicts are rejected **Type:** + boundary **Disposition:** new **Harness:** Trycycle metadata fixture builder + **Preconditions:** Contract completion definition conflicts with finalize + semantics (for example, allows close without terminal PR state or integrated + SHA proof). **Actions:** Evaluate readiness on conflicting completion + definitions. **Expected outcome:** Readiness fails with conflict diagnostics; + compliant definitions pass. Source: implementation plan, Contracts and + Invariants item 7. **Interactions:** Shared completion-definition checker + with lifecycle rules. + +1. **Name:** Guardrails report fails targeted changesets missing contract + payload **Type:** scenario **Disposition:** extend **Harness:** Planner + script harness (`check_guardrails.py` module entrypoint) **Preconditions:** + Guardrails target list contains at least one targeted changeset without + `trycycle.contract_json`. **Actions:** Run `_evaluate_guardrails(...)` and + `main()` paths. **Expected outcome:** Report includes targeted-contract + violation text and retains existing guardrail reporting format. Source: + implementation plan Task 2 and User-Visible Behavior Contract items 2-3. + **Interactions:** Planner contract checks plus trycycle validator reuse. + +1. **Name:** Guardrails accept fully valid targeted planning payload **Type:** + integration **Disposition:** extend **Harness:** Planner script harness + (`check_guardrails.py`) **Preconditions:** Targeted issue has valid contract + payload with required plan-stage semantics for planner readiness. + **Actions:** Run guardrail evaluation for targeted child and epic-as-single + paths. **Expected outcome:** No trycycle-targeted violations are emitted for + valid payloads. Source: implementation plan Task 2 and Strategy Gate Decision + 3\. **Interactions:** Cross-check between authoring-contract and trycycle + checks. + +1. **Name:** Promotion blocks targeted work when contract validation fails + **Type:** scenario **Disposition:** extend **Harness:** Promotion + approval-audit harness (`promote_epic.py` main path) **Preconditions:** + Deferred epic has targeted executable unit with invalid or missing trycycle + contract. **Actions:** Run promotion script with and without `--yes`. + **Expected outcome:** Script exits fail-closed before lifecycle transitions; + stderr includes deterministic trycycle validation failure. Source: + implementation plan Task 3 and User-Visible Behavior Contract item 3. + **Interactions:** Store transition API, preview renderer, shared validator. + +1. **Name:** Promotion records complete approval audit metadata for targeted + child changesets **Type:** integration **Disposition:** extend **Harness:** + Promotion approval-audit harness **Preconditions:** Deferred epic with + deferred targeted child changeset, valid contract, explicit `--yes`. + **Actions:** Run `promote_epic.py --yes`, then inspect persisted metadata. + **Expected outcome:** Promoted targeted child includes + `trycycle.plan_stage=approved`, `approved_by`, `approved_at`, and + `approval_message_id`, and audit evidence is persisted/readable. Source: + implementation plan User-Visible Behavior Contract item 4 and Task 3. + **Interactions:** Description-field updates, message-thread persistence, + lifecycle transition sequencing. + +1. **Name:** Promotion records identical approval metadata for epic-as-single + executable path **Type:** integration **Disposition:** extend **Harness:** + Promotion approval-audit harness **Preconditions:** Deferred targeted epic + has no child changesets and valid contract payload. **Actions:** Run + `promote_epic.py --yes` for single-unit epic path. **Expected outcome:** Epic + record receives the same approval audit fields and evidence message behavior + as child-changeset path. Source: implementation plan Task 3 and File + Structure note for epic-as-single-unit coverage. **Interactions:** + Single-unit execution path in promotion script and store. + +1. **Name:** Next-changeset selection skips unapproved targeted candidates + **Type:** scenario **Disposition:** extend **Harness:** Startup protocol + fake-service extension **Preconditions:** Descendant candidate list contains + targeted unapproved candidate first, followed by eligible non-targeted + candidate. **Actions:** Run `next_changeset_service(...)`. **Expected + outcome:** First candidate is rejected by claim gate; second is returned. + Source: implementation plan User-Visible Behavior Contract item 5. + **Interactions:** Dependency checks, review-state checks, and trycycle gate + order. + +1. **Name:** Explicit-epic startup path fails closed when only targeted + unapproved work exists **Type:** scenario **Disposition:** extend + **Harness:** Startup protocol fake-service extension **Preconditions:** + `explicit_epic_id` points to claimable epic; next executable candidate is + targeted and unapproved. **Actions:** Run `run_startup_contract_service(...)` + with explicit epic mode. **Expected outcome:** Startup does not return a + runnable changeset and exits with deterministic fail-closed reason/output. + Source: implementation plan User-Visible Behavior Contract item 5 and + Contracts and Invariants item 6. **Interactions:** Explicit epic branch, + claimability checks, emission path. + +1. **Name:** Explicit merge-conflict selection cannot bypass trycycle approval + gate **Type:** scenario **Disposition:** extend **Harness:** Startup protocol + fake-service extension **Preconditions:** PR startup context enabled; + merge-conflict selector returns targeted unapproved changeset. **Actions:** + Run startup contract explicit path with `select_conflicted_changeset` + returning that candidate. **Expected outcome:** Startup rejects the candidate + and continues scanning or exits non-claimable; it does not return the blocked + changeset. Source: implementation plan User-Visible Behavior Contract item 5. + **Interactions:** Merge-conflict selector, startup reason routing. + +1. **Name:** Explicit review-feedback selection cannot bypass trycycle approval + gate **Type:** scenario **Disposition:** extend **Harness:** Startup protocol + fake-service extension **Preconditions:** PR startup context enabled; + review-feedback selector returns targeted unapproved changeset. **Actions:** + Run startup contract explicit path with `select_review_feedback_changeset`. + **Expected outcome:** Candidate is rejected by the shared gate; startup does + not claim it. Source: implementation plan User-Visible Behavior Contract item + 5\. **Interactions:** Review-feedback selector and claim gating. + +1. **Name:** Global review-feedback and merge-conflict selectors are gated with + the same trycycle check **Type:** integration **Disposition:** extend + **Harness:** Startup protocol fake-service extension **Preconditions:** + Global selectors return targeted unapproved candidates from other epic + families. **Actions:** Run non-explicit startup contract flow through global + selector stages. **Expected outcome:** Global candidates are rejected + identically to explicit selectors; no bypass path remains. Source: + implementation plan Contracts and Invariants item 4 and Task 4. + **Interactions:** Global selector cache, claimability checks, fallback paths. + +1. **Name:** Approved targeted changesets remain selectable across all claim + sources **Type:** invariant **Disposition:** new **Harness:** Startup + protocol fake-service extension **Preconditions:** Targeted candidates + include complete approved metadata and valid contract payload. **Actions:** + Exercise explicit next-changeset, explicit review-feedback, explicit + merge-conflict, and global selector paths. **Expected outcome:** All paths + accept approved targeted candidates when other lifecycle constraints are + satisfied. Source: implementation plan User-Visible Behavior Contract items + 4-5. **Interactions:** Shared gate with existing claimability predicates. + +1. **Name:** Non-trycycle startup behavior remains unchanged **Type:** + regression **Disposition:** extend **Harness:** Existing startup/lifecycle + matrix harnesses **Preconditions:** Issues do not carry + `trycycle.targeted: true`. **Actions:** Re-run existing startup and lifecycle + matrix scenarios. **Expected outcome:** Existing non-trycycle selection and + lifecycle outcomes remain unchanged. Source: implementation plan User-Visible + Behavior Contract item 6 and Contracts and Invariants item 1. + **Interactions:** Baseline startup selection, lifecycle matrix predicates. + +1. **Name:** Shared-validator differential parity across planner and worker + entrypoints **Type:** differential **Disposition:** new **Harness:** Trycycle + metadata fixture builder + startup fake extensions **Preconditions:** Matrix + of targeted payloads: valid approved, valid unapproved, malformed JSON, + missing fields, completion conflict. **Actions:** Evaluate each payload + through `trycycle_contract.evaluate_issue_trycycle_readiness`, guardrails + integration wrapper, promotion preflight path, and worker claim-eligibility + adapter. **Expected outcome:** All entrypoints agree on eligible vs + non-eligible and deterministic summary class. Source: implementation plan + Strategy Gate Decision 3. **Interactions:** Cross-module contract reuse and + error-summary formatting. + +1. **Name:** Runtime adapters call shared validator (no ad hoc parsing) + **Type:** integration **Disposition:** extend **Harness:** + `test_work_startup_runtime.py` adapter harness **Preconditions:** + Monkeypatched validator helper records calls; adapter service receives + targeted and non-targeted issues. **Actions:** Call + `_NextChangesetService.trycycle_claim_eligible(...)` and + `_StartupContractService.trycycle_claim_eligible(...)`. **Expected outcome:** + Both adapters delegate to shared validator and return normalized + `(eligible, reason)` tuples. Source: implementation plan Task 4 and File + Structure notes. **Interactions:** Runtime adapter boundary to core validator + module. + +1. **Name:** Projected planner scripts keep bootstrap compatibility after adding + shared trycycle contract import **Type:** integration **Disposition:** extend + **Harness:** Projected skill bootstrap shim **Preconditions:** Projected + guardrails and promotion scripts run in agent home with repo/src precedence + test setup. **Actions:** Execute projected script `--help` runs under + bootstrap tests. **Expected outcome:** Scripts import successfully and still + resolve repo source modules before installed packages. Source: implementation + plan Cutover and Regression Risks item 3. **Interactions:** projected + bootstrap, Python path ordering, skill packaging. + +1. **Name:** Planner template and skill docs encode trycycle approval gate + contract **Type:** regression **Disposition:** extend **Harness:** + `tests/atelier/test_skills.py` and + `tests/atelier/test_planner_agents_template.py` **Preconditions:** Updated + `AGENTS.planner.md.tmpl`, `plan-changesets/SKILL.md`, + `plan-changeset-guardrails/SKILL.md`, and `plan-promote-epic/SKILL.md`. + **Actions:** Run skill/template content assertions against packaged files. + **Expected outcome:** Guidance explicitly references trycycle fields, + approval gate, and worker fail-closed behavior. Source: implementation plan + Task 5 and User-Visible Behavior Contract items 3-5. **Interactions:** + Packaged skill sync/install and template rendering paths. + +## Coverage summary + +Covered action space: + +1. Planner guardrail CLI actions: `--epic-id`, `--changeset-id`, targeted + contract validation reporting. +1. Promotion CLI actions: preview vs `--yes` promotion, targeted validation + gate, approval evidence persistence for child and epic-as-single paths. +1. Worker startup claim actions: explicit epic, next-changeset, review-feedback, + merge-conflict, global review-feedback, global merge-conflict, and no-bypass + fail-closed behavior. +1. Shared validator parity: planner, promotion, and worker paths all consume one + deterministic contract checker. +1. Packaging and documentation surface: projected skill bootstrap compatibility + plus planner guidance updates. + +Explicit exclusions (per current implementation strategy): + +1. End-to-end networked GitHub review API behavior is not exercised directly in + these tests. Risk: integration bugs in upstream review fetchers could still + affect startup candidate discovery before gating. +1. Full `atelier work` subprocess E2E is not expanded for this slice; startup + contract/service tests remain the primary behavior proof. Risk: CLI wiring + regressions outside startup contract adapters could require follow-up + command-level tests. +1. No performance benchmark suite is added because this change is contract + gating, not throughput-sensitive logic. Risk: minimal; catastrophic + regressions are still caught by normal test runtime and CI gates. diff --git a/docs/plans/2026-03-26-trycycle-ready-planner-contract.md b/docs/plans/2026-03-26-trycycle-ready-planner-contract.md new file mode 100644 index 00000000..e93462d3 --- /dev/null +++ b/docs/plans/2026-03-26-trycycle-ready-planner-contract.md @@ -0,0 +1,629 @@ +# Trycycle-Ready Planner Contract and Approval Gate Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use trycycle-executing to +> implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for +> tracking. + +**Goal:** Add a fail-closed planner+worker contract so trycycle-targeted +changesets cannot be claimed until a validated plan package and explicit +operator approval are both recorded and auditable. + +**Architecture:** Introduce a typed trycycle plan-contract model and shared +validator in core `src/atelier` code, then route planner guardrails, +promotion-time approval capture, and worker startup selection through that one +validator so readiness semantics are identical everywhere. Keep canonical +lifecycle statuses unchanged (`deferred|open|in_progress|blocked|closed`) and +represent staged review with namespaced metadata fields to avoid destabilizing +existing lifecycle logic. + +**Tech Stack:** Python 3.11, Pydantic models, Atelier store/beads adapters, +planner skill scripts, worker startup pipeline, pytest. + +______________________________________________________________________ + +## User-Visible Behavior Contract + +1. A changeset becomes trycycle-targeted only when it carries + `trycycle.targeted: true` metadata. +1. Trycycle-targeted changesets must include a typed plan payload under + `trycycle.contract_json`. +1. Planner validation must fail closed for trycycle-targeted changesets before + they are promoted to runnable/open work. +1. Operator approval is explicit and auditable. Promotion records: + - `trycycle.plan_stage: approved` + - `trycycle.approved_by` + - `trycycle.approved_at` + - `trycycle.approval_message_id` +1. Worker startup refuses to select or resume trycycle-targeted changesets whose + contract is invalid, missing, or unapproved (including explicit-epic, + next-changeset, merge-conflict, and review-feedback selection paths). +1. Non-trycycle changesets retain current behavior with no new gating. + +## Contracts and Invariants + +1. Lifecycle contract remains canonical and unchanged. No new status values are + added to `LifecycleStatus`. +1. `planning_in_review` is represented as metadata + (`trycycle.plan_stage: planning_in_review`) rather than lifecycle state. +1. Validation is deterministic and shared. The same checker must be reused by: + - planner guardrail scripts + - promotion/approval path + - worker startup claim gate +1. Worker startup claim gate applies to all claim sources, not only + next-changeset selection. +1. Approval is mandatory for trycycle-targeted claim eligibility. +1. Missing or malformed contract metadata is treated as non-runnable for + trycycle-targeted changesets. +1. Completion definition must not conflict with existing finalize semantics: + close only when PR lifecycle is terminal (`merged|closed`) or when + `changeset.integrated_sha` proves integration. + +## Strategy Gate Decisions (Locked) + +1. Use namespaced metadata fields, not a new lifecycle status. Reason: adding + lifecycle states would touch broad store/lifecycle/finalize logic and risks + regressions unrelated to #719. +1. Use one typed JSON payload field (`trycycle.contract_json`) plus small, + auditable metadata fields for stage/approval. Reason: typed payload gives + strict schema guarantees; small scalar fields keep approval evidence easy to + inspect with current description-field tooling. +1. Reuse one shared core validator for planner and worker paths. Reason: avoids + drift where planner and worker disagree on readiness. +1. Make worker gate selective (only `trycycle.targeted: true`). Reason: + preserves existing execution behavior for non-trycycle work. + +## File Structure + +- Create: `src/atelier/trycycle_contract.py` + - Typed models, parser/serializer, validation helpers, completion-definition + conflict checks, evidence summary rendering. +- Create: `tests/atelier/test_trycycle_contract.py` + - Unit tests for parsing, schema validation, quality checks, and lifecycle + completion-definition conflict detection. +- Modify: + `src/atelier/skills/plan-changeset-guardrails/scripts/check_guardrails.py` + - Invoke shared trycycle validator, surface deterministic violations. +- Modify: `tests/atelier/skills/test_plan_changeset_guardrails_script.py` + - Add trycycle-targeted valid/invalid/missing-field coverage. +- Modify: `src/atelier/skills/plan-promote-epic/scripts/promote_epic.py` + - Enforce trycycle validation before promotion, persist approval metadata, + write evidence summary note, emit approval message. +- Modify: `tests/atelier/skills/test_plan_promote_epic_script.py` + - Add tests for required approval markers/evidence persistence. +- Modify: `src/atelier/worker/session/startup.py` + - Extend startup/next-changeset protocols to require trycycle claim + eligibility and apply the same gate to all startup selection paths. +- Modify: `src/atelier/worker/work_startup_runtime.py` + - Implement startup + next-changeset service adapter methods using shared + validator. +- Modify: `tests/atelier/worker/test_session_startup.py` + - Verify worker skips/rejects unapproved or invalid trycycle-targeted + changesets across claim paths. +- Modify: `tests/atelier/worker/test_session_next_changeset.py` + - Verify next-changeset service uses shared trycycle gate for leaf-epic and + descendant candidate selection. +- Modify: `tests/atelier/worker/test_work_startup_runtime.py` + - Verify startup runtime adapters use store/contract helpers (not ad hoc + parsing), including trycycle eligibility adapters. +- Modify: `tests/atelier/worker/test_lifecycle_matrix.py` + - Add matrix coverage for trycycle claim-gating invariants. +- Modify: `tests/atelier/skills/test_projected_skill_runtime_bootstrap.py` + - Keep projected runtime fixtures compatible with any new core imports from + planner scripts. +- Modify: `src/atelier/templates/AGENTS.planner.md.tmpl` + - Add trycycle-ready planning + explicit approval guidance. +- Modify: `src/atelier/skills/plan-changesets/SKILL.md` +- Modify: `src/atelier/skills/plan-changeset-guardrails/SKILL.md` +- Modify: `src/atelier/skills/plan-promote-epic/SKILL.md` + - Document new contract fields and approval gate. +- Modify: `docs/behavior.md` + - Add runtime behavior notes for trycycle-targeted claim gating. + +## Cutover and Regression Risks + +1. Risk: worker refuses all work due to over-broad gating. Mitigation: gate only + when `trycycle.targeted: true`; add explicit tests for unchanged non-trycycle + flow. +1. Risk: startup bypasses gate via review-feedback/merge-conflict selection. + Mitigation: apply one shared `trycycle_claim_eligible` check before every + startup return path that yields a concrete changeset id. +1. Risk: planner script import drift in projected skill runtime tests. + Mitigation: update bootstrap fixture module stubs when introducing + `trycycle_contract` import. +1. Risk: approval metadata write succeeds partially. Mitigation: use store/beads + update + read-after-write verification (existing atomic verification pattern + used in store adapters). +1. Risk: completion-definition checker blocks valid plans. Mitigation: codify + allowed forms aligned to current worker prompt/finalize contract and cover + with focused tests. + +### Task 1: Add Typed Trycycle Contract Model and Shared Validator + +**Files:** + +- Create: `src/atelier/trycycle_contract.py` + +- Test: `tests/atelier/test_trycycle_contract.py` + +- [ ] **Step 1: Identify or write the failing test** + +```python +# tests/atelier/test_trycycle_contract.py +from atelier import trycycle_contract + + +def test_validate_contract_accepts_complete_payload() -> None: + issue = { + "description": ( + "trycycle.targeted: true\n" + "trycycle.plan_stage: planning_in_review\n" + "trycycle.contract_json: {\"objective\":\"...\",...}\n" + ) + } + result = trycycle_contract.evaluate_issue_trycycle_readiness(issue) + assert result.ok is True +``` + +- [ ] **Step 2: Run test to verify it fails** + +Run: `uv run pytest tests/atelier/test_trycycle_contract.py -v` Expected: FAIL +(`ImportError` / missing module or missing API). + +- [ ] **Step 3: Write minimal implementation** + +```python +# src/atelier/trycycle_contract.py +class TrycycleContract(BaseModel): + objective: str + non_goals: tuple[str, ...] + acceptance_criteria: tuple[AcceptanceCriterion, ...] + scope: ScopeBoundary + verification_plan: tuple[str, ...] + risks: tuple[RiskItem, ...] + escalation_conditions: tuple[str, ...] + completion_definition: CompletionDefinition + + +def evaluate_issue_trycycle_readiness(issue: Mapping[str, object]) -> ReadinessResult: + # parse description fields + # if trycycle.targeted != true: return ok/non-targeted + # parse trycycle.contract_json and stage/approval fields + # run schema + quality + lifecycle conflict checks +``` + +- [ ] **Step 4: Run test to verify it passes** + +Run: `uv run pytest tests/atelier/test_trycycle_contract.py -v` Expected: PASS. + +- [ ] **Step 5: Refactor and verify** + +Tighten parse/normalization boundaries and add dedicated tests for: + +- malformed JSON +- missing escalation conditions +- non-testable acceptance criteria (missing evidence) +- conflicting completion definition + +Run: `uv run pytest tests/atelier/test_trycycle_contract.py -v` Expected: all +PASS. + +- [ ] **Step 6: Commit** + +```bash +git add tests/atelier/test_trycycle_contract.py src/atelier/trycycle_contract.py +git commit -m "feat(planner): add typed trycycle contract validator" \ + -m "- add shared trycycle readiness schema and parser\n- add deterministic readiness validation and diagnostics" +``` + +### Task 2: Enforce Trycycle Contract in Planner Guardrail Checks + +**Files:** + +- Modify: + `src/atelier/skills/plan-changeset-guardrails/scripts/check_guardrails.py` + +- Test: `tests/atelier/skills/test_plan_changeset_guardrails_script.py` + +- [ ] **Step 1: Identify or write the failing test** + +```python +# tests/atelier/skills/test_plan_changeset_guardrails_script.py + +def test_guardrails_flags_trycycle_target_missing_contract() -> None: + child = {"id": "at-epic.1", "description": "trycycle.targeted: true"} + report = module._evaluate_guardrails( + epic_issue=None, + child_changesets=[], + target_changesets=[child], + ) + assert any("trycycle.contract_json" in item for item in report.violations) +``` + +- [ ] **Step 2: Run test to verify it fails** + +Run: +`uv run pytest tests/atelier/skills/test_plan_changeset_guardrails_script.py -k trycycle -v` +Expected: FAIL. + +- [ ] **Step 3: Write minimal implementation** + +```python +from atelier import trycycle_contract + +readiness = trycycle_contract.evaluate_issue_trycycle_readiness(issue) +if readiness.targeted and not readiness.contract_present: + violations.append(f"{issue_id}: missing trycycle.contract_json") +if readiness.targeted and readiness.errors: + violations.extend(f"{issue_id}: {err}" for err in readiness.errors) +``` + +- [ ] **Step 4: Run test to verify it passes** + +Run: +`uv run pytest tests/atelier/skills/test_plan_changeset_guardrails_script.py -k trycycle -v` +Expected: PASS. + +- [ ] **Step 5: Refactor and verify** + +Add test coverage for: + +- valid targeted payload (no violations) +- completion-definition conflict violation +- explicit `planning_in_review` stage requirement + +Run: +`uv run pytest tests/atelier/skills/test_plan_changeset_guardrails_script.py -v` +Expected: all PASS. + +- [ ] **Step 6: Commit** + +```bash +git add src/atelier/skills/plan-changeset-guardrails/scripts/check_guardrails.py \ + tests/atelier/skills/test_plan_changeset_guardrails_script.py +git commit -m "feat(planner): validate trycycle-ready contracts in guardrails" \ + -m "- wire guardrail script to shared trycycle readiness validator\n- add deterministic contract violation coverage" +``` + +### Task 3: Add Promotion-Time Approval Capture and Evidence Persistence + +**Files:** + +- Modify: `src/atelier/skills/plan-promote-epic/scripts/promote_epic.py` + +- Test: `tests/atelier/skills/test_plan_promote_epic_script.py` + +- [ ] **Step 1: Identify or write the failing test** + +```python +# tests/atelier/skills/test_plan_promote_epic_script.py + +def test_promote_epic_blocks_trycycle_target_without_valid_contract(...): + # targeted child with invalid payload + # expect SystemExit(1) and explicit trycycle validation error + + +def test_promote_epic_records_trycycle_approval_metadata(...): + # targeted valid child + --yes + # expect description fields include approved_by/approved_at/stage + + +def test_promote_epic_records_trycycle_approval_metadata_for_epic_single_unit(...): + # targeted epic with no child changesets + --yes + # expect same approval fields/message id on epic record +``` + +- [ ] **Step 2: Run test to verify it fails** + +Run: +`uv run pytest tests/atelier/skills/test_plan_promote_epic_script.py -k trycycle -v` +Expected: FAIL. + +- [ ] **Step 3: Write minimal implementation** + +```python +# in promote_epic.py before transition_lifecycle +readiness = trycycle_contract.evaluate_issue_trycycle_readiness(child_issue) +if readiness.targeted and not readiness.ok: + raise RuntimeError(f"{child_id} trycycle readiness failed: {readiness.summary}") + +# on --yes for targeted issue +# set: trycycle.plan_stage=approved +# set: trycycle.approved_by= +# set: trycycle.approved_at= +# set: trycycle.approval_message_id= +# append concise evidence note +# create work-threaded approval message and persist message id +# apply updates to every promoted targeted executable record +# (child changesets and epic-as-single-unit when no children exist) +``` + +- [ ] **Step 4: Run test to verify it passes** + +Run: +`uv run pytest tests/atelier/skills/test_plan_promote_epic_script.py -k trycycle -v` +Expected: PASS. + +- [ ] **Step 5: Refactor and verify** + +Add coverage for: + +- non-targeted changesets unchanged +- target with `planning_in_review` stage transitions to `approved` +- persisted approval message id is included in metadata +- single-unit epic target records identical approval metadata + +Run: `uv run pytest tests/atelier/skills/test_plan_promote_epic_script.py -v` +Expected: all PASS. + +- [ ] **Step 6: Commit** + +```bash +git add src/atelier/skills/plan-promote-epic/scripts/promote_epic.py \ + tests/atelier/skills/test_plan_promote_epic_script.py +git commit -m "feat(planner): require and record trycycle operator approval" \ + -m "- block promotion when trycycle contract validation fails\n- persist approval metadata and planning evidence audit trail" +``` + +### Task 4: Gate Worker Claim Selection on Trycycle Readiness + +**Files:** + +- Modify: `src/atelier/worker/session/startup.py` + +- Modify: `src/atelier/worker/work_startup_runtime.py` + +- Test: `tests/atelier/worker/test_session_startup.py` + +- Test: `tests/atelier/worker/test_session_next_changeset.py` + +- Test: `tests/atelier/worker/test_work_startup_runtime.py` + +- Test: `tests/atelier/worker/test_lifecycle_matrix.py` + +- [ ] **Step 1: Identify or write the failing test** + +```python +# tests/atelier/worker/test_session_startup.py + +def test_next_changeset_skips_unapproved_trycycle_changeset() -> None: + # first candidate targeted+invalid, second candidate non-targeted valid + # expect second candidate selected + + +def test_explicit_epic_with_only_invalid_trycycle_changeset_returns_none() -> None: + # expect no actionable changeset and startup reason remains fail-closed + + +def test_review_feedback_selection_rejects_unapproved_trycycle_changeset() -> None: + # explicit/global review-feedback selection yields a targeted unapproved id + # expect startup keeps searching or exits with deterministic non-claim reason + + +def test_merge_conflict_selection_rejects_unapproved_trycycle_changeset() -> None: + # explicit/global merge-conflict selection yields targeted unapproved id + # expect startup refuses claim and does not bypass gate +``` + +- [ ] **Step 2: Run test to verify it fails** + +Run: `uv run pytest tests/atelier/worker/test_session_startup.py -k trycycle -v` +`uv run pytest tests/atelier/worker/test_session_next_changeset.py -k trycycle -v` +Expected: FAIL. + +- [ ] **Step 3: Write minimal implementation** + +```python +# startup.NextChangesetService + +def trycycle_claim_eligible(self, issue: dict[str, object]) -> tuple[bool, str | None]: ... + +# startup.StartupContractService + +def trycycle_claim_eligible(self, issue: dict[str, object]) -> tuple[bool, str | None]: ... + +# next_changeset_service loop +eligible, reason = service.trycycle_claim_eligible(issue) +if not eligible: + continue + +# run_startup_contract_service paths that return a concrete changeset id +# (explicit/global review-feedback, explicit/global merge-conflict) +# must load the selected issue and apply the same gate before returning. +``` + +```python +# work_startup_runtime._NextChangesetService + +def trycycle_claim_eligible(self, issue): + result = trycycle_contract.evaluate_issue_trycycle_readiness(issue) + if not result.targeted: + return True, None + if result.approved: + return True, None + return False, result.summary + +# work_startup_runtime._StartupContractService +# delegates to the same helper/validator as _NextChangesetService +``` + +- [ ] **Step 4: Run test to verify it passes** + +Run: `uv run pytest tests/atelier/worker/test_session_startup.py -k trycycle -v` +Expected: PASS. + +- [ ] **Step 5: Refactor and verify** + +Add matrix/regression coverage: + +- targeted+approved is selectable +- targeted+missing approval is not selectable +- explicit/global review-feedback and merge-conflict selections cannot bypass + trycycle approval gate +- non-targeted flow unchanged +- fake/test services implement the new protocol method + +Run: + +- `uv run pytest tests/atelier/worker/test_session_startup.py -v` + +- `uv run pytest tests/atelier/worker/test_session_next_changeset.py -v` + +- `uv run pytest tests/atelier/worker/test_work_startup_runtime.py -v` + +- `uv run pytest tests/atelier/worker/test_lifecycle_matrix.py -v` Expected: all + PASS. + +- [ ] **Step 6: Commit** + +```bash +git add src/atelier/worker/session/startup.py src/atelier/worker/work_startup_runtime.py \ + tests/atelier/worker/test_session_startup.py \ + tests/atelier/worker/test_session_next_changeset.py \ + tests/atelier/worker/test_work_startup_runtime.py \ + tests/atelier/worker/test_lifecycle_matrix.py +git commit -m "feat(worker): fail closed on non-approved trycycle changesets" \ + -m "- add startup claim gate for trycycle contract readiness across all claim paths\n- preserve existing non-trycycle claim behavior" +``` + +### Task 5: Keep Skill Runtime Bootstrap and Planner Guidance Consistent + +**Files:** + +- Modify: `tests/atelier/skills/test_projected_skill_runtime_bootstrap.py` + +- Modify: `src/atelier/templates/AGENTS.planner.md.tmpl` + +- Modify: `src/atelier/skills/plan-changesets/SKILL.md` + +- Modify: `src/atelier/skills/plan-changeset-guardrails/SKILL.md` + +- Modify: `src/atelier/skills/plan-promote-epic/SKILL.md` + +- Modify: `docs/behavior.md` + +- Test: `tests/atelier/test_planner_agents_template.py` + +- Test: `tests/atelier/test_skills.py` + +- [ ] **Step 1: Identify or write the failing test** + +```python +# tests/atelier/test_skills.py + +def test_plan_promote_epic_skill_mentions_trycycle_approval_gate() -> None: + text = skills.load_packaged_skills()["plan-promote-epic"].text + assert "trycycle.plan_stage" in text +``` + +- [ ] **Step 2: Run test to verify it fails** + +Run: +`uv run pytest tests/atelier/test_skills.py tests/atelier/test_planner_agents_template.py -v` +Expected: FAIL. + +- [ ] **Step 3: Write minimal implementation** + +Update planner guidance docs/templates to require: + +- `@plan-changeset-guardrails` validation for trycycle-targeted work + +- explicit operator confirmation and metadata fields + +- worker fail-closed expectation for unapproved trycycle changesets + +- [ ] **Step 4: Run test to verify it passes** + +Run: +`uv run pytest tests/atelier/test_skills.py tests/atelier/test_planner_agents_template.py -v` +Expected: PASS. + +- [ ] **Step 5: Refactor and verify** + +Ensure projected bootstrap fixtures still satisfy script imports. + +Run: +`uv run pytest tests/atelier/skills/test_projected_skill_runtime_bootstrap.py -k guardrails -v` +Expected: PASS. + +- [ ] **Step 6: Commit** + +```bash +git add tests/atelier/skills/test_projected_skill_runtime_bootstrap.py \ + src/atelier/templates/AGENTS.planner.md.tmpl \ + src/atelier/skills/plan-changesets/SKILL.md \ + src/atelier/skills/plan-changeset-guardrails/SKILL.md \ + src/atelier/skills/plan-promote-epic/SKILL.md docs/behavior.md \ + tests/atelier/test_planner_agents_template.py tests/atelier/test_skills.py +git commit -m "docs(planner): codify trycycle-ready approval workflow" \ + -m "- align planner template and skill docs with trycycle gating contract\n- update bootstrap and docs coverage for new shared contract usage" +``` + +### Task 6: Run Full Verification and Land + +**Files:** + +- Modify: any files touched by fixes discovered during verification. + +- [ ] **Step 1: Run focused regression suite** + +Run: + +```bash +uv run pytest tests/atelier/test_trycycle_contract.py -v +uv run pytest tests/atelier/skills/test_plan_changeset_guardrails_script.py -v +uv run pytest tests/atelier/skills/test_plan_promote_epic_script.py -v +uv run pytest tests/atelier/worker/test_session_startup.py -v +uv run pytest tests/atelier/worker/test_session_next_changeset.py -v +uv run pytest tests/atelier/worker/test_work_startup_runtime.py -v +uv run pytest tests/atelier/worker/test_lifecycle_matrix.py -v +``` + +Expected: all PASS. + +- [ ] **Step 2: Run canonical project gates** + +Run: + +```bash +just format +just lint +just test +``` + +Expected: all PASS. + +- [ ] **Step 3: If any gate fails, fix root cause (not tests)** + +Prefer production/contract fixes over weakening assertions. + +- [ ] **Step 4: Re-run failing gate(s) and full required gates** + +Re-run whichever command failed, then re-run: +`just format && just lint && just test` Expected: all PASS. + +- [ ] **Step 5: Final review of behavior invariants** + +Confirm before merge: + +- non-trycycle flows unchanged + +- trycycle-targeted flows fail closed unless approved + +- approval/evidence metadata is auditable from beads/messages + +- [ ] **Step 6: Commit verification fixes** + +```bash +git add -A +git commit -m "test(planner): finalize trycycle-ready claim-gate coverage" \ + -m "- close verification gaps across planner validation and worker gating\n- ensure format/lint/test gates pass with fail-closed invariants" +``` + +## Definition of Done + +1. Planner can produce and validate typed trycycle-ready payloads. +1. Promotion path enforces validation and captures explicit approval evidence. +1. Worker startup claim path rejects non-compliant trycycle-targeted changesets. +1. Metadata/messages provide auditable evidence summary + approval record. +1. Existing non-trycycle lifecycle behavior remains unchanged. +1. `just format`, `just lint`, and `just test` pass. diff --git a/src/atelier/refined_planning_contract.py b/src/atelier/refined_planning_contract.py new file mode 100644 index 00000000..12e83d9b --- /dev/null +++ b/src/atelier/refined_planning_contract.py @@ -0,0 +1,392 @@ +"""Planner/worker refined-planning contract models and readiness validators.""" + +from __future__ import annotations + +import datetime as dt +import json +import re +from collections.abc import Mapping +from typing import Literal + +from pydantic import BaseModel, ConfigDict, Field, ValidationError, field_validator + +from . import beads + +_EXECUTION_STRATEGY_FIELD = "execution.strategy" +_CONTRACT_JSON_FIELD = "planning.contract_json" +_PLAN_STAGE_FIELD = "planning.stage" +_APPROVED_BY_FIELD = "planning.approved_by" +_APPROVED_AT_FIELD = "planning.approved_at" +_APPROVAL_MESSAGE_ID_FIELD = "planning.approval_message_id" +_APPROVAL_MESSAGE_ID_PATTERN = re.compile(r"^[a-z0-9][a-z0-9._:-]*$") +_PLANNING_IN_REVIEW = "planning_in_review" +_APPROVED = "approved" +_REFINED_STRATEGY = "refined" + + +class AcceptanceCriterion(BaseModel): + """One measurable acceptance criterion for refined planning.""" + + model_config = ConfigDict(extra="forbid") + + statement: str = Field(min_length=1) + evidence: tuple[str, ...] = Field(min_length=1) + + @field_validator("statement") + @classmethod + def _validate_statement(cls, value: str) -> str: + cleaned = value.strip() + if not cleaned: + raise ValueError("statement must be non-empty") + return cleaned + + @field_validator("evidence") + @classmethod + def _validate_evidence(cls, value: tuple[str, ...]) -> tuple[str, ...]: + cleaned = tuple(item.strip() for item in value if item.strip()) + if not cleaned: + raise ValueError("evidence must include at least one testable artifact") + return cleaned + + +class ScopeBoundary(BaseModel): + """Contract scope boundary describing included/excluded surfaces.""" + + model_config = ConfigDict(extra="forbid") + + includes: tuple[str, ...] = Field(min_length=1) + excludes: tuple[str, ...] = Field(min_length=1) + + @field_validator("includes", "excludes") + @classmethod + def _validate_lines(cls, value: tuple[str, ...]) -> tuple[str, ...]: + cleaned = tuple(item.strip() for item in value if item.strip()) + if not cleaned: + raise ValueError("scope boundaries must include non-empty entries") + return cleaned + + +class RiskItem(BaseModel): + """Risk statement and mitigation plan.""" + + model_config = ConfigDict(extra="forbid") + + risk: str = Field(min_length=1) + mitigation: str = Field(min_length=1) + + @field_validator("risk", "mitigation") + @classmethod + def _validate_text(cls, value: str) -> str: + cleaned = value.strip() + if not cleaned: + raise ValueError("risk and mitigation fields must be non-empty") + return cleaned + + +class CompletionDefinition(BaseModel): + """Completion-rule configuration constrained by worker finalize semantics.""" + + model_config = ConfigDict(extra="forbid") + + requires_terminal_pr_state: bool = True + allowed_terminal_pr_states: tuple[Literal["merged", "closed"], ...] = ("merged", "closed") + allows_integrated_sha_proof: bool = True + allow_close_without_terminal_or_integrated_sha: bool = False + + @field_validator("allowed_terminal_pr_states") + @classmethod + def _validate_terminal_states( + cls, + value: tuple[Literal["merged", "closed"], ...], + ) -> tuple[Literal["merged", "closed"], ...]: + if not value: + raise ValueError("allowed_terminal_pr_states must not be empty") + return value + + +class RefinedPlanningContract(BaseModel): + """Typed refined contract payload stored under ``planning.contract_json``.""" + + model_config = ConfigDict(extra="forbid") + + objective: str = Field(min_length=1) + non_goals: tuple[str, ...] = Field(min_length=1) + acceptance_criteria: tuple[AcceptanceCriterion, ...] = Field(min_length=1) + scope: ScopeBoundary + verification_plan: tuple[str, ...] = Field(min_length=1) + risks: tuple[RiskItem, ...] = Field(min_length=1) + escalation_conditions: tuple[str, ...] = Field(min_length=1) + completion_definition: CompletionDefinition + + @field_validator("objective") + @classmethod + def _validate_objective(cls, value: str) -> str: + cleaned = value.strip() + if not cleaned: + raise ValueError("objective must be non-empty") + return cleaned + + @field_validator("non_goals", "verification_plan", "escalation_conditions") + @classmethod + def _validate_text_arrays(cls, value: tuple[str, ...]) -> tuple[str, ...]: + cleaned = tuple(item.strip() for item in value if item.strip()) + if not cleaned: + raise ValueError("list fields must include at least one non-empty item") + return cleaned + + +class ReadinessResult(BaseModel): + """Structured readiness evaluation outcome for one issue payload.""" + + model_config = ConfigDict(frozen=True) + + refined: bool + contract_present: bool + stage: str | None + ok: bool + approved: bool + claim_eligible: bool + errors: tuple[str, ...] + claim_blockers: tuple[str, ...] + + @property + def summary(self) -> str: + """Return deterministic human-readable readiness diagnostics.""" + + diagnostics: list[str] = list(self.errors) + diagnostics.extend(blocker for blocker in self.claim_blockers if blocker not in diagnostics) + return "; ".join(diagnostics) + + +def parse_contract_json(raw_contract_json: str) -> RefinedPlanningContract: + """Parse and validate a serialized ``planning.contract_json`` payload. + + Args: + raw_contract_json: Raw JSON string captured in issue metadata. + + Returns: + Parsed and validated ``RefinedPlanningContract`` model. + + Raises: + ValueError: Raised when JSON is malformed or schema validation fails. + """ + + try: + payload = json.loads(raw_contract_json) + except json.JSONDecodeError as exc: # pragma: no cover - exercised via wrapper path + raise ValueError("planning.contract_json must be valid JSON") from exc + try: + return RefinedPlanningContract.model_validate(payload) + except ValidationError as exc: # pragma: no cover - exercised via wrapper path + first_error = exc.errors(include_url=False)[0] + location = ".".join(str(part) for part in first_error.get("loc", ()) if part is not None) + detail = first_error.get("msg", "invalid contract payload") + if location: + raise ValueError(f"planning.contract_json invalid at {location}: {detail}") from exc + raise ValueError(f"planning.contract_json invalid: {detail}") from exc + + +def serialize_contract(contract: RefinedPlanningContract) -> str: + """Serialize a validated contract for deterministic metadata persistence. + + Args: + contract: Contract model to serialize. + + Returns: + JSON string sorted by key for stable audits. + """ + + return json.dumps(contract.model_dump(mode="json"), separators=(",", ":"), sort_keys=True) + + +def evaluate_issue_refined_planning_readiness(issue: Mapping[str, object]) -> ReadinessResult: + """Evaluate planner/worker refined readiness for one issue payload. + + Args: + issue: Issue-like mapping containing at least an optional description. + + Returns: + Structured readiness with deterministic validation diagnostics. + """ + + description = issue.get("description") + fields = beads.parse_description_fields(description if isinstance(description, str) else "") + refined = _is_refined_strategy(fields) + if not refined: + return ReadinessResult( + refined=False, + contract_present=False, + stage=None, + ok=True, + approved=False, + claim_eligible=True, + errors=(), + claim_blockers=(), + ) + + errors: list[str] = [] + claim_blockers: list[str] = [] + + stage_raw = fields.get(_PLAN_STAGE_FIELD) + stage = _normalize_field(stage_raw) + if stage not in {_PLANNING_IN_REVIEW, _APPROVED}: + errors.append( + "refined changesets require planning.stage set to 'planning_in_review' or 'approved'" + ) + + contract_raw = fields.get(_CONTRACT_JSON_FIELD) + contract_text = contract_raw.strip() if isinstance(contract_raw, str) else "" + contract_present = bool(contract_text) + contract: RefinedPlanningContract | None = None + if not contract_present: + errors.append("refined changesets require planning.contract_json") + else: + try: + contract = parse_contract_json(contract_text) + except ValueError as exc: + errors.append(str(exc)) + if contract is not None: + errors.extend(_completion_definition_conflicts(contract.completion_definition)) + + approved = stage == _APPROVED + if approved: + approval_errors = _approval_errors(fields) + errors.extend(approval_errors) + else: + claim_blockers.append( + "refined changesets require planning.stage=approved before worker claim" + ) + + ok = not errors + claim_eligible = ok and approved + if not claim_eligible and approved and errors: + claim_blockers.extend(errors) + + return ReadinessResult( + refined=True, + contract_present=contract_present, + stage=stage, + ok=ok, + approved=approved, + claim_eligible=claim_eligible, + errors=tuple(errors), + claim_blockers=tuple(dict.fromkeys(claim_blockers)), + ) + + +def refined_planning_claim_eligible(issue: Mapping[str, object]) -> tuple[bool, str | None]: + """Return worker-claim eligibility and rejection reason for one issue. + + Args: + issue: Issue payload to evaluate. + + Returns: + Tuple ``(eligible, reason)`` where reason is populated only for blocked + refined changesets. + """ + + readiness = evaluate_issue_refined_planning_readiness(issue) + if not readiness.refined or readiness.claim_eligible: + return True, None + reason = readiness.summary or "refined changeset is not claim-eligible" + return False, reason + + +def approval_evidence_summary(issue: Mapping[str, object]) -> str: + """Render a concise audit summary for persisted approval metadata. + + Args: + issue: Issue payload that may contain refined approval metadata fields. + + Returns: + Human-readable approval evidence summary string. + """ + + description = issue.get("description") + fields = beads.parse_description_fields(description if isinstance(description, str) else "") + stage = _normalize_field(fields.get(_PLAN_STAGE_FIELD)) + approved_by = _normalize_field(fields.get(_APPROVED_BY_FIELD)) + approved_at = _normalize_field(fields.get(_APPROVED_AT_FIELD)) + approval_message_id = _normalize_field(fields.get(_APPROVAL_MESSAGE_ID_FIELD)) + parts = [ + f"stage={stage or '(missing)'}", + f"approved_by={approved_by or '(missing)'}", + f"approved_at={approved_at or '(missing)'}", + f"approval_message_id={approval_message_id or '(missing)'}", + ] + return "refined approval evidence: " + ", ".join(parts) + + +def _is_refined_strategy(fields: Mapping[str, str]) -> bool: + strategy = _normalize_field(fields.get(_EXECUTION_STRATEGY_FIELD)) + return strategy == _REFINED_STRATEGY + + +def _normalize_field(raw: str | None) -> str | None: + if raw is None: + return None + cleaned = raw.strip().lower() + if not cleaned or cleaned == "null": + return None + return cleaned + + +def _approval_errors(fields: Mapping[str, str]) -> list[str]: + errors: list[str] = [] + approved_by = _normalize_field(fields.get(_APPROVED_BY_FIELD)) + if approved_by is None: + errors.append("approved stage requires planning.approved_by") + approved_at_raw = _normalize_field(fields.get(_APPROVED_AT_FIELD)) + if approved_at_raw is None: + errors.append("approved stage requires planning.approved_at") + elif not _is_valid_iso_timestamp(approved_at_raw): + errors.append("planning.approved_at must be an ISO-8601 timestamp") + approval_message_id = _normalize_field(fields.get(_APPROVAL_MESSAGE_ID_FIELD)) + if approval_message_id is None: + errors.append("approved stage requires planning.approval_message_id") + elif not _APPROVAL_MESSAGE_ID_PATTERN.fullmatch(approval_message_id): + errors.append("planning.approval_message_id must be an identifier") + return errors + + +def _is_valid_iso_timestamp(value: str) -> bool: + if "t" not in value: + return False + try: + parsed = dt.datetime.fromisoformat(value.replace("z", "+00:00")) + except ValueError: + return False + if parsed.tzinfo is None or parsed.utcoffset() is None: + return False + return True + + +def _completion_definition_conflicts(definition: CompletionDefinition) -> list[str]: + conflicts: list[str] = [] + if definition.allow_close_without_terminal_or_integrated_sha: + conflicts.append( + "completion_definition conflicts with lifecycle finalize semantics: close without " + "terminal PR state or integrated SHA proof is not allowed" + ) + if not definition.requires_terminal_pr_state and not definition.allows_integrated_sha_proof: + conflicts.append( + "completion_definition must require terminal PR state or integrated SHA proof" + ) + if definition.requires_terminal_pr_state and not definition.allowed_terminal_pr_states: + conflicts.append("completion_definition requires non-empty terminal PR states") + return conflicts + + +__all__ = [ + "AcceptanceCriterion", + "CompletionDefinition", + "ReadinessResult", + "RefinedPlanningContract", + "RiskItem", + "ScopeBoundary", + "approval_evidence_summary", + "evaluate_issue_refined_planning_readiness", + "parse_contract_json", + "refined_planning_claim_eligible", + "serialize_contract", +] diff --git a/src/atelier/skills/plan-changeset-guardrails/SKILL.md b/src/atelier/skills/plan-changeset-guardrails/SKILL.md index 663e89a7..2404cce8 100644 --- a/src/atelier/skills/plan-changeset-guardrails/SKILL.md +++ b/src/atelier/skills/plan-changeset-guardrails/SKILL.md @@ -35,6 +35,11 @@ description: >- changeset or stack extension) when thresholds or new domains appear. - Require explicit guidance that review-feedback scope growth is captured immediately as deferred follow-on work or stack extension. +- For refined executable units, require `planning.contract_json` and + `planning.stage: planning_in_review`. +- Surface deterministic errors from the shared refined validator, including + malformed JSON, missing required payload fields, and completion-definition + lifecycle conflicts. ## Steps @@ -52,6 +57,8 @@ description: >- - If a large estimate is found (>800), ensure approval is recorded. - When lifecycle/contract invariant terms are present, verify the invariant impact map and re-split guidance fields are present. + - If `execution.strategy: refined`, run shared refined validation and require + `planning.stage: planning_in_review` before promotion. 1. If an epic has exactly one child changeset, require explicit decomposition rationale in the epic or child notes/description. 1. Summarize any violations and send a message to the planner/overseer with diff --git a/src/atelier/skills/plan-changeset-guardrails/scripts/check_guardrails.py b/src/atelier/skills/plan-changeset-guardrails/scripts/check_guardrails.py index 8641f5d0..62f80ced 100644 --- a/src/atelier/skills/plan-changeset-guardrails/scripts/check_guardrails.py +++ b/src/atelier/skills/plan-changeset-guardrails/scripts/check_guardrails.py @@ -30,6 +30,9 @@ from atelier.bd_invocation import with_bd_mode # noqa: E402 from atelier.beads_context import resolve_runtime_repo_dir_hint # noqa: E402 from atelier.planner_contract import validate_authoring_contract # noqa: E402 +from atelier.refined_planning_contract import ( + evaluate_issue_refined_planning_readiness, # noqa: E402 +) _LOC_TRIGGER = re.compile(r"\b(?:loc|estimate)\b", re.IGNORECASE) _NUMBER = re.compile(r"\b\d{2,5}\b") @@ -194,7 +197,13 @@ def _cross_cutting_corpus( if epic_issue is not None: parts.append(_text_blob(epic_issue)) parts.extend(_text_blob(issue) for issue in target_changesets) - return "\n".join(part for part in parts if part.strip()) + corpus = "\n".join(part for part in parts if part.strip()) + # Ignore raw contract metadata lines that would otherwise trip narrative checks. + return "\n".join( + line + for line in corpus.splitlines() + if not line.strip().lower().startswith("planning.contract_json:") + ) def _subject_id( @@ -270,9 +279,28 @@ def _evaluate_guardrails( ) -> _GuardrailReport: violations: list[str] = [] path_summary: str | None = None + seen_violations: set[str] = set() + + def _append_violation(message: str) -> None: + if message in seen_violations: + return + seen_violations.add(message) + violations.append(message) for issue in target_changesets: issue_id = _issue_id(issue) or "(unknown)" + refined_readiness = evaluate_issue_refined_planning_readiness(issue) + if refined_readiness.refined: + if not refined_readiness.contract_present: + _append_violation(f"{issue_id}: refined changesets require planning.contract_json.") + if refined_readiness.stage != "planning_in_review": + _append_violation( + f"{issue_id}: refined planner payload must set " + "planning.stage: planning_in_review." + ) + for error in refined_readiness.errors: + _append_violation(f"{issue_id}: {error}") + inherited_context = ( [epic_issue] if epic_issue is not None and epic_issue is not issue else [] ) @@ -285,23 +313,23 @@ def _evaluate_guardrails( field for field in missing_contract_fields if field != "done_definition" ) if missing_sections: - violations.append( + _append_violation( f"{issue_id}: missing planner authoring contract fields " f"({', '.join(missing_sections)}); add explicit key/value context to the " "epic or changeset description/notes/design." ) if "done_definition" in missing_contract_fields: - violations.append( + _append_violation( f"{issue_id}: missing explicit done definition; add acceptance criteria or " "`done_definition:` to the executable path." ) text = _text_blob(issue) estimate = _extract_loc_estimate(text) if estimate is None: - violations.append(f"{issue_id}: missing LOC estimate in notes/description.") + _append_violation(f"{issue_id}: missing LOC estimate in notes/description.") continue if estimate > 800 and not _APPROVAL.search(text): - violations.append( + _append_violation( f"{issue_id}: LOC estimate {estimate} exceeds 800 without explicit approval note." ) @@ -321,7 +349,7 @@ def _evaluate_guardrails( ) else: path_summary = f"{epic_id}: one child changeset ({child_id}) without rationale." - violations.append( + _append_violation( f"{epic_id}: one-child anti-pattern; add decomposition rationale " f"for {child_id} or keep the epic as the executable changeset." ) diff --git a/src/atelier/skills/plan-changesets/SKILL.md b/src/atelier/skills/plan-changesets/SKILL.md index 1d524614..a0517809 100644 --- a/src/atelier/skills/plan-changesets/SKILL.md +++ b/src/atelier/skills/plan-changesets/SKILL.md @@ -10,6 +10,9 @@ Only use this when an epic should be decomposed. If the epic itself is already within guardrails, keep the epic as the executable changeset instead of creating children. +Use the shared planning discipline from `plan-refined-deliberation`. Use the +same drafting strategy for all planning passes. + Capture new executable work immediately as deferred changesets (`status=deferred`) when issues are actionable. Do not wait for approval to create/edit deferred work. diff --git a/src/atelier/skills/plan-create-epic/SKILL.md b/src/atelier/skills/plan-create-epic/SKILL.md index 6e95aaa0..8bc3c0da 100644 --- a/src/atelier/skills/plan-create-epic/SKILL.md +++ b/src/atelier/skills/plan-create-epic/SKILL.md @@ -10,6 +10,9 @@ description: >- When a concrete issue is identified, capture it as a deferred epic immediately. Do not request approval to create or edit deferred beads. +Use the shared planning discipline from `plan-refined-deliberation`. Use the +same drafting strategy for all planning passes. + ## Inputs - title: Epic title. diff --git a/src/atelier/skills/plan-promote-epic/SKILL.md b/src/atelier/skills/plan-promote-epic/SKILL.md index 4513687a..d4b4aa05 100644 --- a/src/atelier/skills/plan-promote-epic/SKILL.md +++ b/src/atelier/skills/plan-promote-epic/SKILL.md @@ -34,6 +34,12 @@ description: >- - Do not require child changesets when the epic itself is guardrail-sized. - If the epic has exactly one child changeset, explicit decomposition rationale must be recorded before promotion. +- For refined executable units (`execution.strategy: refined`), contract + validation must pass before promotion. Required planning-stage metadata: + - `planning.contract_json` + - `planning.stage: planning_in_review` +- Worker startup fails closed for refined units unless approval evidence is + recorded. ## Steps @@ -83,6 +89,12 @@ description: >- 1. Promote each fully-defined child changeset from `deferred` to `open` regardless of current dependency blockers (dependency graph still gates runnability). +1. For each promoted refined executable unit, persist approval evidence: + - `planning.stage: approved` + - `planning.approved_by` + - `planning.approved_at` + - `planning.approval_message_id` and record an approval evidence note/thread + message. 1. Let dependency resolution determine runnability (`bd ready`) at worker time. ## Verification diff --git a/src/atelier/skills/plan-promote-epic/scripts/promote_epic.py b/src/atelier/skills/plan-promote-epic/scripts/promote_epic.py index 3425c340..1420e65f 100644 --- a/src/atelier/skills/plan-promote-epic/scripts/promote_epic.py +++ b/src/atelier/skills/plan-promote-epic/scripts/promote_epic.py @@ -5,8 +5,12 @@ import argparse import asyncio +import datetime as dt +import os import sys +from collections.abc import Mapping from pathlib import Path +from typing import Protocol _SHARED_SCRIPTS_ROOT = Path(__file__).resolve().parents[2] / "shared" / "scripts" if str(_SHARED_SCRIPTS_ROOT) not in sys.path: @@ -22,11 +26,30 @@ require_runtime_health=__name__ == "__main__", ) +from atelier import refined_planning_contract # noqa: E402 from atelier.beads_context import ( # noqa: E402 resolve_runtime_repo_dir_hint, resolve_skill_beads_context, ) +from atelier.lib.beads import ShowIssueRequest, UpdateIssueRequest # noqa: E402 from atelier.lib.beads import description_fields as bead_fields # noqa: E402 +from atelier.store import AppendNotesRequest, CreateMessageRequest # noqa: E402 + + +class _ApprovalStore(Protocol): + """Minimal typed boundary for refined approval persistence helpers.""" + + async def create_message(self, request: CreateMessageRequest) -> object: ... + + async def append_notes(self, request: AppendNotesRequest) -> object: ... + + +class _ApprovalClient(Protocol): + """Typed client boundary for description-field persistence.""" + + async def show(self, request: ShowIssueRequest) -> object: ... + + async def update(self, request: UpdateIssueRequest) -> object: ... def _build_store_and_client(*, beads_root: Path, repo_root: Path): @@ -188,6 +211,153 @@ def _render_issue_preview(*, header: str, issue: object) -> str: return "\n".join(lines) +def _issue_metadata_payload(issue: object) -> dict[str, object]: + return { + "id": _issue_text(issue, "id") or "", + "description": _issue_text(issue, "description") or "", + } + + +def _refined_validation_error(issue: object) -> str | None: + issue_id = _issue_text(issue, "id") or "(issue)" + readiness = refined_planning_contract.evaluate_issue_refined_planning_readiness( + _issue_metadata_payload(issue) + ) + if readiness.refined and not readiness.ok: + return f"{issue_id} refined readiness failed: {readiness.summary}" + return None + + +def _issue_thread_kind(issue_id: str): + from atelier.store import MessageThreadKind + + return MessageThreadKind.CHANGESET if "." in issue_id else MessageThreadKind.EPIC + + +def _approval_timestamp() -> str: + return ( + dt.datetime.now(tz=dt.timezone.utc) + .replace(microsecond=0) + .isoformat() + .replace("+00:00", "Z") + ) + + +def _required_operator_id() -> str: + operator_id = str(os.environ.get("ATELIER_AGENT_ID") or "").strip() + if not operator_id: + raise RuntimeError("ATELIER_AGENT_ID must be set for refined approvals") + return operator_id + + +def _record_refined_approval( + *, + store: _ApprovalStore, + client: _ApprovalClient, + issue: object, + beads_root: Path, + repo_root: Path, + operator_id: str, +) -> str: + issue_id = _issue_text(issue, "id") + if issue_id is None: + raise RuntimeError("refined issue is missing id") + + issue_payload = _issue_metadata_payload(issue) + evidence = refined_planning_contract.approval_evidence_summary(issue_payload) + message = asyncio.run( + store.create_message( + CreateMessageRequest( + title=f"Refined approval: {issue_id}", + body=( + f"Operator {operator_id} approved refined promotion for {issue_id}.\n{evidence}" + ), + sender=operator_id, + thread_id=issue_id, + thread_kind=_issue_thread_kind(issue_id), + ) + ) + ) + approval_message_id = str(getattr(message, "id", "")).strip() + if not approval_message_id: + raise RuntimeError(f"{issue_id} refined approval message id missing") + + approved_at = _approval_timestamp() + updated = _update_description_metadata_fields( + client=client, + issue_id=issue_id, + fields={ + "planning.stage": "approved", + "planning.approved_by": operator_id, + "planning.approved_at": approved_at, + "planning.approval_message_id": approval_message_id, + }, + ) + updated_evidence = refined_planning_contract.approval_evidence_summary(updated) + asyncio.run( + store.append_notes( + AppendNotesRequest( + issue_id=issue_id, + notes=(f"Refined approval recorded. {updated_evidence}",), + ) + ) + ) + return approval_message_id + + +def _issue_mapping(issue: object) -> dict[str, object]: + if isinstance(issue, Mapping): + return {str(key): value for key, value in issue.items()} + return {"description": _issue_text(issue, "description") or ""} + + +def _render_description_with_updates( + description: str | None, + *, + fields: Mapping[str, str], +) -> str: + existing_lines = (description or "").splitlines() + update_keys = tuple(fields.keys()) + preserved_lines = [ + line + for line in existing_lines + if not any(line.strip().startswith(f"{key}:") for key in update_keys) + ] + update_lines = [f"{key}: {value}" for key, value in fields.items()] + merged = [*preserved_lines, *update_lines] + return ("\n".join(merged).rstrip("\n") + "\n") if merged else "" + + +def _update_description_metadata_fields( + *, + client: _ApprovalClient, + issue_id: str, + fields: Mapping[str, str], +) -> dict[str, object]: + for _ in range(3): + current = asyncio.run(client.show(ShowIssueRequest(issue_id=issue_id))) + current_description = _issue_text(current, "description") + next_description = _render_description_with_updates( + current_description, + fields=fields, + ) + asyncio.run( + client.update( + UpdateIssueRequest( + issue_id=issue_id, + description=next_description, + ) + ) + ) + refreshed = asyncio.run(client.show(ShowIssueRequest(issue_id=issue_id))) + refreshed_fields = bead_fields.parse_description_fields( + _issue_text(refreshed, "description") or "" + ) + if all(refreshed_fields.get(key) == value for key, value in fields.items()): + return _issue_mapping(refreshed) + raise RuntimeError(f"{issue_id} metadata update could not be verified") + + def main() -> None: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--epic-id", required=True, help="Deferred epic bead id") @@ -244,6 +414,7 @@ def main() -> None: ) preview_blocks = [_render_issue_preview(header=f"EPIC {epic_id}", issue=epic_issue)] child_missing: dict[str, tuple[str, ...]] = {} + child_issues_by_id = {str(getattr(issue, "id")): issue for issue in child_issues} promotable_children: list[str] = [] for record, issue in zip(changesets, child_issues, strict=True): preview_blocks.append( @@ -267,6 +438,15 @@ def main() -> None: problems.append( "incomplete child changesets remain deferred: " + ", ".join(incomplete_children) ) + validation_targets = tuple(child_issues) if changesets else (epic_issue,) + executable_targets = ( + tuple(child_issues_by_id[issue_id] for issue_id in promotable_children) + if promotable_children + else ((epic_issue,) if not changesets and not epic_missing else ()) + ) + for issue in validation_targets: + if (refined_error := _refined_validation_error(issue)) is not None: + problems.append(refined_error) if problems: raise RuntimeError("; ".join(problems)) @@ -275,6 +455,25 @@ def main() -> None: print("confirmation_required: rerun with --yes after explicit operator confirmation") return + approval_targets = [ + issue + for issue in executable_targets + if refined_planning_contract.evaluate_issue_refined_planning_readiness( + _issue_metadata_payload(issue) + ).refined + ] + if approval_targets: + operator_id = _required_operator_id() + for issue in approval_targets: + _record_refined_approval( + store=store, + client=client, + issue=issue, + beads_root=beads_root, + repo_root=repo_root, + operator_id=operator_id, + ) + asyncio.run( store.transition_lifecycle( LifecycleTransitionRequest( diff --git a/src/atelier/skills/plan-refined-deliberation/SKILL.md b/src/atelier/skills/plan-refined-deliberation/SKILL.md new file mode 100644 index 00000000..41456052 --- /dev/null +++ b/src/atelier/skills/plan-refined-deliberation/SKILL.md @@ -0,0 +1,31 @@ +--- +name: plan-refined-deliberation +description: >- + Shared planning discipline for all planning passes, including refinement + rounds. +--- + +# Plan refined deliberation + +Use this shared discipline for every planning pass, including the first draft +and later refinement rounds. + +## Contract + +- Start with the strategy gate before task breakdown. +- Keep a low bar to reframe when a better path exists. +- Keep a high bar to ask the user. +- Treat each refinement round as a fresh, stateless pass over the current plan. +- stop at sign-off or loop limit. +- Target epic and changeset beads, not markdown checklists. +- This discipline applies to all planning, not just refinement rounds. + +## Planning steps + +1. Check the strategy gate before decomposition. +1. If a better path exists, reframe the plan instead of forcing the current one. +1. Only ask the user when the decision truly needs human input. +1. On each refinement round, begin from the current plan and evaluate it afresh. +1. Stop when the plan is approved or the loop budget is exhausted. +1. Record work as epic and changeset beads with explicit intent, rationale, + non-goals, constraints, edge cases, related context, and a done definition. diff --git a/src/atelier/skills/plan-split-tasks/SKILL.md b/src/atelier/skills/plan-split-tasks/SKILL.md index 03173390..5794396b 100644 --- a/src/atelier/skills/plan-split-tasks/SKILL.md +++ b/src/atelier/skills/plan-split-tasks/SKILL.md @@ -10,6 +10,9 @@ description: >- Use this only when the epic should be decomposed. If the epic itself is a single review-sized unit, keep it as the executable changeset. +Use the shared planning discipline from `plan-refined-deliberation`. Use the +same drafting strategy for all planning passes. + ## Inputs - epic_id: Parent epic bead id. diff --git a/src/atelier/templates/AGENTS.planner.md.tmpl b/src/atelier/templates/AGENTS.planner.md.tmpl index df5044ad..aaf7a920 100644 --- a/src/atelier/templates/AGENTS.planner.md.tmpl +++ b/src/atelier/templates/AGENTS.planner.md.tmpl @@ -18,6 +18,8 @@ which version is local, ask the overseer before using a global fallback. The CLI does not enforce planning correctness. You do, via skills. - Run `planner-startup-check` before any planning work. +- Run `plan-refined-deliberation` before any plan breakdown or refinement pass; + the same drafting strategy applies to all planning passes. - Run `plan-changeset-guardrails` after creating or updating changesets. - Use `plan-promote-epic` to promote epics, with explicit user confirmation. - Use `mail-send` for planner-to-worker dispatch on the epic/changeset thread; @@ -27,6 +29,17 @@ The CLI does not enforce planning correctness. You do, via skills. - Do not use `mail-send` without an epic/changeset thread; select the owning work item first so coordination stays durable. +## Shared Planning Discipline + +`plan-refined-deliberation` is the shared planning discipline for every +planning pass. refined means iterative reevaluation of the same plan, not a +different drafting style. + +- Keep the low bar to reframe when a better path exists. +- Keep the high bar to ask the user. +- Keep plan quality separate from bead mechanics. +- Use this shared discipline before any task breakdown or refinement loop. + ## No Approval Step (Except Promotion) There is no approval step for creating or editing deferred work. When you diff --git a/src/atelier/worker/session/startup.py b/src/atelier/worker/session/startup.py index 732d8eb1..fdc4f6cc 100644 --- a/src/atelier/worker/session/startup.py +++ b/src/atelier/worker/session/startup.py @@ -86,6 +86,11 @@ def changeset_integration_signal( def is_changeset_in_progress(self, issue: dict[str, object]) -> bool: ... + def refined_planning_claim_eligible( + self, + issue: dict[str, object], + ) -> tuple[bool, str | None]: ... + def _issue_id(issue: dict[str, object]) -> str | None: issue_id = issue.get("id") @@ -229,6 +234,10 @@ def review_resume_allowed(issue: dict[str, object]) -> bool: git_path=context.git_path, ) + def refined_planning_eligible(issue: dict[str, object]) -> bool: + eligible, _reason = service.refined_planning_claim_eligible(issue) + return eligible + target = service.show_issue(context.epic_id) if target: issue = target @@ -280,7 +289,9 @@ def review_resume_allowed(issue: dict[str, object]) -> bool: or target_recovery_candidate ) ): - if not service.has_open_descendant_changesets(context.epic_id): + if not service.has_open_descendant_changesets( + context.epic_id + ) and refined_planning_eligible(issue): return issue if ( isinstance(issue_id, str) @@ -292,7 +303,9 @@ def review_resume_allowed(issue: dict[str, object]) -> bool: return None if isinstance(issue_id, str) and issue_id == context.epic_id and claimability.role.is_epic: if not explicit_descendants: - return issue + if refined_planning_eligible(issue): + return issue + return None descendants = service.list_descendant_changesets(context.epic_id, include_closed=False) descendants_by_id = { @@ -335,6 +348,8 @@ def review_resume_allowed(issue: dict[str, object]) -> bool: ), ) for issue in prioritized: + if not refined_planning_eligible(issue): + continue issue_id = issue.get("id") if isinstance(issue_id, str) and issue_id: if not service.has_open_descendant_changesets(issue_id): @@ -485,6 +500,11 @@ def emit(self, message: str) -> None: ... def die(self, message: str) -> None: ... + def refined_planning_claim_eligible( + self, + issue: dict[str, object], + ) -> tuple[bool, str | None]: ... + def run_startup_contract_service( *, context: StartupContractContext, service: StartupContractService @@ -520,6 +540,31 @@ def stage_call(stage: str, callback: Callable[[], _StageResult]) -> _StageResult ) return result + def refined_planning_selection_allowed( + *, + changeset_id: str, + stage: str, + ) -> bool: + issue = service.show_issue(changeset_id) + if issue is None: + detail = "unable to load changeset metadata for refined claim gate" + service.emit(f"Skipping {stage} changeset {changeset_id}: {detail}") + atelier_log.warning( + "startup skipping " + f"{stage} changeset={changeset_id} reason=refined_metadata_unavailable" + ) + return False + payload: dict[str, object] = issue + eligible, reason = service.refined_planning_claim_eligible(payload) + if eligible: + return True + detail = reason or "refined changeset is not claim-eligible" + service.emit(f"Skipping {stage} changeset {changeset_id}: {detail}") + atelier_log.warning( + f"startup skipping {stage} changeset={changeset_id} reason=refined_ineligible" + ) + return False + """Apply startup-contract skill ordering to select the next epic.""" selected_epic: str | None = None if explicit_epic_id is not None: @@ -612,13 +657,17 @@ def stage_call(stage: str, callback: Callable[[], _StageResult]) -> _StageResult ), ) if explicit_conflict is not None: - return StartupContractResult( - epic_id=explicit_conflict.epic_id, + if refined_planning_selection_allowed( changeset_id=explicit_conflict.changeset_id, - should_exit=False, - reason="merge_conflict", - reassign_from=explicit_reassign_from, - ) + stage="explicit merge-conflict", + ): + return StartupContractResult( + epic_id=explicit_conflict.epic_id, + changeset_id=explicit_conflict.changeset_id, + should_exit=False, + reason="merge_conflict", + reassign_from=explicit_reassign_from, + ) explicit_feedback = stage_call( "explicit.review-feedback", lambda: service.select_review_feedback_changeset( @@ -627,13 +676,17 @@ def stage_call(stage: str, callback: Callable[[], _StageResult]) -> _StageResult ), ) if explicit_feedback is not None: - return StartupContractResult( - epic_id=explicit_feedback.epic_id, + if refined_planning_selection_allowed( changeset_id=explicit_feedback.changeset_id, - should_exit=False, - reason="review_feedback", - reassign_from=explicit_reassign_from, - ) + stage="explicit review-feedback", + ): + return StartupContractResult( + epic_id=explicit_feedback.epic_id, + changeset_id=explicit_feedback.changeset_id, + should_exit=False, + reason="review_feedback", + reassign_from=explicit_reassign_from, + ) explicit_next_changeset = stage_call( "explicit.next-changeset", lambda: service.next_changeset( @@ -644,6 +697,13 @@ def stage_call(stage: str, callback: Callable[[], _StageResult]) -> _StageResult resume_review=resume_review, ), ) + if explicit_next_changeset is not None: + explicit_issue_id = _issue_id(explicit_next_changeset) + if explicit_issue_id is not None and not refined_planning_selection_allowed( + changeset_id=explicit_issue_id, + stage="explicit next-changeset", + ): + explicit_next_changeset = None if explicit_next_changeset is None: if status in {"in_progress", "hooked"}: service.emit( @@ -863,9 +923,13 @@ def select_conflict_candidate( ), ) if selection is not None: - if select_first_eligible: - return selection - conflict_candidates.append(selection) + if refined_planning_selection_allowed( + changeset_id=selection.changeset_id, + stage="merge-conflict", + ): + if select_first_eligible: + return selection + conflict_candidates.append(selection) if not conflict_candidates: return None conflict_candidates.sort( @@ -918,9 +982,13 @@ def select_feedback_candidate( ), ) if feedback_selection is not None: - if select_first_eligible: - return feedback_selection - feedback_candidates.append(feedback_selection) + if refined_planning_selection_allowed( + changeset_id=feedback_selection.changeset_id, + stage="review-feedback", + ): + if select_first_eligible: + return feedback_selection + feedback_candidates.append(feedback_selection) if not feedback_candidates: return None feedback_candidates.sort( @@ -1016,6 +1084,11 @@ def resume_feedback(selection: ReviewFeedbackSelection) -> StartupContractResult global_conflict.epic_id, stage="global-merge-conflict" ) or is_excluded(global_conflict.epic_id, stage="global-merge-conflict"): global_conflict = None + elif not refined_planning_selection_allowed( + changeset_id=global_conflict.changeset_id, + stage="global merge-conflict", + ): + global_conflict = None if global_conflict is not None: return resume_conflict(global_conflict) feedback = select_feedback_candidate(unhooked_epics) @@ -1032,6 +1105,11 @@ def resume_feedback(selection: ReviewFeedbackSelection) -> StartupContractResult global_feedback = None elif not is_claimable(global_feedback.epic_id, stage="global-review-feedback"): global_feedback = None + elif not refined_planning_selection_allowed( + changeset_id=global_feedback.changeset_id, + stage="global review-feedback", + ): + global_feedback = None if global_feedback is not None: return resume_feedback(global_feedback) diff --git a/src/atelier/worker/work_startup_runtime.py b/src/atelier/worker/work_startup_runtime.py index 0e07b3db..cc65d4b0 100644 --- a/src/atelier/worker/work_startup_runtime.py +++ b/src/atelier/worker/work_startup_runtime.py @@ -6,7 +6,7 @@ from collections.abc import Callable from pathlib import Path -from .. import agent_home, beads, changeset_fields, prs, work_feedback +from .. import agent_home, beads, changeset_fields, prs, refined_planning_contract, work_feedback from ..io import die, prompt, say, select from ..work_feedback import ReviewFeedbackSnapshot from ..worker import prompts as worker_prompts @@ -166,6 +166,31 @@ def startup_finalize_preflight( ) +def _refined_planning_claim_eligibility(issue: dict[str, object]) -> tuple[bool, str | None]: + return refined_planning_contract.refined_planning_claim_eligible(issue) + + +def _hydrate_refined_planning_issue_payload( + issue: dict[str, object], + *, + beads_root: Path, + repo_root: Path, +) -> dict[str, object] | None: + if isinstance(issue.get("description"), str): + return issue + issue_id = issue.get("id") + if not isinstance(issue_id, str) or not issue_id.strip(): + return None + hydrated = worker_store.show_issue( + issue_id, + beads_root=beads_root, + repo_root=repo_root, + ) + if hydrated is None or not isinstance(hydrated.get("description"), str): + return None + return hydrated + + class _NextChangesetService(worker_startup.NextChangesetService): """Concrete next-changeset service implementation for worker startup.""" @@ -277,6 +302,19 @@ def changeset_integration_signal( def is_changeset_in_progress(self, issue: dict[str, object]) -> bool: return is_changeset_in_progress(issue) + def refined_planning_claim_eligible( + self, + issue: dict[str, object], + ) -> tuple[bool, str | None]: + payload = _hydrate_refined_planning_issue_payload( + issue, + beads_root=self._beads_root, + repo_root=self._repo_root, + ) + if payload is None: + return False, "unable to load changeset metadata for refined claim gate" + return _refined_planning_claim_eligibility(payload) + def _no_eligible_epics_summary( *, @@ -914,6 +952,19 @@ def select_global_review_feedback_changeset( ) -> ReviewFeedbackSelection | None: return self._global_startup_candidates(repo_slug=repo_slug).feedback + def refined_planning_claim_eligible( + self, + issue: dict[str, object], + ) -> tuple[bool, str | None]: + payload = _hydrate_refined_planning_issue_payload( + issue, + beads_root=self._beads_root, + repo_root=self._repo_root, + ) + if payload is None: + return False, "unable to load changeset metadata for refined claim gate" + return _refined_planning_claim_eligibility(payload) + def check_inbox_before_claim(self, agent_id: str) -> bool: return check_inbox_before_claim( agent_id, beads_root=self._beads_root, repo_root=self._repo_root diff --git a/tests/atelier/skills/test_plan_changeset_guardrails_script.py b/tests/atelier/skills/test_plan_changeset_guardrails_script.py index eccde62b..89b9d32e 100644 --- a/tests/atelier/skills/test_plan_changeset_guardrails_script.py +++ b/tests/atelier/skills/test_plan_changeset_guardrails_script.py @@ -1,6 +1,7 @@ from __future__ import annotations import importlib.util +import json import subprocess import sys from pathlib import Path @@ -35,6 +36,123 @@ def _planner_contract_text() -> str: ) +def _valid_refined_contract_json() -> str: + return json.dumps( + { + "objective": "Ship fail-closed refined contract enforcement", + "non_goals": ["Do not modify non-refined startup semantics"], + "acceptance_criteria": [ + {"statement": "Reject missing contracts", "evidence": ["guardrails tests"]} + ], + "scope": { + "includes": ["planner guardrails"], + "excludes": ["worker PR flow redesign"], + }, + "verification_plan": ["uv run pytest tests/atelier/skills -k refined -v"], + "risks": [{"risk": "over-blocking", "mitigation": "refined-only gating"}], + "escalation_conditions": ["validator disagreement"], + "completion_definition": { + "requires_terminal_pr_state": True, + "allowed_terminal_pr_states": ["merged", "closed"], + "allows_integrated_sha_proof": True, + "allow_close_without_terminal_or_integrated_sha": False, + }, + }, + separators=(",", ":"), + ) + + +def _refined_description(*, stage: str, contract_json: str) -> str: + return ( + f"{_planner_contract_text()}\n" + "LOC estimate: 220\n" + "done_definition: Done when guardrails and worker startup reject invalid refined work.\n" + "execution.strategy: refined\n" + f"planning.stage: {stage}\n" + f"planning.contract_json: {contract_json}\n" + ) + + +def test_guardrails_flags_refined_target_missing_contract() -> None: + module = _load_script_module() + child = {"id": "at-epic.1", "description": "execution.strategy: refined"} + + report = module._evaluate_guardrails( + epic_issue=None, + child_changesets=[], + target_changesets=[child], + ) + + assert any("planning.contract_json" in item for item in report.violations) + + +def test_guardrails_accepts_valid_refined_payload() -> None: + module = _load_script_module() + child = { + "id": "at-epic.1", + "description": _refined_description( + stage="planning_in_review", + contract_json=_valid_refined_contract_json(), + ), + "acceptance_criteria": "Done when validation fails closed before promotion.", + } + + report = module._evaluate_guardrails( + epic_issue=None, + child_changesets=[], + target_changesets=[child], + ) + + assert not any("planning." in item for item in report.violations) + + +def test_guardrails_flags_refined_completion_definition_conflict() -> None: + module = _load_script_module() + contract_payload = json.loads(_valid_refined_contract_json()) + contract_payload["completion_definition"]["allow_close_without_terminal_or_integrated_sha"] = ( + True + ) + child = { + "id": "at-epic.1", + "description": _refined_description( + stage="planning_in_review", + contract_json=json.dumps(contract_payload, separators=(",", ":")), + ), + "acceptance_criteria": "Done when conflicts are rejected deterministically.", + } + + report = module._evaluate_guardrails( + epic_issue=None, + child_changesets=[], + target_changesets=[child], + ) + + assert any("completion_definition conflicts" in item for item in report.violations) + + +def test_guardrails_require_planning_in_review_for_refined_payloads() -> None: + module = _load_script_module() + child = { + "id": "at-epic.1", + "description": _refined_description( + stage="approved", + contract_json=_valid_refined_contract_json(), + ) + + "planning.approved_by: operator\n" + + "planning.approved_at: 2026-03-26T12:00:00Z\n" + + "planning.approval_message_id: at-msg.1\n", + "acceptance_criteria": "Done when approval metadata is persisted.", + } + + report = module._evaluate_guardrails( + epic_issue=None, + child_changesets=[], + target_changesets=[child], + ) + + assert any("planning_in_review" in item for item in report.violations) + + def test_evaluate_guardrails_accepts_single_unit_epic_path() -> None: module = _load_script_module() epic = { diff --git a/tests/atelier/skills/test_plan_promote_epic_script.py b/tests/atelier/skills/test_plan_promote_epic_script.py index 780e4da0..e4876744 100644 --- a/tests/atelier/skills/test_plan_promote_epic_script.py +++ b/tests/atelier/skills/test_plan_promote_epic_script.py @@ -1,6 +1,7 @@ from __future__ import annotations import importlib.util +import json import sys from pathlib import Path from types import SimpleNamespace @@ -47,6 +48,482 @@ def _issue( ) +def _valid_refined_contract_json() -> str: + return json.dumps( + { + "objective": "Enforce fail-closed refined promotion gate", + "non_goals": ["Do not alter non-refined promotion"], + "acceptance_criteria": [ + {"statement": "Reject invalid refined payloads", "evidence": ["pytest"]} + ], + "scope": { + "includes": ["plan-promote-epic"], + "excludes": ["worker runtime redesign"], + }, + "verification_plan": ["uv run pytest tests/atelier/skills -k refined -v"], + "risks": [{"risk": "over-blocking", "mitigation": "refined-only checks"}], + "escalation_conditions": ["validator disagreement"], + "completion_definition": { + "requires_terminal_pr_state": True, + "allowed_terminal_pr_states": ["merged", "closed"], + "allows_integrated_sha_proof": True, + "allow_close_without_terminal_or_integrated_sha": False, + }, + }, + separators=(",", ":"), + ) + + +def _refined_description(*, contract_json: str, stage: str = "planning_in_review") -> str: + return ( + "related_context: at-context\n" + "changeset_strategy: Keep review scope small.\n" + "execution.strategy: refined\n" + f"planning.stage: {stage}\n" + f"planning.contract_json: {contract_json}\n" + ) + + +def test_promote_epic_blocks_refined_target_without_valid_contract( + monkeypatch, + capsys: pytest.CaptureFixture[str], + tmp_path: Path, +) -> None: + module = _load_script_module() + + monkeypatch.setattr( + module, + "_resolve_context", + lambda **_kwargs: (tmp_path / ".beads", tmp_path / "repo", None), + ) + + epic_issue = _issue( + "at-epic", + title="Epic", + description=( + "changeset_strategy: Keep review scope small.\n" + "related_context: at-context\n" + "promotion_note: ready for confirmation\n" + ), + ) + child_issue = _issue( + "at-epic.1", + title="Child", + description=_refined_description(contract_json="{bad-json}"), + ) + + class FakeStore: + async def get_epic(self, epic_id): + assert epic_id == "at-epic" + from atelier.store import LifecycleStatus + + return SimpleNamespace(id=epic_id, lifecycle=LifecycleStatus.DEFERRED) + + async def list_changesets(self, query): + del query + from atelier.store import LifecycleStatus + + return (SimpleNamespace(id="at-epic.1", lifecycle=LifecycleStatus.DEFERRED),) + + async def transition_lifecycle(self, request): # pragma: no cover - defensive + raise AssertionError(request) + + class FakeClient: + async def show(self, request): + return {"at-epic": epic_issue, "at-epic.1": child_issue}[request.issue_id] + + monkeypatch.setattr( + module, "_build_store_and_client", lambda **_kwargs: (FakeStore(), FakeClient()) + ) + monkeypatch.setattr(sys, "argv", ["promote_epic.py", "--epic-id", "at-epic"]) + + with pytest.raises(SystemExit) as excinfo: + module.main() + + assert excinfo.value.code == 1 + assert "refined readiness failed" in capsys.readouterr().err + + +def test_promote_epic_records_refined_approval_metadata_for_child_changeset( + monkeypatch, + tmp_path: Path, +) -> None: + module = _load_script_module() + transitions: list[object] = [] + created_messages: list[object] = [] + metadata_updates: dict[str, dict[str, str | None]] = {} + + monkeypatch.setattr( + module, + "_resolve_context", + lambda **_kwargs: (tmp_path / ".beads", tmp_path / "repo", None), + ) + monkeypatch.setenv("ATELIER_AGENT_ID", "atelier/planner/codex/p1") + + epic_issue = _issue( + "at-epic", + title="Epic", + description=( + "changeset_strategy: Keep review scope small.\n" + "related_context: at-context\n" + "promotion_note: ready for confirmation\n" + ), + ) + child_issue = _issue( + "at-epic.1", + title="Child", + description=_refined_description(contract_json=_valid_refined_contract_json()), + notes="child notes", + ) + + class FakeStore: + async def get_epic(self, epic_id): + assert epic_id == "at-epic" + from atelier.store import LifecycleStatus + + return SimpleNamespace(id=epic_id, lifecycle=LifecycleStatus.DEFERRED) + + async def list_changesets(self, query): + del query + from atelier.store import LifecycleStatus + + return (SimpleNamespace(id="at-epic.1", lifecycle=LifecycleStatus.DEFERRED),) + + async def create_message(self, request): + created_messages.append(request) + return SimpleNamespace(id="at-msg.1") + + async def append_notes(self, request): + return request + + async def transition_lifecycle(self, request): + transitions.append(request) + return request + + class FakeClient: + async def show(self, request): + return {"at-epic": epic_issue, "at-epic.1": child_issue}[request.issue_id] + + def _record_metadata( + *, + client: object, + issue_id: str, + fields: dict[str, str], + ) -> dict[str, object]: + del client + metadata_updates[issue_id] = fields + return {"id": issue_id} + + monkeypatch.setattr( + module, "_build_store_and_client", lambda **_kwargs: (FakeStore(), FakeClient()) + ) + monkeypatch.setattr(module, "_update_description_metadata_fields", _record_metadata) + monkeypatch.setattr(sys, "argv", ["promote_epic.py", "--epic-id", "at-epic", "--yes"]) + + module.main() + + assert [request.issue_id for request in transitions] == ["at-epic", "at-epic.1"] + assert created_messages + assert metadata_updates["at-epic.1"]["planning.stage"] == "approved" + assert metadata_updates["at-epic.1"]["planning.approved_by"] == "atelier/planner/codex/p1" + assert metadata_updates["at-epic.1"]["planning.approval_message_id"] == "at-msg.1" + assert metadata_updates["at-epic.1"]["planning.approved_at"] is not None + + +def test_promote_epic_requires_explicit_operator_identity_for_refined_approval( + monkeypatch, + capsys: pytest.CaptureFixture[str], + tmp_path: Path, +) -> None: + module = _load_script_module() + transitions: list[object] = [] + created_messages: list[object] = [] + + monkeypatch.setattr( + module, + "_resolve_context", + lambda **_kwargs: (tmp_path / ".beads", tmp_path / "repo", None), + ) + monkeypatch.delenv("ATELIER_AGENT_ID", raising=False) + + epic_issue = _issue( + "at-epic", + title="Epic", + description=( + "changeset_strategy: Keep review scope small.\n" + "related_context: at-context\n" + "promotion_note: ready for confirmation\n" + ), + ) + child_issue = _issue( + "at-epic.1", + title="Child", + description=_refined_description(contract_json=_valid_refined_contract_json()), + notes="child notes", + ) + + class FakeStore: + async def get_epic(self, epic_id): + assert epic_id == "at-epic" + from atelier.store import LifecycleStatus + + return SimpleNamespace(id=epic_id, lifecycle=LifecycleStatus.DEFERRED) + + async def list_changesets(self, query): + del query + from atelier.store import LifecycleStatus + + return (SimpleNamespace(id="at-epic.1", lifecycle=LifecycleStatus.DEFERRED),) + + async def create_message(self, request): + created_messages.append(request) + return SimpleNamespace(id="at-msg.1") + + async def append_notes(self, request): + return request + + async def transition_lifecycle(self, request): + transitions.append(request) + return request + + class FakeClient: + async def show(self, request): + return {"at-epic": epic_issue, "at-epic.1": child_issue}[request.issue_id] + + monkeypatch.setattr( + module, "_build_store_and_client", lambda **_kwargs: (FakeStore(), FakeClient()) + ) + monkeypatch.setattr(sys, "argv", ["promote_epic.py", "--epic-id", "at-epic", "--yes"]) + + with pytest.raises(SystemExit) as excinfo: + module.main() + + assert excinfo.value.code == 1 + assert "ATELIER_AGENT_ID must be set for refined approvals" in capsys.readouterr().err + assert created_messages == [] + assert transitions == [] + + +def test_promote_epic_does_not_transition_lifecycle_when_refined_approval_fails( + monkeypatch, + capsys: pytest.CaptureFixture[str], + tmp_path: Path, +) -> None: + module = _load_script_module() + transitions: list[object] = [] + + monkeypatch.setattr( + module, + "_resolve_context", + lambda **_kwargs: (tmp_path / ".beads", tmp_path / "repo", None), + ) + monkeypatch.setenv("ATELIER_AGENT_ID", "atelier/planner/codex/p1") + + epic_issue = _issue( + "at-epic", + title="Epic", + description=( + "changeset_strategy: Keep review scope small.\n" + "related_context: at-context\n" + "promotion_note: ready for confirmation\n" + ), + ) + child_issue = _issue( + "at-epic.1", + title="Child", + description=_refined_description(contract_json=_valid_refined_contract_json()), + notes="child notes", + ) + + class FakeStore: + async def get_epic(self, epic_id): + assert epic_id == "at-epic" + from atelier.store import LifecycleStatus + + return SimpleNamespace(id=epic_id, lifecycle=LifecycleStatus.DEFERRED) + + async def list_changesets(self, query): + del query + from atelier.store import LifecycleStatus + + return (SimpleNamespace(id="at-epic.1", lifecycle=LifecycleStatus.DEFERRED),) + + async def create_message(self, request): + del request + return SimpleNamespace(id="at-msg.1") + + async def append_notes(self, request): + del request + return SimpleNamespace(id="at-note.1") + + async def transition_lifecycle(self, request): + transitions.append(request) + return request + + class FakeClient: + async def show(self, request): + return {"at-epic": epic_issue, "at-epic.1": child_issue}[request.issue_id] + + monkeypatch.setattr( + module, "_build_store_and_client", lambda **_kwargs: (FakeStore(), FakeClient()) + ) + monkeypatch.setattr( + module, + "_update_description_metadata_fields", + lambda **_kwargs: (_ for _ in ()).throw(RuntimeError("metadata update failed")), + ) + monkeypatch.setattr(sys, "argv", ["promote_epic.py", "--epic-id", "at-epic", "--yes"]) + + with pytest.raises(SystemExit) as excinfo: + module.main() + + assert excinfo.value.code == 1 + assert "metadata update failed" in capsys.readouterr().err + assert transitions == [] + + +def test_promote_epic_records_refined_approval_metadata_for_epic_single_unit( + monkeypatch, + tmp_path: Path, +) -> None: + module = _load_script_module() + created_messages: list[object] = [] + metadata_updates: dict[str, dict[str, str | None]] = {} + + monkeypatch.setattr( + module, + "_resolve_context", + lambda **_kwargs: (tmp_path / ".beads", tmp_path / "repo", None), + ) + monkeypatch.setenv("ATELIER_AGENT_ID", "atelier/planner/codex/p1") + + epic_issue = _issue( + "at-epic", + title="Epic", + description=_refined_description(contract_json=_valid_refined_contract_json()), + notes="epic notes", + ) + + class FakeStore: + async def get_epic(self, epic_id): + assert epic_id == "at-epic" + from atelier.store import LifecycleStatus + + return SimpleNamespace(id=epic_id, lifecycle=LifecycleStatus.DEFERRED) + + async def list_changesets(self, query): + del query + return () + + async def create_message(self, request): + created_messages.append(request) + return SimpleNamespace(id="at-msg.2") + + async def append_notes(self, request): + return request + + async def transition_lifecycle(self, request): + return request + + class FakeClient: + async def show(self, request): + assert request.issue_id == "at-epic" + return epic_issue + + def _record_metadata( + *, + client: object, + issue_id: str, + fields: dict[str, str], + ) -> dict[str, object]: + del client + metadata_updates[issue_id] = fields + return {"id": issue_id} + + monkeypatch.setattr( + module, "_build_store_and_client", lambda **_kwargs: (FakeStore(), FakeClient()) + ) + monkeypatch.setattr(module, "_update_description_metadata_fields", _record_metadata) + monkeypatch.setattr(sys, "argv", ["promote_epic.py", "--epic-id", "at-epic", "--yes"]) + + module.main() + + assert created_messages + assert metadata_updates["at-epic"]["planning.stage"] == "approved" + assert metadata_updates["at-epic"]["planning.approval_message_id"] == "at-msg.2" + + +def test_promote_epic_non_refined_changesets_do_not_write_refined_approval( + monkeypatch, + tmp_path: Path, +) -> None: + module = _load_script_module() + transitions: list[object] = [] + + monkeypatch.setattr( + module, + "_resolve_context", + lambda **_kwargs: (tmp_path / ".beads", tmp_path / "repo", None), + ) + + epic_issue = _issue( + "at-epic", + title="Epic", + description=( + "changeset_strategy: Keep review scope small.\n" + "related_context: at-context\n" + "promotion_note: ready for confirmation\n" + ), + ) + child_issue = _issue( + "at-epic.1", + title="Child", + description="related_context: at-context\n", + notes="child notes", + ) + + class FakeStore: + async def get_epic(self, epic_id): + assert epic_id == "at-epic" + from atelier.store import LifecycleStatus + + return SimpleNamespace(id=epic_id, lifecycle=LifecycleStatus.DEFERRED) + + async def list_changesets(self, query): + del query + from atelier.store import LifecycleStatus + + return (SimpleNamespace(id="at-epic.1", lifecycle=LifecycleStatus.DEFERRED),) + + async def create_message(self, request): # pragma: no cover - defensive + raise AssertionError(request) + + async def append_notes(self, request): # pragma: no cover - defensive + raise AssertionError(request) + + async def transition_lifecycle(self, request): + transitions.append(request) + return request + + class FakeClient: + async def show(self, request): + return {"at-epic": epic_issue, "at-epic.1": child_issue}[request.issue_id] + + monkeypatch.setattr( + module, "_build_store_and_client", lambda **_kwargs: (FakeStore(), FakeClient()) + ) + monkeypatch.setattr( + module, + "_update_description_metadata_fields", + lambda *args, **kwargs: (_ for _ in ()).throw(AssertionError((args, kwargs))), + ) + monkeypatch.setattr(sys, "argv", ["promote_epic.py", "--epic-id", "at-epic", "--yes"]) + + module.main() + + assert [request.issue_id for request in transitions] == ["at-epic", "at-epic.1"] + + def test_promote_epic_preview_requires_confirmation(monkeypatch, capsys, tmp_path: Path) -> None: module = _load_script_module() transitions: list[object] = [] diff --git a/tests/atelier/skills/test_projected_skill_runtime_bootstrap.py b/tests/atelier/skills/test_projected_skill_runtime_bootstrap.py index 0cb51eb7..9355d111 100644 --- a/tests/atelier/skills/test_projected_skill_runtime_bootstrap.py +++ b/tests/atelier/skills/test_projected_skill_runtime_bootstrap.py @@ -583,6 +583,17 @@ def test_projected_check_guardrails_reorders_repo_src_ahead_of_installed_package "planner_contract.py": ( "def validate_authoring_contract(*_args, **_kwargs):\n return []\n" ), + "refined_planning_contract.py": ( + "class _Readiness:\n" + " def __init__(self):\n" + " self.refined = False\n" + " self.contract_present = False\n" + " self.stage = None\n" + " self.errors = ()\n" + "\n" + "def evaluate_issue_refined_planning_readiness(_issue):\n" + " return _Readiness()\n" + ), }, ) installed_root = _fake_installed_package( @@ -610,6 +621,17 @@ def test_projected_check_guardrails_reorders_repo_src_ahead_of_installed_package "planner_contract.py": ( "def validate_authoring_contract(*_args, **_kwargs):\n return []\n" ), + "refined_planning_contract.py": ( + "class _Readiness:\n" + " def __init__(self):\n" + " self.refined = False\n" + " self.contract_present = False\n" + " self.stage = None\n" + " self.errors = ()\n" + "\n" + "def evaluate_issue_refined_planning_readiness(_issue):\n" + " return _Readiness()\n" + ), }, ) sentinel_path = tmp_path / "check-guardrails-sentinel.txt" @@ -662,6 +684,17 @@ def test_projected_check_guardrails_ignores_inherited_pythonpath_when_repo_runti "planner_contract.py": ( "def validate_authoring_contract(*_args, **_kwargs):\n return []\n" ), + "refined_planning_contract.py": ( + "class _Readiness:\n" + " def __init__(self):\n" + " self.refined = False\n" + " self.contract_present = False\n" + " self.stage = None\n" + " self.errors = ()\n" + "\n" + "def evaluate_issue_refined_planning_readiness(_issue):\n" + " return _Readiness()\n" + ), }, ) _link_repo_python(repo_root) @@ -680,6 +713,17 @@ def test_projected_check_guardrails_ignores_inherited_pythonpath_when_repo_runti "planner_contract.py": ( "def validate_authoring_contract(*_args, **_kwargs):\n return []\n" ), + "refined_planning_contract.py": ( + "class _Readiness:\n" + " def __init__(self):\n" + " self.refined = False\n" + " self.contract_present = False\n" + " self.stage = None\n" + " self.errors = ()\n" + "\n" + "def evaluate_issue_refined_planning_readiness(_issue):\n" + " return _Readiness()\n" + ), }, ) sentinel_path = tmp_path / "check-guardrails-pydantic-sentinel.txt" @@ -709,6 +753,123 @@ def test_projected_check_guardrails_ignores_inherited_pythonpath_when_repo_runti ) +def test_projected_promote_epic_reorders_repo_src_ahead_of_installed_package( + tmp_path: Path, +) -> None: + agent_home, projected_script = _install_projected_script( + tmp_path, + skill_name="plan-promote-epic", + script_name="promote_epic.py", + ) + repo_root = _fake_repo( + tmp_path, + sentinel_import="refined_planning_contract", + extra_modules={ + "beads_context.py": ( + "class _Context:\n" + " def __init__(self, repo_dir):\n" + " self.beads_root = repo_dir\n" + " self.repo_root = repo_dir\n" + " self.override_warning = None\n" + "\n" + "def resolve_runtime_repo_dir_hint(*, repo_dir=None, cwd=None, env=None):\n" + " return (repo_dir, None)\n" + "\n" + "def resolve_skill_beads_context(*, beads_dir=None, repo_dir=None):\n" + " return _Context(repo_dir)\n" + ), + "lib/beads.py": ( + "class ShowIssueRequest:\n" + " def __init__(self, issue_id):\n" + " self.issue_id = issue_id\n" + "\n" + "class UpdateIssueRequest:\n" + " def __init__(self, issue_id, description):\n" + " self.issue_id = issue_id\n" + " self.description = description\n" + "\n" + "class description_fields:\n" + " @staticmethod\n" + " def parse_description_fields(_description):\n" + " return {}\n" + ), + "store.py": ( + "class AppendNotesRequest:\n" + " def __init__(self, issue_id, notes):\n" + " self.issue_id = issue_id\n" + " self.notes = notes\n" + "\n" + "class CreateMessageRequest:\n" + " def __init__(self, **kwargs):\n" + " self.kwargs = kwargs\n" + ), + }, + ) + installed_root = _fake_installed_package( + tmp_path, + modules={ + "refined_planning_contract.py": ( + "from pathlib import Path\n" + "import os\n" + "\n" + "Path(os.environ['BOOTSTRAP_SENTINEL']).write_text(__file__, encoding='utf-8')\n" + ), + "beads_context.py": ( + "class _Context:\n" + " def __init__(self, repo_dir):\n" + " self.beads_root = repo_dir\n" + " self.repo_root = repo_dir\n" + " self.override_warning = None\n" + "\n" + "def resolve_runtime_repo_dir_hint(*, repo_dir=None, cwd=None, env=None):\n" + " return (repo_dir, None)\n" + "\n" + "def resolve_skill_beads_context(*, beads_dir=None, repo_dir=None):\n" + " return _Context(repo_dir)\n" + ), + "lib/beads.py": ( + "class ShowIssueRequest:\n" + " pass\n" + "\n" + "class UpdateIssueRequest:\n" + " pass\n" + "\n" + "class description_fields:\n" + " @staticmethod\n" + " def parse_description_fields(_description):\n" + " return {}\n" + ), + "store.py": ( + "class AppendNotesRequest:\n pass\n\nclass CreateMessageRequest:\n pass\n" + ), + }, + ) + sentinel_path = tmp_path / "promote-epic-sentinel.txt" + + completed = subprocess.run( + [ + sys.executable, + str(projected_script), + "--repo-dir", + str(repo_root), + "--help", + ], + check=False, + capture_output=True, + text=True, + cwd=agent_home, + env={ + "BOOTSTRAP_SENTINEL": str(sentinel_path), + "PYTHONPATH": os.pathsep.join([str(installed_root), str(repo_root / "src")]), + }, + ) + + assert completed.returncode == 0 + assert sentinel_path.read_text(encoding="utf-8") == str( + repo_root / "src" / "atelier" / "refined_planning_contract.py" + ) + + def test_projected_render_tickets_section_reorders_repo_src_ahead_of_installed_package( tmp_path: Path, ) -> None: diff --git a/tests/atelier/test_planner_agents_template.py b/tests/atelier/test_planner_agents_template.py index e3376825..562d6857 100644 --- a/tests/atelier/test_planner_agents_template.py +++ b/tests/atelier/test_planner_agents_template.py @@ -21,6 +21,7 @@ def test_planner_agents_template_contains_core_sections() -> None: assert "plan-changeset-guardrails" in content assert "plan-promote-epic" in content assert "planner-startup-check" in content + assert "plan-refined-deliberation" in content assert "mail-send" in content assert "epic-list" in content assert "one child changeset" in content @@ -60,3 +61,9 @@ def test_planner_agents_template_contains_core_sections() -> None: assert "concrete issue, create or update a deferred bead immediately" in content assert "Create or update deferred beads immediately" in content assert "Capture first, then ask only for decisions" in content + assert "same drafting strategy applies to all planning passes" in content + assert "refined means iterative reevaluation of the same plan" in content + assert "low bar to reframe when a better path exists" in content + assert "high bar to ask the user" in content + assert "plan quality separate from bead mechanics" in content + assert "trycycle" not in content.lower() diff --git a/tests/atelier/test_refined_planning_contract.py b/tests/atelier/test_refined_planning_contract.py new file mode 100644 index 00000000..bb94d3fa --- /dev/null +++ b/tests/atelier/test_refined_planning_contract.py @@ -0,0 +1,392 @@ +import importlib.util +import json +import sys +from pathlib import Path +from types import SimpleNamespace + +from atelier import refined_planning_contract +from atelier.worker import work_startup_runtime + + +def _valid_contract_json( + *, + escalation_conditions: list[str] | None = None, + acceptance_evidence: list[str] | None = None, + completion_allow_unsafe_close: bool = False, +) -> str: + escalation = ( + ["validator disagreement"] if escalation_conditions is None else escalation_conditions + ) + evidence = ["tests"] if acceptance_evidence is None else acceptance_evidence + return ( + '{"objective":"Ship fail-closed refined gating",' + '"non_goals":["Do not alter non-refined flows"],' + '"acceptance_criteria":[{"statement":"Gate invalid changesets","evidence":' + f"{json.dumps(evidence)}" + "}]," + '"scope":{"includes":["worker startup"],"excludes":["CLI redesign"]},' + '"verification_plan":["uv run pytest tests/atelier/worker/test_session_startup.py -v"],' + '"risks":[{"risk":"Over-broad blocking","mitigation":"refined-only gate"}],' + f'"escalation_conditions":{json.dumps(escalation)},' + '"completion_definition":{"requires_terminal_pr_state":true,' + '"allowed_terminal_pr_states":["merged","closed"],' + '"allows_integrated_sha_proof":true,' + '"allow_close_without_terminal_or_integrated_sha":' + f"{'true' if completion_allow_unsafe_close else 'false'}" + "}}" + ) + + +def _load_guardrails_script_module(): + script_path = ( + Path(__file__).resolve().parents[2] + / "src" + / "atelier" + / "skills" + / "plan-changeset-guardrails" + / "scripts" + / "check_guardrails.py" + ) + spec = importlib.util.spec_from_file_location("check_guardrails_parity", script_path) + assert spec is not None and spec.loader is not None + module = importlib.util.module_from_spec(spec) + sys.modules[spec.name] = module + spec.loader.exec_module(module) + return module + + +def _load_promote_epic_script_module(): + script_path = ( + Path(__file__).resolve().parents[2] + / "src" + / "atelier" + / "skills" + / "plan-promote-epic" + / "scripts" + / "promote_epic.py" + ) + spec = importlib.util.spec_from_file_location("promote_epic_parity", script_path) + assert spec is not None and spec.loader is not None + module = importlib.util.module_from_spec(spec) + sys.modules[spec.name] = module + spec.loader.exec_module(module) + return module + + +def _planner_contract_text() -> str: + return ( + "intent: Keep refined startup selections fail-closed.\n" + "rationale: Workers need deterministic claim eligibility.\n" + "non_goals: Do not alter non-refined startup behavior.\n" + "constraints: Preserve lifecycle invariants and explicit audits.\n" + "edge_cases: Missing metadata must not silently bypass gates.\n" + "related_context: at-719.\n" + "LOC estimate: 220\n" + "done_definition: Done when startup/promotion parity checks pass.\n" + ) + + +def _refined_description( + *, + contract_json: str, + stage: str, + approved_by: str | None = None, + approved_at: str | None = None, + approval_message_id: str | None = None, +) -> str: + lines = [ + "execution.strategy: refined", + f"planning.stage: {stage}", + f"planning.contract_json: {contract_json}", + ] + if approved_by is not None: + lines.append(f"planning.approved_by: {approved_by}") + if approved_at is not None: + lines.append(f"planning.approved_at: {approved_at}") + if approval_message_id is not None: + lines.append(f"planning.approval_message_id: {approval_message_id}") + return _planner_contract_text() + "\n".join(lines) + "\n" + + +def _guardrails_issue(issue_id: str, description: str) -> dict[str, object]: + return { + "id": issue_id, + "title": issue_id, + "description": description, + "notes": "notes: parity harness", + "acceptance_criteria": "Done when parity checks are deterministic.", + } + + +def test_validate_contract_accepts_complete_payload() -> None: + issue = { + "description": ( + "execution.strategy: refined\n" + "planning.stage: planning_in_review\n" + f"planning.contract_json: {_valid_contract_json()}\n" + ) + } + + result = refined_planning_contract.evaluate_issue_refined_planning_readiness(issue) + + assert result.ok is True + assert result.refined is True + assert result.claim_eligible is False + assert result.claim_blockers == ( + "refined changesets require planning.stage=approved before worker claim", + ) + + +def test_validate_contract_ignores_legacy_trycycle_fields() -> None: + issue = { + "description": ( + "trycycle.targeted: true\n" + "trycycle.plan_stage: planning_in_review\n" + f"trycycle.contract_json: {_valid_contract_json()}\n" + ) + } + + result = refined_planning_contract.evaluate_issue_refined_planning_readiness(issue) + + assert result.ok is True + assert result.refined is False + assert result.claim_eligible is True + assert result.errors == () + assert result.claim_blockers == () + + +def test_validate_contract_rejects_malformed_json() -> None: + issue = { + "description": ( + "execution.strategy: refined\n" + "planning.stage: planning_in_review\n" + "planning.contract_json: {not-json}\n" + ) + } + + result = refined_planning_contract.evaluate_issue_refined_planning_readiness(issue) + + assert result.ok is False + assert "planning.contract_json must be valid JSON" in result.summary + + +def test_validate_contract_rejects_missing_escalation_conditions() -> None: + issue = { + "description": ( + "execution.strategy: refined\n" + "planning.stage: planning_in_review\n" + f"planning.contract_json: {_valid_contract_json(escalation_conditions=[])}\n" + ) + } + + result = refined_planning_contract.evaluate_issue_refined_planning_readiness(issue) + + assert result.ok is False + assert "escalation_conditions" in result.summary + + +def test_validate_contract_rejects_non_testable_acceptance_criteria() -> None: + issue = { + "description": ( + "execution.strategy: refined\n" + "planning.stage: planning_in_review\n" + f"planning.contract_json: {_valid_contract_json(acceptance_evidence=[])}\n" + ) + } + + result = refined_planning_contract.evaluate_issue_refined_planning_readiness(issue) + + assert result.ok is False + assert "evidence" in result.summary + + +def test_validate_contract_rejects_completion_definition_conflicts() -> None: + issue = { + "description": ( + "execution.strategy: refined\n" + "planning.stage: planning_in_review\n" + "planning.contract_json: " + f"{_valid_contract_json(completion_allow_unsafe_close=True)}\n" + ) + } + + result = refined_planning_contract.evaluate_issue_refined_planning_readiness(issue) + + assert result.ok is False + assert "completion_definition conflicts" in result.summary + + +def test_validate_contract_approved_stage_requires_full_evidence_fields() -> None: + issue = { + "description": ( + "execution.strategy: refined\n" + "planning.stage: approved\n" + f"planning.contract_json: {_valid_contract_json()}\n" + "planning.approved_by: operator\n" + ) + } + + result = refined_planning_contract.evaluate_issue_refined_planning_readiness(issue) + + assert result.ok is False + assert "planning.approved_at" in result.summary + assert "planning.approval_message_id" in result.summary + + +def test_validate_contract_rejects_malformed_approved_timestamp() -> None: + issue = { + "description": ( + "execution.strategy: refined\n" + "planning.stage: approved\n" + f"planning.contract_json: {_valid_contract_json()}\n" + "planning.approved_by: atelier/planner/codex/p1\n" + "planning.approved_at: yesterday\n" + "planning.approval_message_id: at-msg.1\n" + ) + } + + result = refined_planning_contract.evaluate_issue_refined_planning_readiness(issue) + + assert result.ok is False + assert "planning.approved_at must be an ISO-8601 timestamp" in result.summary + + +def test_validate_contract_rejects_date_only_approved_timestamp() -> None: + issue = { + "description": ( + "execution.strategy: refined\n" + "planning.stage: approved\n" + f"planning.contract_json: {_valid_contract_json()}\n" + "planning.approved_by: atelier/planner/codex/p1\n" + "planning.approved_at: 2026-03-27\n" + "planning.approval_message_id: at-msg.1\n" + ) + } + + result = refined_planning_contract.evaluate_issue_refined_planning_readiness(issue) + + assert result.ok is False + assert "planning.approved_at must be an ISO-8601 timestamp" in result.summary + + +def test_validate_contract_rejects_naive_approved_timestamp() -> None: + issue = { + "description": ( + "execution.strategy: refined\n" + "planning.stage: approved\n" + f"planning.contract_json: {_valid_contract_json()}\n" + "planning.approved_by: atelier/planner/codex/p1\n" + "planning.approved_at: 2026-03-27T01:00:00\n" + "planning.approval_message_id: at-msg.1\n" + ) + } + + result = refined_planning_contract.evaluate_issue_refined_planning_readiness(issue) + + assert result.ok is False + assert "planning.approved_at must be an ISO-8601 timestamp" in result.summary + + +def test_validate_contract_rejects_malformed_approval_message_id() -> None: + issue = { + "description": ( + "execution.strategy: refined\n" + "planning.stage: approved\n" + f"planning.contract_json: {_valid_contract_json()}\n" + "planning.approved_by: atelier/planner/codex/p1\n" + "planning.approved_at: 2026-03-27T01:00:00Z\n" + "planning.approval_message_id: bad id\n" + ) + } + + result = refined_planning_contract.evaluate_issue_refined_planning_readiness(issue) + + assert result.ok is False + assert "planning.approval_message_id must be an identifier" in result.summary + + +def test_refined_readiness_parity_across_entrypoints() -> None: + guardrails = _load_guardrails_script_module() + promote_epic = _load_promote_epic_script_module() + worker_service = work_startup_runtime._StartupContractService( # pyright: ignore[reportPrivateUsage] + beads_root=Path("/beads"), + repo_root=Path("/repo"), + ) + + conflict_payload = json.loads(_valid_contract_json()) + conflict_payload["completion_definition"]["allow_close_without_terminal_or_integrated_sha"] = ( + True + ) + cases = ( + ( + "valid_approved", + _refined_description( + contract_json=_valid_contract_json(), + stage="approved", + approved_by="atelier/planner/codex/p1", + approved_at="2026-03-27T01:00:00Z", + approval_message_id="at-msg.1", + ), + ), + ( + "valid_unapproved", + _refined_description( + contract_json=_valid_contract_json(), + stage="planning_in_review", + ), + ), + ( + "malformed_json", + _refined_description( + contract_json="{bad-json}", + stage="planning_in_review", + ), + ), + ( + "missing_contract", + _planner_contract_text() + + "execution.strategy: refined\nplanning.stage: planning_in_review\n", + ), + ( + "completion_conflict", + _refined_description( + contract_json=json.dumps(conflict_payload, separators=(",", ":")), + stage="planning_in_review", + ), + ), + ) + + for label, description in cases: + issue_id = f"at-parity.{label}" + issue_payload = {"id": issue_id, "description": description} + readiness = refined_planning_contract.evaluate_issue_refined_planning_readiness( + issue_payload + ) + promotion_error = promote_epic._refined_validation_error( # pyright: ignore[reportPrivateUsage] + SimpleNamespace(id=issue_id, description=description) + ) + worker_eligible, worker_reason = worker_service.refined_planning_claim_eligible( + issue_payload + ) + report = guardrails._evaluate_guardrails( # pyright: ignore[reportPrivateUsage] + epic_issue=None, + child_changesets=[], + target_changesets=[_guardrails_issue(issue_id, description)], + ) + refined_violations = tuple( + violation + for violation in report.violations + if violation.startswith(f"{issue_id}:") + and ("planning." in violation or "completion_definition conflicts" in violation) + ) + + assert (promotion_error is None) is readiness.ok + assert worker_eligible is readiness.claim_eligible + if not worker_eligible: + assert worker_reason is not None + if readiness.ok and readiness.stage == "planning_in_review": + assert refined_violations == () + elif readiness.ok and readiness.stage == "approved": + assert any("planning_in_review" in violation for violation in refined_violations) + else: + assert refined_violations diff --git a/tests/atelier/test_skills.py b/tests/atelier/test_skills.py index 230d316f..1ab01d77 100644 --- a/tests/atelier/test_skills.py +++ b/tests/atelier/test_skills.py @@ -40,6 +40,7 @@ def test_packaged_skills_include_core_set() -> None: "plan-create-epic", "plan-split-tasks", "plan-changesets", + "plan-refined-deliberation", "plan-changeset-guardrails", "plan-promote-epic", "planner-startup-check", @@ -92,6 +93,7 @@ def test_install_workspace_skills_writes_skill_docs() -> None: "plan-create-epic", "plan-split-tasks", "plan-changesets", + "plan-refined-deliberation", "plan-changeset-guardrails", "plan-promote-epic", "planner-startup-check", @@ -164,6 +166,29 @@ def test_packaged_planning_skills_include_scripts() -> None: assert "scripts/claim_message.py" in definitions["mail-queue-claim"].files +def test_refined_deliberation_skill_is_packaged_and_shared() -> None: + definitions = skills.load_packaged_skills() + skill = definitions["plan-refined-deliberation"] + text = skill.files["SKILL.md"].decode("utf-8") + + with tempfile.TemporaryDirectory() as tmp: + workspace_dir = Path(tmp) + metadata = skills.install_workspace_skills(workspace_dir) + + assert "plan-refined-deliberation" in metadata + assert (workspace_dir / "skills" / "plan-refined-deliberation" / "SKILL.md").is_file() + + assert "strategy gate before task breakdown" in text + assert "low bar to reframe when a better path exists" in text + assert "high bar to ask the user" in text + assert "fresh, stateless pass" in text + assert "stop at sign-off or loop limit" in text + assert "epic and changeset beads" in text + assert "applies to all planning, not just refinement rounds" in text + assert "epic and changeset beads, not markdown checklists" in text + assert "trycycle" not in text.lower() + + def test_work_done_skill_references_close_epic_script() -> None: skill = skills.load_packaged_skills()["work-done"] text = skill.files["SKILL.md"].decode("utf-8") @@ -325,6 +350,15 @@ def test_plan_create_epic_skill_captures_drafts_without_approval() -> None: assert "done_definition" in text +def test_planning_skills_reference_shared_discipline() -> None: + for name in ("plan-create-epic", "plan-changesets", "plan-split-tasks"): + text = skills.load_packaged_skills()[name].files["SKILL.md"].decode("utf-8") + assert "plan-refined-deliberation" in text + assert "shared planning discipline" in text + assert "same drafting strategy" in text + assert "trycycle" not in text.lower() + + def test_beads_conventions_reference_includes_concrete_authoring_examples() -> None: reference_path = ( Path(__file__).resolve().parents[2] diff --git a/tests/atelier/worker/test_lifecycle_matrix.py b/tests/atelier/worker/test_lifecycle_matrix.py index 245c82b7..454a8765 100644 --- a/tests/atelier/worker/test_lifecycle_matrix.py +++ b/tests/atelier/worker/test_lifecycle_matrix.py @@ -1,19 +1,27 @@ from __future__ import annotations import datetime as dt +import json +from collections.abc import Callable from pathlib import Path from unittest.mock import patch import pytest -from atelier import config, lifecycle, planner_overview +from atelier import config, lifecycle, planner_overview, refined_planning_contract from atelier.worker import finalize_pipeline, reconcile, selection from atelier.worker.session import startup class _NextChangesetMatrixService(startup.NextChangesetService): - def __init__(self, issue: dict[str, object]) -> None: + def __init__( + self, + issue: dict[str, object], + *, + claim_eligibility: Callable[[dict[str, object]], tuple[bool, str | None]] | None = None, + ) -> None: self._issue = issue + self._claim_eligibility = claim_eligibility or (lambda _issue: (True, None)) def show_issue(self, issue_id: str) -> dict[str, object] | None: if issue_id == "at-epic": @@ -69,6 +77,12 @@ def list_descendant_changesets( def is_changeset_in_progress(self, issue: dict[str, object]) -> bool: return lifecycle.canonical_lifecycle_status(issue.get("status")) == "in_progress" + def refined_planning_claim_eligible( + self, + issue: dict[str, object], + ) -> tuple[bool, str | None]: + return self._claim_eligibility(issue) + class _FinalizeMatrixService(finalize_pipeline.FinalizePipelineService): def __init__(self) -> None: @@ -351,3 +365,91 @@ def test_lifecycle_matrix_reconcile_ignores_terminal_labels_on_active_status() - ) assert candidates == {"at-1": ["at-1.7"]} + + +def _valid_refined_contract_json() -> str: + return json.dumps( + { + "objective": "Protect worker startup from unapproved targeted claims", + "non_goals": ["Do not alter non-refined flows"], + "acceptance_criteria": [ + {"statement": "Skip unapproved targeted work", "evidence": ["startup tests"]} + ], + "scope": { + "includes": ["worker startup"], + "excludes": ["planner UX redesign"], + }, + "verification_plan": ["uv run pytest tests/atelier/worker -k refined -v"], + "risks": [{"risk": "over-blocking", "mitigation": "targeted-only checks"}], + "escalation_conditions": ["validator disagreement"], + "completion_definition": { + "requires_terminal_pr_state": True, + "allowed_terminal_pr_states": ["merged", "closed"], + "allows_integrated_sha_proof": True, + "allow_close_without_terminal_or_integrated_sha": False, + }, + }, + separators=(",", ":"), + ) + + +def test_lifecycle_matrix_refined_unapproved_target_is_not_selected() -> None: + issue = { + "id": "at-epic", + "status": "open", + "labels": ["at:epic"], + "assignee": None, + "description": ( + "execution.strategy: refined\n" + "planning.stage: planning_in_review\n" + f"planning.contract_json: {_valid_refined_contract_json()}\n" + ), + } + startup_context = startup.NextChangesetContext( + epic_id="at-epic", + repo_slug=None, + branch_pr=False, + git_path=None, + ) + + selected = startup.next_changeset_service( + context=startup_context, + service=_NextChangesetMatrixService( + issue, claim_eligibility=refined_planning_contract.refined_planning_claim_eligible + ), + ) + + assert selected is None + + +def test_lifecycle_matrix_refined_approved_target_is_selected() -> None: + issue = { + "id": "at-epic", + "status": "open", + "labels": ["at:epic"], + "assignee": None, + "description": ( + "execution.strategy: refined\n" + "planning.stage: approved\n" + f"planning.contract_json: {_valid_refined_contract_json()}\n" + "planning.approved_by: atelier/planner/codex/p1\n" + "planning.approved_at: 2026-03-26T18:00:00Z\n" + "planning.approval_message_id: at-msg.1\n" + ), + } + startup_context = startup.NextChangesetContext( + epic_id="at-epic", + repo_slug=None, + branch_pr=False, + git_path=None, + ) + + selected = startup.next_changeset_service( + context=startup_context, + service=_NextChangesetMatrixService( + issue, claim_eligibility=refined_planning_contract.refined_planning_claim_eligible + ), + ) + + assert selected is not None + assert selected["id"] == "at-epic" diff --git a/tests/atelier/worker/test_session_next_changeset.py b/tests/atelier/worker/test_session_next_changeset.py index efb3179e..0778abcb 100644 --- a/tests/atelier/worker/test_session_next_changeset.py +++ b/tests/atelier/worker/test_session_next_changeset.py @@ -16,6 +16,7 @@ def __init__( waiting_by_id: dict[str, bool] | None = None, integrated_by_id: dict[str, bool] | None = None, work_children_by_id: dict[str, list[dict[str, object]]] | None = None, + refined_eligibility_by_id: dict[str, tuple[bool, str | None]] | None = None, ) -> None: self._issues_by_id = issues_by_id self._ready_changesets = ready_changesets @@ -24,6 +25,7 @@ def __init__( self._waiting_by_id = waiting_by_id or {} self._integrated_by_id = integrated_by_id or {} self._work_children_by_id = work_children_by_id or {} + self._refined_eligibility_by_id = refined_eligibility_by_id or {} def show_issue(self, issue_id: str) -> dict[str, object] | None: return self._issues_by_id.get(issue_id) @@ -120,6 +122,15 @@ def changeset_integration_signal( def is_changeset_in_progress(self, issue: dict[str, object]) -> bool: return str(issue.get("status") or "").strip().lower() == "in_progress" + def refined_planning_claim_eligible( + self, + issue: dict[str, object], + ) -> tuple[bool, str | None]: + issue_id = issue.get("id") + if not isinstance(issue_id, str): + return True, None + return self._refined_eligibility_by_id.get(issue_id, (True, None)) + def _changeset( issue_id: str, @@ -623,3 +634,19 @@ def test_next_changeset_service_rehydrates_sparse_ready_candidates() -> None: assert selected is not None assert selected["id"] == "at-epic.1" + + +def test_next_changeset_service_skips_unapproved_refined_changeset() -> None: + blocked = _changeset("at-epic.1", status="open", work_branch="feat/at-epic.1") + allowed = _changeset("at-epic.2", status="open", work_branch="feat/at-epic.2") + service = FakeNextChangesetService( + issues_by_id={"at-epic": _epic(), blocked["id"]: blocked, allowed["id"]: allowed}, + ready_changesets=[], + descendants=[blocked, allowed], + refined_eligibility_by_id={"at-epic.1": (False, "missing refined approval")}, + ) + + selected = startup.next_changeset_service(context=_context(), service=service) + + assert selected is not None + assert selected["id"] == "at-epic.2" diff --git a/tests/atelier/worker/test_session_startup.py b/tests/atelier/worker/test_session_startup.py index ef6ad5e0..fb8c05c9 100644 --- a/tests/atelier/worker/test_session_startup.py +++ b/tests/atelier/worker/test_session_startup.py @@ -1,11 +1,13 @@ from __future__ import annotations +import json from collections.abc import Callable from pathlib import Path from types import SimpleNamespace from typing import Any from unittest.mock import patch +from atelier import refined_planning_contract from atelier.worker import integration, work_startup_runtime from atelier.worker.models_boundary import parse_issue_boundary from atelier.worker.review import MergeConflictSelection, ReviewFeedbackSelection @@ -69,6 +71,9 @@ def __init__(self, **overrides: Any) -> None: self._select_global_review_feedback_changeset = overrides.pop( "select_global_review_feedback_changeset", lambda **_kwargs: None ) + self._refined_planning_claim_eligible = overrides.pop( + "refined_planning_claim_eligible", lambda _issue: (True, None) + ) self._check_inbox_before_claim = overrides.pop( "check_inbox_before_claim", lambda *_args: False ) @@ -217,6 +222,12 @@ def select_global_review_feedback_changeset( ) -> ReviewFeedbackSelection | None: return self._select_global_review_feedback_changeset(repo_slug=repo_slug) + def refined_planning_claim_eligible( + self, + issue: dict[str, object], + ) -> tuple[bool, str | None]: + return self._refined_planning_claim_eligible(issue) + def check_inbox_before_claim(self, agent_id: str) -> bool: return self._check_inbox_before_claim(agent_id) @@ -300,6 +311,48 @@ def _run_startup(**overrides: Any) -> startup.StartupContractResult: return startup.run_startup_contract_service(context=context, service=service) +def _valid_refined_contract_json() -> str: + return json.dumps( + { + "objective": "Gate all startup claim paths with shared refined checks", + "non_goals": ["Do not alter non-refined selection"], + "acceptance_criteria": [ + {"statement": "Reject unapproved targeted work", "evidence": ["pytest"]} + ], + "scope": {"includes": ["worker startup"], "excludes": ["planner workflows"]}, + "verification_plan": ["uv run pytest tests/atelier/worker/test_session_startup.py -v"], + "risks": [{"risk": "selection drift", "mitigation": "shared validator parity"}], + "escalation_conditions": ["claim-path disagreement"], + "completion_definition": { + "requires_terminal_pr_state": True, + "allowed_terminal_pr_states": ["merged", "closed"], + "allows_integrated_sha_proof": True, + "allow_close_without_terminal_or_integrated_sha": False, + }, + }, + separators=(",", ":"), + ) + + +def _refined_target_description(*, approved: bool) -> str: + lines = [ + "execution.strategy: refined", + f"planning.contract_json: {_valid_refined_contract_json()}", + ] + if approved: + lines.extend( + [ + "planning.stage: approved", + "planning.approved_by: atelier/planner/codex/p1", + "planning.approved_at: 2026-03-26T18:00:00Z", + "planning.approval_message_id: at-msg.1", + ] + ) + else: + lines.append("planning.stage: planning_in_review") + return "\n".join(lines) + "\n" + + def test_run_startup_contract_service_supports_typed_context() -> None: context, service = _startup_context_service( explicit_epic_id="at-explicit", @@ -412,6 +465,179 @@ def test_run_startup_contract_explicit_epic_prioritizes_merge_conflict() -> None assert result.changeset_id == "at-explicit" +def test_run_startup_contract_explicit_epic_rejects_unapproved_refined_next_changeset() -> None: + emitted: list[str] = [] + + result = _run_startup( + explicit_epic_id="at-explicit", + show_issue=lambda issue_id: { + "id": issue_id, + "status": "open", + "labels": ["at:epic"] if issue_id == "at-explicit" else [], + "issue_type": "task" if issue_id != "at-explicit" else "epic", + }, + next_changeset=lambda **_kwargs: {"id": "at-explicit.1", "status": "open", "labels": []}, + refined_planning_claim_eligible=lambda issue: ( + (False, "missing refined approval") + if issue.get("id") == "at-explicit.1" + else (True, None) + ), + emit=lambda message: emitted.append(message), + ) + + assert result.should_exit is True + assert result.reason == "explicit_epic_not_actionable" + assert any("missing refined approval" in message for message in emitted) + + +def test_run_startup_contract_explicit_review_feedback_rejects_unapproved_refined() -> None: + feedback = ReviewFeedbackSelection( + epic_id="at-explicit", + changeset_id="at-explicit.1", + feedback_at="2026-02-20T00:00:00Z", + ) + emitted: list[str] = [] + + result = _run_startup( + explicit_epic_id="at-explicit", + branch_pr=True, + repo_slug="org/repo", + show_issue=lambda issue_id: { + "id": issue_id, + "status": "open", + "labels": ["at:epic"] if issue_id == "at-explicit" else [], + }, + select_review_feedback_changeset=lambda **_kwargs: feedback, + next_changeset=lambda **_kwargs: None, + refined_planning_claim_eligible=lambda issue: ( + (False, "missing refined approval") + if issue.get("id") == "at-explicit.1" + else (True, None) + ), + emit=lambda message: emitted.append(message), + ) + + assert result.reason == "explicit_epic_not_actionable" + assert any("missing refined approval" in message for message in emitted) + + +def test_run_startup_contract_explicit_review_feedback_fails_closed_on_missing_metadata() -> None: + feedback = ReviewFeedbackSelection( + epic_id="at-explicit", + changeset_id="at-explicit.1", + feedback_at="2026-02-20T00:00:00Z", + ) + emitted: list[str] = [] + + result = _run_startup( + explicit_epic_id="at-explicit", + branch_pr=True, + repo_slug="org/repo", + show_issue=lambda issue_id: ( + {"id": issue_id, "status": "open", "labels": ["at:epic"]} + if issue_id == "at-explicit" + else None + ), + select_review_feedback_changeset=lambda **_kwargs: feedback, + next_changeset=lambda **_kwargs: None, + emit=lambda message: emitted.append(message), + ) + + assert result.reason == "explicit_epic_not_actionable" + assert any( + "unable to load changeset metadata for refined claim gate" in message for message in emitted + ) + + +def test_run_startup_contract_explicit_merge_conflict_rejects_unapproved_refined() -> None: + conflict = MergeConflictSelection( + epic_id="at-explicit", + changeset_id="at-explicit.1", + observed_at="2026-02-20T00:00:00Z", + pr_url="https://github.com/org/repo/pull/110", + ) + emitted: list[str] = [] + + result = _run_startup( + explicit_epic_id="at-explicit", + branch_pr=True, + repo_slug="org/repo", + show_issue=lambda issue_id: { + "id": issue_id, + "status": "open", + "labels": ["at:epic"] if issue_id == "at-explicit" else [], + }, + select_conflicted_changeset=lambda **_kwargs: conflict, + next_changeset=lambda **_kwargs: None, + refined_planning_claim_eligible=lambda issue: ( + (False, "missing refined approval") + if issue.get("id") == "at-explicit.1" + else (True, None) + ), + emit=lambda message: emitted.append(message), + ) + + assert result.reason == "explicit_epic_not_actionable" + assert any("missing refined approval" in message for message in emitted) + + +def test_run_startup_contract_explicit_refined_paths_use_shared_validator() -> None: + scenarios: tuple[tuple[str, bool, str], ...] = ( + ("review", False, "explicit_epic_not_actionable"), + ("review", True, "review_feedback"), + ("conflict", False, "explicit_epic_not_actionable"), + ("conflict", True, "merge_conflict"), + ) + for selector, approved, expected_reason in scenarios: + emitted: list[str] = [] + feedback = ( + ReviewFeedbackSelection( + epic_id="at-explicit", + changeset_id="at-explicit.1", + feedback_at="2026-02-20T00:00:00Z", + ) + if selector == "review" + else None + ) + conflict = ( + MergeConflictSelection( + epic_id="at-explicit", + changeset_id="at-explicit.1", + observed_at="2026-02-20T00:00:00Z", + pr_url="https://github.com/org/repo/pull/110", + ) + if selector == "conflict" + else None + ) + + result = _run_startup( + explicit_epic_id="at-explicit", + branch_pr=True, + repo_slug="org/repo", + show_issue=lambda issue_id: { + "id": issue_id, + "status": "open", + "labels": ["at:epic"] if issue_id == "at-explicit" else [], + "description": ( + _refined_target_description(approved=approved) + if issue_id == "at-explicit.1" + else "" + ), + }, + select_conflicted_changeset=lambda **_kwargs: conflict, + select_review_feedback_changeset=lambda **_kwargs: feedback, + next_changeset=lambda **_kwargs: None, + refined_planning_claim_eligible=refined_planning_contract.refined_planning_claim_eligible, + emit=lambda message: emitted.append(message), + ) + + assert result.reason == expected_reason + if not approved: + assert any("planning.stage=approved" in message for message in emitted) + else: + assert result.changeset_id == "at-explicit.1" + + def test_run_startup_contract_explicit_epic_completed_exits_cleanly() -> None: emitted: list[str] = [] result = _run_startup( @@ -775,6 +1001,12 @@ def next_changeset(**_kwargs: Any) -> dict[str, object] | None: branch_pr=True, repo_slug="org/repo", resolve_hooked_epic=lambda *_args: "at-epic", + show_issue=lambda issue_id: { + "id": issue_id, + "status": "open", + "labels": ["at:epic"] if issue_id == "at-epic" else [], + "description": "", + }, select_review_feedback_changeset=lambda **_kwargs: feedback, next_changeset=next_changeset, list_epics=lambda: [ @@ -901,6 +1133,12 @@ def select_review_feedback_changeset( select="first-eligible", branch_pr=True, repo_slug="org/repo", + show_issue=lambda issue_id: { + "id": issue_id, + "status": "open", + "labels": ["at:epic"] if issue_id in {"at-early", "at-oldest"} else [], + "description": "", + }, list_epics=lambda: [ { "id": "at-early", @@ -952,6 +1190,12 @@ def select_review_feedback_changeset( select="oldest-feedback", branch_pr=True, repo_slug="org/repo", + show_issue=lambda issue_id: { + "id": issue_id, + "status": "open", + "labels": ["at:epic"] if issue_id in {"at-early", "at-oldest"} else [], + "description": "", + }, list_epics=lambda: [ { "id": "at-early", @@ -995,6 +1239,12 @@ def next_changeset(**_kwargs: Any) -> dict[str, object] | None: branch_pr=True, repo_slug="org/repo", resolve_hooked_epic=lambda *_args: "at-epic", + show_issue=lambda issue_id: { + "id": issue_id, + "status": "open", + "labels": ["at:epic"] if issue_id == "at-epic" else [], + "description": "", + }, select_conflicted_changeset=lambda **_kwargs: conflict, next_changeset=next_changeset, list_epics=lambda: [{"id": "at-epic", "assignee": "atelier/worker/codex/p010"}], @@ -1031,6 +1281,12 @@ def select_feedback(*, epic_id: str, repo_slug: str | None) -> ReviewFeedbackSel result = _run_startup( branch_pr=True, repo_slug="org/repo", + show_issue=lambda issue_id: { + "id": issue_id, + "status": "open", + "labels": ["at:epic"] if issue_id in {"at-blocked", "at-claimable"} else [], + "description": "", + }, list_epics=lambda: [ { "id": "at-blocked", @@ -1116,6 +1372,12 @@ def test_run_startup_contract_selects_stale_reclaimable_review_feedback() -> Non result = _run_startup( branch_pr=True, repo_slug="org/repo", + show_issue=lambda issue_id: { + "id": issue_id, + "status": "open", + "labels": ["at:epic"] if issue_id == "at-stale" else [], + "description": "", + }, list_epics=lambda: [stale_issue], stale_family_assigned_epics=lambda _issues, agent_id: [stale_issue], next_changeset=lambda **_kwargs: {"id": "at-stale.2"}, @@ -1172,6 +1434,14 @@ def test_run_startup_contract_resumes_unassigned_draft_pr_review_followup() -> N "atelier.worker.work_startup_runtime.worker_selection.stale_family_assigned_epics", return_value=[], ), + patch( + "atelier.worker.work_startup_runtime.worker_store.show_issue", + side_effect=lambda issue_id, *, beads_root, repo_root: ( + epic + if issue_id == "at-v1se7" + else (changeset if issue_id == "at-v1se7.1" else None) + ), + ), patch( "atelier.worker.work_startup_runtime.select_conflicted_changeset", return_value=None, @@ -1256,6 +1526,158 @@ def test_run_startup_contract_skips_unclaimable_global_review_feedback() -> None assert result.epic_id == "at-claimable" +def test_run_startup_contract_global_review_feedback_rejects_unapproved_refined() -> None: + blocked_feedback = ReviewFeedbackSelection( + epic_id="at-blocked", + changeset_id="at-blocked.1", + feedback_at="2026-02-19T00:00:00Z", + ) + emitted: list[str] = [] + + result = _run_startup( + branch_pr=True, + repo_slug="org/repo", + list_epics=lambda: [ + { + "id": "at-claimable", + "status": "open", + "labels": ["at:epic"], + "assignee": None, + "created_at": "2026-02-21T00:00:00Z", + } + ], + show_issue=lambda issue_id: { + "id": issue_id, + "status": "open", + "labels": ["at:epic"] if issue_id == "at-blocked" else [], + }, + next_changeset=lambda **kwargs: {"id": f"{kwargs['epic_id']}.1"}, + select_review_feedback_changeset=lambda **_kwargs: None, + select_global_review_feedback_changeset=lambda **_kwargs: blocked_feedback, + refined_planning_claim_eligible=lambda issue: ( + (False, "missing refined approval") + if issue.get("id") == "at-blocked.1" + else (True, None) + ), + emit=lambda message: emitted.append(message), + ) + + assert result.reason == "selected_auto" + assert result.epic_id == "at-claimable" + assert any("missing refined approval" in message for message in emitted) + + +def test_run_startup_contract_global_merge_conflict_rejects_unapproved_refined() -> None: + blocked_conflict = MergeConflictSelection( + epic_id="at-blocked", + changeset_id="at-blocked.1", + observed_at="2026-02-19T00:00:00Z", + pr_url="https://github.com/org/repo/pull/404", + ) + emitted: list[str] = [] + + result = _run_startup( + branch_pr=True, + repo_slug="org/repo", + list_epics=lambda: [ + { + "id": "at-claimable", + "status": "open", + "labels": ["at:epic"], + "assignee": None, + "created_at": "2026-02-21T00:00:00Z", + } + ], + show_issue=lambda issue_id: { + "id": issue_id, + "status": "open", + "labels": ["at:epic"] if issue_id == "at-blocked" else [], + }, + next_changeset=lambda **kwargs: {"id": f"{kwargs['epic_id']}.1"}, + select_conflicted_changeset=lambda **_kwargs: None, + select_global_conflicted_changeset=lambda **_kwargs: blocked_conflict, + refined_planning_claim_eligible=lambda issue: ( + (False, "missing refined approval") + if issue.get("id") == "at-blocked.1" + else (True, None) + ), + emit=lambda message: emitted.append(message), + ) + + assert result.reason == "selected_auto" + assert result.epic_id == "at-claimable" + assert any("missing refined approval" in message for message in emitted) + + +def test_run_startup_contract_global_refined_paths_use_shared_validator() -> None: + scenarios: tuple[tuple[str, bool, str], ...] = ( + ("review", False, "selected_auto"), + ("review", True, "review_feedback"), + ("conflict", False, "selected_auto"), + ("conflict", True, "merge_conflict"), + ) + for selector, approved, expected_reason in scenarios: + emitted: list[str] = [] + feedback = ( + ReviewFeedbackSelection( + epic_id="at-blocked", + changeset_id="at-blocked.1", + feedback_at="2026-02-19T00:00:00Z", + ) + if selector == "review" + else None + ) + conflict = ( + MergeConflictSelection( + epic_id="at-blocked", + changeset_id="at-blocked.1", + observed_at="2026-02-19T00:00:00Z", + pr_url="https://github.com/org/repo/pull/404", + ) + if selector == "conflict" + else None + ) + + result = _run_startup( + branch_pr=True, + repo_slug="org/repo", + list_epics=lambda: [ + { + "id": "at-claimable", + "status": "open", + "labels": ["at:epic"], + "assignee": None, + "created_at": "2026-02-21T00:00:00Z", + } + ], + show_issue=lambda issue_id: { + "id": issue_id, + "status": "open", + "labels": ["at:epic"] if issue_id in {"at-blocked", "at-claimable"} else [], + "description": ( + _refined_target_description(approved=approved) + if issue_id == "at-blocked.1" + else "" + ), + }, + next_changeset=lambda **kwargs: {"id": f"{kwargs['epic_id']}.1"}, + select_conflicted_changeset=lambda **_kwargs: None, + select_global_conflicted_changeset=lambda **_kwargs: conflict, + select_review_feedback_changeset=lambda **_kwargs: None, + select_global_review_feedback_changeset=lambda **_kwargs: feedback, + refined_planning_claim_eligible=refined_planning_contract.refined_planning_claim_eligible, + emit=lambda message: emitted.append(message), + ) + + assert result.reason == expected_reason + if not approved: + assert result.epic_id == "at-claimable" + assert any("planning.stage=approved" in message for message in emitted) + else: + assert result.epic_id == "at-blocked" + assert result.changeset_id == "at-blocked.1" + + def test_run_startup_contract_claims_global_feedback_standalone_identity() -> None: standalone_feedback = ReviewFeedbackSelection( epic_id="at-bmu", diff --git a/tests/atelier/worker/test_work_startup_runtime.py b/tests/atelier/worker/test_work_startup_runtime.py index 48228317..8c3b84be 100644 --- a/tests/atelier/worker/test_work_startup_runtime.py +++ b/tests/atelier/worker/test_work_startup_runtime.py @@ -1,4 +1,5 @@ import asyncio +import json from pathlib import Path from atelier.lib.beads import SyncBeadsClient @@ -146,3 +147,123 @@ def _raise_on_raw_beads(*_args, **_kwargs): service.update_changeset_integrated_sha("at-epic.1", "abc1234") assert calls == [("at-epic.1", "abc1234", Path("/beads"), Path("/repo"), True)] + + +def test_next_changeset_service_refined_eligibility_uses_shared_helper(monkeypatch) -> None: + calls: list[dict[str, object]] = [] + issue = {"id": "at-epic.1", "description": "execution.strategy: refined"} + + monkeypatch.setattr( + work_startup_runtime, + "_refined_planning_claim_eligibility", + lambda candidate: (calls.append(candidate), (False, "blocked"))[1], + ) + service = work_startup_runtime._NextChangesetService( # pyright: ignore[reportPrivateUsage] + beads_root=Path("/beads"), + repo_root=Path("/repo"), + ) + + eligible, reason = service.refined_planning_claim_eligible(issue) + + assert (eligible, reason) == (False, "blocked") + assert calls == [issue] + + +def test_startup_contract_service_refined_eligibility_uses_shared_helper(monkeypatch) -> None: + calls: list[dict[str, object]] = [] + issue = {"id": "at-epic.1", "description": "execution.strategy: refined"} + + monkeypatch.setattr( + work_startup_runtime, + "_refined_planning_claim_eligibility", + lambda candidate: (calls.append(candidate), (True, None))[1], + ) + service = work_startup_runtime._StartupContractService( # pyright: ignore[reportPrivateUsage] + beads_root=Path("/beads"), + repo_root=Path("/repo"), + ) + + eligible, reason = service.refined_planning_claim_eligible(issue) + + assert (eligible, reason) == (True, None) + assert calls == [issue] + + +def test_startup_contract_service_refined_eligibility_hydrates_sparse_issue( + monkeypatch, +) -> None: + contract_json = json.dumps( + { + "objective": "Gate startup candidates", + "non_goals": ["Do not alter non-targeted startup paths"], + "acceptance_criteria": [{"statement": "Reject unapproved", "evidence": ["pytest"]}], + "scope": {"includes": ["startup"], "excludes": ["planner workflow"]}, + "verification_plan": ["uv run pytest tests/atelier/worker -k refined -v"], + "risks": [{"risk": "claim drift", "mitigation": "shared validator"}], + "escalation_conditions": ["validator mismatch"], + "completion_definition": { + "requires_terminal_pr_state": True, + "allowed_terminal_pr_states": ["merged", "closed"], + "allows_integrated_sha_proof": True, + "allow_close_without_terminal_or_integrated_sha": False, + }, + }, + separators=(",", ":"), + ) + hydrated_issue = { + "id": "at-epic.1", + "description": ( + "execution.strategy: refined\n" + "planning.stage: planning_in_review\n" + f"planning.contract_json: {contract_json}\n" + ), + } + calls: list[str] = [] + + monkeypatch.setattr( + worker_store, + "show_issue", + lambda issue_id, *, beads_root, repo_root: ( + calls.append(f"{issue_id}|{beads_root}|{repo_root}"), + hydrated_issue, + )[1], + ) + + service = work_startup_runtime._StartupContractService( # pyright: ignore[reportPrivateUsage] + beads_root=Path("/beads"), + repo_root=Path("/repo"), + ) + + eligible, reason = service.refined_planning_claim_eligible({"id": "at-epic.1"}) + + assert (eligible, reason) == ( + False, + "refined changesets require planning.stage=approved before worker claim", + ) + assert calls == ["at-epic.1|/beads|/repo"] + + +def test_startup_contract_service_refined_eligibility_fails_closed_when_hydration_missing( + monkeypatch, +) -> None: + calls: list[str] = [] + monkeypatch.setattr( + worker_store, + "show_issue", + lambda issue_id, *, beads_root, repo_root: ( + calls.append(f"{issue_id}|{beads_root}|{repo_root}"), + None, + )[1], + ) + service = work_startup_runtime._StartupContractService( # pyright: ignore[reportPrivateUsage] + beads_root=Path("/beads"), + repo_root=Path("/repo"), + ) + + eligible, reason = service.refined_planning_claim_eligible({"id": "at-missing.1"}) + + assert (eligible, reason) == ( + False, + "unable to load changeset metadata for refined claim gate", + ) + assert calls == ["at-missing.1|/beads|/repo"]