`75139ae`

Add prompt and artifact diff inspection surfaces

Authored by

espadonne 1 month ago

SHA: 75139aee1829354e1efaee5cb07f9df2feb9050a
Parents: 6f05b29
Tree: 359b983

3 changed files

Status	File	+
M	`src/loader/cli/main.py`	147
M	`src/loader/runtime/inspection.py`	307
M	`tests/test_inspection.py`	178

src/loader/cli/main.pymodified

      DoctorReport,
      PermissionCheckResult,
      PermissionSnapshot,
 +    PromptDiffSnapshot,
      PromptPreview,
      StatusSnapshot,
 +    WorkflowArtifactDiffSnapshot,
      WorkflowTimelineSnapshot,
      collect_doctor_report,
      collect_permission_snapshot,
 +    collect_prompt_diff,
      collect_prompt_preview,
      collect_status_snapshot,
 +    collect_workflow_artifact_diffs,
      collect_workflow_timeline,
      dry_run_permission_check,
      list_session_summaries,
+     )
 +@prompt_cli.command("diff")
 +@click.option("--full", is_flag=True, help="Show the full unified prompt diff")
 +@click.argument("session_id", required=False)
 +def prompt_diff_cli(full: bool, session_id: str | None) -> None:
 +    """Compare the latest persisted prompt contracts."""
++
 +    _prompt_diff_main(session_id=session_id, full=full)
++
++
  @workflow_cli.command("show")
  @click.option(
      "--mode",
      show_default=True,
      help="Show only the most recent matching entries",
+ )
 +@click.option("--diff", "show_diff", is_flag=True, help="Show persisted artifact diffs")
 +@click.option("--full-diff", is_flag=True, help="Show the full unified artifact diffs")
  @click.argument("session_id", required=False)
  def workflow_show_cli(
      mode: str | None,
      kind: str | None,
      limit: int,
 +    show_diff: bool,
 +    full_diff: bool,
      session_id: str | None,
  ) -> None:
      """Show the persisted workflow timeline for the latest or named session."""
          mode=mode,
          kind=kind,
          limit=limit,
 +        show_diff=show_diff,
 +        full_diff=full_diff,
+     )
              "  loader status              Show persisted runtime status",
              "  loader explore <prompt>    Run a fast read-only lookup query",
              "  loader prompt show         Preview the current prompt contract",
 +            "  loader prompt diff         Compare the latest persisted prompt contracts",
              "  loader permissions show    Display normalized permission rules",
              "  loader permissions check   Dry-run one permission decision",
              "  loader workflow show       Show the persisted workflow timeline",
              "",
              "Commands:",
              "  show [task]  Render the current prompt contract without a model call",
 +            "  diff [id]    Compare the latest persisted prompt contracts",
+         ]
+     )
              "",
              "Commands:",
              "  show [id]  Show the persisted workflow timeline with optional filters",
 +            "            Add --diff to compare the latest persisted artifacts",
