skills/codestable__codestable
1
0
Fork 0
mirror of https://github.com/codestable/CodeStable.git synced 2026-09-19 09:03:09 +08:00
Code Issues Packages Projects Releases Wiki Activity
Files
feat/lean-workflow-artifacts
codestable__codestable/tests/test_codestable_workflow_next.py
T

2752 lines
102 KiB
Python
Raw Permalink Blame History

from __future__ import annotations
import importlib.util
import json
import subprocess
import sys
from pathlib import Path
import pytest
TOOLS_DIR = Path(__file__).resolve().parents[1] / "plugins/codestable/skills/cs-onboard/tools"
sys.path.insert(0, str(TOOLS_DIR))
def load_tool():
spec = importlib.util.spec_from_file_location("codestable_workflow_next", TOOLS_DIR / "codestable-workflow-next.py")
assert spec and spec.loader
module = importlib.util.module_from_spec(spec)
sys.modules[spec.name] = module
spec.loader.exec_module(module)
return module
workflow_next = load_tool()
def write(path: Path, text: str) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(text, encoding="utf-8")
def init_repo(tmp_path: Path) -> Path:
repo = tmp_path / "repo"
repo.mkdir()
(repo / ".codestable").mkdir()
return repo
def write_roadmap(repo: Path, status: str = "active") -> Path:
roadmap = repo / ".codestable/roadmap/billing-system"
write(
roadmap / "billing-system-roadmap.md",
f"---\ndoc_type: roadmap\nslug: billing-system\nstatus: {status}\n---\n# Roadmap\n",
)
write(roadmap / "billing-system-roadmap-review.md", "---\ndoc_type: roadmap-review\nstatus: passed\n---\n# Review\n")
write(
roadmap / "billing-system-items.yaml",
"roadmap: billing-system\n"
"items:\n"
" - slug: api-seed\n"
" status: planned\n"
" feature: null\n"
" - slug: ui-seed\n"
" status: planned\n"
" feature: null\n",
)
return roadmap
def write_feature(
repo: Path,
slug: str,
*,
design_status: str = "draft",
review_status: str = "passed",
review_state: str | None = None,
review_reason: str = "",
reviewer_id: str = "",
execution_lane: str | None = None,
execution_lane_reason: str | None = None,
include_roadmap: bool = False,
) -> Path:
feature = repo / ".codestable/features" / f"2026-07-02-{slug}"
roadmap_fields = f"roadmap: billing-system\nroadmap_item: {slug}\n" if include_roadmap else ""
lane_field = f"execution_lane: {execution_lane}\n" if execution_lane else ""
lane_reason_field = f"execution_lane_reason: {execution_lane_reason}\n" if execution_lane_reason else ""
review_state_fields = ""
if review_state is not None:
review_state_fields = (
f"review_state: {review_state}\n"
f'review_reason: "{review_reason}"\n'
f'reviewer_id: "{reviewer_id}"\n'
)
write(
feature / f"{slug}-design.md",
f"---\ndoc_type: feature-design\nfeature: 2026-07-02-{slug}\n"
f"{roadmap_fields}{lane_field}{lane_reason_field}"
f"status: {design_status}\n---\n# Design\n",
)
write(feature / f"{slug}-checklist.yaml", "steps:\n - id: step-1\n status: pending\n")
write(
feature / f"{slug}-design-review.md",
f"---\ndoc_type: feature-design-review\nstatus: {review_status}\n"
f"{review_state_fields}---\n# Review\n",
)
return feature
def write_goal_state(
directory: Path,
*,
status: str,
stage: str | None = None,
driver_kind: str = "none",
driver_id: str = "",
handoff_reason: str = "",
handoff_next: str = "",
acceptance_authorization: str | None = "approved",
acceptance_authorization_ref: str = "approval-report.md#goal-acceptance",
approval_decision_status: str | None = "approved",
commit_authorization: str | None = "approved",
commit_authorization_ref: str = "approval-report.md#goal-commits",
commit_decision_status: str | None = "approved",
) -> None:
lines = [f"status: {status}"]
if stage is not None:
lines.append(f"stage: {stage}")
lines.extend(
[
f"driver_kind: {driver_kind}",
f'driver_id: "{driver_id}"',
f'handoff_reason: "{handoff_reason}"',
f'handoff_next: "{handoff_next}"',
]
)
if acceptance_authorization is not None:
lines.extend(
[
f"acceptance_authorization: {acceptance_authorization}",
f'acceptance_authorization_ref: "{acceptance_authorization_ref}"',
]
)
if commit_authorization is not None:
lines.extend(
[
f"commit_authorization: {commit_authorization}",
f'commit_authorization_ref: "{commit_authorization_ref}"',
]
)
write(directory / "goal-state.yaml", "\n".join(lines) + "\n")
approval_rows: list[str] = []
if acceptance_authorization == "approved" and approval_decision_status is not None:
approval_rows.append(f" goal-acceptance: {approval_decision_status}")
if commit_authorization == "approved" and commit_decision_status is not None:
approval_rows.append(f" goal-commits: {commit_decision_status}")
if approval_rows:
write(
directory / "approval-report.md",
"---\ndoc_type: approval-report\nstatus: approved\napprovals:\n"
+ "\n".join(approval_rows)
+ "\n---\n# Approval\n",
)
def write_code_review(
feature: Path,
slug: str,
*,
status: str = "passed",
reviewer: str | None = "subagent",
doc_type: str = "feature-review",
) -> None:
reviewer_field = f"reviewer: {reviewer}\n" if reviewer is not None else ""
write(
feature / f"{slug}-review.md",
f"---\ndoc_type: {doc_type}\nstatus: {status}\n{reviewer_field}---\n# Review\n",
)
def write_ff_note(feature: Path, slug: str) -> None:
write(
feature / f"{slug}-ff-note.md",
f"---\ndoc_type: feature-ff-note\nfeature: {slug}\ndate: 2026-07-13\n---\n# Fastforward Note\n",
)
def write_roadmap_goal_state(
roadmap: Path,
*,
feature_slug: str = "api-seed",
status: str = "ready-to-dispatch",
driver_kind: str = "host-agent",
driver_id: str = "epic-run-123",
acceptance_authorization: str | None = "approved",
approval_decision_status: str | None = "approved",
commit_authorization: str | None = "approved",
commit_decision_status: str | None = "approved",
execution_confirmation_id: str | None = None,
approval_group_status: str | None = None,
approval_group_confirmation_id: str = "",
) -> None:
lines = [
"roadmap: billing-system",
f"status: {status}",
f"driver_kind: {driver_kind}",
f'driver_id: "{driver_id}"',
]
if execution_confirmation_id is not None:
lines.append(f'execution_confirmation_id: "{execution_confirmation_id}"')
if acceptance_authorization is not None:
lines.extend(
[
f"acceptance_authorization: {acceptance_authorization}",
'acceptance_authorization_ref: "approval-report.md#goal-acceptance"',
]
)
if commit_authorization is not None:
lines.extend(
[
f"commit_authorization: {commit_authorization}",
'commit_authorization_ref: "approval-report.md#goal-commits"',
]
)
lines.extend(
[
"current_feature_index: 0",
"features:",
f" - slug: {feature_slug}",
f" roadmap_item: {feature_slug}",
f" feature_dir: .codestable/features/2026-07-02-{feature_slug}",
" status: implementing",
]
)
write(roadmap / "goal-state.yaml", "\n".join(lines) + "\n")
approval_rows = []
if approval_decision_status is not None:
approval_rows.append(f" goal-acceptance: {approval_decision_status}")
if commit_decision_status is not None:
approval_rows.append(f" goal-commits: {commit_decision_status}")
if approval_rows or approval_group_status is not None:
approvals = (
"approvals: {}\n"
if not approval_rows
else "approvals:\n" + "\n".join(approval_rows) + "\n"
)
approval_group = ""
if approval_group_status is not None:
approval_group = (
"approval_groups:\n"
" goal-execution:\n"
f" status: {approval_group_status}\n"
f' confirmation_id: "{approval_group_confirmation_id}"\n'
" decisions:\n"
" - goal-acceptance\n"
" - goal-commits\n"
)
write(
roadmap / "approval-report.md",
"---\ndoc_type: approval-report\nstatus: approved\n"
+ approvals
+ approval_group
+ "---\n# Approval\n",
)
def write_reverse_owner_state(
repo: Path,
roadmap_slug: str,
*,
state_roadmap: str | None = None,
rows: list[tuple[str | None, str]],
) -> Path:
lines = [
f"roadmap: {state_roadmap or roadmap_slug}",
"status: ready-to-dispatch",
"features:",
]
for item, feature_dir in rows:
lines.append(f" - slug: {item}" if item else " - status: implementing")
if item:
lines.append(f" roadmap_item: {item}")
lines.append(f" feature_dir: {feature_dir}")
if item:
lines.append(" status: implementing")
goal_state = repo / ".codestable/roadmap" / roadmap_slug / "goal-state.yaml"
write(goal_state, "\n".join(lines) + "\n")
return goal_state
def run_cli_json(repo: Path, workflow: str, path: Path) -> tuple[subprocess.CompletedProcess[str], dict[str, object]]:
completed = subprocess.run(
[
sys.executable,
(TOOLS_DIR / "codestable-workflow-next.py").as_posix(),
workflow,
f"--{workflow if workflow == 'feature' else 'roadmap'}",
path.as_posix(),
"--json",
],
cwd=repo,
check=False,
text=True,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
env={"PYTHONDONTWRITEBYTECODE": "1"},
timeout=10,
)
return completed, json.loads(completed.stdout)
def test_epic_continues_when_only_first_child_design_review_passed(tmp_path: Path) -> None:
repo = init_repo(tmp_path)
roadmap = write_roadmap(repo)
write_feature(repo, "api-seed", include_roadmap=True)
result = workflow_next.epic_next(roadmap)
assert result["ok"] is True
assert result["status"] == "continue"
assert result["next_action"] == "cs-feat design/design-review"
assert result["must_continue"] is True
assert result["final_answer_allowed"] is False
assert result["evidence"]["next_item"]["item"] == "ui-seed"
def test_epic_child_batch_design_admits_dependent_after_predecessor_review(tmp_path: Path) -> None:
repo = init_repo(tmp_path)
roadmap = write_roadmap(repo)
write(
roadmap / "billing-system-items.yaml",
"roadmap: billing-system\n"
"items:\n"
" - slug: api-seed\n"
" status: in-progress\n"
" feature: null\n"
" - slug: ui-seed\n"
" status: planned\n"
" depends_on: [api-seed]\n"
" feature: null\n",
)
write_feature(repo, "api-seed", include_roadmap=True)
result = workflow_next.epic_next(roadmap)
assert result["status"] == "continue"
assert result["next_action"] == "cs-feat design/design-review"
assert result["evidence"]["next_item"]["item"] == "ui-seed"
assert result["evidence"]["design_admission"] == {
"dependencies": [
{
"item": "api-seed",
"item_status": "in-progress",
"design_review_status": "passed",
"design_ready": True,
"ready_by": "design-review",
"reason": None,
}
],
"ready": True,
"review_state": "missing",
"review_reason": None,
}
def test_epic_child_batch_selects_unblocked_predecessor_from_unsorted_dag(tmp_path: Path) -> None:
repo = init_repo(tmp_path)
roadmap = write_roadmap(repo)
write(
roadmap / "billing-system-items.yaml",
"roadmap: billing-system\n"
"items:\n"
" - slug: ui-seed\n"
" status: planned\n"
" depends_on: [api-seed]\n"
" feature: null\n"
" - slug: api-seed\n"
" status: planned\n"
" feature: null\n",
)
result = workflow_next.epic_next(roadmap)
assert result["status"] == "continue"
assert result["evidence"]["next_item"]["item"] == "api-seed"
def test_epic_child_batch_blocks_cyclic_design_dependencies(tmp_path: Path) -> None:
repo = init_repo(tmp_path)
roadmap = write_roadmap(repo)
write(
roadmap / "billing-system-items.yaml",
"roadmap: billing-system\n"
"items:\n"
" - slug: api-seed\n"
" status: planned\n"
" depends_on: [ui-seed]\n"
" feature: null\n"
" - slug: ui-seed\n"
" status: planned\n"
" depends_on: [api-seed]\n"
" feature: null\n",
)
result = workflow_next.epic_next(roadmap)
assert result["status"] == "blocked"
assert result["next_action"] == "fix-roadmap-items"
assert result["evidence"]["cycle_items"] == ["api-seed", "ui-seed"]
def test_epic_blocks_cycle_even_when_all_design_reviews_passed(tmp_path: Path) -> None:
repo = init_repo(tmp_path)
roadmap = write_roadmap(repo)
write(
roadmap / "billing-system-items.yaml",
"roadmap: billing-system\nitems:\n"
" - slug: api-seed\n status: in-progress\n depends_on: [ui-seed]\n"
" - slug: ui-seed\n status: in-progress\n depends_on: [api-seed]\n",
)
write_feature(repo, "api-seed", include_roadmap=True)
write_feature(repo, "ui-seed", include_roadmap=True)
result = workflow_next.epic_next(roadmap)
assert result["status"] == "blocked"
assert result["next_action"] == "fix-roadmap-items"
assert "cycle" in result["reason"]
@pytest.mark.parametrize(
("depends_on", "reason"),
[
("[missing-seed]", "missing item missing-seed"),
('"api-seed"', "must be a list"),
("[api-seed, api-seed]", "must not contain duplicate"),
],
)
def test_epic_rejects_invalid_dependency_graph(
tmp_path: Path,
depends_on: str,
reason: str,
) -> None:
repo = init_repo(tmp_path)
roadmap = write_roadmap(repo)
write(
roadmap / "billing-system-items.yaml",
"roadmap: billing-system\nitems:\n"
" - slug: api-seed\n status: planned\n"
f" - slug: ui-seed\n status: planned\n depends_on: {depends_on}\n",
)
result = workflow_next.epic_next(roadmap)
assert result["status"] == "blocked"
assert result["next_action"] == "fix-roadmap-items"
assert reason in result["reason"]
def test_epic_rejects_duplicate_item_slug(tmp_path: Path) -> None:
repo = init_repo(tmp_path)
roadmap = write_roadmap(repo)
write(
roadmap / "billing-system-items.yaml",
"roadmap: billing-system\nitems:\n"
" - slug: api-seed\n status: planned\n"
" - slug: api-seed\n status: planned\n",
)
result = workflow_next.epic_next(roadmap)
assert result["status"] == "blocked"
assert result["next_action"] == "fix-roadmap-items"
assert "duplicate roadmap item slug" in result["reason"]
def test_epic_rejects_whitespace_padded_item_slug_without_traceback(tmp_path: Path) -> None:
repo = init_repo(tmp_path)
roadmap = write_roadmap(repo)
write(
roadmap / "billing-system-items.yaml",
'roadmap: billing-system\nitems:\n - slug: " api-seed "\n status: planned\n',
)
result = workflow_next.epic_next(roadmap)
assert result["status"] == "continue"
assert result["evidence"]["next_item"]["item"] == " api-seed "
def test_epic_dependency_ambiguous_feature_lookup_fails_closed(tmp_path: Path) -> None:
repo = init_repo(tmp_path)
roadmap = write_roadmap(repo)
write(
roadmap / "billing-system-items.yaml",
"roadmap: billing-system\nitems:\n"
" - slug: ui-seed\n status: planned\n depends_on: [api-seed]\n"
" - slug: api-seed\n status: in-progress\n",
)
(repo / ".codestable/features/2026-07-01-api-seed").mkdir(parents=True)
(repo / ".codestable/features/2026-07-02-api-seed").mkdir(parents=True)
result = workflow_next.epic_next(roadmap)
assert result["status"] == "blocked"
assert result["next_action"] == "fix-roadmap-items"
assert "multiple feature directories" in result["reason"]
@pytest.mark.parametrize("dependency_status", ["done", "dropped"])
def test_epic_design_admits_terminal_dependency_state(tmp_path: Path, dependency_status: str) -> None:
repo = init_repo(tmp_path)
roadmap = write_roadmap(repo)
write(
roadmap / "billing-system-items.yaml",
"roadmap: billing-system\nitems:\n"
f" - slug: api-seed\n status: {dependency_status}\n"
" - slug: ui-seed\n status: planned\n depends_on: [api-seed]\n",
)
result = workflow_next.epic_next(roadmap)
assert result["status"] == "continue"
assert result["evidence"]["next_item"]["item"] == "ui-seed"
assert result["evidence"]["design_admission"]["dependencies"][0]["ready_by"] == dependency_status
def test_epic_blocks_dropped_dependency_before_batch_confirmation(tmp_path: Path) -> None:
repo = init_repo(tmp_path)
roadmap = write_roadmap(repo)
write(
roadmap / "billing-system-items.yaml",
"roadmap: billing-system\nitems:\n"
" - slug: api-seed\n status: dropped\n"
" - slug: ui-seed\n status: in-progress\n depends_on: [api-seed]\n",
)
write_feature(repo, "ui-seed", include_roadmap=True)
result = workflow_next.epic_next(roadmap)
assert result["status"] == "blocked"
assert result["next_action"] == "resolve-dropped-implementation-dependencies"
assert result["evidence"]["dependency_blockers"] == [
{"item": "ui-seed", "dropped_dependencies": ["api-seed"]}
]
def test_feature_implementation_gate_requires_all_dependencies_done(tmp_path: Path) -> None:
repo = init_repo(tmp_path)
roadmap = write_roadmap(repo)
write(
roadmap / "billing-system-items.yaml",
"roadmap: billing-system\nitems:\n"
" - slug: api-seed\n status: in-progress\n"
" - slug: ui-seed\n status: in-progress\n depends_on: [api-seed]\n"
" feature: .codestable/features/2026-07-02-ui-seed\n",
)
feature = write_feature(repo, "ui-seed", design_status="approved", include_roadmap=True)
blocked = workflow_next.feature_next(feature, False, require_implementation_ready=True)
assert blocked["status"] == "blocked"
assert blocked["next_action"] == "complete-roadmap-dependencies-before-implementation"
assert blocked["evidence"]["implementation_dependencies"][0]["status"] == "in-progress"
items = roadmap / "billing-system-items.yaml"
write(items, items.read_text(encoding="utf-8").replace("status: in-progress", "status: done", 1))
ready = workflow_next.feature_next(feature, False, require_implementation_ready=True)
assert ready["ok"] is True
assert ready["evidence"]["implementation_ready"] is True
completed = subprocess.run(
[
sys.executable,
(TOOLS_DIR / "codestable-workflow-next.py").as_posix(),
"feature",
"--feature",
feature.as_posix(),
"--require-implementation-ready",
"--json",
],
cwd=repo,
check=False,
text=True,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
env={"PYTHONDONTWRITEBYTECODE": "1"},
timeout=10,
)
assert completed.returncode == 0
assert json.loads(completed.stdout)["evidence"]["implementation_ready"] is True
def test_epic_user_gate_only_after_all_child_design_reviews_passed(tmp_path: Path) -> None:
repo = init_repo(tmp_path)
roadmap = write_roadmap(repo)
write_feature(repo, "api-seed", include_roadmap=True)
write_feature(repo, "ui-seed", include_roadmap=True)
result = workflow_next.epic_next(roadmap)
assert result["status"] == "user_gate"
assert result["next_action"] == "all-feature-designs-confirmation"
assert result["must_continue"] is False
assert result["final_answer_allowed"] is True
assert {item["item"] for item in result["evidence"]["unapproved_items"]} == {"api-seed", "ui-seed"}
def test_epic_goal_package_after_batch_approval(tmp_path: Path) -> None:
repo = init_repo(tmp_path)
roadmap = write_roadmap(repo)
write_feature(repo, "api-seed", design_status="approved", include_roadmap=True)
write_feature(repo, "ui-seed", design_status="approved", include_roadmap=True)
result = workflow_next.epic_next(roadmap)
assert result["status"] == "goal_package"
assert result["next_action"] == "cs-epic goal-package"
assert result["must_continue"] is True
assert result["final_answer_allowed"] is False
def test_feature_epic_child_batch_returns_to_epic_loop(tmp_path: Path) -> None:
repo = init_repo(tmp_path)
write_roadmap(repo)
feature = write_feature(repo, "api-seed", include_roadmap=True)
result = workflow_next.feature_next(feature, epic_child_batch=True)
assert result["status"] == "continue"
assert result["next_action"] == "return-to-cs-epic-batch-loop"
assert result["must_continue"] is True
assert result["final_answer_allowed"] is False
assert result["evidence"]["roadmap_item"] == "api-seed"
assert result["evidence"]["epic_command"] == (
f"python3 {(TOOLS_DIR / 'codestable-workflow-next.py').as_posix()} "
"epic --roadmap .codestable/roadmap/billing-system --json"
)
def test_feature_single_mode_stops_at_design_confirmation(tmp_path: Path) -> None:
repo = init_repo(tmp_path)
feature = write_feature(repo, "api-seed")
result = workflow_next.feature_next(feature, epic_child_batch=False)
assert result["status"] == "user_gate"
assert result["next_action"] == "feature-design-confirmation"
assert result["must_continue"] is False
assert result["final_answer_allowed"] is True
def test_feature_legacy_epic_child_with_roadmap_goal_state_returns_to_epic(tmp_path: Path) -> None:
repo = init_repo(tmp_path)
roadmap = write_roadmap(repo)
write_roadmap_goal_state(roadmap)
feature = write_feature(repo, "api-seed", design_status="approved", include_roadmap=True)
write(feature / "api-seed-checklist.yaml", "steps:\n - id: step-1\n status: done\n")
write_code_review(feature, "api-seed")
write(
feature / "api-seed-acceptance.md",
"---\ndoc_type: feature-acceptance\nstatus: passed\n---\n# Acceptance\n",
)
result = workflow_next.feature_next(feature, epic_child_batch=False)
assert result["status"] == "continue"
assert result["next_action"] == "return-to-cs-epic"
assert result["evidence"]["roadmap_item"] == "api-seed"
assert result["evidence"]["roadmap_goal_state"].endswith("billing-system/goal-state.yaml")
@pytest.mark.parametrize("artifacts_complete", [False, True])
def test_feature_legacy_epic_child_without_roadmap_frontmatter_is_reverse_owned(
tmp_path: Path,
artifacts_complete: bool,
) -> None:
repo = init_repo(tmp_path)
roadmap = write_roadmap(repo)
write_roadmap_goal_state(roadmap)
feature = write_feature(repo, "api-seed", design_status="approved", include_roadmap=False)
if artifacts_complete:
write(feature / "api-seed-checklist.yaml", "steps:\n - id: step-1\n status: done\n")
write_code_review(feature, "api-seed")
write(
feature / "api-seed-acceptance.md",
"---\ndoc_type: feature-acceptance\nstatus: passed\n---\n# Acceptance\n",
)
result = workflow_next.feature_next(feature, epic_child_batch=False)
assert result["status"] == "continue"
assert result["next_action"] == "return-to-cs-epic"
assert result["evidence"]["execution_lane"] == "goal"
assert result["evidence"]["execution_lane_source"] == "roadmap-goal-state"
assert result["next_action"] != "CS_FEATURE_STANDARD_COMPLETE"
@pytest.mark.parametrize(
("design_status", "artifacts_complete"),
[("draft", False), ("approved", True)],
)
def test_feature_pre_goal_package_metadata_less_child_with_items_pointer_returns_to_epic(
tmp_path: Path,
design_status: str,
artifacts_complete: bool,
) -> None:
repo = init_repo(tmp_path)
roadmap = write_roadmap(repo)
write(
roadmap / "billing-system-items.yaml",
"roadmap: billing-system\n"
"items:\n"
" - slug: api-seed\n"
" status: planned\n"
" feature: .codestable/features/2026-07-02-api-seed\n",
)
feature = write_feature(repo, "api-seed", design_status=design_status, include_roadmap=False)
if artifacts_complete:
write(feature / "api-seed-checklist.yaml", "steps:\n - id: step-1\n status: done\n")
write_code_review(feature, "api-seed")
write(
feature / "api-seed-acceptance.md",
"---\ndoc_type: feature-acceptance\nstatus: passed\n---\n# Acceptance\n",
)
result = workflow_next.feature_next(feature, epic_child_batch=False)
epic_result = workflow_next.epic_next(roadmap)
assert result["status"] == "continue"
assert result["next_action"] == "return-to-cs-epic"
assert result["evidence"]["execution_lane"] == "goal"
assert result["evidence"]["execution_lane_source"] == "roadmap-items"
assert result["evidence"]["roadmap_items"].endswith("billing-system-items.yaml")
assert result["next_action"] not in {"feature-design-confirmation", "CS_FEATURE_STANDARD_COMPLETE"}
assert epic_result["next_action"] == (
"all-feature-designs-confirmation" if design_status == "draft" else "cs-epic goal-package"
)
@pytest.mark.parametrize(
("design_status", "artifacts_complete"),
[("draft", False), ("approved", True)],
)
def test_feature_pre_goal_package_metadata_less_child_with_items_glob_returns_to_epic(
tmp_path: Path,
design_status: str,
artifacts_complete: bool,
) -> None:
repo = init_repo(tmp_path)
write_roadmap(repo)
feature = write_feature(repo, "api-seed", design_status=design_status, include_roadmap=False)
if artifacts_complete:
write(feature / "api-seed-checklist.yaml", "steps:\n - id: step-1\n status: done\n")
write_code_review(feature, "api-seed")
write(
feature / "api-seed-acceptance.md",
"---\ndoc_type: feature-acceptance\nstatus: passed\n---\n# Acceptance\n",
)
result = workflow_next.feature_next(feature, epic_child_batch=False)
assert result["status"] == "continue"
assert result["next_action"] == "return-to-cs-epic"
assert result["evidence"]["execution_lane_source"] == "roadmap-items"
assert result["next_action"] not in {"feature-design-confirmation", "CS_FEATURE_STANDARD_COMPLETE"}
@pytest.mark.parametrize("same_items_file", [False, True])
def test_reverse_items_owner_multiple_claims_fail_closed(
tmp_path: Path,
same_items_file: bool,
) -> None:
repo = init_repo(tmp_path)
feature = write_feature(repo, "api-seed", design_status="approved", include_roadmap=False)
feature_dir = ".codestable/features/2026-07-02-api-seed"
first = repo / ".codestable/roadmap/first-roadmap/first-roadmap-items.yaml"
if same_items_file:
write(
first,
"roadmap: first-roadmap\nitems:\n"
f" - slug: api-seed\n feature: {feature_dir}\n"
f" - slug: api-seed-copy\n feature: {feature_dir}\n",
)
expected_paths = [first.relative_to(repo).as_posix()]
else:
second = repo / ".codestable/roadmap/second-roadmap/second-roadmap-items.yaml"
write(first, f"roadmap: first-roadmap\nitems:\n - slug: api-seed\n feature: {feature_dir}\n")
write(second, f"roadmap: second-roadmap\nitems:\n - slug: api-seed\n feature: {feature_dir}\n")
expected_paths = [first.relative_to(repo).as_posix(), second.relative_to(repo).as_posix()]
result = workflow_next.feature_next(feature, epic_child_batch=False)
assert result["status"] == "blocked"
assert result["next_action"] == "inspect-epic-goal-state"
assert "multiple roadmap items claim" in result["reason"]
assert all(path in result["blocking"][0] for path in expected_paths)
@pytest.mark.parametrize("contents", ["items: [unterminated\n", "- api-seed\n"])
def test_reverse_items_owner_invalid_artifact_fails_closed(
tmp_path: Path,
contents: str,
) -> None:
repo = init_repo(tmp_path)
feature = write_feature(repo, "api-seed", design_status="approved", include_roadmap=False)
items_path = repo / ".codestable/roadmap/broken-roadmap/broken-roadmap-items.yaml"
write(items_path, contents)
result = workflow_next.feature_next(feature, epic_child_batch=False)
expected_path = items_path.relative_to(repo).as_posix()
assert result["status"] == "blocked"
assert result["next_action"] == "inspect-epic-goal-state"
assert expected_path in result["blocking"][0]
assert expected_path in result["evidence"]["roadmap_owner_error"]
@pytest.mark.parametrize(
"contents",
[
"roadmap: broken-roadmap\nitems:\n slug: api-seed\n feature: null\n",
"roadmap: broken-roadmap\nitems:\n - slug: api-seed\n feature: []\n",
],
)
def test_reverse_items_owner_invalid_yaml_shape_returns_structured_block(
tmp_path: Path,
contents: str,
) -> None:
repo = init_repo(tmp_path)
feature = write_feature(repo, "api-seed", design_status="approved", include_roadmap=False)
items_path = repo / ".codestable/roadmap/broken-roadmap/broken-roadmap-items.yaml"
write(items_path, contents)
completed, payload = run_cli_json(repo, "feature", feature)
expected_path = items_path.relative_to(repo).as_posix()
assert completed.returncode == 1
assert completed.stderr == ""
assert payload["status"] == "blocked"
assert expected_path in payload["blocking"][0]
@pytest.mark.parametrize(
"contents",
[
"roadmap: billing-system\nitems:\n slug: api-seed\n feature: null\n",
"roadmap: billing-system\nitems:\n - slug: api-seed\n feature: []\n",
],
)
def test_epic_invalid_items_yaml_shape_returns_structured_block(
tmp_path: Path,
contents: str,
) -> None:
repo = init_repo(tmp_path)
roadmap = write_roadmap(repo)
items_path = roadmap / "billing-system-items.yaml"
write(items_path, contents)
completed, payload = run_cli_json(repo, "epic", roadmap)
expected_path = items_path.relative_to(repo).as_posix()
assert completed.returncode == 1
assert completed.stderr == ""
assert payload["status"] == "blocked"
assert payload["next_action"] == "fix-epic-artifact-yaml"
assert expected_path in payload["blocking"][0]
@pytest.mark.parametrize("quick", [False, True])
def test_reverse_items_owner_ignores_unrelated_valid_items(tmp_path: Path, quick: bool) -> None:
repo = init_repo(tmp_path)
if quick:
feature = repo / ".codestable/features/2026-07-13-small-export"
write_ff_note(feature, "small-export")
else:
feature = write_feature(repo, "api-seed", design_status="approved", include_roadmap=False)
write(
repo / ".codestable/roadmap/other-roadmap/other-roadmap-items.yaml",
"roadmap: other-roadmap\n"
"items:\n"
" - slug: other-feature\n"
" feature: .codestable/features/2026-07-02-other-feature\n",
)
result = workflow_next.feature_next(feature, epic_child_batch=False)
assert result["status"] == "continue"
assert result["next_action"] == ("cs-code-review" if quick else "cs-feat implementation")
assert result["evidence"]["execution_lane"] == ("quick" if quick else "standard")
@pytest.mark.parametrize("quick", [False, True])
def test_reverse_items_owner_ignores_slug_suffix_collision(tmp_path: Path, quick: bool) -> None:
repo = init_repo(tmp_path)
if quick:
item_slug = "export"
feature = repo / ".codestable/features/2026-07-13-small-export"
write_ff_note(feature, "small-export")
write_code_review(feature, "small-export")
else:
item_slug = "auth"
feature = write_feature(repo, "user-auth", design_status="approved", include_roadmap=False)
write(
repo / ".codestable/roadmap/platform/platform-items.yaml",
"roadmap: platform\n"
"items:\n"
f" - slug: {item_slug}\n"
" feature: null\n",
)
result = workflow_next.feature_next(feature, epic_child_batch=False)
assert result["status"] == ("complete" if quick else "continue")
assert result["next_action"] == ("CS_FEATURE_QUICK_COMPLETE" if quick else "cs-feat implementation")
assert result["evidence"]["execution_lane"] == ("quick" if quick else "standard")
def test_reverse_items_owner_multiple_exact_slug_directories_fail_closed(tmp_path: Path) -> None:
repo = init_repo(tmp_path)
roadmap = write_roadmap(repo)
feature = write_feature(repo, "api-seed", design_status="approved", include_roadmap=False)
write(repo / ".codestable/features/2026-07-01-api-seed/placeholder.txt", "")
write(
roadmap / "billing-system-items.yaml",
"roadmap: billing-system\nitems:\n - slug: api-seed\n feature: null\n",
)
result = workflow_next.feature_next(feature, epic_child_batch=False)
epic_result = workflow_next.epic_next(roadmap)
assert result["status"] == "blocked"
assert result["next_action"] == "inspect-epic-goal-state"
assert "multiple feature directories match roadmap item api-seed" in result["reason"]
assert epic_result["status"] == "blocked"
assert epic_result["next_action"] == "fix-roadmap-items"
assert "multiple feature directories match roadmap item api-seed" in epic_result["reason"]
def test_reverse_items_owner_does_not_fallback_from_explicit_missing_pointer(tmp_path: Path) -> None:
repo = init_repo(tmp_path)
roadmap = write_roadmap(repo)
feature = write_feature(repo, "api-seed", design_status="approved", include_roadmap=False)
write(
roadmap / "billing-system-items.yaml",
"roadmap: billing-system\n"
"items:\n"
" - slug: api-seed\n"
" feature: .codestable/features/2026-07-02-missing-feature\n",
)
result = workflow_next.feature_next(feature, epic_child_batch=False)
epic_result = workflow_next.epic_next(roadmap)
assert result["status"] == "continue"
assert result["next_action"] == "cs-feat implementation"
assert result["evidence"]["execution_lane"] == "standard"
assert epic_result["status"] == "continue"
assert epic_result["next_action"] == "cs-feat design/design-review"
assert epic_result["evidence"]["next_item"]["feature_dir"] is None
def test_reverse_items_owner_invalid_pointer_returns_structured_block(tmp_path: Path) -> None:
repo = init_repo(tmp_path)
items_path = repo / ".codestable/roadmap/platform/platform-items.yaml"
write(
items_path,
"roadmap: platform\n"
"items:\n"
" - slug: api-seed\n"
' feature: ".codestable/features/2026\\0-07-02-api-seed"\n',
)
feature = write_feature(repo, "api-seed", design_status="approved", include_roadmap=False)
write(feature / "api-seed-checklist.yaml", "steps:\n - id: step-1\n status: done\n")
write_code_review(feature, "api-seed")
write(
feature / "api-seed-acceptance.md",
"---\ndoc_type: feature-acceptance\nstatus: passed\n---\n# Acceptance\n",
)
completed, payload = run_cli_json(repo, "feature", feature)
expected_path = items_path.relative_to(repo).as_posix()
assert completed.returncode == 1
assert completed.stderr == ""
assert payload["status"] == "blocked"
assert expected_path in payload["blocking"][0]
assert payload["next_action"] != "CS_FEATURE_STANDARD_COMPLETE"
def test_reverse_owner_ignores_unrelated_valid_goal_state(tmp_path: Path) -> None:
repo = init_repo(tmp_path)
feature = write_feature(repo, "api-seed", design_status="approved", include_roadmap=False)
write_reverse_owner_state(
repo,
"other-roadmap",
rows=[("other-feature", ".codestable/features/2026-07-02-other-feature")],
)
result = workflow_next.feature_next(feature, epic_child_batch=False)
assert result["status"] == "continue"
assert result["next_action"] == "cs-feat implementation"
assert result["evidence"]["execution_lane"] == "standard"
@pytest.mark.parametrize("quick", [False, True])
def test_reverse_owner_ignores_unrelated_identity_mismatch(tmp_path: Path, quick: bool) -> None:
repo = init_repo(tmp_path)
if quick:
feature = repo / ".codestable/features/2026-07-13-small-export"
write_ff_note(feature, "small-export")
else:
feature = write_feature(repo, "api-seed", design_status="approved", include_roadmap=False)
write_reverse_owner_state(
repo,
"renamed-roadmap",
state_roadmap="old-roadmap",
rows=[("other-feature", ".codestable/features/2026-07-02-other-feature")],
)
result = workflow_next.feature_next(feature, epic_child_batch=False)
assert result["status"] == "continue"
assert result["next_action"] == ("cs-code-review" if quick else "cs-feat implementation")
def test_reverse_owner_identity_mismatch_claiming_feature_fails_closed(tmp_path: Path) -> None:
repo = init_repo(tmp_path)
feature = write_feature(repo, "api-seed", design_status="approved", include_roadmap=False)
goal_state = write_reverse_owner_state(
repo,
"renamed-roadmap",
state_roadmap="old-roadmap",
rows=[("api-seed", ".codestable/features/2026-07-02-api-seed")],
)
result = workflow_next.feature_next(feature, epic_child_batch=False)
expected_path = goal_state.relative_to(repo).as_posix()
assert result["status"] == "blocked"
assert result["next_action"] == "inspect-epic-goal-state"
assert expected_path in result["blocking"][0]
assert expected_path in result["evidence"]["roadmap_owner_error"]
def test_reverse_owner_unparseable_goal_state_fails_closed(tmp_path: Path) -> None:
repo = init_repo(tmp_path)
feature = write_feature(repo, "api-seed", design_status="approved", include_roadmap=False)
goal_state = repo / ".codestable/roadmap/broken-roadmap/goal-state.yaml"
write(goal_state, "roadmap: broken-roadmap\nfeatures: [unterminated\n")
result = workflow_next.feature_next(feature, epic_child_batch=False)
expected_path = goal_state.relative_to(repo).as_posix()
assert result["status"] == "blocked"
assert expected_path in result["blocking"][0]
assert expected_path in result["evidence"]["roadmap_owner_error"]
@pytest.mark.parametrize("same_state", [False, True])
def test_reverse_owner_multiple_claims_fail_closed(tmp_path: Path, same_state: bool) -> None:
repo = init_repo(tmp_path)
feature = write_feature(repo, "api-seed", design_status="approved", include_roadmap=False)
feature_dir = ".codestable/features/2026-07-02-api-seed"
first = write_reverse_owner_state(repo, "first-roadmap", rows=[("api-seed", feature_dir)])
if same_state:
write_reverse_owner_state(
repo,
"first-roadmap",
rows=[("api-seed", feature_dir), ("api-seed-copy", feature_dir)],
)
expected_paths = [first.relative_to(repo).as_posix()]
else:
second = write_reverse_owner_state(repo, "second-roadmap", rows=[("api-seed", feature_dir)])
expected_paths = [first.relative_to(repo).as_posix(), second.relative_to(repo).as_posix()]
result = workflow_next.feature_next(feature, epic_child_batch=False)
assert result["status"] == "blocked"
assert "multiple roadmap goal-states claim" in result["reason"]
assert all(path in result["blocking"][0] for path in expected_paths)
def test_reverse_owner_matching_row_without_item_fails_closed(tmp_path: Path) -> None:
repo = init_repo(tmp_path)
feature = write_feature(repo, "api-seed", design_status="approved", include_roadmap=False)
goal_state = write_reverse_owner_state(
repo,
"broken-roadmap",
rows=[(None, ".codestable/features/2026-07-02-api-seed")],
)
result = workflow_next.feature_next(feature, epic_child_batch=False)
expected_path = goal_state.relative_to(repo).as_posix()
assert result["status"] == "blocked"
assert expected_path in result["blocking"][0]
assert expected_path in result["evidence"]["roadmap_owner_error"]
@pytest.mark.parametrize("forward", [False, True])
def test_goal_state_invalid_feature_dir_returns_structured_block(
tmp_path: Path,
forward: bool,
) -> None:
repo = init_repo(tmp_path)
roadmap = write_roadmap(repo)
goal_state = roadmap / "goal-state.yaml"
write(
goal_state,
"roadmap: billing-system\n"
"status: ready-to-dispatch\n"
"features:\n"
" - slug: api-seed\n"
" roadmap_item: api-seed\n"
' feature_dir: ".codestable/features/2026\\0-07-02-api-seed"\n'
" status: implementing\n",
)
feature = write_feature(
repo,
"api-seed",
design_status="approved",
include_roadmap=forward,
)
completed, payload = run_cli_json(repo, "feature", feature)
expected_path = goal_state.relative_to(repo).as_posix()
assert completed.returncode == 1
assert completed.stderr == ""
assert payload["status"] == "blocked"
assert expected_path in payload["blocking"][0]
@pytest.mark.parametrize(
"features_yaml",
[
"features:\n slug: api-seed\n feature_dir: .codestable/features/2026-07-02-api-seed\n",
"features:\n - api-seed\n",
],
)
def test_reverse_goal_owner_invalid_features_shape_fails_closed(
tmp_path: Path,
features_yaml: str,
) -> None:
repo = init_repo(tmp_path)
roadmap = write_roadmap(repo)
goal_state = roadmap / "goal-state.yaml"
write(goal_state, "roadmap: billing-system\nstatus: ready-to-dispatch\n" + features_yaml)
feature = write_feature(repo, "api-seed", design_status="approved", include_roadmap=False)
write(feature / "api-seed-checklist.yaml", "steps:\n - id: step-1\n status: done\n")
write_code_review(feature, "api-seed")
write(
feature / "api-seed-acceptance.md",
"---\ndoc_type: feature-acceptance\nstatus: passed\n---\n# Acceptance\n",
)
completed, payload = run_cli_json(repo, "feature", feature)
expected_path = goal_state.relative_to(repo).as_posix()
assert completed.returncode == 1
assert completed.stderr == ""
assert payload["status"] == "blocked"
assert expected_path in payload["blocking"][0]
assert payload["next_action"] != "CS_FEATURE_STANDARD_COMPLETE"
def test_forward_metadata_missing_state_detects_external_reverse_claim(tmp_path: Path) -> None:
repo = init_repo(tmp_path)
feature = write_feature(repo, "api-seed", design_status="approved", include_roadmap=True)
write(feature / "api-seed-checklist.yaml", "steps:\n - id: step-1\n status: done\n")
write_code_review(feature, "api-seed")
write(
feature / "api-seed-acceptance.md",
"---\ndoc_type: feature-acceptance\nstatus: passed\n---\n# Acceptance\n",
)
external = write_reverse_owner_state(
repo,
"external-roadmap",
rows=[("api-seed", ".codestable/features/2026-07-02-api-seed")],
)
result = workflow_next.feature_next(feature, epic_child_batch=False)
expected = [
".codestable/roadmap/billing-system/goal-state.yaml",
external.relative_to(repo).as_posix(),
]
assert result["status"] == "blocked"
assert result["next_action"] == "inspect-epic-goal-state"
assert result["next_action"] != "CS_FEATURE_STANDARD_COMPLETE"
assert all(path in result["blocking"][0] for path in expected)
def test_forward_owner_detects_second_external_claim(tmp_path: Path) -> None:
repo = init_repo(tmp_path)
roadmap = write_roadmap(repo)
write_roadmap_goal_state(roadmap)
feature = write_feature(repo, "api-seed", design_status="approved", include_roadmap=True)
external = write_reverse_owner_state(
repo,
"external-roadmap",
rows=[("api-seed", ".codestable/features/2026-07-02-api-seed")],
)
result = workflow_next.feature_next(feature, epic_child_batch=False)
assert result["status"] == "blocked"
assert result["next_action"] == "inspect-epic-goal-state"
assert external.relative_to(repo).as_posix() in result["blocking"][0]
@pytest.mark.parametrize(
("design_status", "artifacts_complete"),
[("draft", False), ("approved", True)],
)
def test_feature_pre_goal_package_roadmap_child_returns_to_epic(
tmp_path: Path,
design_status: str,
artifacts_complete: bool,
) -> None:
repo = init_repo(tmp_path)
write_roadmap(repo)
feature = write_feature(repo, "api-seed", design_status=design_status, include_roadmap=True)
if artifacts_complete:
write(feature / "api-seed-checklist.yaml", "steps:\n - id: step-1\n status: done\n")
write_code_review(feature, "api-seed")
write(
feature / "api-seed-acceptance.md",
"---\ndoc_type: feature-acceptance\nstatus: passed\n---\n# Acceptance\n",
)
result = workflow_next.feature_next(feature, epic_child_batch=False)
assert result["status"] == "continue"
assert result["next_action"] == "return-to-cs-epic"
assert result["evidence"]["roadmap_owner_source"] == "roadmap-items"
assert result["evidence"]["roadmap_items"].endswith("billing-system-items.yaml")
assert result["next_action"] not in {"feature-design-confirmation", "CS_FEATURE_STANDARD_COMPLETE"}
@pytest.mark.parametrize(
("metadata_field", "metadata_value"),
[("roadmap", "billing-system"), ("roadmap_item", "api-seed")],
)
def test_feature_incomplete_roadmap_metadata_fails_closed(
tmp_path: Path,
metadata_field: str,
metadata_value: str,
) -> None:
repo = init_repo(tmp_path)
feature = write_feature(repo, "api-seed", design_status="approved", include_roadmap=False)
write(
feature / "api-seed-design.md",
"---\ndoc_type: feature-design\nfeature: 2026-07-02-api-seed\n"
f"{metadata_field}: {metadata_value}\nstatus: approved\n---\n# Design\n",
)
result = workflow_next.feature_next(feature, epic_child_batch=False)
assert result["status"] == "blocked"
assert result["next_action"] == "fix-feature-roadmap-metadata"
assert "roadmap and roadmap_item together" in result["reason"]
def test_feature_invalid_roadmap_slug_returns_structured_block(tmp_path: Path) -> None:
repo = init_repo(tmp_path)
feature = write_feature(repo, "api-seed", design_status="approved", include_roadmap=False)
write(
feature / "api-seed-design.md",
"---\n"
"doc_type: feature-design\n"
"feature: 2026-07-02-api-seed\n"
'roadmap: "billing\\0x"\n'
"roadmap_item: api-seed\n"
"status: approved\n"
"---\n# Design\n",
)
completed, payload = run_cli_json(repo, "feature", feature)
assert completed.returncode == 1
assert completed.stderr == ""
assert payload["status"] == "blocked"
assert payload["next_action"] == "inspect-epic-goal-state"
assert "invalid roadmap owner path" in payload["reason"]
def test_feature_pre_goal_package_roadmap_without_matching_item_fails_closed(tmp_path: Path) -> None:
repo = init_repo(tmp_path)
roadmap = write_roadmap(repo)
write(
roadmap / "billing-system-items.yaml",
"roadmap: billing-system\nitems:\n - slug: ui-seed\n status: planned\n",
)
feature = write_feature(repo, "api-seed", design_status="approved", include_roadmap=True)
result = workflow_next.feature_next(feature, epic_child_batch=False)
items_path = ".codestable/roadmap/billing-system/billing-system-items.yaml"
assert result["status"] == "blocked"
assert result["next_action"] == "inspect-epic-goal-state"
assert items_path in result["blocking"][0]
assert "do not uniquely own" in result["reason"]
def test_feature_pre_goal_package_roadmap_item_pointer_must_match_feature(tmp_path: Path) -> None:
repo = init_repo(tmp_path)
roadmap = write_roadmap(repo)
write(
roadmap / "billing-system-items.yaml",
"roadmap: billing-system\n"
"items:\n"
" - slug: api-seed\n"
" status: planned\n"
" feature: .codestable/features/2026-07-02-other-feature\n",
)
feature = write_feature(repo, "api-seed", design_status="approved", include_roadmap=True)
result = workflow_next.feature_next(feature, epic_child_batch=False)
items_path = ".codestable/roadmap/billing-system/billing-system-items.yaml"
assert result["status"] == "blocked"
assert result["next_action"] == "inspect-epic-goal-state"
assert "feature pointer does not match" in result["reason"]
assert items_path in result["blocking"][0]
@pytest.mark.parametrize(
"contents",
[
"roadmap: billing-system\nitems:\n slug: api-seed\n feature: null\n",
"roadmap: billing-system\nitems:\n - slug: api-seed\n feature: []\n",
],
)
def test_forward_items_owner_invalid_yaml_shape_returns_structured_block(
tmp_path: Path,
contents: str,
) -> None:
repo = init_repo(tmp_path)
roadmap = write_roadmap(repo)
items_path = roadmap / "billing-system-items.yaml"
write(items_path, contents)
feature = write_feature(repo, "api-seed", design_status="approved", include_roadmap=True)
completed, payload = run_cli_json(repo, "feature", feature)
expected_path = items_path.relative_to(repo).as_posix()
assert completed.returncode == 1
assert completed.stderr == ""
assert payload["status"] == "blocked"
assert payload["next_action"] == "inspect-epic-goal-state"
assert expected_path in payload["blocking"][0]
@pytest.mark.parametrize("same_items_file", [False, True])
def test_forward_items_owner_detects_second_items_claim(
tmp_path: Path,
same_items_file: bool,
) -> None:
repo = init_repo(tmp_path)
roadmap = write_roadmap(repo)
feature_dir = ".codestable/features/2026-07-02-api-seed"
primary_items = roadmap / "billing-system-items.yaml"
if same_items_file:
write(
primary_items,
"roadmap: billing-system\nitems:\n"
f" - slug: api-seed\n feature: {feature_dir}\n"
f" - slug: other-item\n feature: {feature_dir}\n",
)
expected_paths = [primary_items.relative_to(repo).as_posix()]
else:
secondary_items = repo / ".codestable/roadmap/secondary/secondary-items.yaml"
write(
primary_items,
f"roadmap: billing-system\nitems:\n - slug: api-seed\n feature: {feature_dir}\n",
)
write(
secondary_items,
f"roadmap: secondary\nitems:\n - slug: other-item\n feature: {feature_dir}\n",
)
expected_paths = [
primary_items.relative_to(repo).as_posix(),
secondary_items.relative_to(repo).as_posix(),
]
feature = write_feature(repo, "api-seed", design_status="approved", include_roadmap=True)
result = workflow_next.feature_next(feature, epic_child_batch=False)
assert result["status"] == "blocked"
assert result["next_action"] == "inspect-epic-goal-state"
assert "multiple roadmap items claim" in result["reason"]
assert all(path in result["blocking"][0] for path in expected_paths)
@pytest.mark.parametrize("same_items_file", [False, True])
def test_forward_goal_owner_detects_second_items_claim(
tmp_path: Path,
same_items_file: bool,
) -> None:
repo = init_repo(tmp_path)
roadmap = write_roadmap(repo)
write_roadmap_goal_state(roadmap)
feature_dir = ".codestable/features/2026-07-02-api-seed"
primary_items = roadmap / "billing-system-items.yaml"
primary = (
"roadmap: billing-system\nitems:\n"
" - slug: api-seed\n feature: null\n"
)
if same_items_file:
write(primary_items, primary + f" - slug: other-item\n feature: {feature_dir}\n")
expected_paths = [primary_items.relative_to(repo).as_posix()]
else:
secondary_items = repo / ".codestable/roadmap/secondary/secondary-items.yaml"
write(primary_items, primary)
write(
secondary_items,
f"roadmap: secondary\nitems:\n - slug: other-item\n feature: {feature_dir}\n",
)
expected_paths = [
primary_items.relative_to(repo).as_posix(),
secondary_items.relative_to(repo).as_posix(),
]
feature = write_feature(repo, "api-seed", design_status="approved", include_roadmap=True)
result = workflow_next.feature_next(feature, epic_child_batch=False)
assert result["status"] == "blocked"
assert result["next_action"] == "inspect-epic-goal-state"
assert "multiple roadmap items claim" in result["reason"]
assert all(path in result["blocking"][0] for path in expected_paths)
def test_forward_goal_owner_detects_second_claim_in_same_goal_state(tmp_path: Path) -> None:
repo = init_repo(tmp_path)
roadmap = write_roadmap(repo)
feature_dir = ".codestable/features/2026-07-02-api-seed"
goal_state = roadmap / "goal-state.yaml"
write(
goal_state,
"roadmap: billing-system\n"
"status: ready-to-dispatch\n"
"features:\n"
" - slug: api-seed\n"
" roadmap_item: api-seed\n"
f" feature_dir: {feature_dir}\n"
" status: implementing\n"
" - slug: duplicate-claim\n"
" roadmap_item: duplicate-claim\n"
f" feature_dir: {feature_dir}\n"
" status: pending\n",
)
feature = write_feature(repo, "api-seed", design_status="approved", include_roadmap=True)
result = workflow_next.feature_next(feature, epic_child_batch=False)
expected_path = goal_state.relative_to(repo).as_posix()
assert result["status"] == "blocked"
assert result["next_action"] == "inspect-epic-goal-state"
assert "multiple roadmap goal-states claim" in result["reason"]
assert expected_path in result["blocking"][0]
def test_feature_epic_child_batch_never_completes_as_standard(tmp_path: Path) -> None:
repo = init_repo(tmp_path)
write_roadmap(repo)
feature = write_feature(repo, "api-seed", design_status="approved", include_roadmap=True)
write(feature / "api-seed-checklist.yaml", "steps:\n - id: step-1\n status: done\n")
write_code_review(feature, "api-seed")
write(
feature / "api-seed-acceptance.md",
"---\ndoc_type: feature-acceptance\nstatus: passed\n---\n# Acceptance\n",
)
result = workflow_next.feature_next(feature, epic_child_batch=True)
assert result["status"] == "continue"
assert result["next_action"] == "return-to-cs-epic-batch-loop"
assert result["evidence"]["execution_lane"] == "goal"
assert result["evidence"]["execution_lane_source"] == "epic-child-batch"
assert result["next_action"] != "CS_FEATURE_STANDARD_COMPLETE"
def test_feature_draft_owned_by_roadmap_goal_returns_to_epic(tmp_path: Path) -> None:
repo = init_repo(tmp_path)
roadmap = write_roadmap(repo)
write_roadmap_goal_state(roadmap)
feature = write_feature(
repo,
"api-seed",
design_status="draft",
review_status="passed",
include_roadmap=True,
)
result = workflow_next.feature_next(feature, epic_child_batch=False)
assert result["status"] == "continue"
assert result["next_action"] == "return-to-cs-epic"
assert result["next_action"] != "feature-design-confirmation"
assert result["evidence"]["execution_lane_source"] == "roadmap-goal-state"
def test_feature_parent_goal_state_with_mismatched_owner_fails_closed(tmp_path: Path) -> None:
repo = init_repo(tmp_path)
roadmap = write_roadmap(repo)
write_roadmap_goal_state(roadmap, feature_slug="ui-seed")
feature = write_feature(repo, "api-seed", design_status="approved", include_roadmap=True)
result = workflow_next.feature_next(feature, epic_child_batch=False)
assert result["status"] == "blocked"
assert result["next_action"] == "inspect-epic-goal-state"
assert "standard" not in result["next_action"]
def test_feature_quick_ff_note_recovers_review_then_completion(tmp_path: Path) -> None:
repo = init_repo(tmp_path)
feature = repo / ".codestable/features/2026-07-13-small-export"
write_ff_note(feature, "small-export")
review = workflow_next.feature_next(feature, epic_child_batch=False)
assert review["status"] == "continue"
assert review["next_action"] == "cs-code-review"
assert review["evidence"]["execution_lane"] == "quick"
write_code_review(feature, "small-export")
complete = workflow_next.feature_next(feature, epic_child_batch=False)
assert complete["status"] == "complete"
assert complete["next_action"] == "CS_FEATURE_QUICK_COMPLETE"
def test_feature_quick_ff_note_routes_review_fix(tmp_path: Path) -> None:
repo = init_repo(tmp_path)
feature = repo / ".codestable/features/2026-07-13-small-export"
write_ff_note(feature, "small-export")
write_code_review(feature, "small-export", status="changes-requested")
result = workflow_next.feature_next(feature, epic_child_batch=False)
assert result["status"] == "continue"
assert result["next_action"] == "cs-feat --mode quick"
assert "review-fix" in result["reason"]
@pytest.mark.parametrize(
("artifact", "status"),
[("qa", "failed"), ("qa", "blocked"), ("acceptance", "failed"), ("acceptance", "blocked")],
)
def test_feature_quick_rejects_existing_failed_quality_evidence(
tmp_path: Path,
artifact: str,
status: str,
) -> None:
repo = init_repo(tmp_path)
feature = write_feature(
repo,
"api-seed",
design_status="approved",
execution_lane="quick",
execution_lane_reason="owner-requested-after-risk-recheck",
)
write_ff_note(feature, "api-seed")
write_code_review(feature, "api-seed")
doc_type = "feature-qa" if artifact == "qa" else "feature-acceptance"
write(
feature / f"api-seed-{artifact}.md",
f"---\ndoc_type: {doc_type}\nstatus: {status}\n---\n# Evidence\n",
)
result = workflow_next.feature_next(feature, epic_child_batch=False)
artifact_path = f".codestable/features/2026-07-02-api-seed/api-seed-{artifact}.md"
assert result["status"] == "blocked"
assert result["next_action"] == "resolve-quick-quality-conflict"
assert artifact_path in result["blocking"][0]
assert result["next_action"] != "CS_FEATURE_QUICK_COMPLETE"
def test_feature_quick_allows_existing_passed_quality_evidence(tmp_path: Path) -> None:
repo = init_repo(tmp_path)
feature = write_feature(
repo,
"api-seed",
design_status="approved",
execution_lane="quick",
execution_lane_reason="owner-requested-after-risk-recheck",
)
write_ff_note(feature, "api-seed")
write_code_review(feature, "api-seed")
write(feature / "api-seed-qa.md", "---\ndoc_type: feature-qa\nstatus: passed\n---\n# QA\n")
write(
feature / "api-seed-acceptance.md",
"---\ndoc_type: feature-acceptance\nstatus: passed\n---\n# Acceptance\n",
)
result = workflow_next.feature_next(feature, epic_child_batch=False)
assert result["status"] == "complete"
assert result["next_action"] == "CS_FEATURE_QUICK_COMPLETE"
def test_feature_reclassified_quick_design_resumes_fastforward(tmp_path: Path) -> None:
repo = init_repo(tmp_path)
feature = write_feature(
repo,
"api-seed",
design_status="approved",
execution_lane="quick",
execution_lane_reason="owner-requested-after-risk-recheck",
)
result = workflow_next.feature_next(feature, epic_child_batch=False)
assert result["status"] == "continue"
assert result["next_action"] == "cs-feat --mode quick"
assert result["evidence"]["execution_lane_source"] == "design"
def test_feature_reclassified_quick_design_requires_persisted_reason(tmp_path: Path) -> None:
repo = init_repo(tmp_path)
feature = write_feature(repo, "api-seed", design_status="approved", execution_lane="quick")
result = workflow_next.feature_next(feature, epic_child_batch=False)
assert result["status"] == "blocked"
assert result["next_action"] == "fix-feature-execution-lane"
assert "execution_lane_reason" in result["reason"]
def test_feature_ff_note_conflicting_with_recorded_standard_lane_fails_closed(tmp_path: Path) -> None:
repo = init_repo(tmp_path)
feature = write_feature(repo, "api-seed", design_status="approved", execution_lane="standard")
write_ff_note(feature, "api-seed")
result = workflow_next.feature_next(feature, epic_child_batch=False)
assert result["status"] == "blocked"
assert result["next_action"] == "resolve-feature-execution-lane-conflict"
def test_feature_goal_state_overrides_reclassified_quick_lane(tmp_path: Path) -> None:
repo = init_repo(tmp_path)
feature = write_feature(repo, "api-seed", design_status="approved", execution_lane="quick")
write_ff_note(feature, "api-seed")
write_goal_state(feature, stage="implementation", status="ready-to-dispatch")
result = workflow_next.feature_next(feature, epic_child_batch=False)
assert result["status"] == "dispatch_goal"
assert result["evidence"]["execution_lane"] == "goal"
assert result["evidence"]["execution_lane_source"] == "feature-goal-state"
def test_feature_goal_state_requires_explicit_acceptance_authorization(tmp_path: Path) -> None:
repo = init_repo(tmp_path)
feature = write_feature(repo, "api-seed", design_status="approved", execution_lane="goal")
write_goal_state(
feature,
stage="implementation",
status="ready-to-dispatch",
acceptance_authorization=None,
)
missing = workflow_next.feature_next(feature, epic_child_batch=False)
assert missing["status"] == "user_gate"
assert missing["next_action"] == "authorize-feature-goal-acceptance"
write_goal_state(
feature,
stage="implementation",
status="ready-to-dispatch",
acceptance_authorization="rejected",
)
rejected = workflow_next.feature_next(feature, epic_child_batch=False)
assert rejected["status"] == "handoff"
assert rejected["next_action"] == "CS_FEATURE_GOAL_HANDOFF"
write_goal_state(
feature,
stage="complete",
status="passed",
acceptance_authorization=None,
)
legacy_complete = workflow_next.feature_next(feature, epic_child_batch=False)
assert legacy_complete["status"] == "user_gate"
assert legacy_complete["next_action"] == "authorize-feature-goal-acceptance"
write_goal_state(
feature,
stage="implementation",
status="running",
driver_kind="host-agent",
driver_id="feature-run-123",
acceptance_authorization=None,
)
unauthorized_driver = workflow_next.feature_next(feature, epic_child_batch=False)
assert unauthorized_driver["status"] == "user_gate"
assert unauthorized_driver["next_action"] == "authorize-feature-goal-acceptance"
@pytest.mark.parametrize(
("reference", "approval_status", "reason"),
[
("does-not-exist.md#goal-acceptance", "approved", "unit approval-report.md"),
("approval-report.md#goal-acceptance", None, "goal-acceptance is not approved"),
("approval-report.md#goal-acceptance", "pending", "is not approved"),
],
)
def test_feature_goal_state_validates_acceptance_approval_artifact(
tmp_path: Path,
reference: str,
approval_status: str | None,
reason: str,
) -> None:
repo = init_repo(tmp_path)
feature = write_feature(repo, "api-seed", design_status="approved", execution_lane="goal")
write_goal_state(
feature,
stage="implementation",
status="ready-to-dispatch",
acceptance_authorization_ref=reference,
approval_decision_status=approval_status,
)
result = workflow_next.feature_next(feature, epic_child_batch=False)
assert result["status"] == "user_gate"
assert result["next_action"] == "authorize-feature-goal-acceptance"
assert reason in result["reason"]
@pytest.mark.parametrize(
("authorization", "expected_status", "expected_action"),
[
(None, "user_gate", "authorize-epic-goal-execution"),
("rejected", "handoff", "CS_ROADMAP_GOAL_HANDOFF"),
],
)
def test_epic_goal_state_requires_explicit_acceptance_authorization(
tmp_path: Path,
authorization: str | None,
expected_status: str,
expected_action: str,
) -> None:
repo = init_repo(tmp_path)
roadmap = write_roadmap(repo)
write_feature(repo, "api-seed", design_status="approved", include_roadmap=True)
write_feature(repo, "ui-seed", design_status="approved", include_roadmap=True)
write_roadmap_goal_state(roadmap, acceptance_authorization=authorization)
result = workflow_next.epic_next(roadmap)
assert result["status"] == expected_status
assert result["next_action"] == expected_action
def test_epic_pending_package_template_stops_at_single_execution_gate(tmp_path: Path) -> None:
repo = init_repo(tmp_path)
roadmap = write_roadmap(repo)
write_feature(repo, "api-seed", design_status="approved", include_roadmap=True)
write_feature(repo, "ui-seed", design_status="approved", include_roadmap=True)
write_roadmap_goal_state(
roadmap,
status="awaiting-authorization",
driver_kind="none",
driver_id="",
acceptance_authorization="pending",
approval_decision_status="pending",
commit_authorization="pending",
commit_decision_status="pending",
execution_confirmation_id="",
approval_group_status="pending",
)
result = workflow_next.epic_next(roadmap)
assert result["status"] == "user_gate"
assert result["next_action"] == "authorize-epic-goal-execution"
assert "approval group goal-execution is not approved" in result["reason"]
assert "acceptance_authorization is not approved" in result["reason"]
assert "commit_authorization is not approved" in result["reason"]
assert result["evidence"]["acceptance_authorization_ref"] == "approval-report.md#goal-acceptance"
assert result["evidence"]["commit_authorization_ref"] == "approval-report.md#goal-commits"
@pytest.mark.parametrize(
("acceptance", "commit", "state_confirmation_id", "status", "reason"),
[
("pending", "pending", "", "awaiting-authorization", "not synchronized"),
(
"approved",
"pending",
"goal-confirm-1",
"ready-to-dispatch",
"commit_authorization",
),
("approved", "approved", "", "ready-to-dispatch", "not synchronized"),
("approved", "approved", "other-confirmation", "ready-to-dispatch", "not synchronized"),
("approved", "approved", "goal-confirm-1", "awaiting-authorization", "still awaits"),
("rejected", "approved", "goal-confirm-1", "ready-to-dispatch", "projection is rejected"),
],
)
def test_epic_durable_execution_confirmation_repairs_partial_goal_state_without_user_gate(
tmp_path: Path,
acceptance: str,
commit: str,
state_confirmation_id: str,
status: str,
reason: str,
) -> None:
repo = init_repo(tmp_path)
roadmap = write_roadmap(repo)
write_feature(repo, "api-seed", design_status="approved", include_roadmap=True)
write_feature(repo, "ui-seed", design_status="approved", include_roadmap=True)
write_roadmap_goal_state(
roadmap,
status=status,
driver_kind="none",
driver_id="",
acceptance_authorization=acceptance,
commit_authorization=commit,
execution_confirmation_id=state_confirmation_id,
approval_group_status="approved",
approval_group_confirmation_id="goal-confirm-1",
)
result = workflow_next.epic_next(roadmap)
assert result["status"] == "continue"
assert result["next_action"] == "repair-epic-goal-execution-authorization"
assert reason in result["reason"]
assert result["must_continue"] is True
assert result["final_answer_allowed"] is False
assert result["evidence"]["execution_confirmation_id"] == "goal-confirm-1"
assert result["evidence"]["acceptance_authorization"] == "approved"
assert result["evidence"]["commit_authorization"] == "approved"
def test_epic_repaired_execution_confirmation_dispatches_without_another_user_gate(
tmp_path: Path,
) -> None:
repo = init_repo(tmp_path)
roadmap = write_roadmap(repo)
write_feature(repo, "api-seed", design_status="approved", include_roadmap=True)
write_feature(repo, "ui-seed", design_status="approved", include_roadmap=True)
write_roadmap_goal_state(
roadmap,
driver_kind="none",
driver_id="",
execution_confirmation_id="goal-confirm-1",
approval_group_status="approved",
approval_group_confirmation_id="goal-confirm-1",
)
result = workflow_next.epic_next(roadmap)
assert result["status"] == "dispatch_goal"
assert result["next_action"] == "dispatch-epic-goal-driver-or-print-goal"
assert result["evidence"]["execution_confirmation_id"] == "goal-confirm-1"
def test_epic_pending_execution_group_cannot_be_bypassed_by_state_first_updates(
tmp_path: Path,
) -> None:
repo = init_repo(tmp_path)
roadmap = write_roadmap(repo)
write_feature(repo, "api-seed", design_status="approved", include_roadmap=True)
write_feature(repo, "ui-seed", design_status="approved", include_roadmap=True)
write_roadmap_goal_state(
roadmap,
execution_confirmation_id="goal-confirm-1",
approval_group_status="pending",
approval_group_confirmation_id="goal-confirm-1",
)
result = workflow_next.epic_next(roadmap)
assert result["status"] == "user_gate"
assert result["next_action"] == "authorize-epic-goal-execution"
assert "approval group goal-execution is not approved" in result["reason"]
def test_epic_rejected_execution_group_handoffs(tmp_path: Path) -> None:
repo = init_repo(tmp_path)
roadmap = write_roadmap(repo)
write_feature(repo, "api-seed", design_status="approved", include_roadmap=True)
write_feature(repo, "ui-seed", design_status="approved", include_roadmap=True)
write_roadmap_goal_state(
roadmap,
execution_confirmation_id="goal-confirm-1",
approval_group_status="rejected",
approval_group_confirmation_id="goal-confirm-1",
)
result = workflow_next.epic_next(roadmap)
assert result["status"] == "handoff"
assert result["next_action"] == "CS_ROADMAP_GOAL_HANDOFF"
def test_epic_legacy_approved_authorizations_dispatch_with_warning(
tmp_path: Path,
) -> None:
repo = init_repo(tmp_path)
roadmap = write_roadmap(repo)
write_feature(repo, "api-seed", design_status="approved", include_roadmap=True)
write_feature(repo, "ui-seed", design_status="approved", include_roadmap=True)
write_roadmap_goal_state(roadmap, driver_kind="none", driver_id="")
result = workflow_next.epic_next(roadmap)
assert result["status"] == "dispatch_goal"
assert "legacy epic goal approvals" in result["warnings"][-1]
def test_epic_new_state_without_execution_group_cannot_use_legacy_bypass(
tmp_path: Path,
) -> None:
repo = init_repo(tmp_path)
roadmap = write_roadmap(repo)
write_feature(repo, "api-seed", design_status="approved", include_roadmap=True)
write_feature(repo, "ui-seed", design_status="approved", include_roadmap=True)
write_roadmap_goal_state(
roadmap,
driver_kind="none",
driver_id="",
execution_confirmation_id="",
)
result = workflow_next.epic_next(roadmap)
assert result["status"] == "user_gate"
assert result["next_action"] == "authorize-epic-goal-execution"
def test_epic_approved_execution_group_without_confirmation_id_does_not_repair(
tmp_path: Path,
) -> None:
repo = init_repo(tmp_path)
roadmap = write_roadmap(repo)
write_feature(repo, "api-seed", design_status="approved", include_roadmap=True)
write_feature(repo, "ui-seed", design_status="approved", include_roadmap=True)
write_roadmap_goal_state(
roadmap,
driver_kind="none",
driver_id="",
execution_confirmation_id="",
approval_group_status="approved",
approval_group_confirmation_id="",
)
result = workflow_next.epic_next(roadmap)
assert result["status"] == "user_gate"
assert result["next_action"] == "authorize-epic-goal-execution"
assert result["next_action"] != "repair-epic-goal-execution-authorization"
assert "approval group goal-execution has no confirmation_id" in result["reason"]
def test_epic_external_approval_symlink_fails_closed_instead_of_repair_loop(
tmp_path: Path,
) -> None:
repo = init_repo(tmp_path)
roadmap = write_roadmap(repo)
write_feature(repo, "api-seed", design_status="approved", include_roadmap=True)
write_feature(repo, "ui-seed", design_status="approved", include_roadmap=True)
write_roadmap_goal_state(
roadmap,
driver_kind="none",
driver_id="",
execution_confirmation_id="goal-confirm-1",
approval_group_status="approved",
approval_group_confirmation_id="goal-confirm-1",
)
approval = roadmap / "approval-report.md"
external = tmp_path / "external-approval-report.md"
external.write_text(approval.read_text(encoding="utf-8"), encoding="utf-8")
approval.unlink()
approval.symlink_to(external)
result = workflow_next.epic_next(roadmap)
assert result["status"] == "user_gate"
assert result["next_action"] == "authorize-epic-goal-execution"
assert result["next_action"] != "repair-epic-goal-execution-authorization"
assert "approval-report.md escapes the workflow unit" in result["reason"]
def test_epic_legacy_complete_goal_state_still_requires_acceptance_authorization(
tmp_path: Path,
) -> None:
repo = init_repo(tmp_path)
roadmap = write_roadmap(repo)
write_feature(repo, "api-seed", design_status="approved", include_roadmap=True)
write_feature(repo, "ui-seed", design_status="approved", include_roadmap=True)
write_goal_state(roadmap, status="complete", acceptance_authorization=None)
result = workflow_next.epic_next(roadmap)
assert result["status"] == "user_gate"
assert result["next_action"] == "authorize-epic-goal-execution"
def test_epic_goal_state_rejects_unverifiable_acceptance_approval_ref(tmp_path: Path) -> None:
repo = init_repo(tmp_path)
roadmap = write_roadmap(repo)
write_feature(repo, "api-seed", design_status="approved", include_roadmap=True)
write_feature(repo, "ui-seed", design_status="approved", include_roadmap=True)
write_goal_state(
roadmap,
status="ready-to-dispatch",
acceptance_authorization_ref="missing.md#goal-acceptance",
)
result = workflow_next.epic_next(roadmap)
assert result["status"] == "user_gate"
assert result["next_action"] == "authorize-epic-goal-execution"
assert "unit approval-report.md" in result["reason"]
@pytest.mark.parametrize(
("authorization", "expected_status", "expected_action"),
[
(None, "user_gate", "authorize-epic-goal-execution"),
("Approved", "user_gate", "authorize-epic-goal-execution"),
("rejected", "handoff", "CS_ROADMAP_GOAL_HANDOFF"),
],
)
def test_epic_goal_state_requires_independent_commit_authorization(
tmp_path: Path,
authorization: str | None,
expected_status: str,
expected_action: str,
) -> None:
repo = init_repo(tmp_path)
roadmap = write_roadmap(repo)
write_feature(repo, "api-seed", design_status="approved", include_roadmap=True)
write_feature(repo, "ui-seed", design_status="approved", include_roadmap=True)
write_roadmap_goal_state(roadmap, commit_authorization=authorization)
result = workflow_next.epic_next(roadmap)
assert result["status"] == expected_status
assert result["next_action"] == expected_action
def test_epic_complete_state_still_requires_commit_authorization(tmp_path: Path) -> None:
repo = init_repo(tmp_path)
roadmap = write_roadmap(repo)
write_feature(repo, "api-seed", design_status="approved", include_roadmap=True)
write_feature(repo, "ui-seed", design_status="approved", include_roadmap=True)
write_goal_state(roadmap, status="complete", commit_authorization=None)
result = workflow_next.epic_next(roadmap)
assert result["status"] == "user_gate"
assert result["next_action"] == "authorize-epic-goal-execution"
def test_epic_goal_state_rejects_unapproved_commit_decision(tmp_path: Path) -> None:
repo = init_repo(tmp_path)
roadmap = write_roadmap(repo)
write_feature(repo, "api-seed", design_status="approved", include_roadmap=True)
write_feature(repo, "ui-seed", design_status="approved", include_roadmap=True)
write_goal_state(
roadmap,
status="ready-to-dispatch",
commit_decision_status="pending",
)
result = workflow_next.epic_next(roadmap)
assert result["status"] == "user_gate"
assert result["next_action"] == "authorize-epic-goal-execution"
assert "goal-commits is not approved" in result["reason"]
@pytest.mark.parametrize(
("reference", "reason"),
[
("approval-report.md#goal-acceptance", "must be approval-report.md#goal-commits"),
("../approval-report.md#goal-commits", "escapes the workflow unit"),
],
)
def test_epic_goal_state_rejects_unverifiable_commit_approval_ref(
tmp_path: Path,
reference: str,
reason: str,
) -> None:
repo = init_repo(tmp_path)
roadmap = write_roadmap(repo)
write_feature(repo, "api-seed", design_status="approved", include_roadmap=True)
write_feature(repo, "ui-seed", design_status="approved", include_roadmap=True)
write_goal_state(
roadmap,
status="ready-to-dispatch",
commit_authorization_ref=reference,
)
result = workflow_next.epic_next(roadmap)
assert result["status"] == "user_gate"
assert result["next_action"] == "authorize-epic-goal-execution"
assert reason in result["reason"]
def test_epic_goal_state_treats_named_commit_rejection_as_handoff(tmp_path: Path) -> None:
repo = init_repo(tmp_path)
roadmap = write_roadmap(repo)
write_feature(repo, "api-seed", design_status="approved", include_roadmap=True)
write_feature(repo, "ui-seed", design_status="approved", include_roadmap=True)
write_goal_state(
roadmap,
status="ready-to-dispatch",
commit_decision_status="rejected",
)
result = workflow_next.epic_next(roadmap)
assert result["status"] == "handoff"
assert result["next_action"] == "CS_ROADMAP_GOAL_HANDOFF"
def test_epic_goal_rejection_precedes_another_missing_authorization(tmp_path: Path) -> None:
repo = init_repo(tmp_path)
roadmap = write_roadmap(repo)
write_feature(repo, "api-seed", design_status="approved", include_roadmap=True)
write_feature(repo, "ui-seed", design_status="approved", include_roadmap=True)
write_goal_state(
roadmap,
status="ready-to-dispatch",
acceptance_authorization=None,
commit_authorization="rejected",
)
result = workflow_next.epic_next(roadmap)
assert result["status"] == "handoff"
assert result["next_action"] == "CS_ROADMAP_GOAL_HANDOFF"
def test_feature_standard_lane_recovers_without_goal_package_or_standalone_qa(tmp_path: Path) -> None:
repo = init_repo(tmp_path)
feature = write_feature(repo, "api-seed", design_status="approved", execution_lane="standard")
implementation = workflow_next.feature_next(feature, epic_child_batch=False)
assert implementation["status"] == "continue"
assert implementation["next_action"] == "cs-feat implementation"
assert implementation["evidence"]["execution_lane"] == "standard"
write(feature / "api-seed-checklist.yaml", "steps:\n - id: step-1\n status: done\n")
review = workflow_next.feature_next(feature, epic_child_batch=False)
assert review["status"] == "continue"
assert review["next_action"] == "cs-code-review"
write_code_review(feature, "api-seed")
acceptance = workflow_next.feature_next(feature, epic_child_batch=False)
assert acceptance["status"] == "continue"
assert acceptance["next_action"] == "cs-feat --stage accept"
write(
feature / "api-seed-acceptance.md",
"---\ndoc_type: feature-acceptance\nstatus: passed\n---\n# Acceptance\n",
)
complete = workflow_next.feature_next(feature, epic_child_batch=False)
assert complete["status"] == "complete"
assert complete["next_action"] == "CS_FEATURE_STANDARD_COMPLETE"
def test_feature_standard_lane_routes_review_fix_without_goal_state(tmp_path: Path) -> None:
repo = init_repo(tmp_path)
feature = write_feature(repo, "api-seed", design_status="approved")
write(feature / "api-seed-checklist.yaml", "steps:\n - id: step-1\n status: done\n")
write(
feature / "api-seed-review.md",
"---\ndoc_type: feature-review\nstatus: changes-requested\n---\n# Review\n",
)
result = workflow_next.feature_next(feature, epic_child_batch=False)
assert result["status"] == "continue"
assert result["next_action"] == "cs-feat implementation review-fix"
assert result["evidence"]["execution_lane"] == "standard"
@pytest.mark.parametrize("qa_status", ["failed", "blocked"])
def test_feature_standard_lane_routes_failed_qa_to_qa_fix(tmp_path: Path, qa_status: str) -> None:
repo = init_repo(tmp_path)
feature = write_feature(repo, "api-seed", design_status="approved", execution_lane="standard")
write(feature / "api-seed-checklist.yaml", "steps:\n - id: step-1\n status: done\n")
write_code_review(feature, "api-seed")
write(
feature / "api-seed-qa.md",
f"---\ndoc_type: feature-qa\nstatus: {qa_status}\n---\n# QA\n",
)
result = workflow_next.feature_next(feature, epic_child_batch=False)
assert result["status"] == "continue"
assert result["next_action"] == "cs-feat --stage impl qa-fix"
def test_feature_standard_lane_resumes_nonterminal_qa(tmp_path: Path) -> None:
repo = init_repo(tmp_path)
feature = write_feature(repo, "api-seed", design_status="approved", execution_lane="standard")
write(feature / "api-seed-checklist.yaml", "steps:\n - id: step-1\n status: done\n")
write_code_review(feature, "api-seed")
write(feature / "api-seed-qa.md", "---\ndoc_type: feature-qa\nstatus: pending\n---\n# QA\n")
result = workflow_next.feature_next(feature, epic_child_batch=False)
assert result["status"] == "continue"
assert result["next_action"] == "cs-feat --stage qa"
@pytest.mark.parametrize(
("doc_type", "reviewer"),
[
("feature-review", None),
("feature-review", "self"),
("feature-review", "ocr"),
("wrong-review", "subagent"),
],
)
def test_feature_standard_lane_rejects_untrusted_passed_review(
tmp_path: Path,
doc_type: str,
reviewer: str | None,
) -> None:
repo = init_repo(tmp_path)
feature = write_feature(repo, "api-seed", design_status="approved", execution_lane="standard")
write(feature / "api-seed-checklist.yaml", "steps:\n - id: step-1\n status: done\n")
write_code_review(feature, "api-seed", doc_type=doc_type, reviewer=reviewer)
result = workflow_next.feature_next(feature, epic_child_batch=False)
assert result["status"] == "blocked"
assert result["next_action"] == "fix-feature-code-review-evidence"
def test_feature_standard_lane_rejects_reviewer_marker_outside_frontmatter(tmp_path: Path) -> None:
repo = init_repo(tmp_path)
feature = write_feature(repo, "api-seed", design_status="approved", execution_lane="standard")
write(feature / "api-seed-checklist.yaml", "steps:\n - id: step-1\n status: done\n")
write(
feature / "api-seed-review.md",
"---\ndoc_type: feature-review\nstatus: passed\nreviewer: self\n---\n"
"# Review\n\nreviewer: subagent\n",
)
result = workflow_next.feature_next(feature, epic_child_batch=False)
assert result["status"] == "blocked"
assert result["next_action"] == "fix-feature-code-review-evidence"
def test_feature_standard_lane_honors_explicit_self_review_fallback(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
repo = init_repo(tmp_path)
feature = write_feature(repo, "api-seed", design_status="approved", execution_lane="standard")
write(feature / "api-seed-checklist.yaml", "steps:\n - id: step-1\n status: done\n")
write_code_review(feature, "api-seed", reviewer="self")
monkeypatch.setenv("CODESTABLE_ALLOW_SELF_REVIEW_FALLBACK", "1")
result = workflow_next.feature_next(feature, epic_child_batch=False)
assert result["status"] == "continue"
assert result["next_action"] == "cs-feat --stage accept"
def test_feature_execution_lane_is_normalized(tmp_path: Path) -> None:
repo = init_repo(tmp_path)
feature = write_feature(repo, "api-seed", design_status="approved", execution_lane='" Standard "')
result = workflow_next.feature_next(feature, epic_child_batch=False)
assert result["status"] == "continue"
assert result["evidence"]["execution_lane"] == "standard"
def test_feature_standard_lane_rejects_wrong_acceptance_doc_type(tmp_path: Path) -> None:
repo = init_repo(tmp_path)
feature = write_feature(repo, "api-seed", design_status="approved", execution_lane="standard")
write(feature / "api-seed-checklist.yaml", "steps:\n - id: step-1\n status: done\n")
write_code_review(feature, "api-seed")
write(
feature / "api-seed-acceptance.md",
"---\ndoc_type: wrong-acceptance\nstatus: passed\n---\n# Acceptance\n",
)
result = workflow_next.feature_next(feature, epic_child_batch=False)
assert result["status"] == "blocked"
assert result["next_action"] == "fix-feature-acceptance-evidence"
def test_feature_standard_lane_parses_checklist_once(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
repo = init_repo(tmp_path)
feature = write_feature(repo, "api-seed", design_status="approved", execution_lane="standard")
calls = 0
original = workflow_next.all_checklist_steps_done
def counted(checklist: Path | None) -> bool:
nonlocal calls
calls += 1
return original(checklist)
monkeypatch.setattr(workflow_next, "all_checklist_steps_done", counted)
workflow_next.feature_next(feature, epic_child_batch=False)
assert calls == 1
def test_feature_goal_lane_still_requires_goal_package(tmp_path: Path) -> None:
repo = init_repo(tmp_path)
feature = write_feature(repo, "api-seed", design_status="approved", execution_lane="goal")
result = workflow_next.feature_next(feature, epic_child_batch=False)
assert result["status"] == "goal_package"
assert result["next_action"] == "cs-feat goal-package"
assert result["evidence"]["execution_lane"] == "goal"
def test_feature_existing_goal_state_overrides_recorded_standard_lane(tmp_path: Path) -> None:
repo = init_repo(tmp_path)
feature = write_feature(repo, "api-seed", design_status="approved", execution_lane="standard")
write_goal_state(feature, stage="implementation", status="ready-to-dispatch")
result = workflow_next.feature_next(feature, epic_child_batch=False)
assert result["status"] == "dispatch_goal"
assert result["evidence"]["execution_lane"] == "goal"
def test_feature_unknown_execution_lane_fails_closed(tmp_path: Path) -> None:
repo = init_repo(tmp_path)
feature = write_feature(repo, "api-seed", execution_lane="turbo")
result = workflow_next.feature_next(feature, epic_child_batch=False)
assert result["status"] == "blocked"
assert result["next_action"] == "fix-feature-execution-lane"
def test_feature_review_changes_return_to_design(tmp_path: Path) -> None:
repo = init_repo(tmp_path)
feature = write_feature(repo, "changes-requested", review_status="changes-requested")
result = workflow_next.feature_next(feature, epic_child_batch=False)
assert result["status"] == "continue"
assert result["next_action"] == "cs-feat design"
assert result["reason"] == "design-review requested changes"
@pytest.mark.parametrize(
("review_state", "review_reason", "reviewer_id", "expected_status", "expected_action"),
[
("awaiting-reviewer", "", "feature-review-123", "awaiting", "resume-feature-design-reviewer"),
(
"needs-owner-approval",
"independent reviewer unavailable",
"",
"user_gate",
"approve-feature-design-review-fallback",
),
(
"reviewer-failed",
"reviewer process failed",
"",
"blocked",
"retry-feature-design-reviewer",
),
(
"blocked",
"design evidence is invalid",
"",
"blocked",
"resolve-feature-design-review-block",
),
],
)
def test_feature_design_review_state_preserves_recovery_semantics(
tmp_path: Path,
review_state: str,
review_reason: str,
reviewer_id: str,
expected_status: str,
expected_action: str,
) -> None:
repo = init_repo(tmp_path)
feature = write_feature(
repo,
review_state,
review_status="blocked",
review_state=review_state,
review_reason=review_reason,
reviewer_id=reviewer_id,
)
result = workflow_next.feature_next(feature, epic_child_batch=False)
assert workflow_next.feature_design_review_state(
feature / f"{review_state}-design-review.md"
)[0] == review_state
assert result["status"] == expected_status
assert result["next_action"] == expected_action
assert result["evidence"]["design_review_state"] == review_state
def test_feature_legacy_blocked_design_review_fails_closed(tmp_path: Path) -> None:
repo = init_repo(tmp_path)
feature = write_feature(repo, "legacy-blocked", review_status="blocked")
result = workflow_next.feature_next(feature, epic_child_batch=False)
assert result["status"] == "blocked"
assert result["next_action"] == "classify-feature-design-review-block"
assert result["evidence"]["design_review_state"] == "legacy-blocked"
def test_feature_approved_design_requires_passed_review_before_goal_state(tmp_path: Path) -> None:
repo = init_repo(tmp_path)
for review_status in ("missing", "changes-requested"):
slug = f"invalid-{review_status}"
feature = write_feature(
repo,
slug,
design_status="approved",
review_status="passed" if review_status == "missing" else review_status,
)
if review_status == "missing":
(feature / f"{slug}-design-review.md").unlink()
write_goal_state(feature, stage="complete", status="passed")
result = workflow_next.feature_next(feature, epic_child_batch=False)
assert result["status"] == "blocked"
assert result["next_action"] == "fix-feature-design-review-state"
assert result["final_answer_allowed"] is True
assert result["blocking"] == [f"approved design has design-review state: {review_status}"]
def test_feature_goal_runtime_distinguishes_dispatch_and_active_driver(tmp_path: Path) -> None:
repo = init_repo(tmp_path)
feature = write_feature(repo, "api-seed", design_status="approved")
write_goal_state(feature, stage="implementation", status="ready-to-dispatch")
dispatch = workflow_next.feature_next(feature, epic_child_batch=False)
assert dispatch["status"] == "dispatch_goal"
assert dispatch["next_action"] == "dispatch-feature-goal-driver-or-print-goal"
assert dispatch["final_answer_allowed"] is False
assert dispatch["evidence"]["acceptance_authorization_ref"] == "approval-report.md#goal-acceptance"
write_goal_state(
feature,
stage="implementation",
status="running",
driver_kind="host-agent",
driver_id="feature-run-123",
)
active = workflow_next.feature_next(feature, epic_child_batch=False)
assert active["status"] == "awaiting"
assert active["evidence"]["driver_id"] == "feature-run-123"
assert active["evidence"]["acceptance_authorization_ref"] == "approval-report.md#goal-acceptance"
def test_feature_terminal_goal_states_override_stale_driver(tmp_path: Path) -> None:
repo = init_repo(tmp_path)
feature = write_feature(repo, "api-seed", design_status="approved")
write_goal_state(
feature,
stage="complete",
status="passed",
driver_kind="host-agent",
driver_id="feature-run-123",
)
complete = workflow_next.feature_next(feature, epic_child_batch=False)
assert complete["status"] == "complete"
assert complete["next_action"] == "CS_FEATURE_GOAL_COMPLETE"
assert complete["evidence"]["acceptance_authorization_ref"] == "approval-report.md#goal-acceptance"
write_goal_state(
feature,
stage="handoff",
status="blocked",
driver_kind="host-agent",
driver_id="feature-run-123",
handoff_reason="missing production credential",
handoff_next="provide credential",
acceptance_authorization=None,
)
handoff = workflow_next.feature_next(feature, epic_child_batch=False)
assert handoff["status"] == "handoff"
assert handoff["next_action"] == "CS_FEATURE_GOAL_HANDOFF"
assert handoff["reason"] == "missing production credential"
assert handoff["evidence"]["handoff_next"] == "provide credential"
def test_epic_goal_runtime_distinguishes_dispatch_and_active_driver(tmp_path: Path) -> None:
repo = init_repo(tmp_path)
roadmap = write_roadmap(repo)
write_feature(repo, "api-seed", design_status="approved", include_roadmap=True)
write_feature(repo, "ui-seed", design_status="approved", include_roadmap=True)
write_goal_state(roadmap, status="ready-to-dispatch")
dispatch = workflow_next.epic_next(roadmap)
assert dispatch["status"] == "dispatch_goal"
assert dispatch["next_action"] == "dispatch-epic-goal-driver-or-print-goal"
assert dispatch["final_answer_allowed"] is False
assert dispatch["evidence"]["acceptance_authorization_ref"] == "approval-report.md#goal-acceptance"
assert dispatch["evidence"]["commit_authorization_ref"] == "approval-report.md#goal-commits"
write_goal_state(
roadmap,
status="ready-to-dispatch",
driver_kind="host-agent",
driver_id="epic-run-123",
)
active = workflow_next.epic_next(roadmap)
assert active["status"] == "awaiting"
assert active["evidence"]["driver_id"] == "epic-run-123"
assert active["evidence"]["acceptance_authorization_ref"] == "approval-report.md#goal-acceptance"
assert active["evidence"]["commit_authorization_ref"] == "approval-report.md#goal-commits"
def test_epic_active_roadmap_requires_review_artifact_before_goal_state(tmp_path: Path) -> None:
repo = init_repo(tmp_path)
roadmap = write_roadmap(repo)
(roadmap / "billing-system-roadmap-review.md").unlink()
write_goal_state(roadmap, status="complete")
result = workflow_next.epic_next(roadmap)
assert result["status"] == "blocked"
assert result["next_action"] == "fix-roadmap-review-state"
assert result["final_answer_allowed"] is True
assert result["blocking"] == ["active roadmap has no roadmap review artifact"]
@pytest.mark.parametrize(
("review_state", "status", "review_reason", "reviewer_id", "expected_status", "expected_action"),
[
("passed", "passed", "", "", "continue", "cs-feat design/design-review"),
(
"changes-requested",
"changes-requested",
"",
"",
"continue",
"cs-epic planning/update then review",
),
("awaiting-reviewer", "blocked", "", "review-run-123", "awaiting", "wait-roadmap-reviewer"),
(
"needs-owner-approval",
"blocked",
"independent reviewer unavailable",
"",
"user_gate",
"resolve-roadmap-review-approval",
),
(
"reviewer-failed",
"blocked",
"reviewer crashed",
"",
"blocked",
"retry-roadmap-reviewer",
),
(
"blocked",
"blocked",
"explicit reviewer config unavailable",
"",
"blocked",
"resolve-roadmap-review-block",
),
],
)
def test_epic_roadmap_review_state_routes_without_collapsing_failure_modes(
tmp_path: Path,
review_state: str,
status: str,
review_reason: str,
reviewer_id: str,
expected_status: str,
expected_action: str,
) -> None:
repo = init_repo(tmp_path)
roadmap = write_roadmap(repo)
write(
roadmap / "billing-system-roadmap-review.md",
"---\n"
"doc_type: roadmap-review\n"
f"status: {status}\n"
f"review_state: {review_state}\n"
f'review_reason: "{review_reason}"\n'
f'reviewer_id: "{reviewer_id}"\n'
"---\n# Review\n",
)
review_file = roadmap / "billing-system-roadmap-review.md"
result = workflow_next.epic_next(roadmap)
assert result["status"] == expected_status
assert result["next_action"] == expected_action
assert workflow_next.roadmap_review_state(review_file)[0] == review_state
if review_state != "passed":
assert result["evidence"]["roadmap_review_state"] == review_state
def test_epic_legacy_blocked_review_fails_closed(tmp_path: Path) -> None:
repo = init_repo(tmp_path)
roadmap = write_roadmap(repo)
write(
roadmap / "billing-system-roadmap-review.md",
"---\ndoc_type: roadmap-review\nstatus: blocked\n---\n# Legacy Review\n",
)
result = workflow_next.epic_next(roadmap)
assert result["status"] == "blocked"
assert result["next_action"] == "rerun-cs-epic-review-to-migrate-state"
assert result["evidence"]["roadmap_review_state"] == "legacy-blocked"
def test_epic_terminal_goal_states_override_stale_driver(tmp_path: Path) -> None:
repo = init_repo(tmp_path)
roadmap = write_roadmap(repo)
write_feature(repo, "api-seed", design_status="approved", include_roadmap=True)
write_feature(repo, "ui-seed", design_status="approved", include_roadmap=True)
write_goal_state(
roadmap,
status="complete",
driver_kind="host-agent",
driver_id="epic-run-123",
)
complete = workflow_next.epic_next(roadmap)
assert complete["status"] == "complete"
assert complete["next_action"] == "CS_ROADMAP_GOAL_COMPLETE"
assert complete["evidence"]["acceptance_authorization_ref"] == "approval-report.md#goal-acceptance"
assert complete["evidence"]["commit_authorization_ref"] == "approval-report.md#goal-commits"
write_goal_state(
roadmap,
status="handoff",
driver_kind="host-agent",
driver_id="epic-run-123",
handoff_reason="migration environment unavailable",
handoff_next="provision migration environment",
)
handoff = workflow_next.epic_next(roadmap)
assert handoff["status"] == "handoff"
assert handoff["next_action"] == "CS_ROADMAP_GOAL_HANDOFF"
assert handoff["reason"] == "migration environment unavailable"
assert handoff["evidence"]["handoff_next"] == "provision migration environment"
@pytest.mark.parametrize(
"artifact",
["design", "checklist", "review", "qa", "acceptance", "ff-note", "goal-state"],
)
def test_feature_corrupt_artifact_returns_structured_json_block(tmp_path: Path, artifact: str) -> None:
repo = init_repo(tmp_path)
feature = write_feature(repo, "api-seed", design_status="approved", execution_lane="standard")
corrupt = "---\ndoc_type: [unterminated\n---\n# Corrupt\n"
if artifact == "design":
target = feature / "api-seed-design.md"
write(target, corrupt)
elif artifact == "checklist":
target = feature / "api-seed-checklist.yaml"
write(target, "steps: [unterminated\n")
elif artifact == "goal-state":
target = feature / "goal-state.yaml"
write(target, "stage: [unterminated\n")
else:
target = feature / f"api-seed-{artifact}.md"
write(target, corrupt)
result = workflow_next.feature_next(feature, epic_child_batch=False)
expected_path = target.relative_to(repo).as_posix()
assert result["status"] == "blocked"
assert result["next_action"] == "fix-feature-artifact-yaml"
assert expected_path in result["blocking"][0]
assert expected_path in result["evidence"]["invalid_artifact"]
completed, payload = run_cli_json(repo, "feature", feature)
assert completed.returncode == 1
assert "Traceback" not in completed.stderr
assert payload["status"] == "blocked"
assert expected_path in str(payload["reason"])
def test_epic_corrupt_goal_state_returns_structured_json_block(tmp_path: Path) -> None:
repo = init_repo(tmp_path)
roadmap = write_roadmap(repo)
write_feature(repo, "api-seed", design_status="approved", include_roadmap=True)
write_feature(repo, "ui-seed", design_status="approved", include_roadmap=True)
target = roadmap / "goal-state.yaml"
write(target, "status: [unterminated\n")
result = workflow_next.epic_next(roadmap)
expected_path = target.relative_to(repo).as_posix()
assert result["status"] == "blocked"
assert result["next_action"] == "fix-epic-artifact-yaml"
assert expected_path in result["blocking"][0]
completed, payload = run_cli_json(repo, "epic", roadmap)
assert completed.returncode == 1
assert "Traceback" not in completed.stderr
assert payload["status"] == "blocked"
assert expected_path in str(payload["reason"])
def test_cli_accepts_json_before_or_after_subcommand(tmp_path: Path) -> None:
repo = init_repo(tmp_path)
roadmap = write_roadmap(repo)
write_feature(repo, "api-seed", include_roadmap=True)
for command in (
[
sys.executable,
(TOOLS_DIR / "codestable-workflow-next.py").as_posix(),
"--json",
"epic",
"--roadmap",
roadmap.as_posix(),
],
[
sys.executable,
(TOOLS_DIR / "codestable-workflow-next.py").as_posix(),
"epic",
"--roadmap",
roadmap.as_posix(),
"--json",
],
):
completed = subprocess.run(
command,
cwd=repo,
check=True,
text=True,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
env={"PYTHONDONTWRITEBYTECODE": "1"},
)
result = json.loads(completed.stdout)
assert result["status"] == "continue"
assert result["final_answer_allowed"] is False
Reference in New Issue View Git Blame Copy Permalink