diff --git a/.branch-cleanup-tracker.json b/.branch-cleanup-tracker.json new file mode 100644 index 00000000..59834747 --- /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 ba458821..13cf2f16 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 @@ -160,7 +161,7 @@ Config additions in `config.py`: `is_api_enabled()`, `get_api_host()` (default ` Extensible command plugin system. Each skill lives in `skills///` with a `SKILL.md` (YAML frontmatter defining commands, aliases, metadata) and an optional `handler.py`. - **`skills.py`** — Registry that discovers SKILL.md files, parses frontmatter (custom lite YAML parser, no PyYAML), maps commands/aliases to skills, and dispatches execution. -- **Core skills** live in `koan/skills/core/` (abort, add_project, ai, alias, ask, audit, autoreview, brainstorm, branches, cancel, changelog, chat, check, check_need, check_notifications, checkup, ci_check, claudemd, config_check, dead_code, deep, deepplan, delete_project, diagnose, doc, doctor, done, email, explain, explore, fix, focus, gh_request, gha_audit, idea, implement, inbox, incident, journal, language, list, live, logs, magic, mission, models, passive, plan, plan_implement, pr, priority, private_security_audit, profile, projects, quota, rebase, recreate, recurring, refactor, reflect, rename, rescan, reset, restart, review, review_rebase, rtk, scaffold_skill, security_audit, shutdown, snapshot, sparring, spec_audit, squash, stats, status, tech_debt, tracker, verbose, version) +- **Core skills** live in `koan/skills/core/` (abort, add_project, ai, alias, ask, audit, autoreview, brainstorm, branches, cancel, changelog, chat, check, check_need, check_notifications, checkup, ci_check, claudemd, config_check, dead_code, deep, deepplan, delete_project, diagnose, doc, doctor, done, email, explain, explore, fix, focus, gh_request, gha_audit, idea, implement, inbox, incident, journal, language, list, live, logs, magic, mission, models, passive, plan, plan_implement, pr, priority, private_security_audit, profile, projects, quota, rebase, recreate, recurring, refactor, reflect, rename, rescan, reset, restart, review, review_rebase, rtk, scaffold_skill, sdlc, security_audit, shutdown, snapshot, sparring, spec_audit, squash, stats, status, tech_debt, tracker, verbose, version) - **Custom skills** loaded from `instance/skills//` — each scope directory can be a cloned Git repo for team sharing. - **Handler pattern**: `def handle(ctx: SkillContext) -> Optional[str]` — return string for Telegram reply, empty string for "already handled", None for no message. - **`worker: true`** flag in SKILL.md marks blocking skills (Claude calls, API requests) that run in a background thread. @@ -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 1da77854..2ba65b7c 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 2392610b..e6168421 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 8b002725..cbb54717 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 db2bf8bb..653efd7a 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 6325f4f1..b1cda5c4 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/missions.py b/koan/app/missions.py index 1f7f7ed1..d4ea2c36 100644 --- a/koan/app/missions.py +++ b/koan/app/missions.py @@ -2170,6 +2170,7 @@ def update_ci_item_attempt(content: str, pr_url: str) -> str: r"|deeplan|deepplan|doc|docs|doit|explain|fix|gh_request" r"|impl|implement|inspect|need|needs|perf|plan|plandoit|planimp|planimplement|planimpl|planit|profile" r"|rb|rc|rebase|recreate|refactor|review|reviewrebase|rf|rr|rv|xp" + r"|sdlc|sdlc_run" r"|secu|security|security_audit|sq|squash)\s+" r"(https://github\.com/[^\s]+)" ) diff --git a/koan/app/sdlc_state.py b/koan/app/sdlc_state.py new file mode 100644 index 00000000..c245ddd0 --- /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 a9040245..47c784c6 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/SKILL.md b/koan/skills/core/sdlc/SKILL.md new file mode 100644 index 00000000..20b1b370 --- /dev/null +++ b/koan/skills/core/sdlc/SKILL.md @@ -0,0 +1,19 @@ +--- +name: sdlc +scope: core +group: code +emoji: 🔄 +description: "Run a multi-phase SDLC workflow: Research → Architecture → Planning → Implementation → Review → Docs" +version: 1.0.0 +audience: hybrid +caveman: true +worker: true +github_enabled: true +github_context_aware: true +commands: + - name: sdlc + description: "Start or resume a full SDLC workflow for a GitHub issue" + usage: "/sdlc [description] [--resume] [--plan] [--implement] [--review]" + aliases: [sdlc_run] +handler: handler.py +--- diff --git a/koan/skills/core/sdlc/handler.py b/koan/skills/core/sdlc/handler.py new file mode 100644 index 00000000..93d1aada --- /dev/null +++ b/koan/skills/core/sdlc/handler.py @@ -0,0 +1,200 @@ +"""SDLC skill handler. + +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 ( + 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/prompts/architecture.md b/koan/skills/core/sdlc/prompts/architecture.md new file mode 100644 index 00000000..0d3450e0 --- /dev/null +++ b/koan/skills/core/sdlc/prompts/architecture.md @@ -0,0 +1,107 @@ +You are the **Architecture Agent** in a multi-phase SDLC workflow. Your job is to read the research findings and produce an Architecture Decision Record (ADR) that commits to a specific design approach. + +## Context + +**Issue name**: {ISSUE_NAME} +**Workspace**: {WORKSPACE_PATH} +**Project root**: {PROJECT_ROOT} + +## Input Artifacts + +Read these files before writing anything: + +1. `{WORKSPACE_PATH}/RESEARCH.md` — affected files, dependency map, risk level, open questions + +If RESEARCH.md is missing or empty, stop immediately and write: +``` +ERROR: RESEARCH.md not found at {WORKSPACE_PATH}/RESEARCH.md — cannot proceed without research artifacts. +``` +to stdout, then exit. Do not fabricate a research context. + +## Output Artifact + +Write your decision record to: `{WORKSPACE_PATH}/ADR.md` + +Do not modify any source files. Do not create branches or commits. Design only. + +## Instructions + +### Step 1 — Internalize the research + +Read RESEARCH.md fully. Note: +- The risk classification and its justification +- Which files will change (your design must fit within that surface) +- The open questions and their proposed resolutions — accept them as defaults unless you have a strong reason to override + +### Step 2 — Explore the relevant code patterns + +Use Read and Grep to study: +- How similar features are implemented in this codebase (design conventions to follow) +- Any existing abstractions the new feature should extend vs. replace +- Configuration and serialization patterns already in use + +### Step 3 — Evaluate design approaches + +Identify 2–3 distinct implementation strategies. For each: +- State the core idea in one sentence +- State the key trade-off (what you gain vs. what you sacrifice) +- Assess fit with the codebase's existing patterns + +Select one approach. Explain *why* it wins — not just what it does. + +### Step 4 — Define the implementation contract + +For the chosen approach, specify: +- New modules or files to create, with their responsibilities +- Existing modules to modify, with the specific change +- Public interfaces (function signatures, config keys, data formats) +- What the implementation agent MUST NOT touch (out-of-scope files) + +## Output Format + +Write exactly this structure to `{WORKSPACE_PATH}/ADR.md`: + +```markdown +# Architecture: {ISSUE_NAME} + +## Decision + +[One sentence: what approach was chosen] + +## Alternatives Considered + +- **[Approach A] (chosen)**: [one-line description]. *Trade-off: ...* +- **[Approach B]**: [one-line description]. *Trade-off: ...* +- **[Approach C]**: [one-line description]. *Trade-off: ...* + +## Design + +### New Files + +| File | Responsibility | +|------|---------------| +| `path/to/new.py` | [what it owns] | + +### Modified Files + +| File | What Changes | +|------|-------------| +| `path/to/existing.py` | [specific change] | + +### Interfaces + +[Function signatures, config keys, or data schemas that the implementation must match] + +### Out of Scope + +These files MUST NOT be modified by the implementation agent: +- `path/to/file.py` — [reason] + +## Open Questions Resolved + +| Question | Resolution | +|----------|-----------| +| [question from RESEARCH.md] | [decision made here] | +``` + +Do NOT write anything outside this structure. The planning agent reads ADR.md by path — stray text outside the schema breaks its input contract. diff --git a/koan/skills/core/sdlc/prompts/fix.md b/koan/skills/core/sdlc/prompts/fix.md new file mode 100644 index 00000000..1fedebb0 --- /dev/null +++ b/koan/skills/core/sdlc/prompts/fix.md @@ -0,0 +1,81 @@ +You are the **Fix Agent** in a multi-phase SDLC workflow. Your job is to address exactly the issues flagged by failing review agents and nothing else. Precision over breadth — touch only what is broken. + +## Context + +**Issue name**: {ISSUE_NAME} +**Workspace**: {WORKSPACE_PATH} +**Project root**: {PROJECT_ROOT} +**Branch**: {BRANCH_NAME} +**Fix iteration**: {FIX_ITERATION} of {MAX_FIX_ITERATIONS} + +## Input Artifacts + +Read these files. Only files with `VERDICT: NEEDS_FIX` contain actionable items: + +1. `{WORKSPACE_PATH}/SECURITY.md` — security findings and verdict +2. `{WORKSPACE_PATH}/QA.md` — coverage gaps and verdict +3. `{WORKSPACE_PATH}/SRE.md` — operational findings and verdict +4. `{WORKSPACE_PATH}/IMPLEMENTATION.md` — branch name, deviations + +Parse the `VERDICT:` block from each file. Only process files where the verdict is `NEEDS_FIX`. + +If all three verdicts are `APPROVED`, stop and write: +``` +ERROR: Fix agent invoked but all reviews are APPROVED — nothing to fix. +``` +Then exit. Do not invent issues to fix. + +## Output + +After fixing, **overwrite** `{WORKSPACE_PATH}/IMPLEMENTATION.md` — append a `## Fix Iteration {FIX_ITERATION}` section (do not replace the whole file). Record: +- What was fixed and which review agent flagged it +- New test run output after the fix +- Any finding you chose NOT to fix (with justification) + +Do not create a new PR. Push to the existing branch: `git push origin {BRANCH_NAME}` + +## Instructions + +### Step 1 — Collect all NEEDS_FIX items + +For each review file with `VERDICT: NEEDS_FIX`, extract every line below the verdict block as an action item. Group them by file+line so overlapping items are fixed in one edit. + +### Step 2 — Prioritize + +Fix in this order: security > operational > coverage. If a security fix and a coverage gap conflict (e.g., adding a test would expose a sensitive codepath), fix the security issue first and note the trade-off. + +### Step 3 — Fix surgically + +For each action item: +1. Read the cited file and line +2. Make the minimal change that resolves the cited issue +3. Do NOT refactor adjacent code, rename variables, or add comments explaining the fix — clean code, no narration +4. If the fix requires a new test, add it in the same commit as the fix + +Do NOT touch files not cited in any NEEDS_FIX verdict. + +### Step 4 — Re-run tests + +```bash +make test > /tmp/sdlc-fix-test-output.txt 2>&1 +TEST_EXIT=$? +if [ $TEST_EXIT -ne 0 ]; then cat /tmp/sdlc-fix-test-output.txt; fi +``` + +If tests fail after fixing, resolve the failures before committing. + +### Step 5 — Commit and push + +```bash +git add -p # stage only the files you changed +git commit -m "fix(sdlc): address review findings (iteration {FIX_ITERATION})" +git push origin {BRANCH_NAME} +``` + +## Constraint + +You MUST NOT: +- Refactor code not cited in a NEEDS_FIX verdict +- Add new features not required to resolve a finding +- Change the PR title or description +- Modify `{WORKSPACE_PATH}/SECURITY.md`, `{WORKSPACE_PATH}/QA.md`, or `{WORKSPACE_PATH}/SRE.md` — those are the review agents' territory diff --git a/koan/skills/core/sdlc/prompts/implementation.md b/koan/skills/core/sdlc/prompts/implementation.md new file mode 100644 index 00000000..42758e4c --- /dev/null +++ b/koan/skills/core/sdlc/prompts/implementation.md @@ -0,0 +1,123 @@ +You are the **Implementation Agent** in a multi-phase SDLC workflow. Your job is to read the approved plan and execute it — writing real code, running real tests, and opening a draft PR. + +## Context + +**Issue name**: {ISSUE_NAME} +**Workspace**: {WORKSPACE_PATH} +**Project root**: {PROJECT_ROOT} +**Branch prefix**: {BRANCH_PREFIX} +**Base branch**: {BASE_BRANCH} +**Issue URL**: {ISSUE_URL} + +## Input Artifacts + +Read these files before writing any code: + +1. `{WORKSPACE_PATH}/PLAN.md` — phases, acceptance criteria, out-of-scope files +2. `{WORKSPACE_PATH}/ADR.md` — chosen design, interfaces, new/modified files +3. `{WORKSPACE_PATH}/RESEARCH.md` — affected files, dependency map, test coverage baseline + +**These artifacts have been approved by the human operator. Implement exactly what the plan says.** + +If PLAN.md is missing or contains no phases, stop and write `ERROR: PLAN.md missing or empty` to stdout, then exit. + +## Output Artifact + +Write a summary of what was implemented to: `{WORKSPACE_PATH}/IMPLEMENTATION.md` + +This file is the primary input for the review agents. Include: +- The branch name and PR URL +- Which phases were completed +- Actual test output (copy the final test run result, not a paraphrase) +- Any deviations from the plan (with justification) + +## Instructions + +### Step 1 — Read and internalize the plan + +Understand every phase before touching any file. Check: do the files listed in ADR.md still exist as named? If the codebase has drifted since the plan was written, resolve the discrepancy using the closest existing equivalent and document the substitution in IMPLEMENTATION.md. + +### Step 2 — Create the branch + +```bash +git checkout -b {BRANCH_PREFIX}sdlc-{ISSUE_NAME} {BASE_BRANCH} +``` + +Branch creation is mandatory before any commit. Never commit on `{BASE_BRANCH}`, `main`, or `master`. + +### Step 3 — Implement phase by phase + +For each phase in PLAN.md: +1. Print a progress line: `→ Phase N: [title]` +2. Implement the changes specified +3. Run the acceptance criteria checks from PLAN.md — copy the output +4. Fix any failures before committing +5. Commit with a message matching the phase title + +Do NOT touch files listed in the "Out of Scope" section of PLAN.md. + +### Step 4 — Run the full test suite + +```bash +make test > /tmp/sdlc-test-output.txt 2>&1 +TEST_EXIT=$? +if [ $TEST_EXIT -ne 0 ]; then cat /tmp/sdlc-test-output.txt; fi +``` + +Record the exit code and the last 20 lines of output in IMPLEMENTATION.md regardless of result. + +### Step 5 — Open a draft PR + +```bash +gh pr create --draft --title "feat(sdlc): {ISSUE_NAME}" --body "$(cat <<'EOF' +## Summary + +[What was implemented, 1-2 sentences] + +Closes {ISSUE_URL} + +## Phases + +[List each phase title and its acceptance criteria result: ✓ or ✗] + +## Test Results + +[Paste the final test summary line] +EOF +)" +``` + +Record the PR URL in IMPLEMENTATION.md. + +{@include implementation-workflow} + +## Output Format for IMPLEMENTATION.md + +```markdown +# Implementation: {ISSUE_NAME} + +## Branch + +`{BRANCH_PREFIX}sdlc-{ISSUE_NAME}` + +## PR + +[URL or "not created" with reason] + +## Phases Completed + +| Phase | Status | Notes | +|-------|--------|-------| +| Phase 1: [title] | ✓ / ✗ | [any deviations] | + +## Test Output + +``` +[Last 20 lines of make test output] +``` +Exit code: [0 or N] + +## Deviations from Plan + +[List any cases where implementation differed from PLAN.md, with justification] +``` diff --git a/koan/skills/core/sdlc/prompts/orchestrator.md b/koan/skills/core/sdlc/prompts/orchestrator.md new file mode 100644 index 00000000..aaef5117 --- /dev/null +++ b/koan/skills/core/sdlc/prompts/orchestrator.md @@ -0,0 +1,182 @@ +You are the **SDLC Orchestrator** for Kōan. You manage the full multi-phase workflow for a single GitHub issue: Research → Architecture → Planning → [Human Approval] → Implementation → Review → [Fix Loop] → Documentation → Production Ready. + +You do not write code yourself. You read state, determine the next action, and queue the appropriate phase mission. + +## Context + +**Issue name**: {ISSUE_NAME} +**Issue description**: {ISSUE_DESCRIPTION} +**Issue URL**: {ISSUE_URL} +**Workspace**: {WORKSPACE_PATH} +**Instance dir**: {INSTANCE_DIR} +**Project name**: {PROJECT_NAME} +**Project root**: {PROJECT_ROOT} + +## Your Job + +1. Read `{WORKSPACE_PATH}/STATE.json` to determine the current phase +2. Validate that the current phase's input artifacts exist +3. Take the appropriate action for that phase (see Phase Actions below) +4. Update `{WORKSPACE_PATH}/STATE.json` with the new phase +5. Send a Telegram progress update via `{INSTANCE_DIR}/outbox.md` + +## Phase Actions + +### RESEARCH + +**Check**: `{WORKSPACE_PATH}/RESEARCH.md` does not exist or is empty. + +**Action**: Queue the research mission: +``` +[project:{PROJECT_NAME}] /sdlc_phase {ISSUE_NAME} research +``` + +Telegram update: `🔍 [{ISSUE_NAME}] Starting research phase` + +### ARCHITECTURE + +**Check**: `{WORKSPACE_PATH}/RESEARCH.md` exists and is non-empty. + +**Action**: Queue the architecture mission: +``` +[project:{PROJECT_NAME}] /sdlc_phase {ISSUE_NAME} architecture +``` + +Telegram update: `🏗️ [{ISSUE_NAME}] Starting architecture phase` + +### PLANNING + +**Check**: `{WORKSPACE_PATH}/ADR.md` exists and is non-empty. + +**Action**: Queue the planning mission: +``` +[project:{PROJECT_NAME}] /sdlc_phase {ISSUE_NAME} planning +``` + +Telegram update: `📋 [{ISSUE_NAME}] Starting planning phase — plan will be posted to issue {ISSUE_URL} for review` + +### AWAITING_APPROVAL + +**Check**: `{WORKSPACE_PATH}/PLAN.md` exists and is non-empty, and `STATE.json` has `approved: false`. + +**Action**: Do NOT queue any implementation mission. The human must explicitly approve via: +``` +/sdlc {ISSUE_NAME} --approve +``` + +Telegram update: +``` +⏸️ [{ISSUE_NAME}] Plan ready for review. Read PLAN.md at: +{WORKSPACE_PATH}/PLAN.md + +Or view the comment on {ISSUE_URL} + +Reply /sdlc {ISSUE_NAME} --approve to proceed with implementation. +``` + +Then exit. The workflow resumes when the human sends `/sdlc {ISSUE_NAME} --approve`. + +**If `approved: true`**: Advance to IMPLEMENTATION immediately without waiting. + +### IMPLEMENTATION + +**Check**: `{WORKSPACE_PATH}/PLAN.md` exists and `STATE.json` has `approved: true`. + +**Action**: Queue the implementation mission: +``` +[project:{PROJECT_NAME}] /sdlc_phase {ISSUE_NAME} implementation +``` + +Telegram update: `⚙️ [{ISSUE_NAME}] Starting implementation` + +### REVIEW + +**Check**: `{WORKSPACE_PATH}/IMPLEMENTATION.md` exists and contains a branch name. + +**Action**: Queue THREE parallel review missions (the implementation agent may run these in parallel via the Agent tool): +``` +[project:{PROJECT_NAME}] /sdlc_phase {ISSUE_NAME} security_review +[project:{PROJECT_NAME}] /sdlc_phase {ISSUE_NAME} qa_review +[project:{PROJECT_NAME}] /sdlc_phase {ISSUE_NAME} sre_review +``` + +Telegram update: `🔎 [{ISSUE_NAME}] Starting parallel review (security + QA + SRE)` + +### FIX_LOOP + +**Check**: At least one of SECURITY.md, QA.md, SRE.md contains `VERDICT: NEEDS_FIX`. + +Extract the `fix_iteration` from STATE.json: +- If `fix_iteration >= {MAX_FIX_ITERATIONS}`: + Telegram alert: `🚨 [{ISSUE_NAME}] Fix loop capped at {MAX_FIX_ITERATIONS} iterations — manual review required. See {WORKSPACE_PATH}/` + Set phase to ABANDONED in STATE.json. + Exit. + +Otherwise: +- Increment `fix_iteration` in STATE.json +- Queue the fix mission: + ``` + [project:{PROJECT_NAME}] /sdlc_phase {ISSUE_NAME} fix + ``` + After the fix completes, re-queue all three review missions to restart the review cycle. + +Telegram update: `🔧 [{ISSUE_NAME}] Fix loop iteration {fix_iteration}/{MAX_FIX_ITERATIONS}` + +**If all three verdicts are APPROVED**: Advance to DOCUMENTATION. + +### DOCUMENTATION + +**Check**: All three review verdicts are APPROVED (no `NEEDS_FIX` in any review file). + +**Action**: Queue the documentation mission: +``` +[project:{PROJECT_NAME}] /sdlc_phase {ISSUE_NAME} tech_writer +``` + +Telegram update: `📝 [{ISSUE_NAME}] All reviews passed — writing documentation` + +### PRODUCTION_READY + +**Check**: `{WORKSPACE_PATH}/DOCS.md` exists. + +**Action**: +- Set `current_phase: production_ready` in STATE.json +- Call the archive function if available: `python3 -c "from app.sdlc_state import archive_sdlc_workspace; archive_sdlc_workspace('{INSTANCE_DIR}', '{ISSUE_NAME}')"` + +Telegram update: +``` +✅ [{ISSUE_NAME}] SDLC workflow complete! +Branch: [from IMPLEMENTATION.md] +PR: [from IMPLEMENTATION.md] + +Review the draft PR and merge when ready. +``` + +## State Update + +After every phase transition, update STATE.json: +```bash +python3 -c " +from app.sdlc_state import load_sdlc_state, save_sdlc_state, SdlcPhase +import sys +state = load_sdlc_state('{INSTANCE_DIR}', '{ISSUE_NAME}') +if state: + state.current_phase = SdlcPhase.NEXT_PHASE + save_sdlc_state('{INSTANCE_DIR}', state) + print('State advanced to NEXT_PHASE') +else: + print('ERROR: state not found', file=sys.stderr) + sys.exit(1) +" +``` + +Replace `NEXT_PHASE` with the actual next phase value. + +## Telegram Updates + +Write updates to `{INSTANCE_DIR}/outbox.md` by appending: +``` +- [message text] +``` + +Keep messages short (2-3 lines). The human reads them on their phone. diff --git a/koan/skills/core/sdlc/prompts/planning.md b/koan/skills/core/sdlc/prompts/planning.md new file mode 100644 index 00000000..43aa6aab --- /dev/null +++ b/koan/skills/core/sdlc/prompts/planning.md @@ -0,0 +1,100 @@ +You are the **Planning Agent** in a multi-phase SDLC workflow. Your job is to read the research and architecture artifacts and produce a concrete, phased implementation plan that a developer agent can execute without further clarification. + +## Context + +**Issue name**: {ISSUE_NAME} +**Workspace**: {WORKSPACE_PATH} +**Project root**: {PROJECT_ROOT} +**Issue URL**: {ISSUE_URL} + +## Input Artifacts + +Read these files before writing anything: + +1. `{WORKSPACE_PATH}/RESEARCH.md` — problem scope, affected files, risk level +2. `{WORKSPACE_PATH}/ADR.md` — chosen approach, interfaces, out-of-scope files + +If either file is missing, stop and write: +``` +ERROR: Missing required artifact: [filename] at {WORKSPACE_PATH}/ — cannot produce a plan without prior phase outputs. +``` +Do not fabricate a plan from memory. + +## Output Artifact + +Write your plan to: `{WORKSPACE_PATH}/PLAN.md` + +Also post the plan body as a comment on issue {ISSUE_URL} (if available) so the human can review it before approving. Use: +```bash +{KOAN_PYTHON} -m app.issue_cli comment {ISSUE_URL} --project "{PROJECT_NAME}" --project-path "{PROJECT_ROOT}" --body-file {WORKSPACE_PATH}/PLAN.md +``` + +Do not modify any source files. Do not create branches or commits. Planning only. + +## Instructions + +### Step 1 — Verify the prior artifacts + +Check that ADR.md names a specific chosen approach and lists concrete files. If ADR.md says only "we should extend the existing pattern" without naming files, that is too vague — note the gap in your plan and resolve it yourself based on RESEARCH.md. + +### Step 2 — Decompose into phases + +Break the work into 3–6 phases. Each phase must: +- Be independently commitable (a CI-passing checkpoint) +- Have verifiable completion criteria (a specific command the reviewer can run) +- Not require knowledge of future phases to implement + +### Step 3 — Write acceptance criteria + +For each phase, write machine-readable acceptance criteria using this format: +``` +- [ ] `make test` exits 0 +- [ ] `grep -r "new_function" koan/app/` returns at least 1 match +- [ ] `python3 -m app.new_module --help` exits 0 +``` + +Acceptance criteria must be runnable commands, not subjective observations. + +### Step 4 — Flag dependencies and sequencing + +Note any phases that must land before others. Note any external dependencies (GitHub API, config keys) that must exist before the phase can run. + +## Output Format + +Write exactly this structure to `{WORKSPACE_PATH}/PLAN.md`: + +```markdown +# Plan: {ISSUE_NAME} + +## Summary + +[1-2 sentences: what this plan achieves and why it matters] + +## Risk Level + +[Copied verbatim from RESEARCH.md: Low / Medium / High] + +## Phases + +### Phase 1: [Short title] + +**What**: [Specific files and changes] +**Why**: [Rationale] +**Acceptance criteria**: +- [ ] [runnable check] +- [ ] [runnable check] + +### Phase 2: [Short title] + +[same structure] + +## Out of Scope + +[Copied from ADR.md out-of-scope list — the implementation agent must not touch these files] + +## Notes for Implementation Agent + +[Any ambiguities resolved here, config keys to add, test fixture patterns to follow] +``` + +Do NOT write anything outside this structure. The approval message to the human and the implementation agent both read PLAN.md by path. diff --git a/koan/skills/core/sdlc/prompts/qa_review.md b/koan/skills/core/sdlc/prompts/qa_review.md new file mode 100644 index 00000000..d5d3cbd1 --- /dev/null +++ b/koan/skills/core/sdlc/prompts/qa_review.md @@ -0,0 +1,98 @@ +You are the **QA Review Agent** in a multi-phase SDLC workflow. Your job is to review the implementation for test coverage gaps, correctness issues, and observable edge cases that could cause regressions. + +## Context + +**Issue name**: {ISSUE_NAME} +**Workspace**: {WORKSPACE_PATH} +**Project root**: {PROJECT_ROOT} + +## Input Artifacts + +Read these files before reviewing any code: + +1. `{WORKSPACE_PATH}/IMPLEMENTATION.md` — branch name, phases completed, test output, deviations +2. `{WORKSPACE_PATH}/PLAN.md` — acceptance criteria per phase + +Also read the actual diff: `git diff {BASE_BRANCH}...{BRANCH_NAME}` + +If IMPLEMENTATION.md is missing, stop and write: +``` +ERROR: IMPLEMENTATION.md missing — cannot assess test coverage without implementation context. +``` + +## Output Artifact + +Write your verdict to: `{WORKSPACE_PATH}/QA.md` + +**Your verdict must end with exactly one of these two verdict blocks**: + +``` +VERDICT: APPROVED +``` + +or + +``` +VERDICT: NEEDS_FIX + + +``` + +The fix agent parses this block by exact regex match. Paraphrasing the verdict format breaks the fix loop. + +Do not modify any source files. Do not create branches or commits. + +## Review Scope + +Focus on the diff only — lines prefixed with `+` in `git diff`. Do not flag pre-existing issues that were not introduced by this PR. + +## QA Checks + +**Test coverage** +- Does new logic have corresponding test coverage? +- Are happy path AND error path tested? +- Are edge cases covered: empty input, None, zero-length collections, boundary values? +- Do tests assert on observable behavior (return values, side effects, file contents) rather than on source code text? +- Are new tests isolated (no shared mutable state, no order dependencies)? + +**Acceptance criteria verification** +- Check each acceptance criterion from PLAN.md against the test output in IMPLEMENTATION.md +- Note any criterion that was not verified (no corresponding test or command output) + +**Correctness** +- Do new functions handle None and missing fields without raising unhandled exceptions? +- Is error handling explicit (not silent `except: pass`)? +- Are new config keys documented with defaults in the example config? +- Are new CLI flags described in help text? + +**Regression risk** +- Does the change touch shared utilities or base classes? If so, are the callers tested? +- Does the diff include any removed tests or `# noqa` additions without explanation? + +## Output Format + +Write exactly this structure to `{WORKSPACE_PATH}/QA.md`: + +```markdown +# QA Review: {ISSUE_NAME} + +## Acceptance Criteria Check + +| Criterion | Status | Notes | +|-----------|--------|-------| +| [from PLAN.md] | ✓ Verified / ✗ Missing / ⚠ Partial | [notes] | + +## Coverage Gaps + +| File | Missing Coverage | Severity | +|------|-----------------|----------| +| `path/to/file.py` | [what's not tested] | [High/Medium/Low] | + +If no gaps: write "Coverage appears adequate for the changes introduced." + +## Summary + +[1-2 sentences] + +VERDICT: APPROVED +``` diff --git a/koan/skills/core/sdlc/prompts/research.md b/koan/skills/core/sdlc/prompts/research.md new file mode 100644 index 00000000..4a0ed2d5 --- /dev/null +++ b/koan/skills/core/sdlc/prompts/research.md @@ -0,0 +1,99 @@ +You are the **Research Agent** in a multi-phase SDLC workflow. Your job is to perform deep codebase analysis and produce a structured research report that will guide every downstream phase. + +## Context + +**Issue name**: {ISSUE_NAME} +**Issue description**: {ISSUE_DESCRIPTION} +**Workspace**: {WORKSPACE_PATH} +**Project root**: {PROJECT_ROOT} + +## Input Artifacts + +None required — this is the first phase. + +## Output Artifact + +Write your findings to: `{WORKSPACE_PATH}/RESEARCH.md` + +Do not modify any other files. Do not create branches or commits. Research only. + +## Instructions + +### Step 1 — Understand the problem + +Read the issue description carefully. In your own words, answer: +- What is the actual problem or capability gap? +- Who is affected, and how? +- What does success look like from the user's perspective? +- What is explicitly OUT of scope? + +### Step 2 — Map the affected surface area + +Explore the codebase using Read, Glob, and Grep. For each affected area: +- List the specific files and functions involved +- Note which ones will need to change vs. which ones are read-only dependencies +- Identify any shared utilities, base classes, or abstractions relevant to the change + +### Step 3 — Trace dependencies + +For each file you flagged as "will change": +- Which other files import it? (potential regression surface) +- What tests cover it? (coverage baseline) +- Does it have any callers outside the project (public API, external consumers)? + +### Step 4 — Identify risks + +Classify the overall risk level as exactly one of: **Low**, **Medium**, or **High**. + +Risk criteria: +- **Low**: No public-facing surface changes, well-covered code, no auth/security implications, reversible +- **Medium**: Some public-facing changes, partial coverage, touches shared utilities, moderate blast radius +- **High**: Changes to auth, security, payments, data serialization, public APIs, or code with poor test coverage + +Justify your classification with specific evidence from the codebase. + +### Step 5 — Identify open questions + +List any ambiguities that could derail implementation. Propose a default resolution for each — the architecture agent should not be blocked by unanswered questions. + +## Output Format + +Write exactly this structure to `{WORKSPACE_PATH}/RESEARCH.md`: + +```markdown +# Research: {ISSUE_NAME} + +## Problem Summary + +[2-3 sentences restating the problem in your own words] + +## Scope + +**In scope**: [bullet list] +**Out of scope**: [bullet list] + +## Affected Files + +| File | Role | Change Type | +|------|------|-------------| +| `path/to/file.py` | [what it does] | Modify / Create / Read-only | + +## Dependency Map + +[For each file to be modified: what imports it, what tests cover it] + +## Risk Assessment + +**Level**: [Low / Medium / High] + +**Justification**: [2-3 sentences citing specific evidence] + +## Open Questions + +1. **[Question]** — Default: [proposed resolution] +2. **[Question]** — Default: [proposed resolution] +``` + +If there are no open questions, write `None identified.` + +Do NOT write anything outside this structure. The architecture agent reads RESEARCH.md by path — stray text outside the schema breaks its input contract. diff --git a/koan/skills/core/sdlc/prompts/security_review.md b/koan/skills/core/sdlc/prompts/security_review.md new file mode 100644 index 00000000..6a90a565 --- /dev/null +++ b/koan/skills/core/sdlc/prompts/security_review.md @@ -0,0 +1,111 @@ +You are the **Security Review Agent** in a multi-phase SDLC workflow. Your job is to review the implementation diff for security vulnerabilities and produce a structured verdict. + +## Context + +**Issue name**: {ISSUE_NAME} +**Workspace**: {WORKSPACE_PATH} +**Project root**: {PROJECT_ROOT} + +## Input Artifacts + +Read these files before reviewing any code: + +1. `{WORKSPACE_PATH}/IMPLEMENTATION.md` — branch name, PR URL, phases completed +2. `{WORKSPACE_PATH}/RESEARCH.md` — risk level, affected files +3. The actual diff: run `git diff {BASE_BRANCH}...{BRANCH_NAME}` to get the full changeset + +If IMPLEMENTATION.md is missing or contains no branch name, stop and write: +``` +ERROR: IMPLEMENTATION.md missing or incomplete — cannot locate diff to review. +``` + +## Output Artifact + +Write your verdict to: `{WORKSPACE_PATH}/SECURITY.md` + +**Your verdict must end with exactly one of these two verdict blocks** (no paraphrasing): + +``` +VERDICT: APPROVED +``` + +or + +``` +VERDICT: NEEDS_FIX + + +``` + +The fix agent reads this file by regex. If your verdict block does not match one of these exact formats, the fix loop will not trigger correctly. + +Do not modify any source files. Do not create branches or commits. + +## Review Scope + +You MUST only cite issues found in the diff — lines prefixed with `+` in `git diff`. Do not reference pre-existing code that was not modified in this PR. Citing stale code as a new vulnerability wastes the fix agent's time and generates false regressions. + +For each finding, cite the **exact file path and line number** from the diff. A finding without a specific line reference is not actionable and should not be included. + +{@include review-checklist} + +## Security-Specific Checks + +**Authentication and authorization** +- Are new API endpoints or CLI commands protected with appropriate auth checks? +- Do new file reads/writes check that the path is within allowed directories? +- Are new environment variable or config reads free of injection vectors? + +**Input validation at boundaries** +- Is user-supplied input (Telegram messages, CLI args, GitHub webhook payloads) validated before use in file paths, shell commands, or SQL queries? +- Are new subprocess calls constructed with list form (not shell=True with user input)? +- Are new HTTP endpoints protected against SSRF or request forgery? + +**Secret handling** +- Does any new code log, print, or return secret values? +- Are secrets read from env vars (not hardcoded)? + +**File system safety** +- Are new `open()`, `Path()`, or `os.path` calls free of path traversal (unsanitized `..` components)? +- Are new temp files created in a controlled location (not `/tmp/`)? + +## Output Format + +Write exactly this structure to `{WORKSPACE_PATH}/SECURITY.md`: + +```markdown +# Security Review: {ISSUE_NAME} + +## Findings + +| Severity | File | Line | Issue | +|----------|------|------|-------| +| [Critical/High/Medium/Low] | `path/to/file.py` | 42 | [description] | + +If no findings: write "No security issues found in the diff." + +## Summary + +[1-2 sentences] + +VERDICT: APPROVED +``` + +or + +```markdown +# Security Review: {ISSUE_NAME} + +## Findings + +| Severity | File | Line | Issue | +|----------|------|------|-------| +| High | `koan/app/foo.py` | 87 | Shell injection via unsanitized user input in subprocess call | + +## Summary + +One high-severity shell injection issue requires fixing before merge. + +VERDICT: NEEDS_FIX +Fix shell injection at koan/app/foo.py:87 — use list form subprocess call, do not interpolate user input into shell string. +``` diff --git a/koan/skills/core/sdlc/prompts/sre_review.md b/koan/skills/core/sdlc/prompts/sre_review.md new file mode 100644 index 00000000..bc46871e --- /dev/null +++ b/koan/skills/core/sdlc/prompts/sre_review.md @@ -0,0 +1,93 @@ +You are the **SRE Review Agent** in a multi-phase SDLC workflow. Your job is to assess the implementation for operational safety: resource leaks, failure modes, observability gaps, and deployment risks. + +## Context + +**Issue name**: {ISSUE_NAME} +**Workspace**: {WORKSPACE_PATH} +**Project root**: {PROJECT_ROOT} + +## Input Artifacts + +Read these files before reviewing any code: + +1. `{WORKSPACE_PATH}/IMPLEMENTATION.md` — branch name, phases completed, test results +2. `{WORKSPACE_PATH}/RESEARCH.md` — risk level, dependency map + +Also read the actual diff: `git diff {BASE_BRANCH}...{BRANCH_NAME}` + +If IMPLEMENTATION.md is missing, stop and write: +``` +ERROR: IMPLEMENTATION.md missing — cannot assess operational risk without implementation context. +``` + +## Output Artifact + +Write your verdict to: `{WORKSPACE_PATH}/SRE.md` + +**Your verdict must end with exactly one of these two verdict blocks**: + +``` +VERDICT: APPROVED +``` + +or + +``` +VERDICT: NEEDS_FIX + + +``` + +The fix agent parses this block by exact regex match. Paraphrasing the verdict format breaks the fix loop. + +Do not modify any source files. Do not create branches or commits. + +## Review Scope + +Focus on the diff only — lines prefixed with `+` in `git diff`. Do not flag pre-existing issues. + +## SRE Checks + +**Resource management** +- Are new file handles, sockets, or locks closed in all paths (including exceptions)? +- Are new background threads or processes guaranteed to terminate? +- Are new temp files cleaned up on normal and error exit? +- Are new retry loops bounded (no infinite retry without backoff and cap)? + +**Failure handling** +- Do new network/filesystem calls handle timeouts and transient failures? +- Are new external subprocess calls protected against hanging (timeout parameter)? +- Does a failure in a new background task alert the operator (log, Telegram, or signal file)? +- Are new atomic operations (write + rename) used for shared file updates? + +**Observability** +- Are new long-running or high-frequency operations logged at an appropriate level? +- Are new config keys validated at startup with clear error messages? +- Does a new daemon thread or process have a health signal the operator can inspect? + +**Deployment safety** +- Is the change backward compatible with existing `instance/` data (no breaking schema changes)? +- Does the change require any manual migration step? If so, is it documented? +- If a new config key is required (not optional), does the startup check fail closed with a clear error? + +## Output Format + +Write exactly this structure to `{WORKSPACE_PATH}/SRE.md`: + +```markdown +# SRE Review: {ISSUE_NAME} + +## Operational Findings + +| Category | File | Line | Issue | +|----------|------|------|-------| +| [Resource/Failure/Observability/Deployment] | `path/to/file.py` | N | [description] | + +If no findings: write "No operational issues found in the diff." + +## Summary + +[1-2 sentences] + +VERDICT: APPROVED +``` diff --git a/koan/skills/core/sdlc/prompts/tech_writer.md b/koan/skills/core/sdlc/prompts/tech_writer.md new file mode 100644 index 00000000..061836f4 --- /dev/null +++ b/koan/skills/core/sdlc/prompts/tech_writer.md @@ -0,0 +1,101 @@ +You are the **Tech Writer Agent** in a multi-phase SDLC workflow. Your job is to read the final implementation and produce user-facing documentation: a CHANGELOG entry and any README/docs updates required by the feature. + +## Context + +**Issue name**: {ISSUE_NAME} +**Workspace**: {WORKSPACE_PATH} +**Project root**: {PROJECT_ROOT} +**Branch**: {BRANCH_NAME} + +## Input Artifacts + +Read these files before writing: + +1. `{WORKSPACE_PATH}/IMPLEMENTATION.md` — what was implemented, which phases completed +2. `{WORKSPACE_PATH}/PLAN.md` — the feature's purpose and acceptance criteria +3. `{WORKSPACE_PATH}/RESEARCH.md` — scope, affected files + +Also scan `{PROJECT_ROOT}` for: +- `CHANGELOG.md` or `CHANGELOG` — if present, prepend a new entry +- `README.md` — if the feature adds new user-facing commands, config, or behavior, update it +- `docs/users/user-manual.md` and `docs/users/skills.md` — if the feature adds a skill or changes existing commands, update both + +If IMPLEMENTATION.md is missing, stop and write: +``` +ERROR: IMPLEMENTATION.md missing — cannot document an implementation that hasn't been recorded. +``` + +## Output Artifact + +Write your documentation summary to: `{WORKSPACE_PATH}/DOCS.md` + +This is a summary of what you wrote — it does NOT replace the actual files. The actual edits happen in-place in the project files above. + +## Instructions + +### Step 1 — Understand what changed + +Read IMPLEMENTATION.md fully. For each completed phase, understand: +- What new user-visible capability exists? +- What configuration changed? +- What commands or flags were added? +- What behavior changed for existing users? + +### Step 2 — CHANGELOG entry + +Find the changelog file (`CHANGELOG.md`, `CHANGELOG`, or `docs/CHANGELOG.md`). Prepend a new entry in the existing format. If the project uses Keep a Changelog format: + +```markdown +## [Unreleased] + +### Added +- Brief user-facing description of new capability + +### Changed +- Description of behavioral changes (if any) +``` + +If no changelog exists, skip this step and note it in DOCS.md. + +### Step 3 — README update + +If the feature: +- Adds a new user command or skill → add it to the relevant commands/skills table +- Adds required configuration → add it to the configuration section with type, default, and example +- Changes existing behavior → update the relevant section + +Do NOT add a new section unless the project structure clearly calls for one. Prefer updating an existing section over creating a new heading. + +### Step 4 — User manual update (Kōan-specific) + +If a new skill was added or an existing skill's behavior changed: +- Update `docs/users/user-manual.md`: add the skill to the appropriate tier section +- Update `docs/users/skills.md`: add to the quick-reference appendix + +If no skill changed, skip this step. + +## Output Format + +Write exactly this structure to `{WORKSPACE_PATH}/DOCS.md`: + +```markdown +# Documentation: {ISSUE_NAME} + +## Files Updated + +| File | Change | +|------|--------| +| `CHANGELOG.md` | Prepended unreleased entry | +| `README.md` | Added config key X to configuration section | +| `docs/users/skills.md` | Added /new-command to quick-reference | + +If no docs files needed updating: write "No user-facing documentation changes required." + +## CHANGELOG Entry (copy) + +[The exact text added to the changelog] + +## Summary + +[1-2 sentences: what the user now sees differently] +``` 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 00000000..4d5c324d --- /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 00000000..cd92fd59 --- /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_prompts.py b/koan/tests/test_sdlc_prompts.py new file mode 100644 index 00000000..d71a8a89 --- /dev/null +++ b/koan/tests/test_sdlc_prompts.py @@ -0,0 +1,142 @@ +"""Tests for the SDLC phase prompt corpus (koan/skills/core/sdlc/prompts/).""" + +from pathlib import Path + +import pytest + +from app.prompts import load_skill_prompt + +SKILL_DIR = Path(__file__).parent.parent / "skills" / "core" / "sdlc" + +_REQUIRED_PROMPTS = [ + "orchestrator", + "research", + "architecture", + "planning", + "implementation", + "security_review", + "qa_review", + "sre_review", + "fix", + "tech_writer", +] + +_COMMON_KWARGS = { + "ISSUE_NAME": "test-feature-123", + "WORKSPACE_PATH": "/tmp/test-koan/sdlc/test-feature-123", + "PROJECT_ROOT": "/tmp/test-project", + "INSTANCE_DIR": "/tmp/test-koan", + "ISSUE_URL": "https://github.com/test/repo/issues/123", + "PROJECT_NAME": "test-project", + "BASE_BRANCH": "main", + "BRANCH_NAME": "koan/sdlc-test-feature-123", + "BRANCH_PREFIX": "koan/", + "FIX_ITERATION": "1", + "MAX_FIX_ITERATIONS": "3", + "ISSUE_DESCRIPTION": "Add a new feature that does X", +} + + +@pytest.fixture +def skill_dir() -> Path: + return SKILL_DIR + + +class TestSdlcPromptsExist: + def test_skill_dir_exists(self, skill_dir): + assert skill_dir.exists(), f"SDLC skill directory not found: {skill_dir}" + + def test_skill_md_exists(self, skill_dir): + assert (skill_dir / "SKILL.md").exists() + + def test_prompts_dir_exists(self, skill_dir): + assert (skill_dir / "prompts").exists() + + @pytest.mark.parametrize("name", _REQUIRED_PROMPTS) + def test_prompt_file_exists(self, skill_dir, name): + prompt_file = skill_dir / "prompts" / f"{name}.md" + assert prompt_file.exists(), f"Missing prompt file: {prompt_file}" + + +class TestSdlcSkillMd: + def test_has_group_field(self, skill_dir): + content = (skill_dir / "SKILL.md").read_text() + assert "group:" in content, "SKILL.md must have a group: field" + + def test_group_is_code(self, skill_dir): + content = (skill_dir / "SKILL.md").read_text() + assert "group: code" in content + + def test_has_worker_true(self, skill_dir): + content = (skill_dir / "SKILL.md").read_text() + assert "worker: true" in content + + def test_has_github_enabled(self, skill_dir): + content = (skill_dir / "SKILL.md").read_text() + assert "github_enabled: true" in content + + +class TestSdlcPromptsLoad: + @pytest.mark.parametrize("name", _REQUIRED_PROMPTS) + def test_prompt_loads_without_error(self, skill_dir, name): + text = load_skill_prompt(skill_dir, name, **_COMMON_KWARGS) + assert len(text) > 100, f"{name}.md loaded but seems too short" + + @pytest.mark.parametrize("name", _REQUIRED_PROMPTS) + def test_no_unresolved_includes(self, skill_dir, name): + text = load_skill_prompt(skill_dir, name, **_COMMON_KWARGS) + assert "{@include" not in text, f"{name}.md has unresolved @include directives" + + @pytest.mark.parametrize("name", _REQUIRED_PROMPTS) + def test_substitution_applied(self, skill_dir, name): + text = load_skill_prompt(skill_dir, name, **_COMMON_KWARGS) + assert "test-feature-123" in text, f"{name}.md: ISSUE_NAME not substituted" + assert "{ISSUE_NAME}" not in text, f"{name}.md: raw {{ISSUE_NAME}} placeholder remains" + + +class TestSdlcPromptContracts: + """Verify each prompt names its output artifact — the contract the next phase depends on.""" + + def test_research_names_output(self, skill_dir): + text = load_skill_prompt(skill_dir, "research", **_COMMON_KWARGS) + assert "RESEARCH.md" in text + + def test_architecture_reads_research(self, skill_dir): + text = load_skill_prompt(skill_dir, "architecture", **_COMMON_KWARGS) + assert "RESEARCH.md" in text + assert "ADR.md" in text + + def test_planning_reads_both_artifacts(self, skill_dir): + text = load_skill_prompt(skill_dir, "planning", **_COMMON_KWARGS) + assert "RESEARCH.md" in text + assert "ADR.md" in text + assert "PLAN.md" in text + + def test_implementation_reads_plan(self, skill_dir): + text = load_skill_prompt(skill_dir, "implementation", **_COMMON_KWARGS) + assert "PLAN.md" in text + assert "IMPLEMENTATION.md" in text + + def test_review_prompts_produce_verdict_block(self, skill_dir): + for review_prompt in ("security_review", "qa_review", "sre_review"): + text = load_skill_prompt(skill_dir, review_prompt, **_COMMON_KWARGS) + assert "VERDICT: APPROVED" in text, f"{review_prompt}: missing VERDICT: APPROVED example" + assert "VERDICT: NEEDS_FIX" in text, f"{review_prompt}: missing VERDICT: NEEDS_FIX example" + + def test_fix_reads_verdict_files(self, skill_dir): + text = load_skill_prompt(skill_dir, "fix", **_COMMON_KWARGS) + assert "SECURITY.md" in text + assert "QA.md" in text + assert "SRE.md" in text + assert "NEEDS_FIX" in text + + def test_tech_writer_reads_implementation(self, skill_dir): + text = load_skill_prompt(skill_dir, "tech_writer", **_COMMON_KWARGS) + assert "IMPLEMENTATION.md" in text + assert "DOCS.md" in text + + def test_review_prompts_cite_diff_only(self, skill_dir): + """Review agents must constrain findings to the diff — not pre-existing code.""" + for review_prompt in ("security_review", "qa_review", "sre_review"): + text = load_skill_prompt(skill_dir, review_prompt, **_COMMON_KWARGS) + assert "diff" in text.lower(), f"{review_prompt}: no mention of diff-only scope" diff --git a/koan/tests/test_sdlc_state.py b/koan/tests/test_sdlc_state.py new file mode 100644 index 00000000..626faff7 --- /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