+         ]
+     )
      mode: str | None,
      kind: str | None,
      limit: int | None,
 +    show_diff: bool,
 +    full_diff: bool,
  ) -> None:
      try:
          snapshot: WorkflowTimelineSnapshot = collect_workflow_timeline(
              snapshot.workflow_ledger,
              title="[bold blue]Workflow Ledger[/bold blue]",
+         )
 +    if show_diff:
 +        console.print()
 +        artifact_diffs = collect_workflow_artifact_diffs(session_id=session_id)
 +        _print_workflow_artifact_diffs(
 +            artifact_diffs,
 +            show_full=full_diff,
 +        )
      console.print()
      _print_workflow_timeline_entries(
          snapshot.entries,
      _print_prompt_preview(preview)
 +def _prompt_diff_main(
 +    *,
 +    session_id: str | None,
 +    full: bool,
 +) -> None:
 +    try:
 +        snapshot: PromptDiffSnapshot = collect_prompt_diff(session_id=session_id)
 +    except FileNotFoundError:
 +        console.print(f"[red]Session not found:[/red] {session_id}")
 +        raise SystemExit(1) from None
++
 +    if snapshot.session_id is None or snapshot.current is None:
 +        console.print("[yellow]No persisted prompt history found.[/yellow]")
 +        return
++
 +    _print_prompt_diff(snapshot, show_full=full)
++
++
  def _permissions_check_main(
      *,
      tool_name: str,
+     )
 +def _print_prompt_diff(snapshot: PromptDiffSnapshot, *, show_full: bool) -> None:
 +    table = Table(show_header=False, box=None)
 +    table.add_column("Field", style="bold cyan")
 +    table.add_column("Value", style="white")
 +    table.add_row("Workspace", str(snapshot.project_root))
 +    table.add_row("Session", snapshot.session_id or "none")
 +    table.add_row("Task", snapshot.current_task or "none")
 +    table.add_row(
 +        "Current",
 +        _format_prompt_snapshot(snapshot.current) if snapshot.current else "none",
 +    )
 +    table.add_row(
 +        "Previous",
 +        _format_prompt_snapshot(snapshot.previous) if snapshot.previous else "none",
 +    )
 +    console.print(
 +        Panel.fit(
 +            table,
 +            title="[bold blue]Prompt Diff[/bold blue]",
 +            border_style="blue",
 +        )
 +    )
++
 +    if snapshot.highlights:
 +        console.print()
 +        _print_workflow_highlights(
 +            snapshot.highlights,
 +            title="[bold blue]Prompt Changes[/bold blue]",
 +        )
++
 +    if show_full:
 +        body = snapshot.unified_diff or "[dim]No prompt body diff available.[/dim]"
 +        console.print()
 +        console.print(
 +            Panel(
 +                body,
 +                title="[bold blue]Prompt Unified Diff[/bold blue]",
 +                border_style="blue",
 +            )
 +        )
++
++
 +def _format_prompt_snapshot(snapshot) -> str:
 +    parts = [snapshot.workflow_mode, snapshot.permission_mode, snapshot.prompt_format]
 +    if snapshot.prompt_sections:
 +        parts.append(f"sections={len(snapshot.prompt_sections)}")
 +    return " / ".join(parts)
++
++
 +def _print_workflow_artifact_diffs(
 +    snapshot: WorkflowArtifactDiffSnapshot,
 +    *,
 +    show_full: bool,
 +) -> None:
 +    if not snapshot.entries:
 +        console.print("[dim]No persisted artifact diffs are available for this session.[/dim]")
 +        return
++
 +    if snapshot.highlights:
 +        _print_workflow_highlights(
 +            snapshot.highlights,
 +            title="[bold blue]Artifact Changes[/bold blue]",
 +        )
++
 +    table = Table(show_header=True, header_style="bold cyan")
 +    table.add_column("Kind", style="white")
 +    table.add_column("Current", style="white")
 +    table.add_column("Previous", style="white")
 +    table.add_column("Summary", style="dim")
++
 +    for entry in snapshot.entries:
 +        table.add_row(
 +            entry.kind,
 +            entry.current_path.name,
 +            entry.previous_path.name if entry.previous_path else "none",
 +            "; ".join(entry.highlights[:2]) or "none",
 +        )
++
 +    console.print(
 +        Panel(
 +            table,
 +            title="[bold blue]Artifact Diff Summary[/bold blue]",
 +            border_style="blue",
 +        )
 +    )
++
 +    if show_full:
 +        for entry in snapshot.entries:
 +            console.print()
 +            console.print(
 +                Panel(
 +                    entry.unified_diff or "[dim]No diff available.[/dim]",
 +                    title=f"[bold blue]{entry.kind} Diff[/bold blue]",
 +                    border_style="blue",
 +                )
 +            )
++
++
  def _print_workflow_timeline_entries(
      entries: list,
      *,

src/loader/runtime/inspection.pymodified

  from __future__ import annotations
 +import difflib
  import json
  import os
  from dataclasses import dataclass, field
      permission_path_hint,
      summarize_permission_input,
+ )
 +from .prompt_history import PromptSnapshot
  from .prompting import build_system_prompt_result
  from .session import SessionSnapshot, SessionStore
  from .workflow_ledger import WorkflowLedger, workflow_ledger_highlights
      workflow_ledger: WorkflowLedger = field(default_factory=WorkflowLedger)
 +@dataclass(slots=True)
 +class PromptDiffSnapshot:
 +    """Operator-facing diff between persisted prompt contracts."""
++
 +    project_root: Path
 +    session_id: str | None
 +    current_task: str | None
 +    current: PromptSnapshot | None
 +    previous: PromptSnapshot | None
 +    highlights: list[str] = field(default_factory=list)
 +    unified_diff: str = ""
++
++
 +@dataclass(slots=True)
 +class ArtifactDiffEntry:
 +    """One persisted artifact diff entry."""
++
 +    kind: str
 +    current_path: Path
 +    previous_path: Path | None
 +    highlights: list[str] = field(default_factory=list)
 +    unified_diff: str = ""
++
++
 +@dataclass(slots=True)
 +class WorkflowArtifactDiffSnapshot:
 +    """Operator-facing diff across persisted workflow artifacts."""
++
 +    project_root: Path
 +    session_id: str | None
 +    current_task: str | None
 +    entries: list[ArtifactDiffEntry] = field(default_factory=list)
 +    highlights: list[str] = field(default_factory=list)
++
++
  def capability_summary(profile: CapabilityProfile) -> str:
      """Render a short human-readable capability summary."""
+     )
 +def collect_prompt_diff(
 +    session_id: str | None = None,
 +    *,
 +    project_root: Path | str | None = None,
 +) -> PromptDiffSnapshot:
 +    """Load the latest persisted prompt-contract diff for one session."""
