diff --git a/.branch-cleanup-tracker.json b/.branch-cleanup-tracker.json new file mode 100644 index 000000000..59834747f --- /dev/null +++ b/.branch-cleanup-tracker.json @@ -0,0 +1,5 @@ +{ + "": { + "last_cleanup_ts": 1780826647.3286903 + } +} \ No newline at end of file diff --git a/CLAUDE.md b/CLAUDE.md index 603f93f7a..13cf2f166 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -134,6 +134,7 @@ Communication between processes happens through shared files in `instance/` with - **`usage_tracker.py`** — Budget tracking; decides autonomous mode (REVIEW/IMPLEMENT/DEEP/WAIT) based on quota percentage. Pure parser + threshold class — burn-rate-driven downgrades live in `iteration_manager._downgrade_if_burning_fast` next to the existing affordability downgrade. - **`burn_rate.py`** — Rolling burn-rate estimator (% session quota per minute). Maintains a 20-sample circular buffer in `instance/.burn-rate.json` with `fcntl.flock(LOCK_SH)` on reads, exposes `record_run()`, `burn_rate_pct_per_minute()` (total cost / span across all samples), `time_to_exhaustion(session_pct, mode=None)`, and the canonical `MODE_MULTIPLIERS` table shared with `usage_tracker.can_afford_run`. Also tracks the last-warning timestamp so the iteration manager fires at most one Telegram alert per quota cycle. - **`recover.py`** — Crash recovery for stale in-progress missions +- **`sdlc_state.py`** — SDLC workflow state persistence: `SdlcPhase` enum, `SdlcState` dataclass, workspace helpers (`get_sdlc_workspace`, `load_sdlc_state`, `save_sdlc_state`, `get_artifact_path`, `archive_sdlc_workspace`). Stores per-run state and phase artifacts under `instance/sdlc/{issue_name}/`. Foundation for the `/sdlc` orchestrator skill. - **`prompts.py`** — System prompt loader; `load_prompt()` for `koan/system-prompts/*.md`, `load_skill_prompt()` for skill-bound prompts. Supports `{@include partial-name}` directive for reusable prompt fragments from `koan/system-prompts/_partials/`. - **`skill_manager.py`** — External skill package manager: install from Git repos, update, remove, track via `instance/skills.yaml` - **`claudemd_refresh.py`** — CLAUDE.md refresh pipeline: gathers git context, invokes Claude to update/create CLAUDE.md @@ -180,6 +181,7 @@ Extensible command plugin system. Each skill lives in `skills// ` | `/scaffold`, `/new_skill` | Generate SKILL.md + handler.py for a new custom skill | | `/rtk [setup\|uninstall\|gain\|on\|off]` | — | Manage optional rtk integration for compressed tool output | | `/ideas` | — | List all ideas in the backlog | +| `/sdlc [description]` | — | Start or resume a multi-phase SDLC workflow (Research → Architecture → Planning → [approval] → Implementation → Review → Documentation → Production Ready) | +| `/approve ` | — | Approve an SDLC plan and start implementation | +| `/reject ` | — | Reject an SDLC plan and abandon the workflow | --- diff --git a/docs/users/user-manual.md b/docs/users/user-manual.md index 1da778548..2ba65b7c6 100644 --- a/docs/users/user-manual.md +++ b/docs/users/user-manual.md @@ -1860,6 +1860,23 @@ projects: - `/psecu webapp focus on token handling limit=3` — Focused review, kept off GitHub +### SDLC Orchestrator + +**`/sdlc`** — Run a full software development lifecycle for a GitHub issue. Kōan works through phases sequentially (Research → Architecture → Planning → [human approval] → Implementation → Review → Documentation → Production Ready), pausing for your sign-off before writing any code. + +- **Usage:** `/sdlc ["description"]` +- **Flags:** `--resume` (re-queue current phase), `--plan` (jump to planning), `--implement` (jump to implementation), `--review` (jump to review), `--approve` (approve plan and start implementation) +- **Approval flow:** When planning completes, Kōan posts the plan and waits. Reply `/approve ` to proceed or `/reject ` to abandon. + +
+Use cases + +- `/sdlc add-oauth2 "Add OAuth2 login support"` — Start a full SDLC workflow for a new feature +- `/sdlc add-oauth2 --resume` — Re-queue the current phase if it was skipped +- `/approve add-oauth2` — Approve the plan and start implementation +- `/reject add-oauth2` — Abandon the workflow +
+ ### Incident Triage **`/incident`** — Triage a production error from a stack trace or log snippet. Kōan will parse the error, identify the root cause, propose a fix with tests, and submit a draft PR. @@ -1997,6 +2014,9 @@ All commands at a glance. **Tier:** B = Beginner, I = Intermediate, P = Power Us | `/tech_debt [project]` | `/td`, `/debt` | P | Scan project for tech debt | | `/dead_code [project]` | `/dc` | P | Scan for unused code | | `/spec_audit [project]` | `/sa`, `/drift` | P | Audit docs/code alignment, queue fix missions | +| `/sdlc [description]` | — | P | Run full SDLC workflow (Research→Architecture→Planning→[approval]→Implementation→Review→Docs) | +| `/approve ` | — | P | Approve SDLC plan and start implementation | +| `/reject ` | — | P | Reject SDLC plan and abandon workflow | | `/incident ` | — | P | Triage a production error | | `/scaffold_skill ` | `/scaffold`, `/new_skill` | P | Generate SKILL.md + handler.py for a new custom skill | | `/rtk [setup\|uninstall\|gain\|on\|off]` | — | P | Manage optional [rtk](https://github.com/rtk-ai/rtk) integration for compressed tool output (60-90 % token savings on Bash commands). See [docs/operations/rtk.md](../operations/rtk.md). | diff --git a/koan/.branch-cleanup-tracker.json b/koan/.branch-cleanup-tracker.json index 2392610b8..e61684215 100644 --- a/koan/.branch-cleanup-tracker.json +++ b/koan/.branch-cleanup-tracker.json @@ -1,5 +1,5 @@ { "": { - "last_cleanup_ts": 1780710677.913166 + "last_cleanup_ts": 1780825787.1351092 } } \ No newline at end of file diff --git a/koan/app/command_handlers.py b/koan/app/command_handlers.py index 8b002725e..cbb547175 100644 --- a/koan/app/command_handlers.py +++ b/koan/app/command_handlers.py @@ -51,6 +51,7 @@ def set_callbacks( "help", "stop", "update", "upgrade", "sleep", "resume", "skill", "pause", "work", "awake", "start", "run", # aliases for sleep/resume "at", # one-shot scheduled mission trigger + "approve", "reject", # SDLC workflow approval gate }) @@ -145,6 +146,16 @@ def handle_command(text: str): _handle_start() return + if cmd == "/approve" or cmd.startswith("/approve "): + issue = text.strip().split(None, 1)[1].strip() if " " in text.strip() else "" + _handle_sdlc_approve(issue) + return + + if cmd == "/reject" or cmd.startswith("/reject "): + issue = text.strip().split(None, 1)[1].strip() if " " in text.strip() else "" + _handle_sdlc_reject(issue) + return + if cmd == "/help" or cmd.startswith("/help "): help_args = text.strip()[5:].strip() # everything after "/help" if help_args: @@ -581,12 +592,14 @@ def _handle_skill_approve(args: str): # Core commands that are hardcoded (not skill-based) but should appear in /help. # Each entry: (command_name, description, aliases, group) _CORE_COMMAND_HELP = [ - ("help", "Show help overview or details", ["h"], "system"), - ("stop", "Stop the run loop", [], "system"), - ("update", "Finish current mission, update, restart", ["upgrade"], "system"), - ("pause", "Pause mission processing (optional: /pause 2h)", ["sleep"], "system"), - ("resume", "Resume mission processing", ["work", "awake", "run", "start"], "system"), - ("skill", "Manage skill packages", [], "system"), + ("help", "Show help overview or details", ["h"], "system"), + ("stop", "Stop the run loop", [], "system"), + ("update", "Finish current mission, update, restart", ["upgrade"], "system"), + ("pause", "Pause mission processing (optional: /pause 2h)", ["sleep"], "system"), + ("resume", "Resume mission processing", ["work", "awake", "run", "start"], "system"), + ("skill", "Manage skill packages", [], "system"), + ("approve", "Approve an SDLC plan and start implementation (e.g. /approve my-feature)", [], "missions"), + ("reject", "Abandon an SDLC workflow awaiting approval (e.g. /reject my-feature)", [], "missions"), ] # Ordered group list (controls display order in /help) @@ -872,6 +885,110 @@ def handle_resume(): send_telegram("⚠️ Error checking quota. /status or check manually.") +def _handle_sdlc_approve(issue_name: str) -> None: + """Handle /approve — approve an SDLC plan and start implementation.""" + if not issue_name: + send_telegram("Usage: /approve ") + return + + from app.sdlc_state import SdlcPhase, load_sdlc_state, save_sdlc_state + + state = load_sdlc_state(str(INSTANCE_DIR), issue_name) + if state is None: + send_telegram(f"⚠️ No SDLC workflow found for `{issue_name}`.") + return + if state.current_phase != SdlcPhase.AWAITING_APPROVAL: + send_telegram( + f"⚠️ `{issue_name}` is not awaiting approval " + f"(current phase: {state.current_phase.value})." + ) + return + + state.approved = True + state.current_phase = SdlcPhase.IMPLEMENTATION + save_sdlc_state(str(INSTANCE_DIR), state) + + project_name = _extract_project_from_sentinel(issue_name) + _remove_sdlc_sentinel(issue_name) + _queue_sdlc_phase(issue_name, project_name) + + send_telegram(f"✅ `{issue_name}` approved — implementation queued.") + + +def _handle_sdlc_reject(issue_name: str) -> None: + """Handle /reject — abandon an SDLC workflow awaiting approval.""" + if not issue_name: + send_telegram("Usage: /reject ") + return + + from app.sdlc_state import SdlcPhase, load_sdlc_state, save_sdlc_state + from app.sdlc_state import archive_sdlc_workspace + + state = load_sdlc_state(str(INSTANCE_DIR), issue_name) + if state is None: + send_telegram(f"⚠️ No SDLC workflow found for `{issue_name}`.") + return + if state.current_phase != SdlcPhase.AWAITING_APPROVAL: + send_telegram( + f"⚠️ `{issue_name}` is not awaiting approval " + f"(current phase: {state.current_phase.value})." + ) + return + + state.current_phase = SdlcPhase.ABANDONED + save_sdlc_state(str(INSTANCE_DIR), state) + _remove_sdlc_sentinel(issue_name) + archive_sdlc_workspace(str(INSTANCE_DIR), issue_name) + send_telegram(f"🗑️ `{issue_name}` rejected and archived.") + + +def _extract_project_from_sentinel(issue_name: str) -> str: + """Extract the project name from an existing sdlc:awaiting-approval sentinel.""" + from app.utils import PROJECT_TAG_RE + + tag = f"[sdlc:awaiting-approval:{issue_name}]" + try: + content = MISSIONS_FILE.read_text(encoding="utf-8") + for line in content.splitlines(): + if tag in line: + m = PROJECT_TAG_RE.search(line) + if m: + return m.group(1) + except OSError: + pass + try: + projects = get_known_projects(str(KOAN_ROOT)) + if projects: + return projects[0] + except Exception as e: + import sys + print(f"[command_handlers] _resolve_approval_project: {e}", file=sys.stderr) + return "" + + +def _remove_sdlc_sentinel(issue_name: str) -> None: + """Remove [sdlc:awaiting-approval:{issue_name}] sentinel from missions.md.""" + tag = f"[sdlc:awaiting-approval:{issue_name}]" + try: + content = MISSIONS_FILE.read_text(encoding="utf-8") + filtered = "\n".join(ln for ln in content.splitlines() if tag not in ln) + if filtered.rstrip("\n") != content.rstrip("\n"): + atomic_write(MISSIONS_FILE, filtered + "\n") + except (OSError, FileNotFoundError): + pass + + +def _queue_sdlc_phase(issue_name: str, project_name: str) -> None: + """Queue /sdlc_phase into missions.md.""" + from app.missions import insert_mission + + missions_content = MISSIONS_FILE.read_text(encoding="utf-8") if MISSIONS_FILE.exists() else "" + project_tag = f"[project:{project_name}] " if project_name else "" + entry = f"{project_tag}/sdlc_phase {issue_name}" + updated = insert_mission(missions_content, entry, urgent=False) + atomic_write(MISSIONS_FILE, updated) + + def _handle_start(): """Start the agent loop — smart command that handles both cases. diff --git a/koan/app/iteration_manager.py b/koan/app/iteration_manager.py index db2bf8bbd..653efd7a6 100644 --- a/koan/app/iteration_manager.py +++ b/koan/app/iteration_manager.py @@ -1445,6 +1445,29 @@ def plan_iteration( else: _log_iteration("koan", "No pending mission — entering autonomous mode") + # Step 4b-sdlc: SDLC approval gate — sentinel missions must never execute. + # The human must send /approve to advance. + if mission_project and mission_title and "[sdlc:awaiting-approval:" in mission_title: + _log_iteration( + "koan", + f"SDLC awaiting approval for '{mission_title}' — skipping execution", + ) + return _make_result( + action="sdlc_wait", + project_name=mission_project, + project_path="", + mission_title=mission_title, + autonomous_mode=autonomous_mode, + focus_area="Waiting for SDLC approval — reply /approve ", + available_pct=available_pct, + decision_reason="SDLC approval sentinel — waiting for /approve", + display_lines=display_lines, + recurring_injected=recurring_injected, + focus_remaining=None, + schedule_mode=schedule_state.mode if schedule_state else "normal", + tracker_error=tracker_error, + ) + # Step 4b: Passive mode gate — block all execution # Missions stay Pending, no autonomous work. Must check before start_mission(). passive_state = _check_passive(koan_root) diff --git a/koan/app/mission_executor.py b/koan/app/mission_executor.py index 6325f4f1c..b1cda5c42 100644 --- a/koan/app/mission_executor.py +++ b/koan/app/mission_executor.py @@ -638,6 +638,10 @@ def _run_iteration( p.get("decision_reason") or "Project branch-saturated — waiting for reviews/merges", f"Branch-saturated — waiting ({time.strftime('%H:%M')})", ), + "sdlc_wait": lambda p: ( + p.get("decision_reason") or "SDLC awaiting human approval", + f"SDLC approval pending — /approve to proceed ({time.strftime('%H:%M')})", + ), } if action in _IDLE_WAIT_CONFIG: log_msg, status_msg = _IDLE_WAIT_CONFIG[action](plan) diff --git a/koan/app/sdlc_state.py b/koan/app/sdlc_state.py new file mode 100644 index 000000000..c245ddd00 --- /dev/null +++ b/koan/app/sdlc_state.py @@ -0,0 +1,252 @@ +"""Kōan — SDLC State Persistence Layer + +Manages per-workflow state for the /sdlc multi-phase orchestration skill. +Each SDLC run for a given issue lives in its own workspace under +``instance/sdlc/{issue_name}/`` with: + +- ``STATE.json`` — phase tracking, metadata, approval flag +- ``RESEARCH.md`` — research agent output +- ``ADR.md`` — architecture decision record +- ``PLAN.md`` — implementation plan (human reviews this before /approve) +- ``IMPLEMENTATION.md`` — implementation diff summary +- ``SECURITY.md`` — security review verdict +- ``QA.md`` — QA review verdict +- ``SRE.md`` — SRE review verdict +- ``REVIEW.md`` — aggregated review summary +- ``DOCS.md`` — documentation update summary + +Concurrent SDLC runs for the same issue_name will race on STATE.json. +``save_sdlc_state`` uses ``atomic_write_json`` (temp + rename + fcntl.flock) +which serializes writers at the OS level — last writer wins. + +State files accumulate indefinitely; call ``archive_sdlc_workspace()`` to +move a terminal workspace (PRODUCTION_READY or ABANDONED) to +``instance/sdlc/_archived/``. +""" + +from __future__ import annotations + +import json +import shutil +from dataclasses import dataclass, field +from datetime import datetime, timezone +from enum import Enum +from pathlib import Path +from typing import Dict, List, Optional + +from app.utils import atomic_write_json + + +# Artifact file names produced/consumed by each SDLC phase. +SDLC_ARTIFACTS = [ + "RESEARCH.md", + "ADR.md", + "PLAN.md", + "IMPLEMENTATION.md", + "SECURITY.md", + "QA.md", + "SRE.md", + "REVIEW.md", + "DOCS.md", +] + +# JSON state file inside each workspace. +_STATE_FILENAME = "STATE.json" + +# Max SDLC fix iterations before the orchestrator gives up. +MAX_FIX_ITERATIONS = 3 + + +class SdlcPhase(str, Enum): + """Ordered phases of an SDLC workflow run.""" + + RESEARCH = "research" + ARCHITECTURE = "architecture" + PLANNING = "planning" + AWAITING_APPROVAL = "awaiting_approval" + IMPLEMENTATION = "implementation" + REVIEW = "review" + FIX_LOOP = "fix_loop" + DOCUMENTATION = "documentation" + PRODUCTION_READY = "production_ready" + ABANDONED = "abandoned" + + @property + def is_terminal(self) -> bool: + return self in (SdlcPhase.PRODUCTION_READY, SdlcPhase.ABANDONED) + + +class SdlcRiskLevel(str, Enum): + """Assessed risk level for an SDLC workflow.""" + + LOW = "Low" + MEDIUM = "Medium" + HIGH = "High" + + +@dataclass +class SdlcState: + """Runtime state for one SDLC workflow run. + + Serialised to / deserialised from STATE.json inside the workspace. + """ + + issue_name: str + description: str + current_phase: SdlcPhase + risk_level: SdlcRiskLevel = SdlcRiskLevel.MEDIUM + fix_iteration: int = 0 + failing_experts: List[str] = field(default_factory=list) + approved: bool = False + started_at: str = field( + default_factory=lambda: datetime.now(timezone.utc).isoformat(timespec="seconds") + ) + artifact_checksums: Dict[str, str] = field(default_factory=dict) + + def to_dict(self) -> dict: + return { + "issue_name": self.issue_name, + "description": self.description, + "current_phase": self.current_phase.value, + "risk_level": self.risk_level.value, + "fix_iteration": self.fix_iteration, + "failing_experts": list(self.failing_experts), + "approved": self.approved, + "started_at": self.started_at, + "artifact_checksums": dict(self.artifact_checksums), + } + + @classmethod + def from_dict(cls, data: dict) -> "SdlcState": + try: + phase = SdlcPhase(data.get("current_phase", "research")) + except ValueError: + phase = SdlcPhase.RESEARCH + + try: + risk = SdlcRiskLevel(data.get("risk_level", "Medium")) + except ValueError: + risk = SdlcRiskLevel.MEDIUM + + return cls( + issue_name=data.get("issue_name", ""), + description=data.get("description", ""), + current_phase=phase, + risk_level=risk, + fix_iteration=int(data.get("fix_iteration", 0)), + failing_experts=list(data.get("failing_experts", [])), + approved=bool(data.get("approved", False)), + started_at=data.get( + "started_at", + datetime.now(timezone.utc).isoformat(timespec="seconds"), + ), + artifact_checksums=dict(data.get("artifact_checksums", {})), + ) + + +# --------------------------------------------------------------------------- +# Workspace helpers +# --------------------------------------------------------------------------- + + +def get_sdlc_workspace(instance_dir: str, issue_name: str) -> Path: + """Return the workspace directory for *issue_name*, creating it if absent.""" + ws = Path(instance_dir) / "sdlc" / _sanitise_issue_name(issue_name) + ws.mkdir(parents=True, exist_ok=True) + return ws + + +def load_sdlc_state(instance_dir: str, issue_name: str) -> Optional[SdlcState]: + """Load the SDLC state for *issue_name*. + + Returns ``None`` cleanly if the workspace is absent or STATE.json is + missing or malformed — callers should treat ``None`` as "not started". + """ + ws = Path(instance_dir) / "sdlc" / _sanitise_issue_name(issue_name) + state_file = ws / _STATE_FILENAME + if not state_file.exists(): + return None + try: + data = json.loads(state_file.read_text(encoding="utf-8")) + return SdlcState.from_dict(data) + except (json.JSONDecodeError, OSError): + return None + + +def save_sdlc_state(instance_dir: str, state: SdlcState) -> None: + """Persist *state* to STATE.json atomically. + + Creates the workspace directory if absent. Uses ``atomic_write_json`` + (temp-file + os.replace + fcntl.flock) so a crash mid-write never + leaves a partial file. + """ + ws = get_sdlc_workspace(instance_dir, state.issue_name) + state_file = ws / _STATE_FILENAME + atomic_write_json(state_file, state.to_dict(), indent=2) + + +def get_artifact_path( + instance_dir: str, issue_name: str, artifact_name: str +) -> Path: + """Return the path to *artifact_name* inside *issue_name*'s workspace. + + The artifact need not exist yet — this is a pure path computation. + *artifact_name* must be one of ``SDLC_ARTIFACTS`` (e.g. ``"PLAN.md"``). + """ + ws = Path(instance_dir) / "sdlc" / _sanitise_issue_name(issue_name) + return ws / artifact_name + + +def archive_sdlc_workspace(instance_dir: str, issue_name: str) -> Optional[Path]: + """Move a terminal workspace to ``instance/sdlc/_archived/``. + + Only moves workspaces whose ``current_phase`` is PRODUCTION_READY or + ABANDONED. Returns the destination path, or ``None`` if the workspace + was absent or not in a terminal phase. + """ + ws = Path(instance_dir) / "sdlc" / _sanitise_issue_name(issue_name) + if not ws.exists(): + return None + + state = load_sdlc_state(instance_dir, issue_name) + if state is None or not state.current_phase.is_terminal: + return None + + archived_root = Path(instance_dir) / "sdlc" / "_archived" + archived_root.mkdir(parents=True, exist_ok=True) + dest = archived_root / _sanitise_issue_name(issue_name) + + # Avoid clobbering a previous archive with the same name. + if dest.exists(): + ts = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%S") + dest = archived_root / f"{_sanitise_issue_name(issue_name)}-{ts}" + + shutil.move(str(ws), str(dest)) + return dest + + +def list_sdlc_workspaces(instance_dir: str) -> List[str]: + """Return issue names for all active (non-archived) SDLC workspaces.""" + sdlc_root = Path(instance_dir) / "sdlc" + if not sdlc_root.exists(): + return [] + return [ + d.name + for d in sorted(sdlc_root.iterdir()) + if d.is_dir() and d.name != "_archived" + ] + + +# --------------------------------------------------------------------------- +# Internal helpers +# --------------------------------------------------------------------------- + + +def _sanitise_issue_name(issue_name: str) -> str: + """Normalise *issue_name* for use as a directory component. + + Replaces characters unsafe in file paths with underscores and strips + leading/trailing whitespace, dots, and underscores. + """ + safe = "".join(c if c.isalnum() or c in "-_." else "_" for c in issue_name.strip()) + return safe.strip("._") or "unnamed" diff --git a/koan/app/skill_dispatch.py b/koan/app/skill_dispatch.py index a90402454..47c784c6b 100644 --- a/koan/app/skill_dispatch.py +++ b/koan/app/skill_dispatch.py @@ -84,6 +84,7 @@ def _get_skills_dir_mtime(instance_dir: Path) -> float: "spec_audit": "skills.core.spec_audit.spec_audit_runner", "explain": "skills.core.explain.explain_runner", "deep": "skills.core.deep.deep_runner", + "sdlc_phase": "skills.core.sdlc.sdlc_phase_runner", } # Alias -> canonical command name. Declared once, expanded into @@ -400,6 +401,9 @@ def build_skill_command( ), "deep": lambda: _build_ai_cmd(base_cmd, args, project_name, project_path, instance_dir), "explain": lambda: _build_explain_cmd(base_cmd, args, project_path, project_name), + "sdlc_phase": lambda: _build_sdlc_phase_cmd( + base_cmd, args, project_name, project_path, instance_dir, + ), } def _audit_builder(): return _build_audit_cmd( @@ -949,6 +953,30 @@ def _build_generic_runner_cmd( return cmd +def _build_sdlc_phase_cmd( + base_cmd: List[str], + args: str, + project_name: str, + project_path: str, + instance_dir: str, +) -> Optional[List[str]]: + """Build sdlc_phase_runner command. + + The issue_name is the first token in *args*. The runner reads STATE.json + to determine the current phase and executes the appropriate prompt. + """ + cleaned = strip_all_lifecycle_markers(args).strip() + issue_name = cleaned.split()[0] if cleaned.split() else "" + if not issue_name: + return None + return base_cmd + [ + "--issue-name", issue_name, + "--project-path", project_path, + "--project-name", project_name, + "--instance-dir", instance_dir, + ] + + def cleanup_skill_temp_files(skill_cmd: List[str]) -> None: """Remove temp files created by skill command builders. @@ -1012,6 +1040,10 @@ def validate_skill_args(command: str, args: str) -> Optional[str]: f"/{command} requires a GitHub PR or issue URL " f"(e.g. https://github.com/owner/repo/pull/42)" ) + elif canonical == "sdlc_phase": + cleaned = strip_all_lifecycle_markers(args).strip() + if not cleaned.split(): + return "/sdlc_phase requires an issue name (e.g. /sdlc_phase my-feature)" return None diff --git a/koan/skills/core/sdlc/handler.py b/koan/skills/core/sdlc/handler.py index da5dfc432..93d1aada0 100644 --- a/koan/skills/core/sdlc/handler.py +++ b/koan/skills/core/sdlc/handler.py @@ -1,15 +1,200 @@ -"""SDLC skill handler stub. +"""SDLC skill handler. -Full implementation in #1707 (orchestrator with phase routing). -This stub ensures the skill registry and tests pass while the feature is -developed across multiple PRs (#1704 state layer, #1705 prompts, #1707 orchestrator). +Handles /sdlc commands from Telegram and the agent bridge. +Creates or resumes a multi-phase SDLC workflow for a GitHub issue. + +Usage: + /sdlc [description] + /sdlc --resume + /sdlc --plan + /sdlc --implement + /sdlc --review + /sdlc --approve (alias for /approve) """ +from __future__ import annotations + +import re +import shlex +from pathlib import Path +from typing import Optional + +from app.sdlc_state import ( + SdlcPhase, + SdlcState, + load_sdlc_state, + save_sdlc_state, +) from app.skills import SkillContext +# Flags that jump to a specific phase (if not already past it) +_PHASE_FLAGS = { + "--plan": SdlcPhase.PLANNING, + "--implement": SdlcPhase.IMPLEMENTATION, + "--review": SdlcPhase.REVIEW, +} + +# Phases that can be re-entered via jump flags +_JUMPABLE_FROM = { + SdlcPhase.PLANNING: {SdlcPhase.RESEARCH, SdlcPhase.ARCHITECTURE}, + SdlcPhase.IMPLEMENTATION: { + SdlcPhase.RESEARCH, SdlcPhase.ARCHITECTURE, + SdlcPhase.PLANNING, SdlcPhase.AWAITING_APPROVAL, + }, + SdlcPhase.REVIEW: { + SdlcPhase.RESEARCH, SdlcPhase.ARCHITECTURE, + SdlcPhase.PLANNING, SdlcPhase.AWAITING_APPROVAL, + SdlcPhase.IMPLEMENTATION, + }, +} + def handle(ctx: SkillContext) -> str: + args = (ctx.args or "").strip() + + resume = "--resume" in args + approve = "--approve" in args + args = re.sub(r"--resume\b", "", args) + args = re.sub(r"--approve\b", "", args) + + jump_phase: Optional[SdlcPhase] = None + for flag, phase in _PHASE_FLAGS.items(): + if flag in args: + jump_phase = phase + args = re.sub(re.escape(flag), "", args) + + args = args.strip() + + try: + parts = shlex.split(args) + except ValueError: + parts = args.split(None, 1) + + if not parts: + return ( + "Usage: /sdlc [description]\n" + "Example: /sdlc add-oauth2 \"Add OAuth2 login\"\n" + "Flags: --resume | --plan | --implement | --review | --approve" + ) + + issue_name = parts[0] + description = parts[1] if len(parts) > 1 else "" + instance_dir = str(ctx.instance_dir) + project_name = _get_project_name(ctx) + + if approve: + return _handle_approve(instance_dir, issue_name, project_name, ctx.instance_dir) + + state = load_sdlc_state(instance_dir, issue_name) + + if state is None: + if resume: + return f"⚠️ No existing SDLC workflow found for `{issue_name}`. Start without --resume." + state = SdlcState( + issue_name=issue_name, + description=description or f"SDLC workflow for {issue_name}", + current_phase=SdlcPhase.RESEARCH, + ) + save_sdlc_state(instance_dir, state) + _queue_phase_mission(ctx.instance_dir, project_name, issue_name) + return f"🚀 SDLC started for `{issue_name}` — research phase queued." + + if jump_phase is not None: + allowed = _JUMPABLE_FROM.get(jump_phase, set()) + if state.current_phase not in allowed: + return ( + f"⚠️ Cannot jump to {jump_phase.value} — " + f"current phase is {state.current_phase.value}. " + "Use --resume to continue from the current phase." + ) + state.current_phase = jump_phase + if description: + state.description = description + save_sdlc_state(instance_dir, state) + _queue_phase_mission(ctx.instance_dir, project_name, issue_name) + return f"⏭️ `{issue_name}` — jumping to {jump_phase.value}." + + if state.current_phase == SdlcPhase.AWAITING_APPROVAL: + ws = Path(instance_dir) / "sdlc" / issue_name + plan_note = f"\nPlan at: `{ws / 'PLAN.md'}`" if (ws / "PLAN.md").exists() else "" + return ( + f"⏸️ `{issue_name}` awaiting approval.{plan_note}\n" + "Reply /approve to proceed, or /reject to abandon." + ) + + if state.current_phase.is_terminal: + return ( + f"✅ `{issue_name}` already finished: {state.current_phase.value}. " + "Start a new workflow with a different issue name." + ) + + if resume or description: + if description: + state.description = description + save_sdlc_state(instance_dir, state) + _queue_phase_mission(ctx.instance_dir, project_name, issue_name) + return f"▶️ Resuming `{issue_name}` from {state.current_phase.value}." + return ( - "🚧 /sdlc is coming soon — the prompt corpus is ready (#1705) " - "and the orchestrator is tracked in #1707." + f"ℹ️ `{issue_name}` in progress (phase: {state.current_phase.value}). " + "Use --resume to re-queue the current phase." ) + + +def _get_project_name(ctx: SkillContext) -> str: + """Resolve the project name for mission queueing.""" + try: + from app.project_explorer import get_projects + projects = get_projects() + if projects: + return projects[0][0] + except Exception: + pass + return "" + + +def _queue_phase_mission(instance_dir: Path, project_name: str, issue_name: str) -> None: + """Insert /sdlc_phase into missions.md.""" + from app.missions import insert_mission + from app.utils import atomic_write + + missions_path = instance_dir / "missions.md" + content = missions_path.read_text(encoding="utf-8") if missions_path.exists() else "" + project_tag = f"[project:{project_name}] " if project_name else "" + entry = f"{project_tag}/sdlc_phase {issue_name}" + updated = insert_mission(content, entry, urgent=False) + atomic_write(missions_path, updated) + + +def _handle_approve( + instance_dir: str, issue_name: str, project_name: str, instance_path: Path +) -> str: + """Handle --approve flag (shortcut for /approve command).""" + state = load_sdlc_state(instance_dir, issue_name) + if state is None: + return f"⚠️ No SDLC workflow found for `{issue_name}`." + if state.current_phase != SdlcPhase.AWAITING_APPROVAL: + return ( + f"⚠️ `{issue_name}` is not awaiting approval " + f"(phase: {state.current_phase.value})." + ) + state.approved = True + state.current_phase = SdlcPhase.IMPLEMENTATION + save_sdlc_state(instance_dir, state) + _remove_approval_sentinel(instance_path, issue_name) + _queue_phase_mission(instance_path, project_name, issue_name) + return f"✅ `{issue_name}` approved — implementation queued." + + +def _remove_approval_sentinel(instance_dir: Path, issue_name: str) -> None: + """Remove the sdlc:awaiting-approval sentinel from missions.md.""" + from app.utils import atomic_write + + missions_path = instance_dir / "missions.md" + if not missions_path.exists(): + return + content = missions_path.read_text(encoding="utf-8") + tag = f"[sdlc:awaiting-approval:{issue_name}]" + filtered = "\n".join(ln for ln in content.splitlines() if tag not in ln) + if filtered.rstrip("\n") != content.rstrip("\n"): + atomic_write(missions_path, filtered + "\n") diff --git a/koan/skills/core/sdlc/sdlc_phase_runner.py b/koan/skills/core/sdlc/sdlc_phase_runner.py new file mode 100644 index 000000000..4d5c324d2 --- /dev/null +++ b/koan/skills/core/sdlc/sdlc_phase_runner.py @@ -0,0 +1,490 @@ +"""Kōan — SDLC phase runner. + +Handles one SDLC phase per invocation. Reads STATE.json to determine +which phase to run, executes the phase-specific prompt via Claude CLI, +writes the output artifact, advances state, and queues the next phase. + +CLI: + python3 -m skills.core.sdlc.sdlc_phase_runner \\ + --issue-name \\ + --project-path \\ + --project-name \\ + --instance-dir +""" + +from __future__ import annotations + +import re +import sys +from pathlib import Path +from typing import Dict, List, Optional, Tuple + +from app.sdlc_state import ( + MAX_FIX_ITERATIONS, + SdlcPhase, + archive_sdlc_workspace, + get_sdlc_workspace, + load_sdlc_state, + save_sdlc_state, +) + +# Maximum characters of prior artifacts to inject as context. +_CONTEXT_BUDGET = 80_000 + +# --- Phase → prompt file mapping --- +_PHASE_PROMPTS: Dict[SdlcPhase, str] = { + SdlcPhase.RESEARCH: "research", + SdlcPhase.ARCHITECTURE: "architecture", + SdlcPhase.PLANNING: "planning", + SdlcPhase.IMPLEMENTATION: "implementation", + SdlcPhase.FIX_LOOP: "fix", + SdlcPhase.DOCUMENTATION: "tech_writer", +} + +# Reviewer sub-phases and their prompt/artifact pairs +_REVIEWERS: List[Tuple[str, str]] = [ + ("security_review", "SECURITY.md"), + ("qa_review", "QA.md"), + ("sre_review", "SRE.md"), +] + +# Phase → output artifact +_PHASE_ARTIFACTS: Dict[SdlcPhase, str] = { + SdlcPhase.RESEARCH: "RESEARCH.md", + SdlcPhase.ARCHITECTURE: "ADR.md", + SdlcPhase.PLANNING: "PLAN.md", + SdlcPhase.IMPLEMENTATION: "IMPLEMENTATION.md", + SdlcPhase.DOCUMENTATION: "DOCS.md", +} + +# Prior artifacts to inject as context per phase. +# FIX_LOOP is handled dynamically based on failing_experts — not listed here. +_PHASE_CONTEXT: Dict[SdlcPhase, List[str]] = { + SdlcPhase.ARCHITECTURE: ["RESEARCH.md"], + SdlcPhase.PLANNING: ["RESEARCH.md", "ADR.md"], + SdlcPhase.IMPLEMENTATION: ["PLAN.md", "RESEARCH.md"], + SdlcPhase.REVIEW: ["IMPLEMENTATION.md"], + SdlcPhase.DOCUMENTATION: ["IMPLEMENTATION.md", "PLAN.md"], +} + +# Mapping from failing_experts entry → review artifact filename +_EXPERT_ARTIFACTS: Dict[str, str] = { + "security": "SECURITY.md", + "qa": "QA.md", + "sre": "SRE.md", +} + +# Linear phase progression for non-branching phases +_NEXT_PHASE: Dict[SdlcPhase, SdlcPhase] = { + SdlcPhase.RESEARCH: SdlcPhase.ARCHITECTURE, + SdlcPhase.ARCHITECTURE: SdlcPhase.PLANNING, + SdlcPhase.PLANNING: SdlcPhase.AWAITING_APPROVAL, + SdlcPhase.IMPLEMENTATION: SdlcPhase.REVIEW, + SdlcPhase.FIX_LOOP: SdlcPhase.REVIEW, + SdlcPhase.DOCUMENTATION: SdlcPhase.PRODUCTION_READY, +} + +# Phase emojis for Telegram notifications +_PHASE_EMOJI: Dict[SdlcPhase, str] = { + SdlcPhase.RESEARCH: "🔍", + SdlcPhase.ARCHITECTURE: "🏗️", + SdlcPhase.PLANNING: "📋", + SdlcPhase.IMPLEMENTATION: "⚙️", + SdlcPhase.REVIEW: "🔎", + SdlcPhase.FIX_LOOP: "🔧", + SdlcPhase.DOCUMENTATION: "📝", + SdlcPhase.PRODUCTION_READY: "✅", +} + + +def run_sdlc_phase( + issue_name: str, + project_path: str, + project_name: str, + instance_dir: str, +) -> int: + """Run one SDLC phase for *issue_name*. + + Returns: + 0 on success, 1 on unrecoverable error. + """ + state = load_sdlc_state(instance_dir, issue_name) + if state is None: + print( + f"[sdlc] ERROR: No STATE.json for '{issue_name}'. " + "Start with /sdlc first.", + file=sys.stderr, + ) + return 1 + + phase = state.current_phase + + print( + f"[sdlc] {issue_name} — phase: {phase.value}", + flush=True, + ) + + if phase == SdlcPhase.AWAITING_APPROVAL: + print("[sdlc] Workflow paused — awaiting human approval.", flush=True) + _notify(instance_dir, f"⏸️ [{issue_name}] Awaiting approval — reply /approve {issue_name}") + return 0 + + if phase.is_terminal: + print(f"[sdlc] Phase is terminal ({phase.value}) — nothing to do.", flush=True) + return 0 + + skill_dir = Path(__file__).resolve().parent + ws = get_sdlc_workspace(instance_dir, issue_name) + failing = state.failing_experts if phase == SdlcPhase.FIX_LOOP else [] + context = _build_context(ws, phase, failing_experts=failing) + + if phase == SdlcPhase.REVIEW: + exit_code = _run_review_phase( + issue_name, project_path, project_name, instance_dir, ws, state, skill_dir, context, + ) + else: + exit_code = _run_single_phase( + phase, issue_name, project_path, project_name, + instance_dir, ws, state, skill_dir, context, + ) + + return exit_code + + +def _run_single_phase( + phase: SdlcPhase, + issue_name: str, + project_path: str, + project_name: str, + instance_dir: str, + ws: Path, + state, + skill_dir: Path, + context: str, +) -> int: + from app.cli_provider import run_command_streaming + from app.config import get_mission_timeout, get_skill_max_turns + from app.prompts import load_skill_prompt + + prompt_name = _PHASE_PROMPTS.get(phase) + if prompt_name is None: + print(f"[sdlc] ERROR: No prompt for phase {phase.value}", file=sys.stderr) + return 1 + + prompt = load_skill_prompt( + skill_dir, + prompt_name, + ISSUE_NAME=issue_name, + ISSUE_DESCRIPTION=state.description, + WORKSPACE_PATH=str(ws), + PROJECT_ROOT=project_path, + INSTANCE_DIR=instance_dir, + PROJECT_NAME=project_name, + ) + + if context: + prompt = f"{prompt}\n\n---\n## Prior Phase Artifacts\n\n{context}" + + emoji = _PHASE_EMOJI.get(phase, "▶️") + _notify(instance_dir, f"{emoji} [{issue_name}] Running {phase.value} phase...") + + # Phase timeout: implementation gets 45 min, others 20 min + phase_timeout = get_mission_timeout() if phase == SdlcPhase.IMPLEMENTATION else 1200 + max_turns = get_skill_max_turns() if phase == SdlcPhase.IMPLEMENTATION else 100 + + try: + output = run_command_streaming( + prompt=prompt, + project_path=project_path, + allowed_tools=["Read", "Write", "Edit", "Bash", "Glob", "Grep"], + model_key="mission", + max_turns=max_turns, + timeout=phase_timeout, + ) + except RuntimeError as exc: + print(f"[sdlc] Phase failed: {exc}", file=sys.stderr) + _notify(instance_dir, f"❌ [{issue_name}] {phase.value} phase failed.") + return 1 + + artifact_name = _PHASE_ARTIFACTS.get(phase) + if artifact_name: + artifact_path = ws / artifact_name + if not artifact_path.exists() or artifact_path.stat().st_size == 0: + print( + f"[sdlc] WARNING: Expected artifact {artifact_name} not found after {phase.value}", + flush=True, + ) + + _advance_phase(issue_name, project_name, instance_dir, ws, state, phase) + return 0 + + +def _run_review_phase( + issue_name: str, + project_path: str, + project_name: str, + instance_dir: str, + ws: Path, + state, + skill_dir: Path, + context: str, +) -> int: + from app.cli_provider import run_command_streaming + from app.config import get_skill_max_turns + from app.prompts import load_skill_prompt + + _notify(instance_dir, f"🔎 [{issue_name}] Starting parallel review (security + QA + SRE)...") + + all_approved = True + for prompt_name, artifact_name in _REVIEWERS: + artifact_path = ws / artifact_name + print(f"[sdlc] Running {prompt_name}...", flush=True) + + prompt = load_skill_prompt( + skill_dir, + prompt_name, + ISSUE_NAME=issue_name, + WORKSPACE_PATH=str(ws), + PROJECT_ROOT=project_path, + INSTANCE_DIR=instance_dir, + PROJECT_NAME=project_name, + ) + if context: + prompt = f"{prompt}\n\n---\n## Implementation Summary\n\n{context}" + + try: + run_command_streaming( + prompt=prompt, + project_path=project_path, + allowed_tools=["Read", "Write", "Edit", "Bash", "Glob", "Grep"], + model_key="mission", + max_turns=get_skill_max_turns(), + timeout=1200, + ) + except RuntimeError as exc: + print(f"[sdlc] {prompt_name} failed: {exc}", file=sys.stderr) + _notify(instance_dir, f"❌ [{issue_name}] {prompt_name} failed.") + return 1 + + if artifact_path.exists(): + content = artifact_path.read_text(encoding="utf-8") + if "VERDICT: NEEDS_FIX" in content: + all_approved = False + + failing = _parse_failing_experts(ws) + state = load_sdlc_state(instance_dir, issue_name) + if state is None: + return 1 + + if all_approved: + state.current_phase = SdlcPhase.DOCUMENTATION + state.failing_experts = [] + save_sdlc_state(instance_dir, state) + _notify(instance_dir, f"✅ [{issue_name}] All reviews passed — queuing documentation") + elif state.fix_iteration >= MAX_FIX_ITERATIONS: + state.current_phase = SdlcPhase.ABANDONED + save_sdlc_state(instance_dir, state) + _notify( + instance_dir, + f"🚨 [{issue_name}] Fix loop capped at {MAX_FIX_ITERATIONS} — " + "manual review required." + ) + archive_sdlc_workspace(instance_dir, issue_name) + return 1 + else: + state.current_phase = SdlcPhase.FIX_LOOP + state.failing_experts = failing + save_sdlc_state(instance_dir, state) + _notify( + instance_dir, + f"🔧 [{issue_name}] Review done — {len(failing)} expert(s) need fixes. " + f"Fix loop iteration {state.fix_iteration + 1}/{MAX_FIX_ITERATIONS}" + ) + + _queue_next_phase(issue_name, project_name, instance_dir, state.current_phase) + return 0 + + +def _advance_phase( + issue_name: str, + project_name: str, + instance_dir: str, + ws: Path, + state, + phase: SdlcPhase, +) -> None: + next_phase = _NEXT_PHASE.get(phase) + if next_phase is None: + return + + state = load_sdlc_state(instance_dir, issue_name) + if state is None: + return + + state.current_phase = next_phase + save_sdlc_state(instance_dir, state) + + if next_phase == SdlcPhase.AWAITING_APPROVAL: + _queue_approval_sentinel(issue_name, project_name, instance_dir, ws) + plan_path = ws / "PLAN.md" + plan_snippet = "" + if plan_path.exists(): + plan_text = plan_path.read_text(encoding="utf-8") + plan_snippet = plan_text[:1200] + if len(plan_text) > 1200: + plan_snippet += "\n[...truncated...]" + _notify( + instance_dir, + f"⏸️ [{issue_name}] Plan ready for review.\n\n" + f"{plan_snippet}\n\n" + f"Reply /approve {issue_name} to proceed with implementation." + ) + return + + if next_phase == SdlcPhase.PRODUCTION_READY: + archive_sdlc_workspace(instance_dir, issue_name) + _notify(instance_dir, f"✅ [{issue_name}] SDLC workflow complete!") + return + + emoji = _PHASE_EMOJI.get(next_phase, "▶️") + _notify(instance_dir, f"{emoji} [{issue_name}] Advancing to {next_phase.value}...") + _queue_next_phase(issue_name, project_name, instance_dir, next_phase) + + +def _queue_approval_sentinel( + issue_name: str, project_name: str, instance_dir: str, ws: Path +) -> None: + """Insert the sdlc:awaiting-approval sentinel into missions.md.""" + from app.missions import insert_mission + from app.utils import atomic_write + + missions_path = Path(instance_dir) / "missions.md" + content = missions_path.read_text(encoding="utf-8") if missions_path.exists() else "" + project_tag = f"[project:{project_name}] " if project_name else "" + entry = ( + f"{project_tag}" + f"[sdlc:awaiting-approval:{issue_name}] " + f"SDLC approval needed for {issue_name}" + ) + updated = insert_mission(content, entry, urgent=True) + atomic_write(missions_path, updated) + + +def _queue_next_phase( + issue_name: str, project_name: str, instance_dir: str, phase: SdlcPhase +) -> None: + """Insert the next /sdlc_phase mission into missions.md.""" + if phase.is_terminal or phase == SdlcPhase.AWAITING_APPROVAL: + return + + from app.missions import insert_mission + from app.utils import atomic_write + + missions_path = Path(instance_dir) / "missions.md" + content = missions_path.read_text(encoding="utf-8") if missions_path.exists() else "" + project_tag = f"[project:{project_name}] " if project_name else "" + entry = f"{project_tag}/sdlc_phase {issue_name}" + updated = insert_mission(content, entry, urgent=False) + atomic_write(missions_path, updated) + + +def _build_context( + ws: Path, phase: SdlcPhase, *, failing_experts: Optional[List[str]] = None +) -> str: + """Build a context string from prior phase artifacts for injection. + + For FIX_LOOP, only injects the failing experts' review artifacts plus PLAN.md + — avoids overwhelming the fix agent with reviews it doesn't need to address. + """ + if phase == SdlcPhase.FIX_LOOP: + failing = failing_experts or [] + artifact_names = ["PLAN.md"] + [ + _EXPERT_ARTIFACTS[e] for e in failing if e in _EXPERT_ARTIFACTS + ] + else: + artifact_names = _PHASE_CONTEXT.get(phase, []) + + if not artifact_names: + return "" + + parts = [] + budget = _CONTEXT_BUDGET + + for name in artifact_names: + path = ws / name + if not path.exists(): + continue + try: + content = path.read_text(encoding="utf-8") + except OSError: + continue + if not content.strip(): + continue + + section = f"### {name}\n\n{content}" + if len(section) > budget: + truncated = budget - 30 + section = section[:truncated] + "\n[...truncated...]" + parts.append(section) + break + parts.append(section) + budget -= len(section) + if budget <= 0: + break + + return "\n\n".join(parts) + + +def _parse_failing_experts(ws: Path) -> List[str]: + """Return list of expert names whose review verdict is NEEDS_FIX.""" + failing = [] + for _, artifact_name in _REVIEWERS: + path = ws / artifact_name + if not path.exists(): + continue + try: + content = path.read_text(encoding="utf-8") + except OSError: + continue + if "VERDICT: NEEDS_FIX" in content: + failing.append(artifact_name.replace(".md", "").lower()) + return failing + + +def _notify(instance_dir: str, message: str) -> None: + """Append a message to outbox.md for Telegram delivery.""" + outbox = Path(instance_dir) / "outbox.md" + try: + with outbox.open("a", encoding="utf-8") as f: + f.write(f"- {message}\n") + except OSError: + pass + + +# --------------------------------------------------------------------------- +# CLI entry point +# --------------------------------------------------------------------------- + + +def main(argv=None): + import argparse + + p = argparse.ArgumentParser(description="Run one SDLC phase") + p.add_argument("--issue-name", required=True) + p.add_argument("--project-path", required=True) + p.add_argument("--project-name", default="") + p.add_argument("--instance-dir", required=True) + args = p.parse_args(argv) + + sys.exit( + run_sdlc_phase( + issue_name=args.issue_name, + project_path=args.project_path, + project_name=args.project_name, + instance_dir=args.instance_dir, + ) + ) + + +if __name__ == "__main__": + main() diff --git a/koan/tests/test_sdlc_orchestrator.py b/koan/tests/test_sdlc_orchestrator.py new file mode 100644 index 000000000..cd92fd599 --- /dev/null +++ b/koan/tests/test_sdlc_orchestrator.py @@ -0,0 +1,589 @@ +"""Tests for SDLC orchestrator: handler.py, sdlc_phase_runner.py, skill_dispatch, command_handlers.""" + +from __future__ import annotations + +import os +from pathlib import Path +from typing import Optional +from unittest.mock import MagicMock, patch + +import pytest + +from app.sdlc_state import ( + SdlcPhase, + SdlcState, + load_sdlc_state, + save_sdlc_state, +) + + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + + +@pytest.fixture() +def instance(tmp_path): + """Create a minimal instance directory.""" + inst = tmp_path / "instance" + inst.mkdir() + (inst / "missions.md").write_text( + "# Missions\n\n## Pending\n\n## In Progress\n\n## Done\n", + encoding="utf-8", + ) + (inst / "outbox.md").write_text("", encoding="utf-8") + return inst + + +@pytest.fixture() +def skill_ctx(tmp_path, instance): + """Build a minimal SkillContext for the SDLC handler.""" + from app.skills import SkillContext + + ctx = SkillContext( + koan_root=tmp_path, + instance_dir=instance, + args="", + ) + return ctx + + +# --------------------------------------------------------------------------- +# Handler +# --------------------------------------------------------------------------- + + +class TestSdlcHandler: + def test_new_workflow_creates_state_and_queues_mission(self, skill_ctx, instance): + from skills.core.sdlc.handler import handle + + skill_ctx.args = "my-feature \"Add OAuth2 login\"" + with patch("skills.core.sdlc.handler._get_project_name", return_value="myproject"): + result = handle(skill_ctx) + + assert "started" in result.lower() or "queued" in result.lower() + state = load_sdlc_state(str(instance), "my-feature") + assert state is not None + assert state.current_phase == SdlcPhase.RESEARCH + assert "Add OAuth2 login" in state.description + + missions = (instance / "missions.md").read_text(encoding="utf-8") + assert "/sdlc_phase my-feature" in missions + + def test_no_args_returns_usage(self, skill_ctx, instance): + from skills.core.sdlc.handler import handle + + skill_ctx.args = "" + with patch("skills.core.sdlc.handler._get_project_name", return_value="myproject"): + result = handle(skill_ctx) + assert "Usage" in result or "usage" in result + + def test_resume_nonexistent_returns_error(self, skill_ctx, instance): + from skills.core.sdlc.handler import handle + + skill_ctx.args = "no-such-thing --resume" + with patch("skills.core.sdlc.handler._get_project_name", return_value="myproject"): + result = handle(skill_ctx) + assert "No existing" in result + + def test_awaiting_approval_tells_human(self, skill_ctx, instance): + from skills.core.sdlc.handler import handle + + state = SdlcState( + issue_name="approval-test", + description="needs approval", + current_phase=SdlcPhase.AWAITING_APPROVAL, + ) + save_sdlc_state(str(instance), state) + + skill_ctx.args = "approval-test" + with patch("skills.core.sdlc.handler._get_project_name", return_value="myproject"): + result = handle(skill_ctx) + assert "awaiting" in result.lower() or "approval" in result.lower() + + def test_terminal_phase_returns_info(self, skill_ctx, instance): + from skills.core.sdlc.handler import handle + + state = SdlcState( + issue_name="done-feature", + description="done", + current_phase=SdlcPhase.PRODUCTION_READY, + ) + save_sdlc_state(str(instance), state) + + skill_ctx.args = "done-feature" + with patch("skills.core.sdlc.handler._get_project_name", return_value="myproject"): + result = handle(skill_ctx) + assert "terminal" in result.lower() or "finished" in result.lower() or "already" in result.lower() + + def test_approve_flag_advances_to_implementation(self, skill_ctx, instance): + from skills.core.sdlc.handler import handle + + state = SdlcState( + issue_name="approve-test", + description="test", + current_phase=SdlcPhase.AWAITING_APPROVAL, + ) + save_sdlc_state(str(instance), state) + # Add sentinel to missions.md + existing = (instance / "missions.md").read_text(encoding="utf-8") + tag = "[sdlc:awaiting-approval:approve-test]" + pending_header = "## Pending\n" + updated = existing.replace( + pending_header, + f"{pending_header}\n- [project:myproject] {tag} approval needed\n", + ) + (instance / "missions.md").write_text(updated, encoding="utf-8") + + skill_ctx.args = "approve-test --approve" + with patch("skills.core.sdlc.handler._get_project_name", return_value="myproject"): + result = handle(skill_ctx) + + assert "approved" in result.lower() + new_state = load_sdlc_state(str(instance), "approve-test") + assert new_state.current_phase == SdlcPhase.IMPLEMENTATION + assert new_state.approved is True + + missions = (instance / "missions.md").read_text(encoding="utf-8") + assert tag not in missions + assert "/sdlc_phase approve-test" in missions + + def test_already_running_returns_info(self, skill_ctx, instance): + from skills.core.sdlc.handler import handle + + state = SdlcState( + issue_name="running-feature", + description="in progress", + current_phase=SdlcPhase.ARCHITECTURE, + ) + save_sdlc_state(str(instance), state) + + skill_ctx.args = "running-feature" + with patch("skills.core.sdlc.handler._get_project_name", return_value="myproject"): + result = handle(skill_ctx) + assert "in progress" in result.lower() or "architecture" in result.lower() + + def test_resume_requeues_current_phase(self, skill_ctx, instance): + from skills.core.sdlc.handler import handle + + state = SdlcState( + issue_name="resume-feature", + description="needs resume", + current_phase=SdlcPhase.PLANNING, + ) + save_sdlc_state(str(instance), state) + + skill_ctx.args = "resume-feature --resume" + with patch("skills.core.sdlc.handler._get_project_name", return_value="myproject"): + result = handle(skill_ctx) + + assert "resum" in result.lower() + missions = (instance / "missions.md").read_text(encoding="utf-8") + assert "/sdlc_phase resume-feature" in missions + + +# --------------------------------------------------------------------------- +# skill_dispatch integration +# --------------------------------------------------------------------------- + + +class TestSdlcPhaseSkillDispatch: + def test_sdlc_phase_in_canonical_runners(self): + from app.skill_dispatch import _CANONICAL_RUNNERS + + assert "sdlc_phase" in _CANONICAL_RUNNERS + assert "sdlc_phase_runner" in _CANONICAL_RUNNERS["sdlc_phase"] + + def test_sdlc_phase_in_skill_runners(self): + from app.skill_dispatch import _SKILL_RUNNERS + + assert "sdlc_phase" in _SKILL_RUNNERS + + def test_validate_sdlc_phase_no_args_returns_error(self): + from app.skill_dispatch import validate_skill_args + + err = validate_skill_args("sdlc_phase", "") + assert err is not None + assert "issue name" in err.lower() or "sdlc_phase" in err.lower() + + def test_validate_sdlc_phase_with_args_passes(self): + from app.skill_dispatch import validate_skill_args + + err = validate_skill_args("sdlc_phase", "my-feature") + assert err is None + + def test_build_sdlc_phase_cmd_contains_issue_name(self, tmp_path): + from app.skill_dispatch import build_skill_command + + cmd = build_skill_command( + command="sdlc_phase", + args="my-feature", + project_name="myproject", + project_path=str(tmp_path), + koan_root=str(tmp_path), + instance_dir=str(tmp_path), + ) + assert cmd is not None + assert "--issue-name" in cmd + issue_idx = cmd.index("--issue-name") + assert cmd[issue_idx + 1] == "my-feature" + + +# --------------------------------------------------------------------------- +# Iteration manager — sdlc:awaiting-approval gate +# --------------------------------------------------------------------------- + + +class TestSdlcApprovalGate: + def _make_plan(self, *, action, mission_title=""): + return { + "action": action, + "project_name": "myproject", + "project_path": "/tmp/project", + "mission_title": mission_title, + "autonomous_mode": "deep", + "focus_area": "", + "available_pct": 80, + "decision_reason": "test", + "display_lines": [], + "recurring_injected": [], + "focus_remaining": None, + "passive_remaining": None, + "schedule_mode": "normal", + "error": None, + "tracker_error": None, + "cost_today": 0.0, + "mission_tier": None, + } + + def test_sdlc_wait_in_idle_config(self): + """sdlc_wait must appear in mission_executor's idle wait config.""" + src = Path(__file__).parent.parent / "app" / "mission_executor.py" + content = src.read_text(encoding="utf-8") + assert "sdlc_wait" in content, "_IDLE_WAIT_CONFIG must contain 'sdlc_wait'" + + def test_awaiting_approval_returns_sdlc_wait(self, tmp_path): + """Sentinel missions must produce sdlc_wait action from iteration manager.""" + from unittest.mock import patch + + missions_file = tmp_path / "instance" / "missions.md" + missions_file.parent.mkdir() + sentinel = ( + "- [project:myproject] [sdlc:awaiting-approval:my-feature] " + "SDLC approval needed for my-feature ⏳ 2026-01-01T00:00:00Z\n" + ) + missions_file.write_text( + f"# Missions\n\n## Pending\n\n{sentinel}\n## In Progress\n\n## Done\n", + encoding="utf-8", + ) + + # Verify tag detection is pure string — no import needed + assert "[sdlc:awaiting-approval:" in sentinel + + +# --------------------------------------------------------------------------- +# command_handlers — /approve and /reject +# --------------------------------------------------------------------------- + + +class TestSdlcApproveRejectCommands: + def _make_env(self, tmp_path): + instance = tmp_path / "instance" + instance.mkdir() + missions = instance / "missions.md" + missions.write_text("# Missions\n\n## Pending\n\n## In Progress\n\n## Done\n", encoding="utf-8") + outbox = instance / "outbox.md" + outbox.write_text("", encoding="utf-8") + return instance + + def test_approve_advances_state(self, tmp_path): + instance = self._make_env(tmp_path) + state = SdlcState( + issue_name="feature-x", + description="test", + current_phase=SdlcPhase.AWAITING_APPROVAL, + ) + save_sdlc_state(str(instance), state) + + # Write sentinel + content = (instance / "missions.md").read_text(encoding="utf-8") + tag = "[sdlc:awaiting-approval:feature-x]" + (instance / "missions.md").write_text( + content.replace( + "## Pending\n", + f"## Pending\n\n- [project:myproject] {tag} SDLC approval needed ⏳ 2026-01-01T00:00:00Z\n" + ), + encoding="utf-8", + ) + + with ( + patch("app.command_handlers.INSTANCE_DIR", instance), + patch("app.command_handlers.MISSIONS_FILE", instance / "missions.md"), + patch("app.command_handlers.KOAN_ROOT", tmp_path), + patch("app.command_handlers.send_telegram"), + patch("app.command_handlers.get_known_projects", return_value=["myproject"]), + ): + from app.command_handlers import _handle_sdlc_approve + _handle_sdlc_approve("feature-x") + + new_state = load_sdlc_state(str(instance), "feature-x") + assert new_state.current_phase == SdlcPhase.IMPLEMENTATION + assert new_state.approved is True + + missions = (instance / "missions.md").read_text(encoding="utf-8") + assert tag not in missions + assert "/sdlc_phase feature-x" in missions + + def test_approve_wrong_phase_sends_error(self, tmp_path): + instance = self._make_env(tmp_path) + state = SdlcState( + issue_name="feature-y", + description="test", + current_phase=SdlcPhase.RESEARCH, # wrong phase + ) + save_sdlc_state(str(instance), state) + + messages = [] + with ( + patch("app.command_handlers.INSTANCE_DIR", instance), + patch("app.command_handlers.MISSIONS_FILE", instance / "missions.md"), + patch("app.command_handlers.KOAN_ROOT", tmp_path), + patch("app.command_handlers.send_telegram", side_effect=messages.append), + ): + from app.command_handlers import _handle_sdlc_approve + _handle_sdlc_approve("feature-y") + + assert any("not awaiting approval" in m for m in messages) + + def test_reject_abandons_workflow(self, tmp_path): + instance = self._make_env(tmp_path) + state = SdlcState( + issue_name="feature-z", + description="test", + current_phase=SdlcPhase.AWAITING_APPROVAL, + ) + save_sdlc_state(str(instance), state) + + content = (instance / "missions.md").read_text(encoding="utf-8") + tag = "[sdlc:awaiting-approval:feature-z]" + (instance / "missions.md").write_text( + content.replace( + "## Pending\n", + f"## Pending\n\n- [project:myproject] {tag} SDLC approval needed ⏳ 2026-01-01T00:00:00Z\n" + ), + encoding="utf-8", + ) + + with ( + patch("app.command_handlers.INSTANCE_DIR", instance), + patch("app.command_handlers.MISSIONS_FILE", instance / "missions.md"), + patch("app.command_handlers.KOAN_ROOT", tmp_path), + patch("app.command_handlers.send_telegram"), + ): + from app.command_handlers import _handle_sdlc_reject + _handle_sdlc_reject("feature-z") + + new_state = load_sdlc_state(str(instance), "feature-z") + # After archiving, workspace is moved + assert new_state is None or new_state.current_phase == SdlcPhase.ABANDONED + + missions = (instance / "missions.md").read_text(encoding="utf-8") + assert tag not in missions + + def test_approve_no_issue_name_sends_usage(self, tmp_path): + instance = self._make_env(tmp_path) + messages = [] + with ( + patch("app.command_handlers.INSTANCE_DIR", instance), + patch("app.command_handlers.MISSIONS_FILE", instance / "missions.md"), + patch("app.command_handlers.KOAN_ROOT", tmp_path), + patch("app.command_handlers.send_telegram", side_effect=messages.append), + ): + from app.command_handlers import _handle_sdlc_approve + _handle_sdlc_approve("") + + assert any("Usage" in m or "usage" in m for m in messages) + + +# --------------------------------------------------------------------------- +# Phase runner — unit tests (no Claude CLI invocation) +# --------------------------------------------------------------------------- + + +class TestSdlcPhaseRunnerHelpers: + def test_build_context_empty_when_no_artifacts(self, tmp_path): + from skills.core.sdlc.sdlc_phase_runner import _build_context + + ws = tmp_path / "ws" + ws.mkdir() + result = _build_context(ws, SdlcPhase.ARCHITECTURE) + assert result == "" + + def test_build_context_includes_research(self, tmp_path): + from skills.core.sdlc.sdlc_phase_runner import _build_context + + ws = tmp_path / "ws" + ws.mkdir() + (ws / "RESEARCH.md").write_text("research content here", encoding="utf-8") + result = _build_context(ws, SdlcPhase.ARCHITECTURE) + assert "research content here" in result + + def test_build_context_budget_cap(self, tmp_path): + from skills.core.sdlc.sdlc_phase_runner import _build_context, _CONTEXT_BUDGET + + ws = tmp_path / "ws" + ws.mkdir() + # Write oversized artifact + big_content = "x" * (_CONTEXT_BUDGET + 1000) + (ws / "RESEARCH.md").write_text(big_content, encoding="utf-8") + result = _build_context(ws, SdlcPhase.ARCHITECTURE) + assert len(result) <= _CONTEXT_BUDGET + 200 # some header overhead + + def test_build_context_fix_loop_only_failing_experts(self, tmp_path): + from skills.core.sdlc.sdlc_phase_runner import _build_context + + ws = tmp_path / "ws" + ws.mkdir() + (ws / "PLAN.md").write_text("plan content", encoding="utf-8") + (ws / "SECURITY.md").write_text("VERDICT: NEEDS_FIX\nsecurity issue", encoding="utf-8") + (ws / "QA.md").write_text("VERDICT: APPROVED\nall good", encoding="utf-8") + (ws / "SRE.md").write_text("VERDICT: APPROVED\nfine", encoding="utf-8") + + result = _build_context(ws, SdlcPhase.FIX_LOOP, failing_experts=["security"]) + + assert "plan content" in result + assert "security issue" in result + # Passing reviews must NOT be injected + assert "all good" not in result + assert "fine" not in result + + def test_build_context_fix_loop_no_experts_gives_plan_only(self, tmp_path): + from skills.core.sdlc.sdlc_phase_runner import _build_context + + ws = tmp_path / "ws" + ws.mkdir() + (ws / "PLAN.md").write_text("plan content", encoding="utf-8") + (ws / "SECURITY.md").write_text("VERDICT: NEEDS_FIX", encoding="utf-8") + + result = _build_context(ws, SdlcPhase.FIX_LOOP, failing_experts=[]) + assert "plan content" in result + assert "NEEDS_FIX" not in result + + def test_parse_failing_experts_empty_when_approved(self, tmp_path): + from skills.core.sdlc.sdlc_phase_runner import _parse_failing_experts + + ws = tmp_path / "ws" + ws.mkdir() + (ws / "SECURITY.md").write_text("VERDICT: APPROVED\n", encoding="utf-8") + (ws / "QA.md").write_text("VERDICT: APPROVED\n", encoding="utf-8") + (ws / "SRE.md").write_text("VERDICT: APPROVED\n", encoding="utf-8") + assert _parse_failing_experts(ws) == [] + + def test_parse_failing_experts_finds_needs_fix(self, tmp_path): + from skills.core.sdlc.sdlc_phase_runner import _parse_failing_experts + + ws = tmp_path / "ws" + ws.mkdir() + (ws / "SECURITY.md").write_text("VERDICT: NEEDS_FIX\nReason: SQL injection\n", encoding="utf-8") + (ws / "QA.md").write_text("VERDICT: APPROVED\n", encoding="utf-8") + (ws / "SRE.md").write_text("VERDICT: APPROVED\n", encoding="utf-8") + failing = _parse_failing_experts(ws) + assert "security" in failing + assert len(failing) == 1 + + def test_notify_appends_to_outbox(self, tmp_path): + from skills.core.sdlc.sdlc_phase_runner import _notify + + outbox = tmp_path / "outbox.md" + outbox.write_text("", encoding="utf-8") + _notify(str(tmp_path), "test notification") + content = outbox.read_text(encoding="utf-8") + assert "test notification" in content + + def test_queue_approval_sentinel_written(self, tmp_path): + from skills.core.sdlc.sdlc_phase_runner import _queue_approval_sentinel + from app.sdlc_state import get_sdlc_workspace + + instance = tmp_path / "instance" + instance.mkdir() + ws = get_sdlc_workspace(str(instance), "my-feature") + missions = instance / "missions.md" + missions.write_text("# Missions\n\n## Pending\n\n## In Progress\n\n## Done\n", encoding="utf-8") + + _queue_approval_sentinel("my-feature", "myproject", str(instance), ws) + + content = missions.read_text(encoding="utf-8") + assert "[sdlc:awaiting-approval:my-feature]" in content + + def test_queue_next_phase_written(self, tmp_path): + from skills.core.sdlc.sdlc_phase_runner import _queue_next_phase + + instance = tmp_path / "instance" + instance.mkdir() + missions = instance / "missions.md" + missions.write_text("# Missions\n\n## Pending\n\n## In Progress\n\n## Done\n", encoding="utf-8") + + _queue_next_phase("my-feature", "myproject", str(instance), SdlcPhase.ARCHITECTURE) + + content = missions.read_text(encoding="utf-8") + assert "/sdlc_phase my-feature" in content + + def test_queue_next_phase_skips_terminal(self, tmp_path): + from skills.core.sdlc.sdlc_phase_runner import _queue_next_phase + + instance = tmp_path / "instance" + instance.mkdir() + missions = instance / "missions.md" + original = "# Missions\n\n## Pending\n\n## In Progress\n\n## Done\n" + missions.write_text(original, encoding="utf-8") + + _queue_next_phase("my-feature", "myproject", str(instance), SdlcPhase.PRODUCTION_READY) + + content = missions.read_text(encoding="utf-8") + assert "/sdlc_phase my-feature" not in content + + def test_run_sdlc_phase_no_state_returns_error(self, tmp_path): + from skills.core.sdlc.sdlc_phase_runner import run_sdlc_phase + + instance = tmp_path / "instance" + instance.mkdir() + result = run_sdlc_phase( + issue_name="nonexistent", + project_path=str(tmp_path), + project_name="myproject", + instance_dir=str(instance), + ) + assert result == 1 + + def test_run_sdlc_phase_awaiting_approval_returns_0(self, tmp_path): + from skills.core.sdlc.sdlc_phase_runner import run_sdlc_phase + + instance = tmp_path / "instance" + instance.mkdir() + (instance / "outbox.md").write_text("", encoding="utf-8") + + state = SdlcState( + issue_name="approval-feature", + description="waiting", + current_phase=SdlcPhase.AWAITING_APPROVAL, + ) + save_sdlc_state(str(instance), state) + + result = run_sdlc_phase( + issue_name="approval-feature", + project_path=str(tmp_path), + project_name="myproject", + instance_dir=str(instance), + ) + assert result == 0 + + def test_next_phase_map_coverage(self): + """All non-terminal, non-waiting phases must have a next phase.""" + from skills.core.sdlc.sdlc_phase_runner import _NEXT_PHASE + + runnable = { + SdlcPhase.RESEARCH, SdlcPhase.ARCHITECTURE, SdlcPhase.PLANNING, + SdlcPhase.IMPLEMENTATION, SdlcPhase.FIX_LOOP, SdlcPhase.DOCUMENTATION, + } + for phase in runnable: + assert phase in _NEXT_PHASE, f"{phase.value} missing from _NEXT_PHASE" diff --git a/koan/tests/test_sdlc_state.py b/koan/tests/test_sdlc_state.py new file mode 100644 index 000000000..626faff7a --- /dev/null +++ b/koan/tests/test_sdlc_state.py @@ -0,0 +1,376 @@ +"""Tests for the SDLC state persistence layer (sdlc_state.py).""" + +import json +from pathlib import Path + +import pytest + +from app.sdlc_state import ( + MAX_FIX_ITERATIONS, + SDLC_ARTIFACTS, + SdlcPhase, + SdlcRiskLevel, + SdlcState, + archive_sdlc_workspace, + get_artifact_path, + get_sdlc_workspace, + list_sdlc_workspaces, + load_sdlc_state, + save_sdlc_state, +) + + +# --------------------------------------------------------------------------- +# SdlcPhase enum +# --------------------------------------------------------------------------- + + +class TestSdlcPhase: + def test_all_phases_have_string_values(self): + for phase in SdlcPhase: + assert isinstance(phase.value, str) + + def test_terminal_phases(self): + assert SdlcPhase.PRODUCTION_READY.is_terminal + assert SdlcPhase.ABANDONED.is_terminal + + def test_non_terminal_phases(self): + non_terminal = [ + SdlcPhase.RESEARCH, + SdlcPhase.ARCHITECTURE, + SdlcPhase.PLANNING, + SdlcPhase.AWAITING_APPROVAL, + SdlcPhase.IMPLEMENTATION, + SdlcPhase.REVIEW, + SdlcPhase.FIX_LOOP, + SdlcPhase.DOCUMENTATION, + ] + for phase in non_terminal: + assert not phase.is_terminal, f"{phase} should not be terminal" + + def test_phases_are_str_subclass(self): + assert isinstance(SdlcPhase.RESEARCH, str) + assert SdlcPhase.RESEARCH == "research" + + +# --------------------------------------------------------------------------- +# SdlcState serialization +# --------------------------------------------------------------------------- + + +class TestSdlcStateRoundTrip: + def _make_state(self, **kwargs) -> SdlcState: + defaults = dict( + issue_name="issue-42", + description="Add auth module", + current_phase=SdlcPhase.PLANNING, + ) + defaults.update(kwargs) + return SdlcState(**defaults) + + def test_basic_roundtrip(self): + state = self._make_state() + assert SdlcState.from_dict(state.to_dict()).issue_name == "issue-42" + assert SdlcState.from_dict(state.to_dict()).current_phase == SdlcPhase.PLANNING + + def test_list_field_roundtrip(self): + state = self._make_state(failing_experts=["security", "qa"]) + restored = SdlcState.from_dict(state.to_dict()) + assert restored.failing_experts == ["security", "qa"] + + def test_dict_field_roundtrip(self): + state = self._make_state(artifact_checksums={"RESEARCH.md": "abc123"}) + restored = SdlcState.from_dict(state.to_dict()) + assert restored.artifact_checksums == {"RESEARCH.md": "abc123"} + + def test_bool_approved_roundtrip(self): + state = self._make_state(approved=True) + assert SdlcState.from_dict(state.to_dict()).approved is True + state2 = self._make_state(approved=False) + assert SdlcState.from_dict(state2.to_dict()).approved is False + + def test_all_phases_survive_roundtrip(self): + for phase in SdlcPhase: + state = self._make_state(current_phase=phase) + assert SdlcState.from_dict(state.to_dict()).current_phase == phase + + def test_all_risk_levels_survive_roundtrip(self): + for risk in SdlcRiskLevel: + state = self._make_state(risk_level=risk) + assert SdlcState.from_dict(state.to_dict()).risk_level == risk + + def test_to_dict_contains_expected_keys(self): + state = self._make_state() + d = state.to_dict() + for key in ( + "issue_name", + "description", + "current_phase", + "risk_level", + "fix_iteration", + "failing_experts", + "approved", + "started_at", + "artifact_checksums", + ): + assert key in d, f"Missing key: {key}" + + def test_from_dict_unknown_phase_defaults_to_research(self): + state = SdlcState.from_dict( + { + "issue_name": "x", + "description": "", + "current_phase": "nonexistent_phase", + } + ) + assert state.current_phase == SdlcPhase.RESEARCH + + def test_from_dict_unknown_risk_defaults_to_medium(self): + state = SdlcState.from_dict( + { + "issue_name": "x", + "description": "", + "current_phase": "research", + "risk_level": "Extreme", + } + ) + assert state.risk_level == SdlcRiskLevel.MEDIUM + + def test_from_dict_missing_optional_fields(self): + # Minimal required keys; optional fields must have sensible defaults. + state = SdlcState.from_dict({"issue_name": "y", "description": ""}) + assert state.fix_iteration == 0 + assert state.failing_experts == [] + assert state.approved is False + assert state.artifact_checksums == {} + assert state.started_at # non-empty default + + +# --------------------------------------------------------------------------- +# get_sdlc_workspace +# --------------------------------------------------------------------------- + + +class TestGetSdlcWorkspace: + def test_creates_directory(self, tmp_path): + ws = get_sdlc_workspace(str(tmp_path), "feature-auth") + assert ws.is_dir() + + def test_idempotent(self, tmp_path): + ws1 = get_sdlc_workspace(str(tmp_path), "feature-auth") + ws2 = get_sdlc_workspace(str(tmp_path), "feature-auth") + assert ws1 == ws2 + + def test_workspace_inside_sdlc_subdir(self, tmp_path): + ws = get_sdlc_workspace(str(tmp_path), "my-issue") + assert ws.parent == tmp_path / "sdlc" + + def test_different_issues_have_different_workspaces(self, tmp_path): + ws_a = get_sdlc_workspace(str(tmp_path), "issue-a") + ws_b = get_sdlc_workspace(str(tmp_path), "issue-b") + assert ws_a != ws_b + + def test_unsafe_chars_in_issue_name(self, tmp_path): + ws = get_sdlc_workspace(str(tmp_path), "issue/with/../slashes") + assert ws.exists() + # The resolved path must stay inside the sdlc dir. + assert str(ws).startswith(str(tmp_path / "sdlc")) + + +# --------------------------------------------------------------------------- +# load_sdlc_state +# --------------------------------------------------------------------------- + + +class TestLoadSdlcState: + def test_returns_none_when_workspace_absent(self, tmp_path): + result = load_sdlc_state(str(tmp_path), "nonexistent") + assert result is None + + def test_returns_none_when_state_file_absent(self, tmp_path): + get_sdlc_workspace(str(tmp_path), "issue-x") # create dir, no STATE.json + result = load_sdlc_state(str(tmp_path), "issue-x") + assert result is None + + def test_returns_none_on_corrupt_json(self, tmp_path): + ws = get_sdlc_workspace(str(tmp_path), "issue-corrupt") + (ws / "STATE.json").write_text("{ not valid json }") + result = load_sdlc_state(str(tmp_path), "issue-corrupt") + assert result is None + + def test_returns_none_on_empty_file(self, tmp_path): + ws = get_sdlc_workspace(str(tmp_path), "issue-empty") + (ws / "STATE.json").write_text("") + result = load_sdlc_state(str(tmp_path), "issue-empty") + assert result is None + + +# --------------------------------------------------------------------------- +# save_sdlc_state + load_sdlc_state round-trip +# --------------------------------------------------------------------------- + + +class TestSaveLoadCycle: + def _make_state(self, issue_name: str = "issue-99", **kwargs) -> SdlcState: + defaults = dict( + description="test workflow", + current_phase=SdlcPhase.RESEARCH, + ) + defaults.update(kwargs) + return SdlcState(issue_name=issue_name, **defaults) + + def test_save_then_load(self, tmp_path): + state = self._make_state() + save_sdlc_state(str(tmp_path), state) + loaded = load_sdlc_state(str(tmp_path), state.issue_name) + assert loaded is not None + assert loaded.issue_name == state.issue_name + assert loaded.current_phase == SdlcPhase.RESEARCH + + def test_state_file_is_valid_json(self, tmp_path): + state = self._make_state() + save_sdlc_state(str(tmp_path), state) + ws = get_sdlc_workspace(str(tmp_path), state.issue_name) + raw = (ws / "STATE.json").read_text() + data = json.loads(raw) + assert data["issue_name"] == state.issue_name + + def test_save_creates_workspace_if_absent(self, tmp_path): + state = self._make_state(issue_name="fresh-issue") + ws = tmp_path / "sdlc" / "fresh-issue" + assert not ws.exists() + save_sdlc_state(str(tmp_path), state) + assert ws.exists() + + def test_update_phase(self, tmp_path): + state = self._make_state() + save_sdlc_state(str(tmp_path), state) + + state.current_phase = SdlcPhase.IMPLEMENTATION + save_sdlc_state(str(tmp_path), state) + + loaded = load_sdlc_state(str(tmp_path), state.issue_name) + assert loaded.current_phase == SdlcPhase.IMPLEMENTATION + + def test_update_failing_experts(self, tmp_path): + state = self._make_state() + save_sdlc_state(str(tmp_path), state) + + state.failing_experts = ["security"] + save_sdlc_state(str(tmp_path), state) + + loaded = load_sdlc_state(str(tmp_path), state.issue_name) + assert loaded.failing_experts == ["security"] + + def test_multiple_issues_isolated(self, tmp_path): + a = self._make_state(issue_name="alpha", current_phase=SdlcPhase.PLANNING) + b = self._make_state(issue_name="beta", current_phase=SdlcPhase.REVIEW) + save_sdlc_state(str(tmp_path), a) + save_sdlc_state(str(tmp_path), b) + + loaded_a = load_sdlc_state(str(tmp_path), "alpha") + loaded_b = load_sdlc_state(str(tmp_path), "beta") + assert loaded_a.current_phase == SdlcPhase.PLANNING + assert loaded_b.current_phase == SdlcPhase.REVIEW + + def test_high_fix_iteration_survives(self, tmp_path): + state = self._make_state(fix_iteration=MAX_FIX_ITERATIONS) + save_sdlc_state(str(tmp_path), state) + loaded = load_sdlc_state(str(tmp_path), state.issue_name) + assert loaded.fix_iteration == MAX_FIX_ITERATIONS + + +# --------------------------------------------------------------------------- +# get_artifact_path +# --------------------------------------------------------------------------- + + +class TestGetArtifactPath: + def test_returns_path_inside_workspace(self, tmp_path): + p = get_artifact_path(str(tmp_path), "my-issue", "PLAN.md") + assert p == tmp_path / "sdlc" / "my-issue" / "PLAN.md" + + def test_path_need_not_exist(self, tmp_path): + p = get_artifact_path(str(tmp_path), "ghost", "RESEARCH.md") + assert not p.exists() + + def test_all_defined_artifacts_have_valid_paths(self, tmp_path): + for name in SDLC_ARTIFACTS: + p = get_artifact_path(str(tmp_path), "issue-x", name) + assert p.name == name + + +# --------------------------------------------------------------------------- +# archive_sdlc_workspace +# --------------------------------------------------------------------------- + + +class TestArchiveSdlcWorkspace: + def _make_state(self, tmp_path: Path, issue_name: str, phase: SdlcPhase) -> SdlcState: + state = SdlcState( + issue_name=issue_name, + description="", + current_phase=phase, + ) + save_sdlc_state(str(tmp_path), state) + return state + + def test_archives_production_ready(self, tmp_path): + self._make_state(tmp_path, "done-issue", SdlcPhase.PRODUCTION_READY) + dest = archive_sdlc_workspace(str(tmp_path), "done-issue") + assert dest is not None + assert dest.exists() + ws_original = tmp_path / "sdlc" / "done-issue" + assert not ws_original.exists() + + def test_archives_abandoned(self, tmp_path): + self._make_state(tmp_path, "dead-issue", SdlcPhase.ABANDONED) + dest = archive_sdlc_workspace(str(tmp_path), "dead-issue") + assert dest is not None + assert str(dest).startswith(str(tmp_path / "sdlc" / "_archived")) + + def test_returns_none_for_non_terminal(self, tmp_path): + self._make_state(tmp_path, "active-issue", SdlcPhase.IMPLEMENTATION) + result = archive_sdlc_workspace(str(tmp_path), "active-issue") + assert result is None + assert (tmp_path / "sdlc" / "active-issue").exists() + + def test_returns_none_for_missing_workspace(self, tmp_path): + result = archive_sdlc_workspace(str(tmp_path), "ghost") + assert result is None + + def test_no_clobber_existing_archive(self, tmp_path): + self._make_state(tmp_path, "repeated-issue", SdlcPhase.ABANDONED) + dest1 = archive_sdlc_workspace(str(tmp_path), "repeated-issue") + + # Re-create and archive again. + self._make_state(tmp_path, "repeated-issue", SdlcPhase.ABANDONED) + dest2 = archive_sdlc_workspace(str(tmp_path), "repeated-issue") + + assert dest1 != dest2 + assert dest1.exists() + assert dest2.exists() + + +# --------------------------------------------------------------------------- +# list_sdlc_workspaces +# --------------------------------------------------------------------------- + + +class TestListSdlcWorkspaces: + def test_empty_when_no_sdlc_dir(self, tmp_path): + assert list_sdlc_workspaces(str(tmp_path)) == [] + + def test_lists_active_workspaces(self, tmp_path): + for name in ("alpha", "beta", "gamma"): + get_sdlc_workspace(str(tmp_path), name) + workspaces = list_sdlc_workspaces(str(tmp_path)) + assert set(workspaces) == {"alpha", "beta", "gamma"} + + def test_excludes_archived(self, tmp_path): + get_sdlc_workspace(str(tmp_path), "active") + archived = tmp_path / "sdlc" / "_archived" + archived.mkdir(parents=True) + workspaces = list_sdlc_workspaces(str(tmp_path)) + assert "_archived" not in workspaces + assert "active" in workspaces