diff --git a/CLAUDE.md b/CLAUDE.md index 415a843..27fc9fd 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -6,7 +6,7 @@ A Claude Code plugin that orchestrates the full SDLC lifecycle using company-con ## Architecture - `plugin.json` — Plugin manifest (entry point for Claude Code) - `SKILL.md` — Main skill definition (loaded when plugin activates) -- `commands/` — 21 slash commands (`/sdlc`, `/sdlc-setup`, `/sdlc-status`, `/sdlc-next`, `/sdlc-gate`, `/sdlc-enhance`, `/sdlc-coach`, `/sdlc-review`, `/sdlc-intake`, `/sdlc-brief`, `/sdlc-spec`, `/sdlc-phase-report`, `/sdlc-audit`, `/sdlc-feature`, `/sdlc-experience`, `/sdlc-data`, `/sdlc-rules`, `/sdlc-channel`, `/sdlc-evals`, `/sdlc-harness`, `/sdlc-upgrade`) +- `commands/` — 23 slash commands (`/sdlc`, `/sdlc-setup`, `/sdlc-status`, `/sdlc-next`, `/sdlc-gate`, `/sdlc-enhance`, `/sdlc-coach`, `/sdlc-review`, `/sdlc-intake`, `/sdlc-brief`, `/sdlc-spec`, `/sdlc-phase-report`, `/sdlc-audit`, `/sdlc-feature`, `/sdlc-experience`, `/sdlc-data`, `/sdlc-rules`, `/sdlc-channel`, `/sdlc-evals`, `/sdlc-harness`, `/sdlc-upgrade`, `/sdlc-revise`, `/sdlc-audit-artifacts`) - `agents/` — 13 agents (orchestrator, requirements-analyst, compliance-checker, section-evaluator, narrative-enhancer, gate-repair, multi-reviewer, discovery-analyst, feature-architect, visual-designer, conversation-designer, data-analyst, bizreq-analyst) - `channels/` — Cross-profile channel descriptor library (`_schema.yaml`, `ag-ui.yaml`, `voice.yaml`, `chat.yaml`) — the delivery-surface vocabulary the channel layer reads - `profiles/` — Company/stack YAML configs with compliance gates @@ -59,6 +59,7 @@ When adding a new agent or command, document both modes in its file. `discovery- - **Finding memory & disposition tracking (Context Repair, Increment A)** — `/sdlc-review` now writes a machine-readable `## Gate Results` block (id/category/severity/target/disposition/detail) that `record_findings.py` parses into an append-only ledger `.sdlc/metrics/findings-log.jsonl`, so a finding survives the report being overwritten and its disposition is tracked across rounds. `findings_model.py` is the single source of truth for the severity↔gate mapping and the disposition state machine (FIXED/SPLIT/ACCEPTED_RISK/POSTPONED/OPEN) with honest counting — a mislabeled disposition (SPLIT without id+owner, an AI signing ACCEPTED_RISK) still counts as debt. The review stays **advisory** (the grader advises, never blocks); only the **FIXED-claim check** may block (`report --strict` exits 2) — a finding marked FIXED whose target file never changed is a factual false claim, not a judgment. The ledger is what later makes a recurring finding promotable into a permanent check (Phase C: "findings become new checks"). Runs standalone (`--repo`) or in-workflow (`--state`). See `docs/proposals/context-repair-loop.md` - **Multi-discipline channel layer** — an additive layer giving Data, Design, and Bizreq first-class seats and organizing a feature around its **customer channel of use**. `channels/` holds schema-guarded YAML descriptors (`ag-ui`, `voice`, `chat`) whose acceptance dimensions ride a spec's *existing* `## Acceptance Checks`; `validate_channel.py` guards the library and `check_channel.py` is an advisory (exit 0) DoR lint that never changes a ready/not-ready verdict. Six commands (`/sdlc-feature`, `/sdlc-experience`, `/sdlc-data`, `/sdlc-rules`, `/sdlc-channel`, `/sdlc-evals`, `/sdlc-harness`, `/sdlc-upgrade`) drive five interview-driven discipline agents (`feature-architect`, `visual-designer`, `conversation-designer`, `data-analyst`, `bizreq-analyst`) that also serve as `/sdlc-review` council lenses (7 viewpoints). A spec's optional `channel:` frontmatter field binds the surface; a phase-spanning `.sdlc/decision-log.md` (owner + 2-business-day clock, surfaced in `/sdlc-status` via `track_decisions.py`) and optional discipline sign-offs on the state sign-off record round it out. The protected core (`check_spec.py`, `check_gates.py`, `section-evaluator`, `harness/**`, `phase_model.py`, `phase-registry.yaml`, `/sdlc-coach`, `/sdlc-spec`) is byte-for-byte unchanged. See `references/channel-model.md` and `references/team-model.md` - **Close handoff-report generation** — Phase C Step 4 ("Hand over the record") is a two-pass draft: `generate_handoff_report.py` does the deterministic assembly first — phase report index, per-phase gate/sign-off table (from `state.yaml`), metrics history (reusing `scorecard.py`), spec backlog (reusing `track_specs.py`) — filling the existing `final-handoff-report.md` template and marking the judgment sections (outcomes vs the Phase 0 statement, debt log, open items, dashboard handover) with `[Fill: ...]` slots for the Explore agent to enrich. Honest by design (missing data reads "no data", never a fabricated zero) and refuses to clobber a human-edited report without `--force`. Runs standalone (`--repo`) or in-workflow (`--state`) +- **Artifact update & audit** — an additive, advisory layer for *changing* a pre-Build artifact after the fact and auditing the trail. `artifact_model.py` owns the change-ledger entry shape and the staleness disposition state machine (`OPEN → REFRESHED | ACKNOWLEDGED | NOT_AFFECTED`) with honest counting (ACKNOWLEDGED needs an owner, NOT_AFFECTED a reason, or it still counts as debt); `artifact_lineage.py` harvests declared `upstream → downstream` edges (frozen-layer `source_artifacts`, id references, explicit markdown paths) with a labeled **coarse** phase-order fallback and cycle-safe traversal. `audit_artifacts.py` (`record` / `impact` / `report`, **exit 0 always**) records changes to an append-only `.sdlc/metrics/artifact-log.jsonl` — its **own** JSONL, never inside `gate_results`, so `/sdlc-audit` output is byte-identical with or without it — and flags a downstream as a *stale candidate* when an upstream changed after it last did. `/sdlc-revise ` is the PM-facing write path (discipline agent proposes, named human decides; records the why to the ledger **and** a linked `DL-NN` decision-log item; re-gates; shows impact to disposition). `/sdlc-audit-artifacts` is the read-only sibling to `/sdlc-audit` (freshness dashboard, `--impact`, `--history`). A `record --scan` step in `/sdlc-next` captures direct edits at advance time; `/sdlc-status` surfaces a stale rollup. The protected core (incl. `advance_phase.py`) is byte-for-byte unchanged. See `references/artifact-lifecycle.md` ## Testing ```bash @@ -77,5 +78,8 @@ uv run scripts/scorecard.py report --state /tmp/test/.sdlc/state.yaml --window-d uv run scripts/generate_handoff_report.py --state /tmp/test/.sdlc/state.yaml uv run scripts/record_findings.py record --report /tmp/test/.sdlc/artifacts/02-design/review-report.md --repo /tmp/test uv run scripts/record_findings.py report --repo /tmp/test --strict -uv run --project scripts --extra test python -m pytest scripts/tests/ -q +uv run scripts/audit_artifacts.py record --scan --repo /tmp/test +uv run scripts/audit_artifacts.py impact FR-012 --repo /tmp/test +uv run scripts/audit_artifacts.py report --repo /tmp/test --json +uv run --project scripts python -m pytest scripts/tests/ -q ``` diff --git a/README.md b/README.md index 70a9342..08a7cc0 100644 --- a/README.md +++ b/README.md @@ -109,7 +109,7 @@ For in-depth technical documentation, see the guides in [`docs/`](docs/): | [Phase Lifecycle](docs/phase-lifecycle.md) | All 9 phases in depth — workflows, artifacts, HITL gates, skills, agents, handoff protocol, project type adaptations | | [Gate System](docs/gate-system.md) | 7-gate validation — integrity, completeness, metrics, compliance, consistency, quality, exit criteria — severity levels, override protocol | | [Profiles](docs/profiles.md) | Schema reference (every field), built-in profiles, custom profile creation, compliance framework integration, evaluation criteria | -| [Commands](docs/commands.md) | All 21 slash commands — internal flow, state changes, Python scripts called, error scenarios, examples | +| [Commands](docs/commands.md) | All 23 slash commands — internal flow, state changes, Python scripts called, error scenarios, examples | | [Agents](docs/agents.md) | 13 custom agents + built-in subagent orchestration, phase-to-agent mapping, parallel execution rules, mandatory spawns | | [State Machine](docs/state-machine.md) | state.yaml format, transition rules, history tracking, session-handoff.json, the spec backlog | | [Templates & Artifacts](docs/templates-artifacts.md) | Template directory structure, per-phase artifact details, handoff document protocol, artifact lifecycle | @@ -228,7 +228,7 @@ Gates have severity levels: claude-code-sdlc/ ├── plugin.json # Plugin manifest ├── SKILL.md # Main skill entry point -├── commands/ # 21 slash commands (/sdlc, /sdlc-setup, /sdlc-status, /sdlc-next, /sdlc-gate, /sdlc-enhance, /sdlc-coach, /sdlc-review, /sdlc-intake, /sdlc-brief, /sdlc-spec, /sdlc-phase-report, /sdlc-audit, /sdlc-feature, /sdlc-experience, /sdlc-data, /sdlc-rules, /sdlc-channel, /sdlc-evals, /sdlc-harness, /sdlc-upgrade) +├── commands/ # 23 slash commands (/sdlc, /sdlc-setup, /sdlc-status, /sdlc-next, /sdlc-gate, /sdlc-enhance, /sdlc-coach, /sdlc-review, /sdlc-intake, /sdlc-brief, /sdlc-spec, /sdlc-phase-report, /sdlc-audit, /sdlc-feature, /sdlc-experience, /sdlc-data, /sdlc-rules, /sdlc-channel, /sdlc-evals, /sdlc-harness, /sdlc-upgrade, /sdlc-revise, /sdlc-audit-artifacts) ├── agents/ # 13 agents (orchestrator, requirements-analyst, compliance-checker, section-evaluator, narrative-enhancer, gate-repair, multi-reviewer, discovery-analyst, feature-architect, visual-designer, conversation-designer, data-analyst, bizreq-analyst) ├── profiles/ # Company/stack YAML profiles ├── channels/ # Channel descriptor library (ag-ui, voice, chat) + schema diff --git a/commands/sdlc-audit-artifacts.md b/commands/sdlc-audit-artifacts.md new file mode 100644 index 0000000..a1737cd --- /dev/null +++ b/commands/sdlc-audit-artifacts.md @@ -0,0 +1,87 @@ +# /sdlc-audit-artifacts — Artifact Freshness, Impact & History + +Audit the **content** of your pre-Build artifacts: which ones have gone stale relative to the things +they depend on, what a change would put at risk, and the full change trail of any one artifact. + +This is a **sibling** to `/sdlc-audit`, not a replacement. `/sdlc-audit` audits *gate effectiveness* +(are the gates well-calibrated). This command audits *artifact staleness and change history*. They +answer different questions and never overlap. + +Everything here is **advisory** — it surfaces candidates a human dispositions; it never blocks a +gate, changes a verdict, or edits your artifacts, specs, or `state.yaml`. + +## Instructions + +1. **Resolve mode and repo root:** + - **Workflow mode** (default): look for `.sdlc/state.yaml`. Pass `--state .sdlc/state.yaml` to the + script. If not found, tell the user to run `/sdlc-setup` first. + - **Standalone mode** (`--repo `, or no `.sdlc/` found): run against any repo containing a + `.sdlc/` directory. History and staleness are only as deep as the ledger in that repo. + +2. **Freshen the change-ledger** (so staleness reflects what is actually on disk): + ```bash + uv run --project ${CLAUDE_PLUGIN_ROOT}/scripts ${CLAUDE_PLUGIN_ROOT}/scripts/audit_artifacts.py record --scan --state .sdlc/state.yaml + ``` + This appends **only** to the change-ledger (`.sdlc/metrics/artifact-log.jsonl`) — it hashes the + artifact tree and records any new or drifted files. It never touches your artifacts, specs, or + `state.yaml`. On the very first run it seeds a baseline ("history starts now") — it does not + invent a past. In standalone mode pass `--repo ` instead of `--state`. + +3. **Pick the lens** from the user's request: + + - **Freshness dashboard (default)** — per artifact: when it last changed and by whom (from the + ledger), its sign-off status (from `state.yaml`), and **FRESH** or **STALE** vs. its declared + sources. Scope with `--phase `, `--artifact `, or `--since `: + ```bash + uv run --project ${CLAUDE_PLUGIN_ROOT}/scripts ${CLAUDE_PLUGIN_ROOT}/scripts/audit_artifacts.py report --state .sdlc/state.yaml + ``` + + - **`--impact `** (forward) — *"I'm about to change FR-012 — what depends on it?"* Walks + the lineage graph and lists every downstream artifact and the path, **before** you commit: + ```bash + uv run --project ${CLAUDE_PLUGIN_ROOT}/scripts ${CLAUDE_PLUGIN_ROOT}/scripts/audit_artifacts.py impact FR-012 --state .sdlc/state.yaml + ``` + + - **`--history `** (backward) — the full change trail of one artifact from the ledger: + every revision, when, who, why, and any linked `DL-NN`: + ```bash + uv run --project ${CLAUDE_PLUGIN_ROOT}/scripts ${CLAUDE_PLUGIN_ROOT}/scripts/audit_artifacts.py report --history FR-012 --state .sdlc/state.yaml + ``` + +4. **Display results** with the honest confidence labels intact: + - Every downstream edge is tagged **declared** (a written-down link: a frozen layer's + `source_artifacts`, a spec's `source`, an `FR-012`/`BR-04` id reference, an explicit markdown + path) or **coarse** (a phase-order *guess*, used only where nothing was declared). Never present + a coarse inference as a declared link. + - A STALE artifact is a **candidate**, never "broken" — it may already account for the change. + - Report the honest counts: `N stale · X open · Y dispositioned`. "no data" reads as no data, + never a fabricated zero. + +5. **Offer to disposition open staleness** (optional). For each OPEN candidate the human wants to + resolve, record their judgement (this is the only write, and only to the ledger): + ```bash + uv run --project ${CLAUDE_PLUGIN_ROOT}/scripts ${CLAUDE_PLUGIN_ROOT}/scripts/audit_artifacts.py \ + record --disposition ACKNOWLEDGED --downstream --upstream --owner "" --state .sdlc/state.yaml + ``` + - `REFRESHED` is **derived** — update the downstream artifact and the next scan clears it + automatically; you don't record it by hand. + - `ACKNOWLEDGED` needs an **owner** or it still counts as debt. `NOT_AFFECTED` needs a **reason**. + You cannot clear debt by typing a word — honest counting enforces this. + +## Arguments + +- No arguments: freshness dashboard for the whole project (workflow mode). +- `--repo `: standalone mode — audit any repo with a `.sdlc/` present. +- `--impact `: forward lineage — what a change here could make stale. +- `--history `: backward change trail for one artifact. +- `--phase ` / `--artifact ` / `--since `: scope the dashboard. + +## Important + +- The user runs `/sdlc-audit-artifacts` — never `audit_artifacts.py` by hand. The command owns the + scan, the lens selection, and the disposition recording. +- **Read-only with respect to your project.** The only thing this command ever writes is the + change-ledger (`.sdlc/metrics/artifact-log.jsonl`) — the audit trail itself. It never edits an + artifact, a spec, or `state.yaml`. To *change* an artifact, use `/sdlc-revise`. +- **Advisory by construction** — `audit_artifacts.py` exits 0 always. Staleness is never a gate; a + candidate is a prompt for a human's judgement, not a verdict. diff --git a/commands/sdlc-next.md b/commands/sdlc-next.md index 7a5fb56..bc4cf74 100644 --- a/commands/sdlc-next.md +++ b/commands/sdlc-next.md @@ -62,8 +62,19 @@ Run exit gate checks for the current phase and advance to the next phase if all 7. If validation fails, fix issues and re-validate before proceeding 8. See `references/frozen-layers.md` for format details and condensation strategy -6. **Advance phase:** Perform the advance via `advance_phase.py` (it applies the state updates below - and records any discipline sign-offs captured in step 4 on the phase's existing sign-off record): +6. **Advance phase:** + + - **First, snapshot the artifact ledger (advisory, exit 0):** capture any direct edits to this + phase's artifacts before they freeze, so the change history and staleness stay current even for + edits made outside `/sdlc-revise`: + ```bash + uv run --project ${CLAUDE_PLUGIN_ROOT}/scripts ${CLAUDE_PLUGIN_ROOT}/scripts/audit_artifacts.py record --scan --state .sdlc/state.yaml + ``` + This appends only to `.sdlc/metrics/artifact-log.jsonl` (the audit trail) — it never modifies + artifacts or `state.yaml`, and never blocks the advance. If the script is absent, skip it. + + Then perform the advance via `advance_phase.py` (it applies the state updates below and records any + discipline sign-offs captured in step 4 on the phase's existing sign-off record): ```bash uv run --project ${CLAUDE_PLUGIN_ROOT}/scripts ${CLAUDE_PLUGIN_ROOT}/scripts/advance_phase.py \ --state .sdlc/state.yaml --confirmed \ diff --git a/commands/sdlc-revise.md b/commands/sdlc-revise.md new file mode 100644 index 0000000..8d5d6d0 --- /dev/null +++ b/commands/sdlc-revise.md @@ -0,0 +1,121 @@ +# /sdlc-revise — Change One Specific Artifact (and record why) + +Adjust a *thing*, not a whole file. Artifacts already carry ids, so you revise `FR-012` or `BR-04` +or one section — the command routes the change to the discipline that owns it, records the why with +an owner and a clock, re-gates, and shows you exactly what the change put at risk. + +``` +/sdlc-revise FR-012 +/sdlc-revise BR-04 +/sdlc-revise requirements.md#non-functional-requirements +``` + +The **One Rule** holds throughout: the discipline agent **proposes** the change; a **named human +decides** it. No agent silently rewrites an artifact. + +## Instructions + +1. **Resolve mode and repo root:** + - **Workflow mode** (default): look for `.sdlc/state.yaml`; pass `--state .sdlc/state.yaml` to the + scripts. If not found, tell the user to run `/sdlc-setup` first. + - **Standalone mode** (`--repo `, or no `.sdlc/`): operate on the given repo. Note the + missing engagement context in the change reason. + +2. **Resolve the target — id first, heading fallback.** + - If the argument is an id (`FR-012`, `BR-04`, `EP-3`, `US-007`, `SCEN-02`, `FE-01`, `ADR-004`), + find the artifact that declares it. + - If it is `file.md#heading-slug`, resolve to that file and section. + - If it is a bare file, ask which section/id inside it the user means. + - If ambiguous, **ask** (`AskUserQuestion`) — never guess which artifact. + +3. **Preview the blast radius before you change anything:** + ```bash + uv run --project ${CLAUDE_PLUGIN_ROOT}/scripts ${CLAUDE_PLUGIN_ROOT}/scripts/audit_artifacts.py impact --state .sdlc/state.yaml + ``` + Show the human what depends on this target (declared vs. coarse) so they revise with eyes open. + +4. **Interview to change it — route to the owning discipline agent.** The agent proposes concrete + wording; the human confirms or edits. Then apply the confirmed edit to the artifact file. + + | Target | Owning discipline agent | + |--------|-------------------------| + | `FR-`, `NFR-`, `EP-`, `US-`, `requirements.md`, `epics.md`, `user-stories.md`, `non-functional-requirements.md` | `requirements-analyst` | + | `BR-`, `business-rules.md`, `SCEN-`, `golden-scenarios.md` | `bizreq-analyst` | + | `FE-`, `feature-brief.md`, the Spec decomposition | `feature-architect` | + | user journey / surface layout / interaction spec for **web/visual** (ag-ui) | `visual-designer` | + | user journey / turn-script / message-flow for **voice or chat** | `conversation-designer` | + | `data-contract`, data-readiness, lineage-audit | `data-analyst` | + + If no discipline clearly owns it, do the edit conversationally with the user — the One Rule still + holds (human decides). + +5. **Record the why — ledger + decision-log (both, always):** + + a. Append the change to the artifact ledger: + ```bash + uv run --project ${CLAUDE_PLUGIN_ROOT}/scripts ${CLAUDE_PLUGIN_ROOT}/scripts/audit_artifacts.py \ + record --artifact --target --event revised \ + --actor "" --reason "" --decision-ref DL-NN --state .sdlc/state.yaml + ``` + + b. Open a linked decision-log item so the change carries an **owner and a 2-business-day clock**. + Read `.sdlc/decision-log.md`, allocate the next `DL-NN` (2-digit; if the file or its table is + missing, create it from `${CLAUDE_PLUGIN_ROOT}/templates/phases/01-requirements/decision-log.md`), + and append one row — the `id` must match the `--decision-ref` you passed above: + + ```markdown + | DL-NN | Revised : | | | | open | + ``` + + `track_decisions.py` then surfaces it in `/sdlc-status`; close it (`status: decided`) once the + downstream ripple is dispositioned. + +6. **Re-gate the affected phase** — dirty-tracking re-validates only the changed artifact: + ``` + /sdlc-gate + ``` + Run the gate for the phase the artifact belongs to and report PASS/FAIL. Fix any MUST failures the + change introduced before moving on. + +7. **Show impact and disposition the ripple.** Re-run the freshness view scoped to what this change + touched and walk the human through each newly-stale downstream: + ```bash + uv run --project ${CLAUDE_PLUGIN_ROOT}/scripts ${CLAUDE_PLUGIN_ROOT}/scripts/audit_artifacts.py impact --state .sdlc/state.yaml + ``` + For each downstream the human wants to resolve now, record the disposition (the only other write, + and only to the ledger): + ```bash + uv run --project ${CLAUDE_PLUGIN_ROOT}/scripts ${CLAUDE_PLUGIN_ROOT}/scripts/audit_artifacts.py \ + record --disposition \ + --downstream --upstream --owner "" --reason "" --state .sdlc/state.yaml + ``` + - **Refreshed?** If you fix the downstream now (or re-run `/sdlc-revise` on it), you do **not** + record anything — a later change hash proves it, so the next scan clears the item automatically. + `REFRESHED` is derived, never typed. + - `ACKNOWLEDGED` — accept the debt for now; **names an owner** or it still counts as debt. + - `NOT_AFFECTED` — the change doesn't ripple here; **give a reason** or it still counts. + +8. **Report:** + ``` + Revised in (actor: , DL-NN) + Re-gate: PASS | FAIL () + Impact: downstream — refreshed, acknowledged, open + ``` + +## Arguments + +- ``: the thing to revise (id preferred — stable; heading slug is a fallback). +- No target: ask the user what to revise. +- `--repo `: standalone mode (no `.sdlc/` present). + +## Important + +- `/sdlc-revise` writes **only** three things: the artifact, the change-ledger, and the decision-log. + It **never** touches `state.yaml`'s gate results or sign-off records — advancing and sign-off remain + `/sdlc-next`'s job, unchanged. +- **Agent proposes, human decides.** The discipline agent drafts the new wording; a named human + confirms it and owns the decision-log item. This is the same One Rule as `/sdlc-spec`'s risk tier. +- **Staleness is advisory.** A downstream flagged after your change is a *candidate* — it may already + account for the change. Disposition it; the tool never blocks on it. +- When behavior changes, the artifact changes here — and if a spec or code already realizes it, that + change belongs in the **same PR** as the code. A stale artifact lies to the next agent and human. diff --git a/commands/sdlc-status.md b/commands/sdlc-status.md index 7cdb7fc..ad00b76 100644 --- a/commands/sdlc-status.md +++ b/commands/sdlc-status.md @@ -38,6 +38,16 @@ Display the current SDLC progress for this project. shown). If `track_decisions.py` or `.sdlc/decision-log.md` is absent, read `.sdlc/decision-log.md` directly for open items, or skip silently if neither exists. Advisory only (exit 0) — never blocks. + - **Artifact staleness rollup** — surface how many artifacts have drifted out of date with the + things they depend on: + ```bash + uv run --project ${CLAUDE_PLUGIN_ROOT}/scripts ${CLAUDE_PLUGIN_ROOT}/scripts/audit_artifacts.py report --state .sdlc/state.yaml --json + ``` + If `has_history` is true, render one line — `Artifacts: N stale (X acknowledged, Y open)` — + using `stale`, `acknowledged`, and `open` from the JSON. If `has_history` is false (no ledger + yet) or the script is absent, skip silently — no baseline has been recorded. Advisory only + (exit 0) — never blocks. See `/sdlc-audit-artifacts` for the full dashboard. + 6. **Suggest next action:** Based on current phase status: - If phase is `active`: suggest running `/sdlc` for phase guidance - If all gates would pass: suggest running `/sdlc-next` to advance diff --git a/docs/commands.md b/docs/commands.md index dda4803..431de95 100644 --- a/docs/commands.md +++ b/docs/commands.md @@ -715,7 +715,7 @@ The Build loop's Intent beat — before building any change. A spec that `check_ ## Additional Commands (summaries) -Eight commands have their full flow documented in their command files rather than here. One +Ten commands have their full flow documented in their command files rather than here. One line each; see `commands/.md` for the complete instructions. | Command | What it does | @@ -728,6 +728,8 @@ line each; see `commands/.md` for the complete instructions. | `/sdlc-data` | Author the data contract, readiness, and lineage for a feature | | `/sdlc-rules` | Author business rules and golden scenarios | | `/sdlc-evals` | Author the versioned golden set for an LLM-powered spec | +| `/sdlc-revise` | Change one specific artifact (id or section) — discipline agent proposes, human decides; records the why to the change-ledger + a linked `DL-NN`, re-gates, shows downstream staleness to disposition | +| `/sdlc-audit-artifacts` | Read-only sibling to `/sdlc-audit`: artifact freshness dashboard, forward `--impact`, and `--history` change trail (advisory; never blocks) | --- diff --git a/references/artifact-lifecycle.md b/references/artifact-lifecycle.md new file mode 100644 index 0000000..14400e3 --- /dev/null +++ b/references/artifact-lifecycle.md @@ -0,0 +1,92 @@ +# Artifact Lifecycle — changing an artifact after the fact, and auditing the trail + +The plugin is strong at *creating* artifacts and *gating* them. This layer is about what happens +**after** an artifact exists and someone needs to change it — a requirement turns out wrong, an SLA +moves, a business rule is refined in design. Left implicit, that change is invisible: nothing flags +the frozen layer or the downstream design doc as out of date, and there is no record of what changed, +when, by whom, or why. + +This is the conceptual reference for the artifact-update-audit layer. The commands are +`/sdlc-revise` (change one thing) and `/sdlc-audit-artifacts` (audit freshness, impact, history); +the engine is `scripts/audit_artifacts.py` over two models (`artifact_model.py`, +`artifact_lineage.py`). Everything here is **advisory** — it tells a human something worth knowing; +it never decides. + +--- + +## The core insight + +The tool already knows the two facts needed to detect staleness — it just never connected them: + +- **When each artifact last changed** — every artifact is hashed (SHA-256). The change-ledger + records each new/changed hash with a timestamp, an actor, and a reason. +- **What depends on what** — traceability is *already declared* across the corpus: a frozen layer's + `source_artifacts:`, a spec's `source:`, the feature-brief "Traces to" column, and the + `FR-NN` / `EP-NN` / `US-NN` / `BR-NN` / `SCEN-NN` id vocabulary. + +Connect "what changed" with "what depends on it" and staleness falls out: if an **upstream** artifact +changed *after* a **downstream** one last did, the downstream is a **stale candidate**. + +## The change-ledger + +An append-only JSONL at `.sdlc/metrics/artifact-log.jsonl` — the artifact **history** the tool +otherwise lacks. It lives in its own file, never inside `state.yaml`'s `gate_results` (which is +gate-audit territory), so `/sdlc-audit` is byte-for-byte identical with or without this layer. + +Two entry kinds share the file, told apart by `event`: + +- a **change** entry (`created` / `revised` / `refreshed` / `snapshot`) records that an artifact's + content moved: `{artifact, target, phase, hash, prev_hash, actor, reason, decision_ref}`. +- a **disposition** entry records a human's judgement about one staleness candidate. + +Changes reach the ledger three ways, all command-driven (never a hook into `advance_phase.py`): + +1. **`/sdlc-revise`** appends an explicit `revised` entry for the thing you changed. +2. **`record --scan`** hash-walks the artifact tree and records any new (`created`) or drifted + (`snapshot`) file. It runs at the top of `/sdlc-audit-artifacts` and at advance time in + `/sdlc-next`, so direct edits (made outside `/sdlc-revise`) are captured too. +3. The **first-ever scan** seeds a baseline — "history starts now". Nothing about the past is + fabricated; missing data reads as no data. + +## Lineage — declared, with a labeled coarse fallback + +`artifact_lineage.py` harvests `upstream → downstream` edges, each tagged with a **confidence**: + +- **declared** (high confidence) — the link is written down: a frozen layer's `source_artifacts`, an + id reference (the file that *declares* `FR-012` is upstream of any file that *references* it), or an + explicit markdown path pointing from a later-phase file to an earlier-phase one. +- **coarse** (a guess) — where an artifact declares *nothing*, phase order supplies a fallback edge + (an artifact in the prior phase is assumed upstream). Always labeled, so a coarse inference is never + mistaken for a declared link. + +Traversal is cycle-safe (a visited set), so a mutual reference can never loop forever. + +## The staleness disposition state machine + +A stale candidate is `(downstream, upstream@hash)` — pinned to the upstream's content hash so a +*later* upstream change re-opens the item rather than silently inheriting an old disposition. Its +disposition follows the same honest-counting discipline as review findings — you cannot clear debt by +typing a word: + +| Disposition | Counts as debt? | Rule | +|-------------|-----------------|------| +| `OPEN` | **yes** | untouched. | +| `REFRESHED` | no | the downstream changed *after* the upstream — **derived** from a later ledger hash, not claimed. A re-scan clears it automatically. | +| `ACKNOWLEDGED` | no **only** with an owner | a human accepts the debt for now; without an `owner` it still counts. | +| `NOT_AFFECTED` | no **only** with a reason | a human judges no ripple; without a `reason` it still counts. | + +A mislabeled disposition (ACKNOWLEDGED with no owner) counts exactly like OPEN. + +## How it stays additive + +- **Advisory, never blocks.** `audit_artifacts.py` exits 0 always. Staleness is a REVIEW aid a human + dispositions — the tool flags **candidates**, it never asserts breakage or changes a gate verdict. +- **Protected core untouched.** `check_spec.py`, `check_gates.py`, `section-evaluator`, `harness/**`, + `phase_model.py`, `phase-registry.yaml`, `/sdlc-coach`, `/sdlc-spec`, and `advance_phase.py` are all + byte-for-byte unchanged. The new modules *read* `check_spec` helpers; they never modify them. +- **No phantom rows.** The ledger is a standalone JSONL; nothing is written into `gate_results`. +- **Dual-mode + graceful degradation.** Every script runs standalone (`--repo`) or in-workflow + (`--state`). With no ledger yet, the first run snapshots a baseline and reports "no history yet". + +*If this whole layer were deleted, the plugin would gate, freeze, and sign off exactly as it does +today. The artifact-update-audit surface only ever tells a human something worth knowing.* diff --git a/scripts/artifact_lineage.py b/scripts/artifact_lineage.py new file mode 100644 index 0000000..0b4f6e7 --- /dev/null +++ b/scripts/artifact_lineage.py @@ -0,0 +1,304 @@ +"""artifact_lineage.py — harvest the upstream→downstream dependency graph the staleness engine walks. + +Traceability is already *declared* all over an SDLC corpus; this module reads it into one graph of +edges `upstream → downstream` ("if upstream changes, downstream may be stale"). Each edge is tagged +with a `confidence`: + + - "declared" — the link is written down somewhere (a frozen layer's `source_artifacts:`, a spec's + `source:`, an `FR-012` / `BR-04` id reference, an explicit markdown path). High confidence. + - "coarse" — nothing was declared, so phase order supplies a fallback edge (an artifact in the + prior phase is *assumed* upstream of one in the next). Always labeled so a human never mistakes + an inference for a declared link. Only ever added where an artifact declares nothing. + +Nodes are repo-relative POSIX paths (e.g. `.sdlc/artifacts/01-requirements/requirements.md`, +`specs/0001-foo.md`) so artifacts inside and outside `.sdlc/` share one namespace with the ledger. + +No writes, no gate — this is pure graph construction the advisory audit surface consumes. +""" + +import re +from pathlib import Path + +import yaml + +PLUGIN_ROOT = Path(__file__).resolve().parent.parent +import sys + +sys.path.insert(0, str(Path(__file__).resolve().parent)) +import phase_model as pm + +# Permissive id matcher across the whole vocabulary (widths vary per prefix; see references). +ID_RE = re.compile(r"\b(FR|EP|US|BR|SCEN|DL|DOC|FE|NFR|ADR)-[A-Z]*\d+\b") + +# id prefix -> the filename stem that *declares* that id family (the owning template artifact). +# An id appearing in any OTHER file is a reference, i.e. that file depends on the owner. +ID_OWNER_STEM = { + "FR": "requirements", + "NFR": "non-functional-requirements", + "EP": "epics", + "US": "user-stories", + "BR": "business-rules", + "SCEN": "golden-scenarios", + "DOC": "document-registry", + "FE": "feature-brief", + "ADR": "adr-registry", + # DL (decision-log) items are process bookkeeping, not an artifact-lineage source — skipped. +} + +# A markdown path reference: [text](path.md) or a bare "*.md" mention. +MD_PATH_RE = re.compile(r"\[[^\]]*\]\(([^)]+\.md)\)|(? str: + try: + return path.resolve().relative_to(repo_root.resolve()).as_posix() + except ValueError: + return path.as_posix() + + +def discover_nodes(repo_root: Path, sdlc_dir: Path) -> list[str]: + """Every markdown artifact worth tracking: phase artifacts, frozen layers, specs.""" + nodes: set[str] = set() + for base in (sdlc_dir / "artifacts", sdlc_dir / "context" / "layers"): + if base.exists(): + for p in base.rglob("*.md"): + if p.is_file(): + nodes.add(_rel(repo_root, p)) + specs = repo_root / "specs" + if specs.exists(): + for p in specs.glob("*.md"): + if p.is_file(): + nodes.add(_rel(repo_root, p)) + return sorted(nodes) + + +def read_yaml_frontmatter(text: str) -> dict: + """YAML frontmatter as a dict (handles list-valued keys like source_artifacts:). Empty on miss. + + The shared check_spec.parse_frontmatter is a flat string parser and CANNOT read lists, so + lineage — which needs `source_artifacts: [...]` — uses real YAML here (as validate_frozen_layer + does).""" + m = re.match(r"^---\s*\n(.*?)\n---", text, re.DOTALL) + if not m: + return {} + try: + data = yaml.safe_load(m.group(1)) + return data if isinstance(data, dict) else {} + except yaml.YAMLError: + return {} + + +# --- Phase mapping (for coarse fallback and cross-ref direction) ------------------------------ + +def _slug_to_order() -> dict[str, int]: + return {p["slug"]: p["order"] for p in pm.all_phases()} + + +def node_phase_order(node: str) -> int | None: + """Lifecycle order of the phase a node belongs to, or None if it can't be placed.""" + parts = node.split("/") + if ".sdlc" in parts and "artifacts" in parts: + i = parts.index("artifacts") + if i + 1 < len(parts): + return _slug_to_order().get(parts[i + 1]) + m = re.search(r"context/layers/phase([^/-]+)-", node) + if m: + return pm.phase_order(m.group(1)) + if node.startswith("specs/"): + return pm.phase_order("build") + return None + + +# --- Edge harvesting -------------------------------------------------------------------------- + +def _edge(upstream: str, downstream: str, basis: str, confidence: str) -> dict: + return {"upstream": upstream, "downstream": downstream, "basis": basis, "confidence": confidence} + + +def find_id_declarations(repo_root: Path, nodes: list[str]) -> dict[str, str]: + """Map each id (FR-012, BR-04, …) to the node that declares it (its owning artifact). + + Ownership is by the id's owning filename stem (requirements.md owns FR, business-rules.md owns + BR, …) — NOT by where the id happens to appear. If several files share a stem, the shortest path + wins (the canonical top-level one).""" + stem_to_node: dict[str, str] = {} + for node in nodes: + # Owners are the template artifacts that LIVE under .sdlc/artifacts/. A file elsewhere + # (a spec, a frozen layer) that merely shares a stem — e.g. specs/requirements.md — must + # not capture ownership, or it would produce a reversed edge against the real declarer. + parts = node.split("/") + if not (".sdlc" in parts and "artifacts" in parts): + continue + stem = Path(node).stem.lower() + for owner_stem in set(ID_OWNER_STEM.values()): + if (stem == owner_stem or stem.startswith(owner_stem)) and ( + owner_stem not in stem_to_node or len(node) < len(stem_to_node[owner_stem]) + ): + stem_to_node[owner_stem] = node + + owners: dict[str, str] = {} + for node in nodes: + try: + text = (repo_root / node).read_text(encoding="utf-8", errors="replace") + except OSError: + continue + for m in ID_RE.finditer(text): + stem = ID_OWNER_STEM.get(m.group(1)) + if stem and stem in stem_to_node: + owners.setdefault(m.group(0), stem_to_node[stem]) + return owners + + +def harvest_edges(repo_root: Path, sdlc_dir: Path, nodes: list[str]) -> list[dict]: + """All DECLARED edges (frozen-layer sources, spec sources, id references, cross-references).""" + edges: list[dict] = [] + node_set = set(nodes) + id_owner = find_id_declarations(repo_root, nodes) + + def add(up: str, down: str, basis: str): + if up and down and up != down and up in node_set and down in node_set: + edges.append(_edge(up, down, basis, "declared")) + + for node in nodes: + try: + text = (repo_root / node).read_text(encoding="utf-8", errors="replace") + except OSError: + continue + + # (1) Frozen-layer source_artifacts: bare filenames under the phase's artifact dir. + if "context/layers/phase" in node: + fm = read_yaml_frontmatter(text) + srcs = fm.get("source_artifacts") or [] + order = node_phase_order(node) + slug = next((p["slug"] for p in pm.all_phases() if p["order"] == order), None) + if isinstance(srcs, list) and slug: + for fn in srcs: + up = f".sdlc/artifacts/{slug}/{str(fn).strip()}" + add(up, node, "frozen-layer-source") + + # (2) id references: the file declaring an id is upstream of any file that references it. + for m in ID_RE.finditer(text): + owner = id_owner.get(m.group(0)) + if owner and owner != node: + add(owner, node, "id-reference") + + # (3) explicit markdown path references, directed by phase order (upstream = earlier phase). + down_order = node_phase_order(node) + for m in MD_PATH_RE.finditer(text): + ref = (m.group(1) or m.group(2) or "").strip() + target = _resolve_path_ref(repo_root, sdlc_dir, node, ref, node_set) + if not target or target == node: + continue + up_order = node_phase_order(target) + if up_order is not None and down_order is not None and up_order < down_order: + add(target, node, "cross-reference") + + # De-duplicate (same up→down can be found by several bases; keep the first/strongest basis). + seen: set[tuple[str, str]] = set() + deduped: list[dict] = [] + for e in edges: + key = (e["upstream"], e["downstream"]) + if key not in seen: + seen.add(key) + deduped.append(e) + return deduped + + +def _resolve_path_ref(repo_root: Path, sdlc_dir: Path, from_node: str, ref: str, node_set: set) -> str | None: + """Resolve a markdown path reference to a known node, if possible.""" + ref = ref.split("#", 1)[0].strip() + if not ref: + return None + from_dir = (repo_root / from_node).parent + candidates = [ + (from_dir / ref), + (repo_root / ref), + ] + base = Path(ref).name + for cand in candidates: + rel = _rel(repo_root, cand) + if rel in node_set: + return rel + # Fall back to a unique basename match among known nodes. + matches = [n for n in node_set if Path(n).name == base] + return matches[0] if len(matches) == 1 else None + + +def coarse_edges(nodes: list[str], declared: list[dict]) -> list[dict]: + """Phase-order fallback: for a node with NO declared upstream, connect every artifact in the + immediately-prior populated phase to it. Always labeled 'coarse'.""" + has_declared_up = {e["downstream"] for e in declared} + by_order: dict[int, list[str]] = {} + for n in nodes: + o = node_phase_order(n) + if o is not None: + by_order.setdefault(o, []).append(n) + orders = sorted(by_order) + prev_of = {orders[i]: orders[i - 1] for i in range(1, len(orders))} + + edges: list[dict] = [] + for order in orders: + prev = prev_of.get(order) + if prev is None: + continue + for down in by_order[order]: + if down in has_declared_up: + continue + for up in by_order[prev]: + if up != down: + edges.append(_edge(up, down, "coarse-phase-order", "coarse")) + return edges + + +def build_graph(repo_root: Path, sdlc_dir: Path) -> list[dict]: + """The full edge list: declared edges first, then coarse fallback for undeclared downstreams.""" + nodes = discover_nodes(repo_root, sdlc_dir) + declared = harvest_edges(repo_root, sdlc_dir, nodes) + return declared + coarse_edges(nodes, declared) + + +# --- Traversal (cycle-safe) ------------------------------------------------------------------- + +def _walk(edges: list[dict], start: str, forward: bool) -> list[dict]: + """BFS from `start`, following edges forward (downstream) or backward (upstream). Cycle-safe. + + Returns one row per reachable node: {node, path (list of intermediate nodes), confidence, + bases}. `confidence` is 'coarse' if ANY edge on the shortest path is coarse, else 'declared'.""" + key = "upstream" if forward else "downstream" + other = "downstream" if forward else "upstream" + adj: dict[str, list[dict]] = {} + for e in edges: + adj.setdefault(e[key], []).append(e) + + results: list[dict] = [] + visited = {start} + queue: list[tuple[str, list[str], str, list[str]]] = [(start, [start], "declared", [])] + while queue: + node, path, conf, bases = queue.pop(0) + for e in adj.get(node, []): + nxt = e[other] + if nxt in visited: + continue + visited.add(nxt) + new_conf = "coarse" if (conf == "coarse" or e["confidence"] == "coarse") else "declared" + new_bases = bases + [e["basis"]] + results.append({ + "node": nxt, + "path": path + [nxt], + "confidence": new_conf, + "bases": new_bases, + }) + queue.append((nxt, path + [nxt], new_conf, new_bases)) + return results + + +def downstream_of(edges: list[dict], node: str) -> list[dict]: + """Everything reachable *downstream* of `node` (if `node` changes, these may go stale).""" + return _walk(edges, node, forward=True) + + +def upstream_of(edges: list[dict], node: str) -> list[dict]: + """Everything `node` depends on (its sources).""" + return _walk(edges, node, forward=False) diff --git a/scripts/artifact_model.py b/scripts/artifact_model.py new file mode 100644 index 0000000..88d5828 --- /dev/null +++ b/scripts/artifact_model.py @@ -0,0 +1,224 @@ +"""artifact_model.py — Single source of truth for the artifact change-ledger entry shape and the +staleness disposition state machine (with honest counting). + +The artifact-update-audit layer records every change to a pre-Build artifact (a requirement, an +epic, a design doc) as an append-only ledger entry, and — when an *upstream* artifact changes after +a *downstream* one last did — surfaces the downstream as *potentially stale*. This module owns two +things and does NO I/O (pure functions, so it is trivially testable and safe to import anywhere): + + 1. The ledger entry vocabulary. Two entry kinds share one JSONL, told apart by `event`: + - a CHANGE entry (event ∈ {created, revised, refreshed, snapshot}) records that an artifact's + content changed: {artifact, target, phase, hash, prev_hash, actor, reason, decision_ref}. + - a DISPOSITION entry (event == "disposition") records a human's judgement about one staleness + candidate: {downstream, upstream, upstream_hash, disposition, owner, reason, actor}. + + 2. The staleness disposition state machine — the review-closure discipline made mechanical so a + stale candidate can't be waved away by relabelling it (mirrors findings_model.py's anti- + relabelling rule): + + - OPEN — untouched. Counts as debt. + - REFRESHED — the downstream was updated *after* the upstream changed. Does NOT count. This + is DERIVED from the ledger (a later change hash), never a word someone types, + so it is trustworthy by construction. + - ACKNOWLEDGED — a human accepts the debt for now. Off the books ONLY with an `owner`; + otherwise it counts exactly like OPEN. + - NOT_AFFECTED — a human judges the change doesn't ripple to this downstream. Off the books + ONLY with a `reason`; otherwise it counts. + +This layer is advisory: nothing here gates. It only lets a human see, and honestly count, what a +change put at risk. Timestamps are always supplied by the caller (never minted here) so the model +stays pure and deterministic under test. +""" + +# --- Ledger event vocabulary ------------------------------------------------------------------ + +# A change entry records that an artifact's *content* moved. +# created — first time this artifact is seen by the ledger (baseline / new file) +# revised — an explicit, human-driven change (the /sdlc-revise write) +# refreshed — a downstream updated in response to an upstream change (closes a staleness item) +# snapshot — drift picked up by a hash-scan (a direct edit the ledger had not yet seen) +CHANGE_EVENTS = ("created", "revised", "refreshed", "snapshot") +DISPOSITION_EVENT = "disposition" +EVENTS = CHANGE_EVENTS + (DISPOSITION_EVENT,) + +DISPOSITIONS = { + "OPEN": "untouched — counts as debt", + "REFRESHED": "downstream updated after the upstream change — DERIVED from the ledger only", + "ACKNOWLEDGED": "human accepts the debt — off the books only with an owner", + "NOT_AFFECTED": "human judges no ripple — off the books only with a reason", +} + +# The dispositions a human may RECORD. REFRESHED is deliberately excluded: it is derived from the +# ledger (a downstream that changed after its upstream simply stops being a stale candidate — see +# the timestamp filter in audit_artifacts.compute_staleness), never a word someone types. A typed +# REFRESHED is exactly the relabelling this state machine exists to reject. +SETTABLE_DISPOSITIONS = ("OPEN", "ACKNOWLEDGED", "NOT_AFFECTED") + +# Fields a disposition must carry to be "off the books". OPEN can never be off the books; a recorded +# REFRESHED is not trusted (see validate_disposition). +_REQUIRED_FIELDS = { + "ACKNOWLEDGED": ("owner",), + "NOT_AFFECTED": ("reason",), +} + + +# --- Normalization ---------------------------------------------------------------------------- + +def normalize_event(event) -> str | None: + """Canonical lower-case event, or None if not one of EVENTS.""" + if event is None: + return None + e = str(event).strip().lower() + return e if e in EVENTS else None + + +def normalize_disposition(disp) -> str | None: + """Canonical upper-case disposition, or None if not one of DISPOSITIONS.""" + if disp is None: + return None + s = str(disp).strip().upper() + return s if s in DISPOSITIONS else None + + +def is_change_entry(entry) -> bool: + return isinstance(entry, dict) and normalize_event(entry.get("event")) in CHANGE_EVENTS + + +def is_disposition_entry(entry) -> bool: + return isinstance(entry, dict) and normalize_event(entry.get("event")) == DISPOSITION_EVENT + + +# --- Identity --------------------------------------------------------------------------------- + +def staleness_key(downstream: str, upstream: str, upstream_hash: str = "") -> str: + """Stable identity for one staleness candidate: a downstream artifact vs. a specific upstream + *change* (pinned by the upstream's content hash so a later upstream change re-opens the item + rather than silently inheriting the old disposition).""" + down = str(downstream or "").strip() + up = str(upstream or "").strip() + h = str(upstream_hash or "").strip() + return f"{down}<-{up}@{h}" if h else f"{down}<-{up}" + + +# --- Honest counting -------------------------------------------------------------------------- + +def validate_disposition(item: dict) -> tuple[bool, str]: + """Is this staleness item 'off the books' (does not count toward debt), and why / why not. + + A mislabeled disposition — ACKNOWLEDGED without an owner, NOT_AFFECTED without a reason — is NOT + off the books; it counts exactly like OPEN. You cannot clear debt by typing a word. + """ + disp = normalize_disposition(item.get("disposition")) + if disp is None: + return False, f"unknown disposition {item.get('disposition')!r}" + if disp == "OPEN": + return False, "open" + if disp == "REFRESHED": + # REFRESHED is DERIVED, not recordable. A genuinely refreshed downstream never reaches here + # (it is filtered out of the stale set by its later change timestamp). So a stale item that + # carries REFRESHED was hand-typed — we do not take the word for it; it still counts as debt. + return False, "REFRESHED is derived from the ledger, not a recordable disposition — still counts" + missing = [f for f in _REQUIRED_FIELDS.get(disp, ()) if not str(item.get(f, "")).strip()] + if missing: + return False, f"{disp} missing required field(s): {', '.join(missing)}" + return True, disp.lower() + + +def counts_as_debt(item: dict) -> bool: + """A staleness item is open debt unless it is legitimately off the books.""" + off_books, _ = validate_disposition(item) + return not off_books + + +def open_debt(items: list[dict]) -> list[dict]: + """The subset of staleness items that still count as debt.""" + return [i for i in items if counts_as_debt(i)] + + +# --- Entry builders (the canonical ledger shape) ---------------------------------------------- + +def change_entry( + *, + ts: str, + artifact: str, + event: str = "revised", + target: str = "", + phase: str = "", + hash: str | None = None, + prev_hash: str | None = None, + actor: str = "", + reason: str = "", + decision_ref: str = "", +) -> dict: + """Build a CHANGE ledger entry. `event` is normalized (falls back to 'revised' if unknown).""" + entry = { + "ts": ts, + "event": normalize_event(event) or "revised", + "artifact": artifact, + "target": target, + "phase": str(phase) if phase != "" and phase is not None else "", + "hash": hash, + "prev_hash": prev_hash, + "actor": actor, + "reason": reason, + } + if decision_ref: + entry["decision_ref"] = decision_ref + return entry + + +def disposition_entry( + *, + ts: str, + downstream: str, + upstream: str, + disposition: str, + upstream_hash: str = "", + owner: str = "", + reason: str = "", + actor: str = "", +) -> dict: + """Build a DISPOSITION ledger entry for one staleness candidate.""" + return { + "ts": ts, + "event": DISPOSITION_EVENT, + "downstream": downstream, + "upstream": upstream, + "upstream_hash": upstream_hash, + "disposition": normalize_disposition(disposition) or str(disposition or "").strip().upper(), + "owner": owner, + "reason": reason, + "actor": actor, + "key": staleness_key(downstream, upstream, upstream_hash), + } + + +# --- Ledger projections (still pure — operate on an in-memory list) ---------------------------- + +def latest_change_per_artifact(ledger: list[dict]) -> dict[str, dict]: + """Latest CHANGE entry per artifact path — the artifact's current state. The ledger is + append-only in time order, so the last write wins.""" + latest: dict[str, dict] = {} + for e in ledger: + if is_change_entry(e): + art = e.get("artifact") + if art: + latest[art] = e + return latest + + +def latest_dispositions(ledger: list[dict]) -> dict[str, dict]: + """Latest DISPOSITION entry per staleness key.""" + latest: dict[str, dict] = {} + for e in ledger: + if is_disposition_entry(e): + key = e.get("key") or staleness_key( + e.get("downstream", ""), e.get("upstream", ""), e.get("upstream_hash", "") + ) + latest[key] = e + return latest + + +def changes_for(ledger: list[dict], artifact: str) -> list[dict]: + """Every CHANGE entry for one artifact, in ledger (time) order — its history trail.""" + return [e for e in ledger if is_change_entry(e) and e.get("artifact") == artifact] diff --git a/scripts/audit_artifacts.py b/scripts/audit_artifacts.py new file mode 100644 index 0000000..b70003a --- /dev/null +++ b/scripts/audit_artifacts.py @@ -0,0 +1,559 @@ +"""audit_artifacts.py — the advisory engine behind /sdlc-audit-artifacts and /sdlc-revise. + +Connects two facts the tool already has — *when each artifact last changed* (SHA-256 hashes) and +*what depends on what* (declared traceability, via artifact_lineage) — into staleness detection, an +append-only change history, and a forward "what would this change put at risk" impact view. + +Three subcommands, all **exit 0 always** (advisory — staleness is never a gate): + + record append to the ledger. Three modes: + --scan hash-walk every artifact; append `created` for new files and + `snapshot` for drifted ones. The first-ever scan is the + baseline ("history starts now" — the retro-baseline). + --artifact P [--target ID] one explicit change entry (the /sdlc-revise write). + --actor A --reason R [--event revised] [--decision-ref DL-NN] [--phase P] + --disposition SET a human's judgement on one staleness candidate. + --downstream D --upstream U [--owner O] [--reason R] [--actor A] + impact forward lineage: everything downstream that a change here could make stale. + report freshness dashboard (default) | --history | --json | --since YYYY-MM-DD + +The ledger lives in its OWN JSONL (.sdlc/metrics/artifact-log.jsonl), never inside state.yaml's +gate_results (that dict is expanded into phantom rows by audit_gates.extract_gate_history — the trap +this layer must not fall into). + +Standalone or Workflow (CLAUDE.md design rule): + --repo standalone (reads /.sdlc) | --state .sdlc/state.yaml in-workflow +""" + +import argparse +import json +import sys +from datetime import datetime, timezone +from pathlib import Path + +import yaml + +sys.path.insert(0, str(Path(__file__).resolve().parent)) +import artifact_lineage as al +import artifact_model as am +import phase_model as pm +from track_artifacts import compute_checksum + +LEDGER_NAME = "artifact-log.jsonl" + + +# --- Path / state plumbing -------------------------------------------------------------------- + +def resolve_paths(args) -> tuple[Path, Path, Path]: + """(base_dir, sdlc_dir, metrics_dir) from --state (the .sdlc beside it) or --repo (/.sdlc).""" + if getattr(args, "state", None): + state_path = Path(args.state) + if not state_path.exists(): + print(f"Error: State file not found: {state_path}") + sys.exit(0) # advisory — never a hard failure + sdlc = state_path.resolve().parent + return sdlc.parent, sdlc, sdlc / "metrics" + repo = Path(args.repo).resolve() + return repo, repo / ".sdlc", repo / ".sdlc" / "metrics" + + +def now_iso() -> str: + return datetime.now(timezone.utc).isoformat() + + +def load_ledger(ledger_path: Path) -> list[dict]: + if not ledger_path.exists(): + return [] + out = [] + for line in ledger_path.read_text(encoding="utf-8", errors="replace").splitlines(): + line = line.strip() + if line: + try: + obj = json.loads(line) + except json.JSONDecodeError: + continue + # Only object entries are ledger records. A valid-JSON non-object line (null, 42, a + # bare string/array — e.g. from a hand-edit or partial write) is skipped, not appended, + # so downstream .get() calls never hit a non-dict and the exit-0 invariant holds. + if isinstance(obj, dict): + out.append(obj) + return out + + +def append_entries(metrics_dir: Path, entries: list[dict]) -> Path: + metrics_dir.mkdir(parents=True, exist_ok=True) + ledger_path = metrics_dir / LEDGER_NAME + with open(ledger_path, "a", encoding="utf-8") as f: + for e in entries: + f.write(json.dumps(e) + "\n") + return ledger_path + + +def load_state(sdlc_dir: Path) -> dict: + state_path = sdlc_dir / "state.yaml" + if not state_path.exists(): + return {} + try: + return yaml.safe_load(state_path.read_text(encoding="utf-8", errors="replace")) or {} + except yaml.YAMLError: + return {} + + +# --- Artifact scanning & identity ------------------------------------------------------------- + +def scan_hashes(base_dir: Path, sdlc_dir: Path) -> dict[str, str]: + """Current SHA of every tracked node (same node namespace as artifact_lineage).""" + hashes: dict[str, str] = {} + for node in al.discover_nodes(base_dir, sdlc_dir): + p = base_dir / node + if p.is_file(): + try: + hashes[node] = compute_checksum(p) + except OSError: + continue # unreadable file (e.g. mode 000) — skip, never crash the advisory scan + return hashes + + +_SLUG_TO_ID = {p["slug"]: pm.normalize_id(p["id"]) for p in pm.all_phases()} + + +def phase_id_of(node: str) -> str: + """Best-effort phase id a node belongs to (for the ledger's `phase` field). '' if unknown.""" + parts = node.split("/") + if ".sdlc" in parts and "artifacts" in parts: + i = parts.index("artifacts") + if i + 1 < len(parts): + return _SLUG_TO_ID.get(parts[i + 1], "") or "" + if "context/layers/phase" in node: + import re + m = re.search(r"context/layers/phase([^/-]+)-", node) + if m and pm.get_phase(m.group(1)): + return pm.normalize_id(m.group(1)) + if node.startswith("specs/"): + return "build" + return "" + + +def resolve_target_node(base_dir: Path, sdlc_dir: Path, target: str, nodes: list[str]) -> str | None: + """Resolve an id (FR-012) or a path to a known node. None if it can't be placed.""" + target = (target or "").strip() + if not target: + return None + if al.ID_RE.fullmatch(target): + return al.find_id_declarations(base_dir, nodes).get(target) + # A path (absolute, repo-relative, or bare filename). + return al._resolve_path_ref(base_dir, sdlc_dir, nodes[0] if nodes else ".", target, set(nodes)) \ + or _rel_if_node(base_dir, target, set(nodes)) + + +def _rel_if_node(base_dir: Path, target: str, node_set: set) -> str | None: + try: + rel = (base_dir / target).resolve().relative_to(base_dir.resolve()).as_posix() + except ValueError: + rel = target + return rel if rel in node_set else None + + +def _parse_ts(ts: str): + try: + dt = datetime.fromisoformat(str(ts)) + except (ValueError, TypeError): + return None + # Ledger timestamps are tz-aware (now_iso); a bare-date --since (the documented YYYY-MM-DD form) + # parses naive. Coerce naive to UTC so `da < db` never mixes aware/naive and raises TypeError. + return dt.replace(tzinfo=timezone.utc) if dt.tzinfo is None else dt + + +def _ts_lt(a: str, b: str) -> bool: + """True if timestamp a is strictly before b (parsed; falls back to string order).""" + da, db = _parse_ts(a), _parse_ts(b) + if da is not None and db is not None: + return da < db + return str(a) < str(b) + + +# --- Staleness --------------------------------------------------------------------------------- + +def compute_staleness(base_dir: Path, sdlc_dir: Path, ledger: list[dict]) -> list[dict]: + """One item per (upstream→downstream) edge where the upstream changed *after* the downstream + last did. Each item carries its resolved disposition (recorded, else OPEN) so honest counting + can run over the list.""" + graph = al.build_graph(base_dir, sdlc_dir) + latest = am.latest_change_per_artifact(ledger) + dispositions = am.latest_dispositions(ledger) + + items: list[dict] = [] + for e in graph: + up, down = e["upstream"], e["downstream"] + cu, cd = latest.get(up), latest.get(down) + if not cu or not cd: + continue + t_up, t_down = cu.get("ts", ""), cd.get("ts", "") + if not _ts_lt(t_down, t_up): + continue # downstream is at least as fresh as the upstream change — not stale + up_hash = cu.get("hash") or "" + key = am.staleness_key(down, up, up_hash) + recorded = dispositions.get(key) + item = { + "downstream": down, + "upstream": up, + "upstream_hash": up_hash, + "basis": e["basis"], + "confidence": e["confidence"], + "upstream_changed": t_up, + "downstream_changed": t_down, + "disposition": "OPEN", + } + if recorded: + item["disposition"] = am.normalize_disposition(recorded.get("disposition")) or "OPEN" + for k in ("owner", "reason"): + if recorded.get(k): + item[k] = recorded[k] + items.append(item) + return items + + +# --- record ------------------------------------------------------------------------------------ + +def do_scan(base_dir: Path, sdlc_dir: Path, metrics_dir: Path, actor: str) -> int: + ledger = load_ledger(metrics_dir / LEDGER_NAME) + first_run = not any(am.is_change_entry(e) for e in ledger) + latest = am.latest_change_per_artifact(ledger) + hashes = scan_hashes(base_dir, sdlc_dir) + ts = now_iso() + actor = actor or ("baseline" if first_run else "scan") + + new_entries: list[dict] = [] + created = drifted = 0 + for node, sha in sorted(hashes.items()): + prev = latest.get(node) + if prev is None: + new_entries.append(am.change_entry( + ts=ts, artifact=node, event="created", phase=phase_id_of(node), + hash=sha, actor=actor, reason="baseline" if first_run else "first seen")) + created += 1 + elif prev.get("hash") != sha: + new_entries.append(am.change_entry( + ts=ts, artifact=node, event="snapshot", phase=phase_id_of(node), + hash=sha, prev_hash=prev.get("hash"), actor=actor, reason="drift detected by scan")) + drifted += 1 + + if not new_entries: + print("Artifact scan: no changes since the last ledger entry (nothing to record).") + return 0 + append_entries(metrics_dir, new_entries) + if first_run: + print(f"Baseline recorded: {created} artifact(s) hashed — history starts now.") + else: + print(f"Recorded {len(new_entries)} change(s): {created} new, {drifted} drifted.") + return 0 + + +def do_change(args, base_dir: Path, sdlc_dir: Path, metrics_dir: Path) -> int: + node = _normalize_artifact_arg(base_dir, args.artifact) + ledger = load_ledger(metrics_dir / LEDGER_NAME) + prev = am.latest_change_per_artifact(ledger).get(node) + p = base_dir / node + try: + sha = compute_checksum(p) if p.is_file() else None + except OSError: + sha = None # unreadable file — record the change with an unknown hash, never crash + entry = am.change_entry( + ts=now_iso(), artifact=node, event=args.event, target=args.target or "", + phase=args.phase or phase_id_of(node), hash=sha, + prev_hash=prev.get("hash") if prev else None, + actor=args.actor or "", reason=args.reason or "", decision_ref=args.decision_ref or "") + append_entries(metrics_dir, [entry]) + tgt = f" ({args.target})" if args.target else "" + print(f"Recorded {entry['event']} of {node}{tgt} by {args.actor or 'unknown'}" + + (f" — ref {args.decision_ref}" if args.decision_ref else "")) + return 0 + + +def do_disposition(args, base_dir: Path, sdlc_dir: Path, metrics_dir: Path) -> int: + disp = am.normalize_disposition(args.disposition) + if disp not in am.SETTABLE_DISPOSITIONS: + print(f"record --disposition: '{args.disposition}' is not a recordable disposition " + f"(choose one of {', '.join(am.SETTABLE_DISPOSITIONS)}). REFRESHED is DERIVED from the " + f"ledger — update the downstream artifact and re-scan; you cannot type it. Nothing recorded.") + return 0 + nodes = al.discover_nodes(base_dir, sdlc_dir) + down = resolve_target_node(base_dir, sdlc_dir, args.downstream, nodes) or args.downstream + up = resolve_target_node(base_dir, sdlc_dir, args.upstream, nodes) or args.upstream + ledger = load_ledger(metrics_dir / LEDGER_NAME) + up_change = am.latest_change_per_artifact(ledger).get(up) + up_hash = args.upstream_hash or (up_change.get("hash") if up_change else "") or "" + entry = am.disposition_entry( + ts=now_iso(), downstream=down, upstream=up, upstream_hash=up_hash, + disposition=args.disposition, owner=args.owner or "", reason=args.reason or "", + actor=args.actor or "") + append_entries(metrics_dir, [entry]) + off, why = am.validate_disposition(entry) + tag = "off the books" if off else f"STILL COUNTS as debt ({why})" + print(f"Recorded disposition {entry['disposition']} for {down} <- {up}: {tag}.") + return 0 + + +def _normalize_artifact_arg(base_dir: Path, artifact: str) -> str: + try: + return (base_dir / artifact).resolve().relative_to(base_dir.resolve()).as_posix() + except ValueError: + return artifact.replace("\\", "/") + + +def cmd_record(args) -> int: + base_dir, sdlc_dir, metrics_dir = resolve_paths(args) + if args.disposition: + if not (args.downstream and args.upstream): + print("record --disposition needs --downstream and --upstream (nothing recorded).") + return 0 + return do_disposition(args, base_dir, sdlc_dir, metrics_dir) + if args.scan: + return do_scan(base_dir, sdlc_dir, metrics_dir, args.actor or "") + if args.artifact: + return do_change(args, base_dir, sdlc_dir, metrics_dir) + print("record: choose a mode — --scan, --artifact , or --disposition " + "(nothing recorded).") + return 0 + + +# --- impact ------------------------------------------------------------------------------------ + +def cmd_impact(args) -> int: + base_dir, sdlc_dir, _ = resolve_paths(args) + nodes = al.discover_nodes(base_dir, sdlc_dir) + node = resolve_target_node(base_dir, sdlc_dir, args.target, nodes) + if not node: + print(f"Impact — could not resolve '{args.target}' to a known artifact.") + print("(advisory — nothing to assess)") + return 0 + graph = al.build_graph(base_dir, sdlc_dir) + rows = al.downstream_of(graph, node) + if args.json: + print(json.dumps({"target": node, "downstream": rows}, indent=2)) + return 0 + print(f"Impact — if {node} changes, {len(rows)} downstream artifact(s) may go stale:") + if not rows: + print(" (no declared or inferred downstream)") + for r in sorted(rows, key=lambda x: (x["confidence"], x["node"])): + label = "declared" if r["confidence"] == "declared" else "COARSE (phase-order guess)" + path = " -> ".join(r["path"][1:]) if len(r["path"]) > 2 else r["node"] + print(f" • {r['node']} [{label}]" + (f" via {path}" if len(r["path"]) > 2 else "")) + print("(advisory — a downstream may already account for the change; disposition each in " + "/sdlc-audit-artifacts)") + return 0 + + +# --- report ------------------------------------------------------------------------------------ + +def cmd_report(args) -> int: + base_dir, sdlc_dir, metrics_dir = resolve_paths(args) + ledger = load_ledger(metrics_dir / LEDGER_NAME) + + if args.history: + return _report_history(base_dir, sdlc_dir, ledger, args) + + full_latest = am.latest_change_per_artifact(ledger) + has_history = bool(full_latest) + items = compute_staleness(base_dir, sdlc_dir, ledger) + latest, items, scope = _apply_scope(base_dir, sdlc_dir, full_latest, items, args) + + debt = am.open_debt(items) + # "dispositioned" == legitimately off the books (ACKNOWLEDGED w/ owner, NOT_AFFECTED w/ reason). + # Defined as the complement of debt so the footer count always reconciles with the per-node line. + acknowledged = [i for i in items if not am.counts_as_debt(i)] + stale_downstreams = sorted({i["downstream"] for i in items}) + + if args.json: + print(json.dumps({ + "has_history": has_history, + "scope": scope, + "artifacts_tracked": len(full_latest), + "artifacts_shown": len(latest), + "stale": len(stale_downstreams), + "open": len(debt), + "acknowledged": len(acknowledged), + "items": items, + }, indent=2)) + return 0 + + print(_format_freshness(sdlc_dir, latest, items, debt, acknowledged, scope, has_history)) + return 0 + + +def _apply_scope(base_dir: Path, sdlc_dir: Path, latest: dict, items: list[dict], args): + """Restrict the freshness view to --artifact, --phase, and/or --since. Returns + (scoped_latest, scoped_items, scope_label).""" + scope = "all" + if getattr(args, "artifact", None): + nodes = al.discover_nodes(base_dir, sdlc_dir) + node = resolve_target_node(base_dir, sdlc_dir, args.artifact, nodes) or \ + _normalize_artifact_arg(base_dir, args.artifact) + latest = {k: v for k, v in latest.items() if k == node} + items = [i for i in items if i["downstream"] == node] + scope = f"artifact:{node}" + elif getattr(args, "phase", None): + pid = pm.normalize_id(args.phase) + latest = {k: v for k, v in latest.items() if phase_id_of(k) == pid} + items = [i for i in items if phase_id_of(i["downstream"]) == pid] + scope = f"phase:{pid}" + if getattr(args, "since", None): + latest = {k: v for k, v in latest.items() if not _ts_lt(v.get("ts", ""), args.since)} + items = [i for i in items if i["downstream"] in latest] + scope = f"{scope}+since:{args.since}" if scope != "all" else f"since:{args.since}" + return latest, items, scope + + +def _report_history(base_dir: Path, sdlc_dir: Path, ledger: list[dict], args) -> int: + nodes = al.discover_nodes(base_dir, sdlc_dir) + node = resolve_target_node(base_dir, sdlc_dir, args.history, nodes) or \ + _normalize_artifact_arg(base_dir, args.history) + trail = am.changes_for(ledger, node) + if args.since: + trail = [e for e in trail if not _ts_lt(e.get("ts", ""), args.since)] + if args.json: + print(json.dumps({"artifact": node, "history": trail}, indent=2)) + return 0 + print(f"History — {node}") + print("=" * 50) + if not trail: + print(" (no recorded changes — run `record --scan` to seed a baseline)") + return 0 + for e in trail: + who = e.get("actor") or "unknown" + tgt = f" [{e['target']}]" if e.get("target") else "" + why = f' — "{e["reason"]}"' if e.get("reason") else "" + ref = f" ({e['decision_ref']})" if e.get("decision_ref") else "" + print(f" {e.get('ts', '?')[:19]} {e.get('event', '?'):9}{tgt} by {who}{why}{ref}") + return 0 + + +def _glyph(preferred: str, fallback: str) -> str: + """`preferred` if the console can encode it, else a plain-ASCII `fallback`. + + A default Windows console is cp1252 and cannot encode '←' or '✓'. Printing one raises + UnicodeEncodeError and kills the process with a NON-ZERO exit — precisely what this advisory + module promises never to do — and it fires on the happy path, since the arrow prints for every + stale item and the check for every signed-off phase. So the tool would die exactly when it had + something to report. Resolved per call rather than at import because stdout may be replaced + after this module loads (test capture, redirection, a wrapping harness). + """ + encoding = getattr(sys.stdout, "encoding", None) or "utf-8" + try: + preferred.encode(encoding) + except (UnicodeEncodeError, LookupError): + return fallback + return preferred + + +def _format_freshness(sdlc_dir, latest, items, debt, acknowledged, scope, has_history) -> str: + state = load_state(sdlc_dir) + phases = state.get("phases", {}) if isinstance(state, dict) else {} + + header = "Artifact Freshness" + ("" if scope == "all" else f" (scope: {scope})") + lines = [header, "=" * 50] + if not latest: + if not has_history: + lines.append("No artifact history yet.") + lines.append("Run `record --scan` to seed a baseline (history starts now); nothing is") + lines.append("fabricated about the past.") + else: + lines.append(f"No tracked artifacts match this scope ({scope}).") + lines.append("=" * 50) + lines.append("ADVISORY — nothing to assess (advisory surface — never blocks).") + return "\n".join(lines) + + arrow = _glyph("←", "<-") + stale_by_down: dict[str, list[dict]] = {} + for i in items: + stale_by_down.setdefault(i["downstream"], []).append(i) + + for node in sorted(latest): + c = latest[node] + when = c.get("ts", "?")[:10] + who = c.get("actor") or "?" + node_items = stale_by_down.get(node, []) + if not node_items: + status = "FRESH" + else: + n_open = len(am.open_debt(node_items)) + status = f"STALE ({n_open} open, {len(node_items) - n_open} dispositioned)" + signed = _signoff_note(node, phases) + lines.append(f" {node}") + lines.append(f" changed {when} by {who} {signed} {status}") + for i in node_items: + disp = am.normalize_disposition(i.get("disposition")) or "OPEN" + owner = f" {i['owner']}" if i.get("owner") else "" + conf = "" if i["confidence"] == "declared" else " [coarse]" + lines.append(f" {arrow} {i['upstream']}{conf} ({disp}{owner})") + + lines.append("=" * 50) + lines.append(f"{len(latest)} artifact(s) tracked · " + f"{len({i['downstream'] for i in items})} stale · " + f"{len(debt)} open · {len(acknowledged)} dispositioned") + lines.append("ADVISORY — staleness is a candidate a human dispositions, never a gate (exit 0).") + return "\n".join(lines) + + +def _signoff_note(node: str, phases: dict) -> str: + pid = phase_id_of(node) + pdata = phases.get(pid) if isinstance(phases, dict) else None + if isinstance(pdata, dict): + if pdata.get("sign_off") or pdata.get("signed_by"): + return f"signed-off {_glyph('✓', '(y)')}" + if pdata.get("status") == "completed": + return "phase completed" + return "unsigned" + + +# --- CLI --------------------------------------------------------------------------------------- + +def main() -> None: + parser = argparse.ArgumentParser( + description="Advisory artifact change-ledger, staleness, and impact (never blocks; exit 0)") + sub = parser.add_subparsers(dest="command", required=True) + + common = argparse.ArgumentParser(add_help=False) + src = common.add_mutually_exclusive_group() + src.add_argument("--state", help="Path to .sdlc/state.yaml (workflow mode)") + src.add_argument("--repo", default=".", help="Repo root containing .sdlc/ (standalone; default cwd)") + + p_rec = sub.add_parser("record", parents=[common], help="Append change or disposition entries") + p_rec.add_argument("--scan", action="store_true", help="Hash-walk artifacts; record new/drifted files") + p_rec.add_argument("--artifact", help="Record one explicit change to this artifact path") + p_rec.add_argument("--target", help="Sub-artifact id the change touched (e.g. FR-012)") + p_rec.add_argument("--event", default="revised", help="Change event (created/revised/refreshed/snapshot)") + p_rec.add_argument("--phase", help="Phase id the artifact belongs to (inferred if omitted)") + p_rec.add_argument("--actor", help="Who made the change") + p_rec.add_argument("--reason", help="Why the change was made") + p_rec.add_argument("--decision-ref", dest="decision_ref", help="Linked decision-log id (DL-NN)") + p_rec.add_argument("--disposition", help="Record a staleness disposition (OPEN/ACKNOWLEDGED/NOT_AFFECTED; REFRESHED is derived, not recordable)") + p_rec.add_argument("--downstream", help="Downstream artifact for a disposition") + p_rec.add_argument("--upstream", help="Upstream artifact for a disposition") + p_rec.add_argument("--upstream-hash", dest="upstream_hash", help="Pin the disposition to this upstream hash") + p_rec.add_argument("--owner", help="Owner (required for ACKNOWLEDGED to be off the books)") + + p_imp = sub.add_parser("impact", parents=[common], help="Forward: what a change here could make stale") + p_imp.add_argument("target", help="An id (FR-012) or an artifact path") + p_imp.add_argument("--json", action="store_true", help="Emit JSON") + + p_rep = sub.add_parser("report", parents=[common], help="Freshness dashboard / history") + p_rep.add_argument("--history", help="Backward change trail for an id or artifact path") + p_rep.add_argument("--artifact", help="Scope the dashboard to one artifact (id or path)") + p_rep.add_argument("--phase", help="Scope the dashboard to one phase id") + p_rep.add_argument("--since", help="Filter to changes on/after this ISO date") + p_rep.add_argument("--json", action="store_true", help="Emit JSON") + + args = parser.parse_args() + if args.command == "record": + rc = cmd_record(args) + elif args.command == "impact": + rc = cmd_impact(args) + else: + rc = cmd_report(args) + # Advisory by construction — never a non-zero exit. + sys.exit(0 if rc is None else 0) + + +if __name__ == "__main__": + main() diff --git a/scripts/tests/test_artifact_lineage.py b/scripts/tests/test_artifact_lineage.py new file mode 100644 index 0000000..9331302 --- /dev/null +++ b/scripts/tests/test_artifact_lineage.py @@ -0,0 +1,178 @@ +"""Tests for artifact_lineage — declared vs coarse edge harvest and cycle-safe traversal.""" + +from pathlib import Path + +from artifact_lineage import ( + build_graph, + coarse_edges, + discover_nodes, + downstream_of, + find_id_declarations, + harvest_edges, + node_phase_order, + upstream_of, +) + + +def _write(p: Path, text: str) -> None: + p.parent.mkdir(parents=True, exist_ok=True) + p.write_text(text, encoding="utf-8") + + +def build_corpus(root: Path) -> Path: + """A tiny but realistic corpus with every declared edge kind + one undeclared downstream.""" + sdlc = root / ".sdlc" + req = sdlc / "artifacts" / "01-requirements" + _write(req / "requirements.md", "# Requirements\n\n## FR-001 Login\nUser can log in.\n") + # epics.md references FR-001 -> declared id-reference edge requirements -> epics + _write(req / "epics.md", "# Epics\n\n## EP-001\nRealizes FR-001.\n") + # A design artifact that declares nothing -> should get a coarse edge only. + _write(sdlc / "artifacts" / "02-design" / "design.md", "# Design\n\nNo ids referenced here.\n") + # A frozen layer whose source_artifacts lists requirements.md -> declared frozen-layer-source edge + _write( + sdlc / "context" / "layers" / "phase1-requirements.md", + "---\nphase: 1\nsource_artifacts:\n - requirements.md\n---\n\n## Decision\nx\n", + ) + # A spec whose source cites FR-001 -> declared id-reference edge requirements -> spec + _write( + root / "specs" / "0001-login.md", + '---\nspec: "0001"\nname: "login"\nsource: "FR-001 (requirements.md)"\n---\n\n# Spec\n', + ) + return sdlc + + +class TestDiscovery: + def test_finds_all_markdown_nodes(self, tmp_path): + sdlc = build_corpus(tmp_path) + nodes = discover_nodes(tmp_path, sdlc) + assert ".sdlc/artifacts/01-requirements/requirements.md" in nodes + assert ".sdlc/context/layers/phase1-requirements.md" in nodes + assert "specs/0001-login.md" in nodes + + def test_phase_order_placement(self, tmp_path): + assert node_phase_order(".sdlc/artifacts/01-requirements/requirements.md") is not None + assert node_phase_order(".sdlc/context/layers/phase1-requirements.md") == \ + node_phase_order(".sdlc/artifacts/01-requirements/requirements.md") + assert node_phase_order("specs/0001-login.md") is not None + + +class TestIdDeclarations: + def test_ids_owned_by_their_template_stem(self, tmp_path): + sdlc = build_corpus(tmp_path) + nodes = discover_nodes(tmp_path, sdlc) + owners = find_id_declarations(tmp_path, nodes) + assert owners["FR-001"] == ".sdlc/artifacts/01-requirements/requirements.md" + assert owners["EP-001"] == ".sdlc/artifacts/01-requirements/epics.md" + + def test_stem_collision_outside_artifacts_does_not_hijack_ownership(self, tmp_path): + # A file that merely SHARES a stem but lives outside .sdlc/artifacts/ (here specs/requirements.md, + # a shorter path) must NOT capture FR ownership — that produced a reversed declared edge. + sdlc = build_corpus(tmp_path) + _write(tmp_path / "specs" / "requirements.md", "# Not the real owner\nMentions FR-001.\n") + nodes = discover_nodes(tmp_path, sdlc) + owners = find_id_declarations(tmp_path, nodes) + assert owners["FR-001"] == ".sdlc/artifacts/01-requirements/requirements.md" + # And no reversed edge from the impostor into the real declarer. + edges = harvest_edges(tmp_path, sdlc, nodes) + assert not [e for e in edges + if e["upstream"] == "specs/requirements.md" + and e["downstream"] == ".sdlc/artifacts/01-requirements/requirements.md"] + + +class TestDeclaredEdges: + def test_all_three_declared_bases_present(self, tmp_path): + sdlc = build_corpus(tmp_path) + nodes = discover_nodes(tmp_path, sdlc) + edges = harvest_edges(tmp_path, sdlc, nodes) + req = ".sdlc/artifacts/01-requirements/requirements.md" + pairs = {(e["upstream"], e["downstream"]): e for e in edges} + + # id-reference: requirements -> epics and requirements -> spec + assert pairs[(req, ".sdlc/artifacts/01-requirements/epics.md")]["basis"] == "id-reference" + assert pairs[(req, "specs/0001-login.md")]["basis"] == "id-reference" + # frozen-layer-source: requirements -> its layer + assert pairs[(req, ".sdlc/context/layers/phase1-requirements.md")]["basis"] == "frozen-layer-source" + assert all(e["confidence"] == "declared" for e in edges) + + def test_no_self_edges(self, tmp_path): + sdlc = build_corpus(tmp_path) + nodes = discover_nodes(tmp_path, sdlc) + for e in harvest_edges(tmp_path, sdlc, nodes): + assert e["upstream"] != e["downstream"] + + +class TestCoarseFallback: + def test_undeclared_downstream_gets_coarse_edge(self, tmp_path): + sdlc = build_corpus(tmp_path) + graph = build_graph(tmp_path, sdlc) + design = ".sdlc/artifacts/02-design/design.md" + coarse_into_design = [e for e in graph if e["downstream"] == design and e["confidence"] == "coarse"] + assert coarse_into_design, "design.md declares nothing → must get a labeled coarse edge" + assert all(e["basis"] == "coarse-phase-order" for e in coarse_into_design) + + def test_declared_downstream_gets_no_coarse(self, tmp_path): + sdlc = build_corpus(tmp_path) + graph = build_graph(tmp_path, sdlc) + spec = "specs/0001-login.md" # has a declared upstream (FR-001) + assert not [e for e in graph if e["downstream"] == spec and e["confidence"] == "coarse"] + + +class TestTraversal: + def test_downstream_of_requirements(self, tmp_path): + sdlc = build_corpus(tmp_path) + graph = build_graph(tmp_path, sdlc) + req = ".sdlc/artifacts/01-requirements/requirements.md" + reachable = {r["node"] for r in downstream_of(graph, req)} + assert ".sdlc/artifacts/01-requirements/epics.md" in reachable + assert "specs/0001-login.md" in reachable + assert ".sdlc/context/layers/phase1-requirements.md" in reachable + + def test_upstream_of_spec_is_requirements(self, tmp_path): + sdlc = build_corpus(tmp_path) + graph = build_graph(tmp_path, sdlc) + up = {r["node"] for r in upstream_of(graph, "specs/0001-login.md")} + assert ".sdlc/artifacts/01-requirements/requirements.md" in up + + def test_confidence_is_coarse_when_any_edge_on_path_is_coarse(self, tmp_path): + sdlc = build_corpus(tmp_path) + graph = build_graph(tmp_path, sdlc) + req = ".sdlc/artifacts/01-requirements/requirements.md" + rows = {r["node"]: r for r in downstream_of(graph, req)} + design = ".sdlc/artifacts/02-design/design.md" + assert rows[design]["confidence"] == "coarse" + + +class TestCycleSafety: + def test_walk_terminates_on_a_cycle(self): + edges = [ + {"upstream": "a.md", "downstream": "b.md", "basis": "id-reference", "confidence": "declared"}, + {"upstream": "b.md", "downstream": "a.md", "basis": "id-reference", "confidence": "declared"}, + ] + down = {r["node"] for r in downstream_of(edges, "a.md")} + assert down == {"b.md"} # reaches b, does not loop back into a forever + + def test_confidence_taints_across_a_multi_hop_path(self): + # A--coarse-->B--declared-->C : C must be labelled coarse (any coarse edge on the path taints + # the whole chain), and B (reached via the coarse edge) is coarse too. + edges = [ + {"upstream": "a.md", "downstream": "b.md", "basis": "coarse-phase-order", "confidence": "coarse"}, + {"upstream": "b.md", "downstream": "c.md", "basis": "id-reference", "confidence": "declared"}, + ] + rows = {r["node"]: r for r in downstream_of(edges, "a.md")} + assert rows["b.md"]["confidence"] == "coarse" + assert rows["c.md"]["confidence"] == "coarse" + + def test_pure_declared_multi_hop_stays_declared(self): + edges = [ + {"upstream": "a.md", "downstream": "b.md", "basis": "id-reference", "confidence": "declared"}, + {"upstream": "b.md", "downstream": "c.md", "basis": "id-reference", "confidence": "declared"}, + ] + rows = {r["node"]: r for r in downstream_of(edges, "a.md")} + assert rows["c.md"]["confidence"] == "declared" + + def test_coarse_edges_only_for_undeclared(self): + nodes = ["p1/x.md", "p2/y.md"] + declared = [{"upstream": "p1/x.md", "downstream": "p2/y.md", + "basis": "id-reference", "confidence": "declared"}] + # y already has a declared upstream, so coarse_edges must add nothing for it. + assert coarse_edges(nodes, declared) == [] diff --git a/scripts/tests/test_artifact_model.py b/scripts/tests/test_artifact_model.py new file mode 100644 index 0000000..e64c8f6 --- /dev/null +++ b/scripts/tests/test_artifact_model.py @@ -0,0 +1,145 @@ +"""Tests for artifact_model — the ledger vocabulary and the staleness honest-counting rules.""" + +from artifact_model import ( + CHANGE_EVENTS, + DISPOSITION_EVENT, + change_entry, + changes_for, + counts_as_debt, + disposition_entry, + is_change_entry, + is_disposition_entry, + latest_change_per_artifact, + latest_dispositions, + normalize_disposition, + normalize_event, + open_debt, + staleness_key, + validate_disposition, +) + +TS = "2026-07-28T00:00:00+00:00" + + +class TestNormalization: + def test_events(self): + assert normalize_event("Revised") == "revised" + assert normalize_event("SNAPSHOT") == "snapshot" + assert normalize_event("nonsense") is None + assert normalize_event(None) is None + + def test_dispositions(self): + assert normalize_disposition("acknowledged") == "ACKNOWLEDGED" + assert normalize_disposition(" not_affected ") == "NOT_AFFECTED" + assert normalize_disposition("bogus") is None + + +class TestEntryKinds: + def test_change_vs_disposition(self): + ch = change_entry(ts=TS, artifact="a.md", event="revised") + dp = disposition_entry(ts=TS, downstream="b.md", upstream="a.md", disposition="OPEN") + assert is_change_entry(ch) and not is_disposition_entry(ch) + assert is_disposition_entry(dp) and not is_change_entry(dp) + assert dp["event"] == DISPOSITION_EVENT + + def test_change_events_are_all_change_kind(self): + for ev in CHANGE_EVENTS: + assert is_change_entry(change_entry(ts=TS, artifact="a.md", event=ev)) + + def test_change_entry_unknown_event_falls_back_to_revised(self): + assert change_entry(ts=TS, artifact="a.md", event="wat")["event"] == "revised" + + def test_change_entry_carries_decision_ref_only_when_present(self): + assert "decision_ref" not in change_entry(ts=TS, artifact="a.md") + assert change_entry(ts=TS, artifact="a.md", decision_ref="DL-07")["decision_ref"] == "DL-07" + + +class TestHonestCounting: + def test_open_is_debt(self): + item = {"disposition": "OPEN"} + off, _ = validate_disposition(item) + assert off is False + assert counts_as_debt(item) is True + + def test_recorded_refreshed_still_counts_as_debt(self): + # REFRESHED is derived-only. A genuinely refreshed downstream never reaches this function + # (it drops out of the stale set by timestamp). So a stale item labelled REFRESHED was + # hand-typed — the anti-relabelling rule says it still counts. You cannot type it away. + item = {"disposition": "REFRESHED"} + off, why = validate_disposition(item) + assert off is False and "derived" in why + assert counts_as_debt(item) is True + + def test_refreshed_is_not_a_settable_disposition(self): + from artifact_model import SETTABLE_DISPOSITIONS + assert "REFRESHED" not in SETTABLE_DISPOSITIONS + assert set(SETTABLE_DISPOSITIONS) == {"OPEN", "ACKNOWLEDGED", "NOT_AFFECTED"} + + def test_acknowledged_needs_owner(self): + assert counts_as_debt({"disposition": "ACKNOWLEDGED"}) is True # no owner -> debt + assert counts_as_debt({"disposition": "ACKNOWLEDGED", "owner": ""}) is True + assert counts_as_debt({"disposition": "ACKNOWLEDGED", "owner": "jane"}) is False + + def test_not_affected_needs_reason(self): + assert counts_as_debt({"disposition": "NOT_AFFECTED"}) is True + assert counts_as_debt({"disposition": "NOT_AFFECTED", "reason": "no data field touched"}) is False + + def test_unknown_disposition_counts_as_debt(self): + off, why = validate_disposition({"disposition": "WISHED_AWAY"}) + assert off is False and "unknown" in why + assert counts_as_debt({"disposition": "WISHED_AWAY"}) is True + + def test_open_debt_filters(self): + items = [ + {"id": 1, "disposition": "OPEN"}, + {"id": 2, "disposition": "NOT_AFFECTED", "reason": "no shared field"}, # off books + {"id": 3, "disposition": "ACKNOWLEDGED"}, # no owner -> debt + {"id": 4, "disposition": "ACKNOWLEDGED", "owner": "x"}, # off books + ] + assert sorted(i["id"] for i in open_debt(items)) == [1, 3] + + +class TestStalenessKey: + def test_pins_on_upstream_hash(self): + k1 = staleness_key("b.md", "a.md", "sha256:aaa") + k2 = staleness_key("b.md", "a.md", "sha256:bbb") + assert k1 != k2 # a later upstream change re-opens the item + assert staleness_key("b.md", "a.md") == "b.md<-a.md" + + def test_disposition_entry_sets_key(self): + dp = disposition_entry(ts=TS, downstream="b.md", upstream="a.md", + upstream_hash="sha256:aaa", disposition="ACKNOWLEDGED", owner="jane") + assert dp["key"] == staleness_key("b.md", "a.md", "sha256:aaa") + + +class TestLedgerProjections: + def test_latest_change_last_write_wins(self): + ledger = [ + change_entry(ts="1", artifact="a.md", event="created", hash="h1"), + change_entry(ts="2", artifact="a.md", event="revised", hash="h2"), + change_entry(ts="1", artifact="b.md", event="created", hash="hb"), + ] + latest = latest_change_per_artifact(ledger) + assert latest["a.md"]["hash"] == "h2" + assert latest["b.md"]["hash"] == "hb" + + def test_latest_dispositions_keyed(self): + ledger = [ + disposition_entry(ts="1", downstream="b.md", upstream="a.md", + upstream_hash="h", disposition="OPEN"), + disposition_entry(ts="2", downstream="b.md", upstream="a.md", + upstream_hash="h", disposition="ACKNOWLEDGED", owner="jane"), + ] + latest = latest_dispositions(ledger) + assert len(latest) == 1 + assert next(iter(latest.values()))["disposition"] == "ACKNOWLEDGED" + + def test_changes_for_history_trail(self): + ledger = [ + change_entry(ts="1", artifact="a.md", event="created"), + disposition_entry(ts="1", downstream="a.md", upstream="x.md", disposition="OPEN"), + change_entry(ts="2", artifact="a.md", event="revised"), + change_entry(ts="2", artifact="b.md", event="created"), + ] + trail = changes_for(ledger, "a.md") + assert [e["event"] for e in trail] == ["created", "revised"] diff --git a/scripts/tests/test_audit_artifacts.py b/scripts/tests/test_audit_artifacts.py new file mode 100644 index 0000000..f355a7f --- /dev/null +++ b/scripts/tests/test_audit_artifacts.py @@ -0,0 +1,354 @@ +"""Tests for audit_artifacts — record/impact/report, the staleness scenario, and exit-0 invariant.""" + +import io +import json +import os +import sys +from pathlib import Path + +import pytest + +import artifact_model as am +import audit_artifacts as aa + +REQ = ".sdlc/artifacts/01-requirements/requirements.md" +EPICS = ".sdlc/artifacts/01-requirements/epics.md" +GLOSSARY = ".sdlc/artifacts/01-requirements/glossary.md" +LAYER = ".sdlc/context/layers/phase1-requirements.md" +SPEC = "specs/0001-login.md" +DESIGN = ".sdlc/artifacts/02-design/design.md" + + +def _write(p: Path, text: str) -> None: + p.parent.mkdir(parents=True, exist_ok=True) + p.write_text(text, encoding="utf-8") + + +def build_corpus(root: Path) -> None: + sdlc = root / ".sdlc" + req = sdlc / "artifacts" / "01-requirements" + _write(req / "requirements.md", "# Requirements\n\n## FR-001 Login\nUser can log in.\n") + _write(req / "epics.md", "# Epics\n\n## EP-001\nRealizes FR-001.\n") # -> declared from req + _write(req / "glossary.md", "# Glossary\n\nTerms. No ids referenced.\n") # -> unrelated to req + _write(sdlc / "artifacts" / "02-design" / "design.md", "# Design\n\nNothing declared.\n") # -> coarse + _write(sdlc / "context" / "layers" / "phase1-requirements.md", + "---\nphase: 1\nsource_artifacts:\n - requirements.md\n---\n\n## Decision\nx\n") + _write(root / "specs" / "0001-login.md", + '---\nspec: "0001"\nname: "login"\nsource: "FR-001 (requirements.md)"\n---\n\n# Spec\n') + + +def write_ledger(root: Path, entries: list[dict]) -> Path: + ledger = root / ".sdlc" / "metrics" / "artifact-log.jsonl" + ledger.parent.mkdir(parents=True, exist_ok=True) + ledger.write_text("".join(json.dumps(e) + "\n" for e in entries), encoding="utf-8") + return ledger + + +def run_cli(argv, capsys) -> tuple[int, str]: + """Invoke main() with argv; return (exit_code, stdout). main() always exits 0 (advisory).""" + old = sys.argv + sys.argv = ["audit_artifacts.py"] + argv + try: + with pytest.raises(SystemExit) as ei: + aa.main() + code = ei.value.code + finally: + sys.argv = old + return code, capsys.readouterr().out + + +# --- baseline / scan -------------------------------------------------------------------------- + +class TestScan: + def test_first_scan_seeds_baseline(self, tmp_path, capsys): + build_corpus(tmp_path) + code, out = run_cli(["record", "--scan", "--repo", str(tmp_path)], capsys) + assert code == 0 + assert "Baseline recorded" in out and "history starts now" in out + ledger = aa.load_ledger(tmp_path / ".sdlc" / "metrics" / "artifact-log.jsonl") + arts = {e["artifact"] for e in ledger if am.is_change_entry(e)} + assert REQ in arts and SPEC in arts and LAYER in arts + assert all(e["event"] == "created" for e in ledger) + + def test_second_scan_no_changes(self, tmp_path, capsys): + build_corpus(tmp_path) + run_cli(["record", "--scan", "--repo", str(tmp_path)], capsys) + code, out = run_cli(["record", "--scan", "--repo", str(tmp_path)], capsys) + assert code == 0 and "no changes" in out.lower() + + def test_scan_detects_drift_as_snapshot(self, tmp_path, capsys): + build_corpus(tmp_path) + run_cli(["record", "--scan", "--repo", str(tmp_path)], capsys) + (tmp_path / REQ).write_text("# Requirements\n\n## FR-001 Login\nSLA 4h.\n", encoding="utf-8") + code, out = run_cli(["record", "--scan", "--repo", str(tmp_path)], capsys) + assert code == 0 and "drifted" in out + ledger = aa.load_ledger(tmp_path / ".sdlc" / "metrics" / "artifact-log.jsonl") + snaps = [e for e in ledger if e.get("event") == "snapshot" and e["artifact"] == REQ] + assert len(snaps) == 1 and snaps[0].get("prev_hash") + + +# --- empty ledger ----------------------------------------------------------------------------- + +class TestEmptyLedger: + def test_report_no_history(self, tmp_path, capsys): + build_corpus(tmp_path) + code, out = run_cli(["report", "--repo", str(tmp_path)], capsys) + assert code == 0 and "No artifact history yet" in out + + def test_report_json_has_history_false(self, tmp_path, capsys): + build_corpus(tmp_path) + code, out = run_cli(["report", "--repo", str(tmp_path), "--json"], capsys) + payload = json.loads(out) + assert payload["has_history"] is False and payload["stale"] == 0 + + +# --- the staleness scenario (controlled timestamps) ------------------------------------------- + +def _baseline_and_upstream_change(tmp_path): + """Baseline everything at t0, then a later change to requirements.md at t1 (hash req2).""" + build_corpus(tmp_path) + t0 = "2026-07-01T00:00:00+00:00" + t1 = "2026-07-10T00:00:00+00:00" + entries = [ + am.change_entry(ts=t0, artifact=REQ, event="created", hash="sha256:req1", actor="kai"), + am.change_entry(ts=t0, artifact=EPICS, event="created", hash="sha256:ep1", actor="kai"), + am.change_entry(ts=t0, artifact=GLOSSARY, event="created", hash="sha256:gl1", actor="kai"), + am.change_entry(ts=t0, artifact=LAYER, event="created", hash="sha256:ly1", actor="kai"), + am.change_entry(ts=t0, artifact=SPEC, event="created", hash="sha256:sp1", actor="kai"), + am.change_entry(ts=t0, artifact=DESIGN, event="created", hash="sha256:dz1", actor="kai"), + am.change_entry(ts=t1, artifact=REQ, event="revised", target="FR-001", + hash="sha256:req2", prev_hash="sha256:req1", actor="jane", reason="SLA 8h->4h"), + ] + write_ledger(tmp_path, entries) + return t1 + + +class TestStaleness: + def test_declared_downstreams_flagged_unrelated_not(self, tmp_path, capsys): + _baseline_and_upstream_change(tmp_path) + code, out = run_cli(["report", "--repo", str(tmp_path), "--json"], capsys) + payload = json.loads(out) + stale = {i["downstream"] for i in payload["items"]} + assert {EPICS, SPEC, LAYER} <= stale # declared downstreams of requirements + assert DESIGN in stale # coarse phase-order downstream + assert GLOSSARY not in stale # no path from requirements -> never stale + assert payload["open"] == len(payload["items"]) # all OPEN before disposition + + def test_declared_vs_coarse_labeled(self, tmp_path, capsys): + _baseline_and_upstream_change(tmp_path) + code, out = run_cli(["report", "--repo", str(tmp_path), "--json"], capsys) + items = {i["downstream"]: i for i in json.loads(out)["items"]} + assert items[EPICS]["confidence"] == "declared" + assert items[DESIGN]["confidence"] == "coarse" + + def test_acknowledge_with_owner_clears_debt(self, tmp_path, capsys): + _baseline_and_upstream_change(tmp_path) + # Disposition epics <- requirements with an owner -> off the books. + run_cli(["record", "--repo", str(tmp_path), "--disposition", "ACKNOWLEDGED", + "--downstream", EPICS, "--upstream", REQ, "--owner", "jane"], capsys) + code, out = run_cli(["report", "--repo", str(tmp_path), "--json"], capsys) + payload = json.loads(out) + epics_item = next(i for i in payload["items"] if i["downstream"] == EPICS) + assert epics_item["disposition"] == "ACKNOWLEDGED" + assert not am.counts_as_debt(epics_item) + assert payload["acknowledged"] >= 1 + assert payload["open"] < len(payload["items"]) + + def test_acknowledge_without_owner_still_debt(self, tmp_path, capsys): + _baseline_and_upstream_change(tmp_path) + run_cli(["record", "--repo", str(tmp_path), "--disposition", "ACKNOWLEDGED", + "--downstream", SPEC, "--upstream", REQ], capsys) # no owner + code, out = run_cli(["report", "--repo", str(tmp_path), "--json"], capsys) + spec_item = next(i for i in json.loads(out)["items"] if i["downstream"] == SPEC) + assert spec_item["disposition"] == "ACKNOWLEDGED" + assert am.counts_as_debt(spec_item) # mislabeled (no owner) -> still debt + + def test_typed_refreshed_is_rejected_and_item_stays_debt(self, tmp_path, capsys): + # The anti-relabelling invariant: you cannot clear staleness debt by typing REFRESHED. + _baseline_and_upstream_change(tmp_path) + code, out = run_cli(["record", "--repo", str(tmp_path), "--disposition", "REFRESHED", + "--downstream", EPICS, "--upstream", REQ], capsys) + assert code == 0 and "not a recordable disposition" in out.lower() + # Nothing was recorded, so the item is still OPEN debt. + _, rep = run_cli(["report", "--repo", str(tmp_path), "--json"], capsys) + epics_item = next(i for i in json.loads(rep)["items"] if i["downstream"] == EPICS) + assert am.counts_as_debt(epics_item) is True + + def test_refresh_downstream_drops_it_from_stale(self, tmp_path, capsys): + _baseline_and_upstream_change(tmp_path) + # epics changes again AFTER the requirements change -> no longer behind -> not stale. + ledger = aa.load_ledger(tmp_path / ".sdlc" / "metrics" / "artifact-log.jsonl") + ledger.append(am.change_entry(ts="2026-07-20T00:00:00+00:00", artifact=EPICS, + event="revised", hash="sha256:ep2", actor="jane")) + write_ledger(tmp_path, ledger) + code, out = run_cli(["report", "--repo", str(tmp_path), "--json"], capsys) + stale = {i["downstream"] for i in json.loads(out)["items"]} + assert EPICS not in stale + + +class TestScope: + def test_phase_scope_limits_rows(self, tmp_path, capsys): + _baseline_and_upstream_change(tmp_path) + code, out = run_cli(["report", "--repo", str(tmp_path), "--phase", "1", "--json"], capsys) + payload = json.loads(out) + assert payload["scope"] == "phase:1" + # design.md is phase 2 -> excluded from a phase-1 scope. + assert all(i["downstream"] != DESIGN for i in payload["items"]) + assert any(i["downstream"] == EPICS for i in payload["items"]) + assert payload["artifacts_tracked"] >= payload["artifacts_shown"] + + def test_artifact_scope_single(self, tmp_path, capsys): + _baseline_and_upstream_change(tmp_path) + code, out = run_cli(["report", "--repo", str(tmp_path), "--artifact", EPICS, "--json"], capsys) + payload = json.loads(out) + assert payload["scope"] == f"artifact:{EPICS}" + assert payload["artifacts_shown"] == 1 + assert all(i["downstream"] == EPICS for i in payload["items"]) + + +# --- impact & history ------------------------------------------------------------------------- + +class TestImpact: + def test_impact_by_id_lists_downstream(self, tmp_path, capsys): + build_corpus(tmp_path) + code, out = run_cli(["impact", "FR-001", "--repo", str(tmp_path), "--json"], capsys) + payload = json.loads(out) + assert payload["target"] == REQ + down = {r["node"] for r in payload["downstream"]} + assert EPICS in down and SPEC in down + + def test_impact_unresolvable_is_graceful(self, tmp_path, capsys): + build_corpus(tmp_path) + code, out = run_cli(["impact", "FR-999", "--repo", str(tmp_path)], capsys) + assert code == 0 and "could not resolve" in out.lower() + + +class TestHistory: + def test_history_trail(self, tmp_path, capsys): + _baseline_and_upstream_change(tmp_path) + code, out = run_cli(["report", "--repo", str(tmp_path), "--history", "FR-001", "--json"], capsys) + payload = json.loads(out) + assert payload["artifact"] == REQ + events = [e["event"] for e in payload["history"]] + assert events == ["created", "revised"] + assert payload["history"][-1]["actor"] == "jane" + + +# --- robustness / exit-0 under adverse inputs (adversarial-review regressions) ---------------- + +class TestSinceTimezone: + def test_since_bare_date_does_not_crash(self, tmp_path, capsys): + # Ledger timestamps are tz-aware; a documented bare-date --since parses naive. The compare + # must not raise (exit-0 invariant). Covers the dashboard and --history paths. + _baseline_and_upstream_change(tmp_path) + code, out = run_cli(["report", "--repo", str(tmp_path), "--since", "2026-07-05", "--json"], capsys) + assert code == 0 + assert json.loads(out)["scope"].endswith("since:2026-07-05") + code2, _ = run_cli(["report", "--repo", str(tmp_path), "--history", REQ, "--since", "2026-07-05"], capsys) + assert code2 == 0 + + +class TestCorruptLedger: + def test_non_dict_json_lines_do_not_crash(self, tmp_path, capsys): + build_corpus(tmp_path) + ledger = tmp_path / ".sdlc" / "metrics" / "artifact-log.jsonl" + ledger.parent.mkdir(parents=True, exist_ok=True) + good = am.change_entry(ts="2026-07-01T00:00:00+00:00", artifact=REQ, event="created", hash="h1") + # Valid-JSON but non-object lines interleaved with a real entry (hand-edit / partial write). + ledger.write_text("null\n42\n\"text\"\n[1,2,3]\n" + json.dumps(good) + "\n", encoding="utf-8") + for argv in (["report"], ["report", "--history", REQ], ["record", "--scan"]): + code, _ = run_cli(argv + ["--repo", str(tmp_path)], capsys) + assert code == 0 + # The one valid entry is still readable past the garbage. + assert any(e.get("artifact") == REQ for e in aa.load_ledger(ledger)) + + +class TestUnreadableArtifact: + @pytest.mark.skipif(os.name == "nt" or (hasattr(os, "geteuid") and os.geteuid() == 0), + reason="chmod 000 is not enforced on Windows or for root") + def test_scan_skips_unreadable_file(self, tmp_path, capsys): + build_corpus(tmp_path) + victim = tmp_path / REQ + victim.chmod(0o000) + try: + code, out = run_cli(["record", "--scan", "--repo", str(tmp_path)], capsys) + finally: + victim.chmod(0o644) + assert code == 0 # unreadable artifact skipped, never crashes the scan + + +# --- exit-0 invariant ------------------------------------------------------------------------- + +class TestExitZero: + @pytest.mark.parametrize("argv", [ + ["record"], # no mode + ["record", "--disposition", "ACKNOWLEDGED"], # missing downstream/upstream + ["impact", "nonsense"], + ["report"], + ["report", "--history", "does-not-exist.md"], + ]) + def test_never_nonzero(self, tmp_path, capsys, argv): + build_corpus(tmp_path) + code, _ = run_cli(argv + ["--repo", str(tmp_path)], capsys) + assert code == 0 + + +# --- narrow console encodings (Windows cp1252) ------------------------------------------------- + +def _run_on_cp1252_console(argv, monkeypatch) -> tuple[int, str]: + """Invoke main() with stdout bound to a real cp1252 stream — a default Windows console. + + capsys buffers as UTF-8, so it can NEVER surface an encoding fault; CI is Linux/UTF-8 for the + same reason. This helper is the only instrument in the suite that reproduces what a Windows + user actually sees.""" + buf = io.BytesIO() + stream = io.TextIOWrapper(buf, encoding="cp1252", newline="") + monkeypatch.setattr(sys, "stdout", stream) + sys.argv = ["audit_artifacts.py"] + argv + with pytest.raises(SystemExit) as ei: + aa.main() + stream.flush() + return ei.value.code, buf.getvalue().decode("cp1252") + + +class TestNarrowConsoleEncoding: + """The freshness dashboard must render on a console that cannot encode '<-' or a checkmark. + + Both glyphs sit on the ordinary happy path — the arrow prints for every stale item, the + checkmark for every signed-off phase — so on a default Windows console the tool used to die + with UnicodeEncodeError and a NON-ZERO exit precisely when it had something to report. That + breaks the exit-0 advisory guarantee this module is built on. + """ + + def test_stale_items_render_on_cp1252(self, tmp_path, monkeypatch): + _baseline_and_upstream_change(tmp_path) + code, out = _run_on_cp1252_console(["report", "--repo", str(tmp_path)], monkeypatch) + assert code == 0 + assert "STALE" in out # it still reports the staleness it found + assert "<-" in out # via the ASCII fallback, not a dropped line + + def test_signoff_marker_renders_on_cp1252(self, tmp_path, monkeypatch): + """A signed-off phase alone is enough — no staleness required.""" + build_corpus(tmp_path) + write_ledger(tmp_path, [ + am.change_entry(ts="2026-07-01T00:00:00+00:00", artifact=REQ, + event="created", hash="sha256:req1", actor="kai"), + ]) + (tmp_path / ".sdlc" / "state.yaml").write_text( + "phases:\n '1':\n sign_off: matt\n", encoding="utf-8") + code, out = _run_on_cp1252_console(["report", "--repo", str(tmp_path)], monkeypatch) + assert code == 0 + assert "signed-off" in out + + def test_utf8_console_keeps_the_nicer_glyphs(self, tmp_path, monkeypatch): + """The fallback is a degradation for narrow consoles only — a UTF-8 console is unchanged.""" + _baseline_and_upstream_change(tmp_path) + buf = io.BytesIO() + stream = io.TextIOWrapper(buf, encoding="utf-8", newline="") + monkeypatch.setattr(sys, "stdout", stream) + sys.argv = ["audit_artifacts.py", "report", "--repo", str(tmp_path)] + with pytest.raises(SystemExit) as ei: + aa.main() + stream.flush() + assert ei.value.code == 0 + assert "←" in buf.getvalue().decode("utf-8")