++
 +    resolved_root = Path(project_root or Path.cwd()).expanduser().resolve()
 +    store = SessionStore(resolved_root)
 +    snapshot = store.load(session_id) if session_id else store.load_latest()
 +    if snapshot is None:
 +        return PromptDiffSnapshot(
 +            project_root=resolved_root,
 +            session_id=None,
 +            current_task=None,
 +            current=None,
 +            previous=None,
 +            highlights=[],
 +            unified_diff="",
 +        )
++
 +    current = snapshot.prompt_history[-1] if snapshot.prompt_history else None
 +    previous = _previous_prompt_snapshot(snapshot.prompt_history)
 +    highlights = _prompt_diff_highlights(previous, current)
 +    unified_diff = _unified_diff(
 +        previous.content if previous is not None else "",
 +        current.content if current is not None else "",
 +        from_label=_prompt_snapshot_label(previous, fallback="previous prompt"),
 +        to_label=_prompt_snapshot_label(current, fallback="current prompt"),
 +    )
 +    return PromptDiffSnapshot(
 +        project_root=resolved_root,
 +        session_id=snapshot.session_id,
 +        current_task=snapshot.current_task,
 +        current=current,
 +        previous=previous,
 +        highlights=highlights,
 +        unified_diff=unified_diff,
 +    )
++
++
 +def collect_workflow_artifact_diffs(
 +    session_id: str | None = None,
 +    *,
 +    project_root: Path | str | None = None,
 +) -> WorkflowArtifactDiffSnapshot:
 +    """Load persisted workflow-artifact diffs for the latest or named session."""
++
 +    resolved_root = Path(project_root or Path.cwd()).expanduser().resolve()
 +    store = SessionStore(resolved_root)
 +    snapshot = store.load(session_id) if session_id else store.load_latest()
 +    if snapshot is None:
 +        return WorkflowArtifactDiffSnapshot(
 +            project_root=resolved_root,
 +            session_id=None,
 +            current_task=None,
 +            entries=[],
 +            highlights=[],
 +        )
++
 +    dod = _load_dod(snapshot.active_dod_path, project_root=resolved_root)
 +    if dod is None:
 +        return WorkflowArtifactDiffSnapshot(
 +            project_root=resolved_root,
 +            session_id=snapshot.session_id,
 +            current_task=snapshot.current_task,
 +            entries=[],
 +            highlights=[],
 +        )
