From 1ede25139be84d4b5b172c69c4f2bc199c196b9e Mon Sep 17 00:00:00 2001 From: Rickard von Haugwitz Date: Fri, 29 May 2026 16:42:22 +0200 Subject: [PATCH] Require planning promotion for epic candidates --- .../epic-promotion-guardrail.plan.json | 314 ++++++++++++++++++ .../workspace_runtime_primitives.py | 235 ++++++++++++- tests/test_workspace_implement_cli.py | 59 +++- tests/test_workspace_start_preflight_cli.py | 95 +++++- 4 files changed, 696 insertions(+), 7 deletions(-) create mode 100644 .agentic-workspace/planning/execplans/archive/epic-promotion-guardrail.plan.json diff --git a/.agentic-workspace/planning/execplans/archive/epic-promotion-guardrail.plan.json b/.agentic-workspace/planning/execplans/archive/epic-promotion-guardrail.plan.json new file mode 100644 index 00000000..b091be77 --- /dev/null +++ b/.agentic-workspace/planning/execplans/archive/epic-promotion-guardrail.plan.json @@ -0,0 +1,314 @@ +{ + "kind": "planning-execplan/v1", + "title": "Epic candidate-lane promotion guardrail", + "execplan_profile": { + "schema": "execplan-profile/v1", + "task_shape": "bounded", + "required_core": [ + "kind", + "title", + "canonical_core", + "goal", + "non_goals", + "active_milestone", + "validation_commands", + "completion_criteria" + ], + "optional_sections": [ + "intent_continuity", + "intent_interpretation", + "execution_bounds", + "stop_conditions", + "context_budget", + "delegated_judgment", + "post_decomposition_delegation" + ], + "projection_rule": "canonical_core is authoritative for intent, scope, next action, proof, continuation, and closeout; legacy fields remain compatibility projections." + }, + "canonical_core": { + "requested_outcome": "GitHub #1215", + "hard_constraints": "Keep scope bounded to the promoted TODO item and its stated touched paths.", + "agent_may_decide": "Bounded decomposition, touched-path narrowing, validation tightening, and plan-local residue routing.", + "escalate_when": "A better-looking fix changes the requested outcome, owned surface, time horizon, or meaningful validation story.", + "next_action": "Fill in execution bounds, touched paths, and validation before implementation starts.", + "proof_expectations": [ + "uv run pytest tests/test_workspace_start_preflight_cli.py tests/test_workspace_implement_cli.py -q; make test-workspace; make lint-workspace" + ], + "touched_scope": [ + "closeout scope recorded in closure_check and generated_closeout." + ], + "completion_criteria": [ + "Epic candidate-lane promotion guardrail is implemented, validated, and closed out honestly." + ], + "continuation_owner": "none", + "closeout_decision": "archive-and-close" + }, + "goal": [ + "GitHub #1215" + ], + "non_goals": [ + "Leave adjacent backlog or follow-on work out of this plan." + ], + "machine_readable_contract": { + "intent": { + "outcome": "GitHub #1215", + "constraints": "Keep scope bounded to the promoted TODO item and its stated touched paths.", + "latitude": "Bounded decomposition, touched-path narrowing, validation tightening, and plan-local residue routing.", + "escalation": "Escalate when a better-looking fix changes the requested outcome, owned surface, time horizon, or meaningful validation story.", + "proof": "uv run pytest tests/test_workspace_start_preflight_cli.py tests/test_workspace_implement_cli.py -q; make test-workspace; make lint-workspace" + }, + "execution": { + "milestone": "epic-promotion-guardrail", + "status": "active", + "next_step": "Fill in execution bounds, touched paths, and validation before implementation starts.", + "proof": "uv run pytest tests/test_workspace_start_preflight_cli.py tests/test_workspace_implement_cli.py -q; make test-workspace; make lint-workspace" + }, + "scope": { + "touched": [ + "closeout scope recorded in closure_check and generated_closeout." + ], + "invariants": [ + "Preserve the planning contract and keep the work bounded to this plan." + ] + } + }, + "intent_continuity": { + "larger intended outcome": "GitHub #1215", + "this slice completes the larger intended outcome": "yes", + "continuation surface": "none" + }, + "required_continuation": { + "required follow-on for the larger intended outcome": "no", + "owner surface": "none", + "activation trigger": "none" + }, + "iterative_follow_through": { + "what this slice enabled": "none yet", + "intentionally deferred": "none", + "discovered implications": "none yet", + "proof achieved now": "yes; planning closeout recorded explicit proof input.", + "validation still needed": "current milestone validation remains pending", + "next likely slice": "continue the current milestone until the completion criteria are met" + }, + "intent_interpretation": { + "literal request": "Epic candidate-lane promotion guardrail", + "inferred intended outcome": "GitHub #1215", + "chosen concrete what": "Fill in execution bounds, touched paths, and validation before implementation starts.", + "interpretation distance": "low", + "review guidance": "Confirm the scaffolded plan still matches the promoted item before broad implementation." + }, + "execution_bounds": { + "allowed paths": "closeout scope recorded in closure_check and generated_closeout.", + "max changed files": "closed slice; no further writes expected without reopening a fresh plan.", + "required validation commands": "uv run pytest tests/test_workspace_start_preflight_cli.py tests/test_workspace_implement_cli.py -q; make test-workspace; make lint-workspace", + "ask-before-refactor threshold": "Ask before broadening beyond the promoted item.", + "stop before touching": "Unrelated backlog, adjacent modules, or canonical contracts not named by this plan." + }, + "stop_conditions": { + "stop when": "The work no longer matches the promoted item or its completion criteria.", + "escalate when boundary reached": "A correct fix requires changing the requested outcome or ownership boundary.", + "escalate on scope drift": "Implementation needs files outside the filled execution bounds.", + "escalate on proof failure": "The selected proof cannot demonstrate the completion criteria." + }, + "context_budget": { + "live working set": "This execplan, the promoted item, and the narrow files needed for the current implementation step.", + "recoverable later": "Repo background, historical reviews, and deferred backlog unless compact outputs point there.", + "externalize before shift": "Update execution_run, proof_report, finished_run_review, and closeout_distillation before pausing.", + "pre-work config pull": "Use compact config/startup/summary outputs before opening raw planning or routing files.", + "pre-work memory pull": "Route to the narrowest relevant memory only when the task needs durable repo knowledge.", + "tiny resumability note": "Fill in execution bounds, touched paths, and validation before implementation starts.", + "context-shift triggers": "Proof failure, scope drift, interruption, handoff, or closeout." + }, + "delegated_judgment": { + "requested outcome": "GitHub #1215", + "hard constraints": "Keep scope bounded to the promoted TODO item and its stated touched paths.", + "agent may decide locally": "Bounded decomposition, touched-path narrowing, validation tightening, and plan-local residue routing.", + "escalate when": "A better-looking fix changes the requested outcome, owned surface, time horizon, or meaningful validation story." + }, + "post_decomposition_delegation": { + "status": "recorded", + "decision rule": "After this slice is bounded, decide whether direct work, read-only exploration, implementation handoff, validation handoff, or stronger review improves quality or saves tokens safely.", + "route candidates": "keep-local|delegate-exploration|delegate-implementation|delegate-validation|escalate-review|no-safe-route", + "required evidence": "slice id, route, reason, quality risk, token-saving class, read-first refs, write scope, proof burden, stop conditions, and return contract", + "route chosen": "keep-local", + "route skipped reason": "Single bounded routing change needs local integration across start and implement surfaces; delegation would add handoff overhead without reducing proof burden.", + "decision command": "agentic-planning delegation-decision", + "planning revision observed": "76e1a176029a371b", + "recorded at": "2026-05-29T14:36:51+00:00" + }, + "system_intent_alignment": { + "relevant system intent": "Preserve the larger intended outcome separately from this bounded slice.", + "slice shaping bias": "Keep the slice bounded while carrying any larger follow-on through explicit continuation fields.", + "broader-lane validation question": "Did this slice advance the declared larger outcome, or only complete the local task?", + "intent evidence source": ".agentic-workspace/docs/system-intent-contract.md" + }, + "references": [ + { + "kind": "source", + "target": "GitHub #1215", + "label": "GitHub #1215", + "role": "intake", + "locator": "" + } + ], + "active_milestone": { + "id": "epic-promotion-guardrail", + "status": "completed", + "scope": "Keep this execution thread bounded to the promoted TODO item.", + "ready": "ready", + "blocked": "none", + "optional_deps": "none" + }, + "immediate_next_action": [ + "Fill in execution bounds, touched paths, and validation before implementation starts." + ], + "blockers": [ + "None." + ], + "touched_paths": [ + "closeout scope recorded in closure_check and generated_closeout." + ], + "invariants": [ + "Preserve the planning contract and keep the work bounded to this plan." + ], + "validation_commands": [ + "uv run pytest tests/test_workspace_start_preflight_cli.py tests/test_workspace_implement_cli.py -q; make test-workspace; make lint-workspace" + ], + "required_tools": [ + "None." + ], + "completion_criteria": [ + "Epic candidate-lane promotion guardrail is implemented, validated, and closed out honestly." + ], + "execution_run": { + "run status": "completed", + "executor": "agentic-planning closeout", + "handoff source": "agentic-planning new-plan", + "what happened": "Implemented Planning candidate pressure and issue-scope evidence in start/implement planning safety. Broad or lane-shaped work with open roadmap/decomposition candidates now blocks for lane promotion or decomposition, while bare issue refs without cached external evidence are marked high-risk unknown instead of silently bounded.", + "scope touched": "Workspace runtime routing and start/implement CLI guardrail tests.", + "changed surfaces": "src/agentic_workspace/workspace_runtime_primitives.py; tests/test_workspace_start_preflight_cli.py; tests/test_workspace_implement_cli.py", + "validations run": "uv run pytest tests/test_workspace_start_preflight_cli.py tests/test_workspace_implement_cli.py -q; make test-workspace; make lint-workspace", + "result for continuation": "bounded closeout complete", + "next step": "archive this execplan" + }, + "finished_run_review": { + "review status": "complete", + "scope respected": "Scope stayed inside routing projection and tests; existing direct-work paths remain clear unless Planning evidence or issue-scope uncertainty applies.", + "proof status": "passed", + "intent served": "yes", + "config compliance": "used planning closeout command-owned writer", + "misinterpretation risk": "low", + "follow-on decision": "none" + }, + "delegation_outcome_feedback": { + "route chosen": "keep-local", + "route skipped reason": "Single bounded routing change needs local integration across start and implement surfaces; delegation would add handoff overhead without reducing proof burden.", + "expected savings": "unknown", + "actual friction": "none recorded", + "proof result": "yes; planning closeout recorded explicit proof input.", + "quality concern": "none recorded", + "decomposition adjustment": "none" + }, + "proof_report": { + "validation proof": "uv run pytest tests/test_workspace_start_preflight_cli.py tests/test_workspace_implement_cli.py -q; make test-workspace; make lint-workspace", + "proof achieved now": "yes; planning closeout recorded explicit proof input.", + "evidence for \"proof achieved\" state": "uv run pytest tests/test_workspace_start_preflight_cli.py tests/test_workspace_implement_cli.py -q; make test-workspace; make lint-workspace" + }, + "intent_satisfaction": { + "original intent": "GitHub #1215", + "was original intent fully satisfied?": "yes", + "evidence of intent satisfaction": "uv run pytest tests/test_workspace_start_preflight_cli.py tests/test_workspace_implement_cli.py -q; make test-workspace; make lint-workspace", + "unsolved intent passed to": "none" + }, + "execution_summary": { + "outcome delivered": "AW now makes epic candidate-lane promotion hard to miss before broad implementation and exposes issue-ref scope uncertainty.", + "validation confirmed": "uv run pytest tests/test_workspace_start_preflight_cli.py tests/test_workspace_implement_cli.py -q; make test-workspace; make lint-workspace", + "follow-on routed to": "none", + "post-work posterity capture": "archive closeout distillation", + "knowledge promoted (memory/docs/config)": "none", + "resume from": "archive", + "knowledge promoted (Memory/Docs/Config)": "none" + }, + "durable_residue": { + "status": "none", + "learned constraint": "No future-relevant learning was identified beyond the closeout evidence.", + "motivation worth preserving": "Closeout reviewed durable residue and found no live follow-up.", + "canonical owner now": "archive", + "promotion trigger": "none", + "retention after promotion": "retain" + }, + "task_intent_promotion": { + "decision": "do-not-promote", + "accepted values": "do-not-promote|memory|subsystem-intent|system-intent|refine-existing-intent|supersede-existing-intent", + "evidence source": "archive-plan --prepare-closeout", + "target scope": "archive", + "proposed durable intent": "Closeout reviewed durable residue and found no live follow-up.", + "confidence": "low", + "needs review": true, + "owner surface": "archive" + }, + "closure_check": { + "closeout scope": "slice", + "slice status": "completed", + "larger-intent status": "closed", + "closure decision": "archive-and-close", + "why this decision is honest": "planning closeout accepted a slice claim with intent-status satisfied.", + "evidence carried forward": "uv run pytest tests/test_workspace_start_preflight_cli.py tests/test_workspace_implement_cli.py -q; make test-workspace; make lint-workspace", + "reopen trigger": "None unless new evidence shows the closeout was incomplete." + }, + "improvement_signal_review": { + "status": "not_checked", + "accepted statuses": "not_checked|signals_routed|signals_fixed|signals_dismissed|no_signal_found", + "guidance": "At closeout, report AW smoothness/helpfulness gaps, better-way signals, unused-feature reflections, and places AW could help more. Route each concrete signal to exactly one owner class unless explicitly split, or mark no_signal_found after checking.", + "source": "operating_posture", + "owner classes": [ + "issue", + "Memory", + "Planning", + "docs/checks/contracts", + "direct fix", + "dismissed with reason" + ], + "ordinary output cap": 3, + "signals found": [], + "signals fixed": [], + "signals routed": [], + "signals dismissed": [], + "next owner": "agent closeout reflection" + }, + "closeout_distillation": { + "buckets": { + "discard": [ + { + "summary": "No Memory, docs, or config promotion was needed for local execution detail.", + "owner": "discard", + "source": "execution_summary.knowledge promoted (Memory/Docs/Config)" + } + ], + "continuation": [], + "memory": [], + "config_check": [], + "docs": [], + "issue_follow_up": [] + } + }, + "drift_log": [ + "2026-05-29: Scaffolded by agentic-planning new-plan.", + "2026-05-29: Recorded delegation decision route=keep-local." + ], + "memory_learning_capture": { + "status": "reviewed", + "memory consult recommended?": "review startup/report memory_consult", + "memory notes read": "not recorded", + "future agents should not rediscover": "no", + "decision": "none", + "target": "none", + "reason": "Derived from durable_residue during closeout preparation." + }, + "generated_closeout": { + "status": "generated", + "source": "archive-plan --prepare-closeout", + "authority": "derived adapter; intent_satisfaction, closure_check, proof_report, durable_residue, execution_run, and execution_summary remain authoritative", + "text": "Generated closeout adapter; structured execplan fields are authoritative.\nIntent: GitHub #1215\nIntent satisfied: yes\nArchive decision: archive-and-close\nProof: uv run pytest tests/test_workspace_start_preflight_cli.py tests/test_workspace_implement_cli.py -q; make test-workspace; make lint-workspace\nChanged surfaces: src/agentic_workspace/workspace_runtime_primitives.py; tests/test_workspace_start_preflight_cli.py; tests/test_workspace_implement_cli.py\nDurable residue: none (archive)\nMemory learning: none\nFollow-up: none" + } +} diff --git a/src/agentic_workspace/workspace_runtime_primitives.py b/src/agentic_workspace/workspace_runtime_primitives.py index 939f2337..0f455972 100644 --- a/src/agentic_workspace/workspace_runtime_primitives.py +++ b/src/agentic_workspace/workspace_runtime_primitives.py @@ -13434,7 +13434,13 @@ def _next_safe_action_packet( forbidden_actions.extend(["begin implementation", "create planning artifact before clarified intent is captured"]) if "closeout" in action: forbidden_actions.append("claim completion before closeout trust is reconciled") - if decision in {"active-execplan-required", "planning-escalation-required", "implementation-owner-missing"}: + if decision in { + "active-execplan-required", + "candidate-lane-promotion-required", + "parent-decomposition-decision-required", + "planning-escalation-required", + "implementation-owner-missing", + }: forbidden_actions.append("continue implementation without active planning ownership") memory_status = str((memory_consult or {}).get("status", "unknown")) command_effect = "none" @@ -14852,6 +14858,8 @@ def _selector_first_planning_safety_gate(gate: Any) -> dict[str, Any]: "changed_path_classification", "active_delegation_requirement", "active_parent_decomposition_requirement", + "candidate_pressure", + "issue_scope_evidence", "repair_route", "work_shape_guidance", ): @@ -15487,6 +15495,200 @@ def _active_decomposition_delegation_payload(*, target_root: Path) -> dict[str, } +def _planning_roadmap_candidates(target_root: Path) -> list[dict[str, Any]]: + state_path = target_root / ".agentic-workspace" / "planning" / "state.toml" + if not state_path.exists(): + return [] + try: + state = tomllib.loads(state_path.read_text(encoding="utf-8-sig")) + except (OSError, tomllib.TOMLDecodeError, UnicodeDecodeError): + return [] + roadmap = state.get("roadmap", {}) if isinstance(state, dict) else {} + raw_candidates = roadmap.get("candidates", []) if isinstance(roadmap, dict) else [] + candidates: list[dict[str, Any]] = [] + for candidate in raw_candidates if isinstance(raw_candidates, list) else []: + if not isinstance(candidate, dict): + continue + status = str(candidate.get("status", "")).strip().lower() + maturity = str(candidate.get("maturity", "")).strip().lower() + if status in {"done", "closed", "retired", "superseded", "deferred"}: + continue + candidates.append( + { + "id": str(candidate.get("id", "")).strip(), + "title": str(candidate.get("title", "")).strip(), + "refs": str(candidate.get("refs", "")).strip(), + "priority": str(candidate.get("priority", "")).strip(), + "status": status or "unknown", + "maturity": maturity or "unknown", + "outcome": str(candidate.get("outcome", "")).strip(), + "promotion_signal": str(candidate.get("promotion_signal", "")).strip(), + "suggested_first_slice": str(candidate.get("suggested_first_slice", "")).strip(), + } + ) + return [candidate for candidate in candidates if candidate.get("id") or candidate.get("title")] + + +def _candidate_promotion_command(*, candidate_id: str, config: WorkspaceConfig, planning_revision: dict[str, Any]) -> str: + return str( + _command_with_expected_planning_revision( + _command_with_cli_invoke( + command=f"agentic-workspace planning promote-to-plan --item-id {candidate_id} --target . --format json", + cli_invoke=config.cli_invoke, + ), + planning_revision=planning_revision, + ) + ) + + +def _planning_candidate_pressure_payload( + *, + target_root: Path, + config: WorkspaceConfig, + issue_refs: list[str], + work_shape: str | None, + decomposition_delegation: dict[str, Any], + planning_revision: dict[str, Any], +) -> dict[str, Any]: + roadmap_candidates = _planning_roadmap_candidates(target_root) + decomposition_candidates = ( + [candidate for candidate in decomposition_delegation.get("candidates", []) if isinstance(candidate, dict)] + if isinstance(decomposition_delegation, dict) + else [] + ) + issue_ref_set = set(issue_refs) + matched_roadmap = [ + candidate for candidate in roadmap_candidates if issue_ref_set and issue_ref_set.intersection(_candidate_refs(candidate)) + ] + broad_shape = work_shape in {"lane", "epic"} + promotion_required = False + reasons: list[str] = [] + if broad_shape and decomposition_candidates: + promotion_required = True + reasons.append("open decomposition lane candidates exist for broad or lane-shaped work") + if broad_shape and len(roadmap_candidates) >= 2: + promotion_required = True + reasons.append("multiple roadmap candidates exist while the requested work is broad or lane-shaped") + if len(matched_roadmap) >= 2: + promotion_required = True + reasons.append("multiple roadmap candidates match the requested external issue refs") + + include_candidate_detail = promotion_required or bool(matched_roadmap) + top_roadmap = (matched_roadmap or roadmap_candidates) if include_candidate_detail else [] + route_options: list[dict[str, Any]] = [] + for candidate in top_roadmap[:3]: + candidate_id = str(candidate.get("id", "")).strip() + if not candidate_id: + continue + route_options.append( + { + "kind": "roadmap-candidate", + "id": candidate_id, + "title": candidate.get("title", ""), + "refs": candidate.get("refs", ""), + "command": _candidate_promotion_command( + candidate_id=candidate_id, + config=config, + planning_revision=planning_revision, + ), + } + ) + for candidate in decomposition_candidates[:3]: + lane_id = str(candidate.get("lane_id", "")).strip() + if not lane_id: + continue + route_options.append( + { + "kind": "decomposition-lane", + "id": lane_id, + "title": candidate.get("title", ""), + "decomposition": candidate.get("decomposition", ""), + "command": _candidate_promotion_command(candidate_id=lane_id, config=config, planning_revision=planning_revision), + } + ) + + status = "promotion-required" if promotion_required else "observed" if roadmap_candidates or decomposition_candidates else "none" + return { + "kind": "agentic-workspace/planning-candidate-pressure/v1", + "status": status, + "work_shape": work_shape or "unknown", + "roadmap_candidate_count": len(roadmap_candidates), + "matched_roadmap_candidate_count": len(matched_roadmap), + "decomposition_candidate_count": len(decomposition_candidates), + "candidate_count": len(roadmap_candidates) + len(decomposition_candidates), + "candidate_ids": [ + *[str(candidate.get("id", "")) for candidate in (matched_roadmap or roadmap_candidates)[:5] if candidate.get("id")], + *[str(candidate.get("lane_id", "")) for candidate in decomposition_candidates[:5] if str(candidate.get("lane_id", "")).strip()], + ] + if include_candidate_detail + else [], + "reasons": reasons, + "route_options": route_options, + "required_before_implementation": [ + "promote a roadmap candidate or decomposition lane to an active execplan", + "create a parent decomposition and bounded lane execplans", + "or record an explicit bounded-slice exception that does not claim parent epic closure", + ] + if promotion_required + else [], + "rule": "Checked-in Planning candidate evidence can require promotion before broad implementation; prompt text alone must not authorize closing an epic.", + } + + +def _issue_scope_evidence_payload(*, target_root: Path, config: WorkspaceConfig, issue_refs: list[str]) -> dict[str, Any]: + if not issue_refs: + return {"kind": "agentic-workspace/issue-scope-evidence/v1", "status": "not-applicable", "issue_refs": []} + path, relative_path, storage = _external_intent_evidence_read_location(target_root) + payload: dict[str, Any] = {} + if path.exists(): + try: + payload = json.loads(path.read_text(encoding="utf-8-sig")) + except (OSError, json.JSONDecodeError, UnicodeDecodeError): + payload = {} + items = [item for item in _list_payload(payload.get("items")) if isinstance(item, dict)] + by_id = {str(item.get("id", "")).strip(): item for item in items if str(item.get("id", "")).strip()} + evidence: list[dict[str, Any]] = [] + missing: list[str] = [] + for issue_ref in issue_refs: + item = by_id.get(issue_ref) + if not item: + missing.append(issue_ref) + continue + evidence.append( + { + "id": issue_ref, + "system": str(item.get("system", "")).strip(), + "status": str(item.get("status", "")).strip(), + "kind": str(item.get("kind", "")).strip(), + "parent_id": str(item.get("parent_id", "")).strip(), + "planning_residue_expected": str(item.get("planning_residue_expected", "")).strip(), + "negative_invariant_count": len(_list_payload(item.get("negative_invariants"))), + "title": str(item.get("title", "")).strip(), + } + ) + if missing and evidence: + status = "partial" + elif missing: + status = "unknown" + else: + status = "available" + return { + "kind": "agentic-workspace/issue-scope-evidence/v1", + "status": status, + "issue_refs": issue_refs, + "evidence": evidence, + "missing_issue_refs": missing, + "source_path": relative_path if path.exists() else "", + "storage": storage if path.exists() else "none", + "risk": "high" if missing else "evidence-backed", + "refresh_command": _command_with_cli_invoke( + command="agentic-workspace external-intent refresh-github --target . --state all --format json", + cli_invoke=config.cli_invoke, + ), + "rule": "Issue refs are external-intent handles until cached provider-agnostic evidence or active Planning owns the scope.", + } + + def _planning_state_has_active_items(*, target_root: Path) -> bool: active_summary = _fast_planning_active_summary(target_root=target_root) return bool( @@ -17635,6 +17837,9 @@ def _tiny_implement_payload(payload: dict[str, Any]) -> dict[str, Any]: compact_gate = _selector_first_planning_safety_gate(planning_safety_gate) compact_gate.pop("planning_revision", None) compact_gate.pop("work_shape_guidance", None) + if compact_gate.get("status") in {"clear", "satisfied"}: + compact_gate.pop("candidate_pressure", None) + compact_gate.pop("issue_scope_evidence", None) projected["context"]["planning_safety_gate"] = compact_gate if isinstance(intent_acknowledgement, dict) and intent_acknowledgement.get("decision") == "proceed-with-stated-assumption": projected["context"]["intent_acknowledgement"] = { @@ -18425,6 +18630,15 @@ def _planning_safety_gate_payload( path_classification = _allow_ancillary_memory_feedback_path(path_classification) path_classification = _allow_issue_scoped_planning_state_reconciliation(path_classification, issue_refs=issue_refs) planning_revision = _planning_revision_payload(target_root=target_root) + issue_scope_evidence = _issue_scope_evidence_payload(target_root=target_root, config=config, issue_refs=issue_refs) + candidate_pressure = _planning_candidate_pressure_payload( + target_root=target_root, + config=config, + issue_refs=issue_refs, + work_shape=work_shape, + decomposition_delegation=decomposition_delegation if isinstance(decomposition_delegation, dict) else {}, + planning_revision=planning_revision, + ) promotion_command = _planning_safety_promotion_command( config=config, decomposition_delegation=decomposition_delegation if isinstance(decomposition_delegation, dict) else {}, @@ -18482,6 +18696,23 @@ def _planning_safety_gate_payload( reason = "Implementation paths are mixed with planning recovery paths without active planning ownership." required_next_action = "checkpoint-planning-before-implementation" workflow_sufficient = False + elif (not active_planning_present) and candidate_pressure.get("status") == "promotion-required": + status = "blocked" + decision = "candidate-lane-promotion-required" + reason = "Checked-in Planning candidates indicate broad or lane-shaped work; promote or decompose a bounded lane first." + required_next_action = "select-or-promote-candidate-lane" + workflow_sufficient = False + elif ( + (not active_planning_present) + and issue_refs + and (not changed_paths) + and issue_scope_evidence.get("status") in {"unknown", "partial"} + ): + status = "attention" + decision = "external-issue-scope-unknown" + reason = "The task references external issue id(s), but AW has no complete cached intent evidence for their scope." + required_next_action = "refresh-external-intent-or-state-bounded-slice" + workflow_sufficient = True elif path_classification["implementation_paths"] and path_classification["scope_growth_detected"]: status = "attention" decision = "agent-work-shape-decision-required" @@ -18514,6 +18745,8 @@ def _planning_safety_gate_payload( "active_plan_reliance": active_plan_reliance, "active_state_summary": active_summary, "issue_refs": issue_refs, + "issue_scope_evidence": issue_scope_evidence, + "candidate_pressure": candidate_pressure, "repair_route": { "status": "retired", "fit_criteria": [ diff --git a/tests/test_workspace_implement_cli.py b/tests/test_workspace_implement_cli.py index c5af14d3..4ab55ec2 100644 --- a/tests/test_workspace_implement_cli.py +++ b/tests/test_workspace_implement_cli.py @@ -1024,12 +1024,65 @@ def test_implement_task_allows_narrow_single_issue_context(tmp_path: Path, capsy payload = json.loads(capsys.readouterr().out) assert "task_routing" not in payload - assert payload["planning_safety_gate"]["status"] == "clear" + assert payload["planning_safety_gate"]["status"] == "attention" + assert payload["planning_safety_gate"]["decision"] == "external-issue-scope-unknown" + assert payload["planning_safety_gate"]["implementation_allowed"] is True + assert payload["planning_safety_gate"]["issue_scope_evidence"]["missing_issue_refs"] == ["#424"] assert payload["planning_safety_gate"]["work_shape_guidance"]["scope_factors"]["issue_refs"] == ["#424"] assert payload["planning_safety_gate"]["work_shape_guidance"]["agent_decision_required"] is True assert payload["next_allowed_action"] == "Provide --changed paths or use start/preflight before broad implementation." +def test_implement_blocks_epic_work_with_multiple_roadmap_candidates(tmp_path: Path, capsys) -> None: + _init_git_repo(tmp_path) + _write( + tmp_path / ".agentic-workspace" / "planning" / "state.toml", + """ +kind = "agentic-planning-state" +schema_version = "planning-state/v1" + +[todo] +active_items = [] +queued_items = [] + +[roadmap] +lanes = [] +candidates = [ + { id = "github-1201-command-package", maturity = "candidate", status = "next", priority = "P1", refs = "GitHub #1201", title = "Command package extraction", outcome = "Extract the command package.", reason = "Open issue.", promotion_signal = "Promote before implementation.", suggested_first_slice = "Shape a bounded lane." }, + { id = "github-1202-runtime-parity", maturity = "candidate", status = "next", priority = "P1", refs = "GitHub #1202", title = "Runtime parity", outcome = "Prove generated runtime parity.", reason = "Open issue.", promotion_signal = "Promote before implementation.", suggested_first_slice = "Shape a bounded lane." }, +] +""", + ) + + assert ( + cli.main( + [ + "implement", + "--target", + str(tmp_path), + "--changed", + "src/agentic_workspace/workspace_runtime_primitives.py", + "--task", + "Implement the command generation extraction epic", + "--format", + "json", + ] + ) + == 0 + ) + + payload = json.loads(capsys.readouterr().out) + gate = payload["context"]["planning_safety_gate"] + assert gate["status"] == "blocked" + assert gate["decision"] == "candidate-lane-promotion-required" + assert gate["implementation_allowed"] is False + assert gate["candidate_pressure"]["candidate_ids"] == [ + "github-1201-command-package", + "github-1202-runtime-parity", + ] + assert payload["context"]["workflow_sufficiency"]["decision"] == "candidate-lane-promotion-required" + + def test_implement_with_explicit_target_ignores_checkout_active_plan(tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys) -> None: active_checkout = tmp_path / "active-checkout" isolated_target = tmp_path / "isolated-target" @@ -1079,7 +1132,9 @@ def test_implement_with_explicit_target_ignores_checkout_active_plan(tmp_path: P payload = json.loads(capsys.readouterr().out) assert "task_routing" not in payload - assert payload["planning_safety_gate"]["status"] == "clear" + assert payload["planning_safety_gate"]["status"] == "attention" + assert payload["planning_safety_gate"]["decision"] == "external-issue-scope-unknown" + assert payload["planning_safety_gate"]["implementation_allowed"] is True assert payload["next_allowed_action"] == "Provide --changed paths or use start/preflight before broad implementation." diff --git a/tests/test_workspace_start_preflight_cli.py b/tests/test_workspace_start_preflight_cli.py index d4dd0693..7a3d6da8 100644 --- a/tests/test_workspace_start_preflight_cli.py +++ b/tests/test_workspace_start_preflight_cli.py @@ -1570,7 +1570,7 @@ def test_start_decomposition_only_delegation_requires_lane_promotion_before_hand assert "Select or promote" in decision["delegation_next_step"]["precondition"] -def test_start_exposes_decomposition_without_prompt_text_blocking(tmp_path: Path, capsys) -> None: +def test_start_blocks_broad_work_when_decomposition_lane_needs_promotion(tmp_path: Path, capsys) -> None: _init_git_repo(tmp_path) _write( tmp_path / ".agentic-workspace" / "planning" / "decompositions" / "dogfood.decomposition.json", @@ -1620,13 +1620,100 @@ def test_start_exposes_decomposition_without_prompt_text_blocking(tmp_path: Path ) payload = json.loads(capsys.readouterr().out) - assert "planning_safety_gate" not in _start_context(payload).get("planning", {}) - assert _start_workflow_sufficiency(payload)["decision"] == "enough-for-first-contact-routing" - assert _start_primary_action(payload)["action"] == "choose-smallest-workflow-shape" + gate = _start_planning_safety_gate(payload) + assert gate["status"] == "blocked" + assert gate["decision"] == "candidate-lane-promotion-required" + assert gate["implementation_allowed"] is False + assert gate["candidate_pressure"]["status"] == "promotion-required" + assert gate["candidate_pressure"]["candidate_ids"] == ["safety-slice"] + assert "promote-to-plan --item-id safety-slice" in gate["promotion_command"] + assert _start_workflow_sufficiency(payload)["decision"] == "candidate-lane-promotion-required" + assert _start_primary_action(payload)["action"] == "select-or-promote-candidate-lane" + assert payload["next_safe_action"]["implementation_allowed"] is False decision = _start_context_value(payload, "delegation_decision") assert decision["decomposition_delegation"]["status"] == "available-without-active-planning" +def test_start_blocks_epic_work_with_multiple_roadmap_candidates(tmp_path: Path, capsys) -> None: + _init_git_repo(tmp_path) + _write( + tmp_path / ".agentic-workspace" / "planning" / "state.toml", + """ +kind = "agentic-planning-state" +schema_version = "planning-state/v1" + +[todo] +active_items = [] +queued_items = [] + +[roadmap] +lanes = [] +candidates = [ + { id = "github-1201-command-package", maturity = "candidate", status = "next", priority = "P1", refs = "GitHub #1201", title = "Command package extraction", outcome = "Extract the command package.", reason = "Open issue.", promotion_signal = "Promote before implementation.", suggested_first_slice = "Shape a bounded lane." }, + { id = "github-1202-runtime-parity", maturity = "candidate", status = "next", priority = "P1", refs = "GitHub #1202", title = "Runtime parity", outcome = "Prove generated runtime parity.", reason = "Open issue.", promotion_signal = "Promote before implementation.", suggested_first_slice = "Shape a bounded lane." }, +] +""", + ) + + assert ( + cli.main( + [ + "start", + "--target", + str(tmp_path), + "--task", + "Implement the command generation extraction epic", + "--format", + "json", + ] + ) + == 0 + ) + + payload = json.loads(capsys.readouterr().out) + gate = _start_planning_safety_gate(payload) + assert gate["decision"] == "candidate-lane-promotion-required" + assert gate["implementation_allowed"] is False + assert gate["candidate_pressure"]["roadmap_candidate_count"] == 2 + assert gate["candidate_pressure"]["candidate_ids"] == [ + "github-1201-command-package", + "github-1202-runtime-parity", + ] + assert "planning promote-to-plan --item-id github-1201-command-package" in gate["candidate_pressure"]["route_options"][0]["command"] + assert payload["next_safe_action"]["implementation_allowed"] is False + + +def test_start_marks_bare_issue_ref_scope_unknown_without_external_evidence(tmp_path: Path, capsys) -> None: + _init_git_repo(tmp_path) + _write( + tmp_path / ".agentic-workspace" / "planning" / "state.toml", + """ +kind = "agentic-planning-state" +schema_version = "planning-state/v1" + +[todo] +active_items = [] +queued_items = [] + +[roadmap] +lanes = [] +candidates = [] +""", + ) + + assert cli.main(["start", "--target", str(tmp_path), "--task", "Implement #1189", "--format", "json"]) == 0 + + payload = json.loads(capsys.readouterr().out) + gate = _start_planning_safety_gate(payload) + assert gate["status"] == "attention" + assert gate["decision"] == "external-issue-scope-unknown" + assert gate["implementation_allowed"] is True + assert gate["issue_scope_evidence"]["status"] == "unknown" + assert gate["issue_scope_evidence"]["risk"] == "high" + assert gate["issue_scope_evidence"]["missing_issue_refs"] == ["#1189"] + assert "external-intent refresh-github" in gate["issue_scope_evidence"]["refresh_command"] + + def test_implement_flags_scope_growth_without_active_execplan(tmp_path: Path, capsys) -> None: _init_git_repo(tmp_path)