++
 +    entries: list[ArtifactDiffEntry] = []
 +    for kind, path_str in (
 +        ("clarify_brief", dod.clarify_brief),
 +        ("implementation_plan", dod.implementation_plan),
 +        ("verification_plan", dod.verification_plan),
 +    ):
 +        if not path_str:
 +            continue
 +        current_path = Path(path_str)
 +        if not current_path.exists():
 +            continue
 +        previous_path = _previous_artifact_path(current_path)
 +        previous_text = previous_path.read_text() if previous_path and previous_path.exists() else ""
 +        current_text = current_path.read_text()
 +        entries.append(
 +            ArtifactDiffEntry(
 +                kind=kind,
 +                current_path=current_path,
 +                previous_path=previous_path,
 +                highlights=_artifact_diff_highlights(
 +                    kind=kind,
 +                    current_path=current_path,
 +                    previous_path=previous_path,
 +                    previous_text=previous_text,
 +                    current_text=current_text,
 +                ),
 +                unified_diff=_unified_diff(
 +                    previous_text,
 +                    current_text,
 +                    from_label=str(previous_path) if previous_path else f"previous {kind}",
 +                    to_label=str(current_path),
 +                ),
 +            )
 +        )
++
 +    highlights = [entry.highlights[0] for entry in entries if entry.highlights]
 +    return WorkflowArtifactDiffSnapshot(
 +        project_root=resolved_root,
 +        session_id=snapshot.session_id,
 +        current_task=snapshot.current_task,
 +        entries=entries,
 +        highlights=highlights,
 +    )
++
++
  def collect_permission_snapshot(
      project_root: Path | str | None = None,
      *,
+     )
 +def _previous_prompt_snapshot(
 +    history: list[PromptSnapshot],
 +) -> PromptSnapshot | None:
 +    if len(history) < 2:
 +        return None
 +    current = history[-1]
 +    for snapshot in reversed(history[:-1]):
 +        if not snapshot.matches_contract(current):
 +            return snapshot
 +    return history[-2]
++
++
 +def _prompt_snapshot_label(
 +    snapshot: PromptSnapshot | None,
 +    *,
 +    fallback: str,
 +) -> str:
 +    if snapshot is None:
 +        return fallback
 +    return (
 +        f"{snapshot.timestamp} "
 +        f"{snapshot.workflow_mode}/{snapshot.permission_mode}/{snapshot.prompt_format}"
 +    )
++
++
 +def _prompt_diff_highlights(
 +    previous: PromptSnapshot | None,
 +    current: PromptSnapshot | None,
 +) -> list[str]:
 +    if current is None:
 +        return []
 +    if previous is None:
 +        return ["No earlier prompt snapshot is available for comparison."]
++
 +    highlights: list[str] = []
 +    if previous.workflow_mode != current.workflow_mode:
 +        highlights.append(
 +            f"Workflow mode changed: {previous.workflow_mode} -> {current.workflow_mode}"
 +        )
 +    if previous.permission_mode != current.permission_mode:
 +        highlights.append(
 +            "Permission mode changed: "
 +            f"{previous.permission_mode} -> {current.permission_mode}"
 +        )
 +    if previous.prompt_format != current.prompt_format:
 +        highlights.append(
 +            f"Prompt format changed: {previous.prompt_format} -> {current.prompt_format}"
 +        )
 +    if previous.current_task != current.current_task:
 +        highlights.append("Task framing changed across prompt snapshots.")
 +    added_sections = [item for item in current.prompt_sections if item not in previous.prompt_sections]
 +    removed_sections = [item for item in previous.prompt_sections if item not in current.prompt_sections]
 +    if added_sections:
 +        highlights.append("Added sections: " + ", ".join(added_sections))
 +    if removed_sections:
 +        highlights.append("Removed sections: " + ", ".join(removed_sections))
 +    additions, removals = _line_change_counts(previous.content, current.content)
 +    highlights.append(f"Prompt body lines changed: +{additions} / -{removals}")
 +    return highlights
++
++
 +def _previous_artifact_path(current_path: Path) -> Path | None:
 +    name = current_path.name
 +    if name in {"implementation.md", "verification.md"}:
 +        parent = current_path.parent
 +        if "-" not in parent.name:
 +            return None
 +        slug = parent.name.split("-", maxsplit=1)[1]
 +        candidates = sorted(
 +            item
 +            for item in parent.parent.glob(f"*-{slug}")
 +            if item.is_dir() and (item / name).exists()
 +        )
 +        previous_dir = _previous_sorted_item(candidates, parent)
 +        if previous_dir is None:
 +            return None
 +        return previous_dir / name
++
 +    stem = current_path.stem
 +    if "-" not in stem:
 +        return None
 +    slug = stem.split("-", maxsplit=1)[1]
 +    candidates = sorted(
 +        item
 +        for item in current_path.parent.glob(f"*-{slug}{current_path.suffix}")
 +        if item.is_file()
 +    )
 +    return _previous_sorted_item(candidates, current_path)
++
++
 +def _previous_sorted_item(items: list[Any], current: Any) -> Any | None:
 +    previous: Any | None = None
 +    for item in items:
 +        if item == current:
 +            return previous
 +        previous = item
 +    return previous
++
++
 +def _artifact_diff_highlights(
 +    *,
 +    kind: str,
 +    current_path: Path,
 +    previous_path: Path | None,
 +    previous_text: str,
 +    current_text: str,
 +) -> list[str]:
 +    label = kind.replace("_", " ")
 +    if previous_path is None:
 +        return [f"{label}: no previous artifact version is available."]
 +    additions, removals = _line_change_counts(previous_text, current_text)
 +    if not additions and not removals:
 +        return [f"{label}: no content changes between persisted versions."]
 +    return [
 +        f"{label}: +{additions} / -{removals} lines vs {previous_path.name}",
 +        f"current={current_path.name}",
 +    ]
++
++
 +def _line_change_counts(previous_text: str, current_text: str) -> tuple[int, int]:
 +    additions = 0
 +    removals = 0
 +    diff_lines = difflib.unified_diff(
 +        previous_text.splitlines(),
 +        current_text.splitlines(),
 +        lineterm="",
 +    )
 +    for line in diff_lines:
 +        if line.startswith(("---", "+++", "@@")):
 +            continue
 +        if line.startswith("+"):
 +            additions += 1
 +        elif line.startswith("-"):
 +            removals += 1
 +    return additions, removals
++
++
 +def _unified_diff(
 +    previous_text: str,
 +    current_text: str,
 +    *,
 +    from_label: str,
 +    to_label: str,
 +) -> str:
 +    return "\n".join(
 +        difflib.unified_diff(
 +            previous_text.splitlines(),
 +            current_text.splitlines(),
 +            fromfile=from_label,
 +            tofile=to_label,
 +            lineterm="",
 +        )
 +    )
++
++
  def _coerce_permission_mode(value: PermissionMode | str) -> PermissionMode:
      if isinstance(value, PermissionMode):
          return value

tests/test_inspection.pymodified

      CheckStatus,
      collect_doctor_report,
      collect_permission_snapshot,
 +    collect_prompt_diff,
      collect_prompt_preview,
      collect_status_snapshot,
 +    collect_workflow_artifact_diffs,
      collect_workflow_timeline,
      dry_run_permission_check,
      list_session_summaries,
      load_session_detail,
+ )
 +from loader.runtime.prompt_history import PromptSnapshot
  from loader.runtime.session import SessionSnapshot, SessionStore
  from loader.runtime.workflow_ledger import WorkflowLedger, WorkflowLedgerItem
  from loader.runtime.workflow_policy import WorkflowTimelineEntry
          permission_rules_source=str(temp_dir / ".loader" / "permission-rules.json"),
          prompt_format="native",
          prompt_sections=["Runtime Config", "Workflow Context", "Mode Guidance"],
 +        prompt_history=[
 +            PromptSnapshot(
 +                timestamp="2026-04-06T12:04:00Z",
 +                workflow_mode="verify",
 +                permission_mode="prompt",
 +                current_task="Fix the failing tests",
 +                prompt_format="native",
 +                prompt_sections=["Runtime Config", "Workflow Context", "Mode Guidance"],
 +                content="# Introduction\nverify parser fix\n",
 +            ),
 +            PromptSnapshot(
 +                timestamp="2026-04-06T12:05:00Z",
 +                workflow_mode="execute",
 +                permission_mode="prompt",
 +                current_task="Fix the failing tests",
 +                prompt_format="native",
 +                prompt_sections=[
 +                    "Runtime Config",
 +                    "Workflow Context",
 +                    "Mode Guidance",
 +                    "Project Context",
 +                ],
 +                content="# Introduction\nexecute parser fix\n# Project Context\npython\n",
 +            ),
 +        ],
          workflow_reason_code="verification_failed_reentry",
          workflow_reason_summary="verification failed; returning to execute for fixes",
          workflow_decision_kind="reentry",
  def _persist_session_with_rich_workflow(temp_dir: Path) -> str:
 +    slug = "tighten-loader-workflow-behavior"
 +    brief_old = temp_dir / ".loader" / "briefs" / f"20260406T150000Z-{slug}.md"
 +    brief_new = temp_dir / ".loader" / "briefs" / f"20260406T150200Z-{slug}.md"
 +    brief_old.write_text(
 +        "# Task Brief\n\n## Likely Touchpoints\n- planned.txt\n\n## Acceptance Criteria\n- planned.txt exists.\n"
 +    )
 +    brief_new.write_text(
 +        "# Task Brief\n\n## Likely Touchpoints\n- notes.txt\n\n## Acceptance Criteria\n- notes.txt exists.\n"
 +    )
 +    plan_old_root = temp_dir / ".loader" / "plans" / f"20260406T150100Z-{slug}"
 +    plan_new_root = temp_dir / ".loader" / "plans" / f"20260406T150300Z-{slug}"
 +    plan_old_root.mkdir(parents=True, exist_ok=True)
 +    plan_new_root.mkdir(parents=True, exist_ok=True)
 +    (plan_old_root / "implementation.md").write_text(
 +        "# Implementation Plan\n\n## File Changes\n- Create planned.txt.\n"
 +    )
 +    (plan_old_root / "verification.md").write_text(
 +        "# Verification Plan\n\n## Acceptance Criteria\n- planned.txt exists.\n"
 +    )
 +    (plan_new_root / "implementation.md").write_text(
 +        "# Implementation Plan\n\n## File Changes\n- Keep notes.txt as the runtime artifact.\n"
 +    )
 +    (plan_new_root / "verification.md").write_text(
 +        "# Verification Plan\n\n## Acceptance Criteria\n- notes.txt exists.\n"
 +    )
++
 +    dod = create_definition_of_done("Tighten Loader workflow behavior")
 +    dod.status = "fixing"
 +    dod.clarify_brief = str(brief_new)
 +    dod.implementation_plan = str(plan_new_root / "implementation.md")
 +    dod.verification_plan = str(plan_new_root / "verification.md")
 +    dod.acceptance_criteria = ["notes.txt exists in the workspace root."]
 +    dod_path = DefinitionOfDoneStore(temp_dir).save(dod)
++
      snapshot = SessionSnapshot(
          session_id="20260406T150000Z-feedface",
          created_at="2026-04-06T15:00:00Z",
              Message(role=Role.USER, content="Tighten Loader workflow behavior"),
              Message(role=Role.ASSISTANT, content="I refreshed the workflow contract."),
          ],
 +        active_dod_path=str(dod_path),
          current_task="Tighten Loader workflow behavior",
          workflow_mode="execute",
          permission_mode="prompt",
          permission_prompting_enabled=True,
 +        prompt_format="native",
 +        prompt_sections=["Runtime Config", "Workflow Context", "Mode Guidance"],
 +        prompt_history=[
 +            PromptSnapshot(
 +                timestamp="2026-04-06T15:02:00Z",
 +                workflow_mode="plan",
 +                permission_mode="prompt",
 +                current_task="Tighten Loader workflow behavior",
 +                prompt_format="native",
 +                prompt_sections=["Runtime Config", "Workflow Context", "Mode Guidance"],
 +                content="# Introduction\nplan around planned.txt\n",
 +            ),
 +            PromptSnapshot(
 +                timestamp="2026-04-06T15:04:00Z",
 +                workflow_mode="execute",
 +                permission_mode="prompt",
 +                current_task="Tighten Loader workflow behavior",
 +                prompt_format="native",
 +                prompt_sections=[
 +                    "Runtime Config",
 +                    "Workflow Context",
 +                    "Mode Guidance",
 +                    "Project Context",
 +                ],
 +                content="# Introduction\nexecute around notes.txt\n# Project Context\npython\n",
 +            ),
 +        ],
          workflow_reason_code="full_replan_completed",
          workflow_reason_summary="clarify and plan artifacts refreshed; returning to execute",
          workflow_decision_kind="handoff",
                      ),
                  ],
                  signal_summary=["recent_reentry=1", "stale_plan=true"],
 +                artifact_paths=[
 +                    str(brief_new),
 +                    str(plan_new_root / "implementation.md"),
 +                    str(plan_new_root / "verification.md"),
 +                ],
              ),
              WorkflowTimelineEntry(
                  timestamp="2026-04-06T15:03:00Z",
+     )
 +def test_collect_prompt_diff_uses_persisted_prompt_history(temp_dir: Path) -> None:
 +    _write_python_workspace(temp_dir)
 +    _ensure_loader_dirs(temp_dir)
 +    session_id, _ = _persist_session_with_dod(temp_dir)
++
 +    diff = collect_prompt_diff(project_root=temp_dir)
++
 +    assert diff.session_id == session_id
 +    assert diff.previous is not None
 +    assert diff.current is not None
 +    assert diff.current.workflow_mode == "execute"
 +    assert diff.previous.workflow_mode == "verify"
 +    assert any("Workflow mode changed:" in item for item in diff.highlights)
 +    assert "---" in diff.unified_diff
 +    assert "execute parser fix" in diff.unified_diff
++
++
 +def test_collect_workflow_artifact_diffs_reads_versioned_artifacts(
 +    temp_dir: Path,
 +) -> None:
 +    _write_python_workspace(temp_dir)
 +    _ensure_loader_dirs(temp_dir)
 +    session_id = _persist_session_with_rich_workflow(temp_dir)
++
 +    snapshot = collect_workflow_artifact_diffs(project_root=temp_dir)
++
 +    assert snapshot.session_id == session_id
 +    assert len(snapshot.entries) == 3
 +    assert {entry.kind for entry in snapshot.entries} == {
 +        "clarify_brief",
 +        "implementation_plan",
 +        "verification_plan",
 +    }
 +    assert any("notes.txt" in entry.unified_diff for entry in snapshot.entries)
 +    assert snapshot.highlights
++
++
  def test_status_and_session_commands_render_persisted_state(
      temp_dir: Path,
      monkeypatch: pytest.MonkeyPatch,
      assert "gates=non_goals,decision_boundaries" in clarify_result.output
 +def test_workflow_show_can_render_artifact_diffs(
 +    temp_dir: Path,
 +    monkeypatch: pytest.MonkeyPatch,
 +) -> None:
 +    _write_python_workspace(temp_dir)
 +    _ensure_loader_dirs(temp_dir)
 +    _persist_session_with_rich_workflow(temp_dir)
 +    runner = CliRunner()
++
 +    monkeypatch.chdir(temp_dir)
++
 +    result = runner.invoke(
 +        cli_main_module.workflow_cli,
 +        ["show", "--diff", "--full-diff"],
 +    )
++
 +    assert result.exit_code == 0
 +    assert "Artifact Changes" in result.output
 +    assert "Artifact Diff Summary" in result.output
 +    assert "clarify_brief" in result.output
 +    assert "implementation_plan" in result.output
 +    assert "verification_plan" in result.output
 +    assert "notes.txt" in result.output
++
++
  def test_collect_prompt_preview_uses_persisted_runtime_state(temp_dir: Path) -> None:
      _write_python_workspace(temp_dir)
      _ensure_loader_dirs(temp_dir)
      assert "Execute Mode" in result.output
 +def test_prompt_diff_command_renders_persisted_prompt_changes(
 +    temp_dir: Path,
 +    monkeypatch: pytest.MonkeyPatch,
 +) -> None:
 +    _write_python_workspace(temp_dir)
 +    _ensure_loader_dirs(temp_dir)
 +    _persist_session_with_dod(temp_dir)
 +    runner = CliRunner()
++
 +    monkeypatch.chdir(temp_dir)
++
 +    result = runner.invoke(cli_main_module.prompt_cli, ["diff", "--full"])
++
 +    assert result.exit_code == 0
 +    assert "Prompt Diff" in result.output
 +    assert "Prompt Changes" in result.output
 +    assert "Workflow mode changed:" in result.output
 +    assert "Prompt Unified Diff" in result.output
 +    assert "execute parser fix" in result.output
++
++
  def test_permission_snapshot_and_dry_run_reflect_rules(temp_dir: Path) -> None:
      _write_python_workspace(temp_dir)
      _ensure_loader_dirs(temp_dir)