#!/usr/bin/env python3
"""Delivery Workbench roadmap maintenance CLI.

Thin adapter over the ``dw_pmo`` core package. All parsing, validation,
trace, and mutation behavior lives in ``pmo-roadmap/lib/dw_pmo`` so the
CLI, the workbench server, and future adapters share one implementation.
"""

from __future__ import annotations

import argparse
import json
import sys
from datetime import datetime, timedelta, timezone
from pathlib import Path


def _bootstrap_core() -> None:
    """Make dw_pmo importable in both source and installed layouts.

    Source layout: this file is pmo-roadmap/bin/dw and the package is
    pmo-roadmap/lib/dw_pmo. Installed layout: this file is
    <target>/.githooks/dw and the package is <target>/.githooks/dw_pmo.
    """
    here = Path(__file__).resolve().parent
    for candidate in (here.parent / "lib", here / "lib", here):
        if (candidate / "dw_pmo" / "__init__.py").is_file():
            if str(candidate) not in sys.path:
                sys.path.insert(0, str(candidate))
            return


_bootstrap_core()

try:
    from dw_pmo import (
        DwError,
        __version__ as DW_VERSION,
        append_trailers,
        apply_plan,
        apply_setup,
        apply_program_act,
        apply_run_act,
        apply_step,
        board_model,
        build_context_payload,
        acknowledge_notification,
        build_notifications,
        pending_deliveries,
        record_delivery,
        build_signals_inventory,
        github_provider_for,
        observe_signals,
        resolve_signal_channel,
        FixtureProvider,
        build_status,
        read_symbol_map,
        refresh_symbol_map,
        ground_project_story,
        grounding_warnings,
        build_delivery_setup,
        preview_setup,
        build_action_presentation,
        build_live_presentation,
        build_lesson_inventory,
        build_start_presentation,
        build_step_result_presentation,
        build_step,
        build_run_plan,
        build_program_plan,
        build_program_start_plan,
        build_program_act_preview,
        build_program_view,
        build_run_act_preview,
        build_run_view,
        contract_digest,
        compile_score_path,
        discover_phases,
        discover_projects,
        find_score_path,
        find_organization_path,
        find_workflow_path,
        find_program_path,
        find_rubric_path,
        find_root,
        get_phase,
        get_project,
        parse_story_rows,
        plan_phase_close,
        plan_phase_create,
        plan_phase_pause,
        plan_phase_resume,
        plan_story_create,
        plan_story_evidence,
        plan_story_status,
        read_text,
        read_program_stream,
        read_run_stream,
        tail_program_events,
        tail_run_events,
        tail_signal_events,
        next_story,
        parked_headline,
        parked_lines,
        parked_summary,
        parse_contract_facts,
        render_board,
        render_status,
        render_delivery_setup,
        render_delivery_setup_pointer,
        render_presentation,
        render_step,
        replay_run,
        run_adoption,
        fix_hooks_path,
        render_doctor,
        render_gate_failure,
        render_gate_porcelain,
        render_verify,
        render_verify_porcelain,
        run_capture,
        run_doctor,
        run_gate,
        run_verify,
        story_detail,
        story_num_from_file,
        score_inventory,
        simulate_score,
        simulate_workflow,
        start_run,
        supervise_run,
        run_inventory,
        validate_score,
        validate_workflow,
        load_score,
        program_inventory,
        program_summary_inventory,
        start_program_by_id,
        rubric_inventory,
        organization_inventory,
        workflow_inventory,
        simulate_organization,
        simulate_program,
        validate_organization,
        validate_program_path,
        load_scaffold_answers,
        scaffold_program,
        validate_rubric,
        write_agent_docs,
        write_contract,
        help_text,
    )
except ImportError as exc:  # pragma: no cover - environment failure path
    print(
        "dw: cannot import the dw_pmo core package "
        f"({exc}); expected it next to this script or under ../lib",
        file=sys.stderr,
    )
    raise SystemExit(1)


def die(message: str, code: int = 1) -> None:
    print(f"dw: {message}", file=sys.stderr)
    raise SystemExit(code)


def read_body_arg(body: str | None, from_file: Path | None) -> str:
    if body and from_file:
        die("pass either --body or --from-file, not both")
    if from_file:
        return read_text(from_file)
    return body or ""


def command_projects(args: argparse.Namespace) -> int:
    for project in discover_projects(args.root):
        print(f"{project.slug}\t{project.prefix}\t{project.path.relative_to(args.root)}")
    return 0


def command_signals_list(args: argparse.Namespace) -> int:
    inventory = build_signals_inventory(args.root, remote=args.remote, branch=args.branch)
    if args.json:
        print(json.dumps(inventory, sort_keys=True))
        return 0
    channels = inventory["channels"]
    if not channels:
        print("dw signals: no observed channels under .git/pmo-signals", file=sys.stderr)
        return 2
    for channel in channels:
        print(
            f"{channel['remote']}\t{channel['branch']}\t{channel['status']}"
            f"\t{channel['ledger_events']}\t{channel['last_observed']}"
        )
    return 0


def command_signals_observe(args: argparse.Namespace) -> int:
    remote, branch = resolve_signal_channel(args.root, args.remote, args.branch)
    if args.provider == "fixture":
        if not args.fixture_file:
            raise DwError("--provider fixture requires --fixture-file")
        provider = FixtureProvider(args.fixture_file)
    else:
        provider = github_provider_for(args.root, remote, branch)
    result = observe_signals(args.root, provider, remote, branch)
    if args.json:
        print(json.dumps(result, sort_keys=True))
        return 0
    refusal = result["refusal"] or "-"
    print(
        f"{remote}\t{branch}\t{result['status']}\t+{result['appended']}"
        f"\trefusal={refusal}\tnot_modified={str(result['not_modified']).lower()}"
    )
    return 0


def command_notifications_list(args: argparse.Namespace) -> int:
    inventory = build_notifications(args.root)
    if args.json:
        print(json.dumps(inventory, sort_keys=True))
        return 0
    items = inventory["notifications"]
    if not items:
        print("dw notifications: none pending", file=sys.stderr)
        return 2
    for item in items:
        flag = "unread" if item["unread"] else "acked"
        delivered = "delivered" if item["delivered"] else f"attempts={item['delivery_attempts']}"
        print(f"{item['id']}\t{item['kind']}\t{flag}\t{delivered}\t{item['run_id'] or item['node']}")
    return 0


def command_notifications_ack(args: argparse.Namespace) -> int:
    now_ts = datetime.now(timezone.utc).replace(microsecond=0).strftime("%Y-%m-%dT%H:%M:%SZ")
    result = acknowledge_notification(args.root, args.notification_id, now_ts)
    print(json.dumps(result, sort_keys=True))
    return 0


def command_notifications_delivered(args: argparse.Namespace) -> int:
    now_ts = datetime.now(timezone.utc).replace(microsecond=0).strftime("%Y-%m-%dT%H:%M:%SZ")
    record_delivery(
        args.root, args.notification_id, args.channel,
        not args.failed, args.failed or "sent", now_ts,
    )
    print(json.dumps({"id": args.notification_id, "ok": not args.failed}, sort_keys=True))
    return 0


def command_orchestration_list(args: argparse.Namespace) -> int:
    inventory = score_inventory(args.root)
    if args.json:
        print(json.dumps(inventory, sort_keys=True))
        return 0
    scores = inventory["scores"]
    if not scores:
        print("dw orchestration list: no scores under pm/orchestration", file=sys.stderr)
        return 0
    for score in scores:
        state = "valid" if score["valid"] else "invalid"
        semantic_hash = score.get("semantic_hash", "-")
        print(f"{score['slug'] or score['name']}\t{state}\t{score['path']}\t{semantic_hash}\t{score['title'] or '-'}")
    return 0


def command_orchestration_show(args: argparse.Namespace) -> int:
    path = find_score_path(args.root, args.score)
    compiled = compile_score_path(path)
    if args.json:
        print(json.dumps(compiled, sort_keys=True))
        return 0
    score = compiled["score"]
    print(f"{score['slug']} — {score['title']}")
    print(f"path\t{path.relative_to(args.root)}")
    print(f"semantic-hash\t{compiled['semantic_hash']}")
    print(f"document-hash\t{compiled['document_hash']}")
    print(f"nodes\t{len(score['nodes'])}")
    print(json.dumps(score, indent=2, sort_keys=True))
    return 0


def command_orchestration_validate(args: argparse.Namespace) -> int:
    path = find_score_path(args.root, args.score)
    document = validate_score(load_score(path))
    if args.json:
        print(json.dumps(document, sort_keys=True))
    elif document["valid"]:
        compiled = compile_score_path(path)
        print(f"valid\t{path.relative_to(args.root)}\t{compiled['semantic_hash']}")
    else:
        for diagnostic in document["diagnostics"]:
            print(
                f"ERROR {diagnostic['pointer']} [{diagnostic['code']}] "
                f"{diagnostic['message']}; {diagnostic['remediation']}"
            )
    return 0 if document["valid"] else 1


def command_orchestration_simulate(args: argparse.Namespace) -> int:
    path = find_score_path(args.root, args.score)
    simulation = simulate_score(load_score(path))
    if args.json:
        print(json.dumps(simulation, sort_keys=True))
        return 0
    print(f"score\t{args.score}\t{simulation['semantic_hash']}")
    for wave in simulation["waves"]:
        scheduled = ", ".join(wave["scheduled"])
        eligible = ", ".join(wave["eligible"])
        locks = ", ".join(wave["resource_groups"]) or "-"
        print(f"wave {wave['wave']}\tscheduled={scheduled}\teligible={eligible}\tlocks={locks}")
    for branch in simulation["failure_branches"]:
        detail = branch.get("node") or branch.get("checkpoint") or "-"
        print(f"failure\t{branch['source']}\t{branch['action']}\t{detail}")
    for terminal in simulation["terminals"]:
        print(f"terminal\t{terminal['node']}\t{terminal['meaning']}")
    return 0


def command_organization_list(args: argparse.Namespace) -> int:
    inventory = organization_inventory(args.root)
    if args.json:
        print(json.dumps(inventory, sort_keys=True))
        return 0
    organizations = inventory["organizations"]
    if not organizations:
        print("dw organization list: no organizations configured (ordinary mode is healthy)")
        return 0
    for organization in organizations:
        state = "valid" if organization["valid"] else "invalid"
        print(
            f"{organization['slug'] or organization['name']}\t{state}"
            f"\t{organization['path']}\t{organization.get('semantic_hash', '-')}"
            f"\t{organization['title'] or '-'}"
        )
    return 0


def command_organization_validate(args: argparse.Namespace) -> int:
    path = find_organization_path(args.root, args.organization)
    document = validate_organization(args.root, path)
    if args.json:
        print(json.dumps(document, sort_keys=True))
    elif document["valid"]:
        compiled = document["compiled"]
        print(
            f"valid\t{path.relative_to(args.root)}\t{compiled['semantic_hash']}"
        )
    else:
        for diagnostic in document["diagnostics"]:
            print(
                f"ERROR {diagnostic['source']}:{diagnostic['pointer']} "
                f"[{diagnostic['code']}] {diagnostic['message']}; "
                f"{diagnostic['remediation']}"
            )
    return 0 if document["valid"] else 1


def command_organization_simulate(args: argparse.Namespace) -> int:
    find_organization_path(args.root, args.organization)
    document = simulate_organization(args.root, args.organization)
    if args.json:
        print(json.dumps(document, sort_keys=True))
        return 0
    organization = document["organization"]
    print(f"organization\t{organization['slug']}\t{organization['semantic_hash']}")
    for team in document["teams"]:
        proof = team["logical_assignment_proof"]
        print(
            f"team\t{team['id']}\trequired_slots={proof['required_slots']}"
            f"\tsatisfiable={str(proof['satisfiable']).lower()}"
        )
        for index, wave in enumerate(team["concurrency_waves"]):
            print(f"wave\t{team['id']}\t{index}\t{','.join(wave)}")
    for council in document["councils"]:
        print(
            f"council\t{council['id']}\tmembers={council['member_cardinality']}"
            f"\tquorum={council['quorum']}\tjudge={council['judge']}"
            f"\tdecision={council['decision']['method']}"
            f"\taudit={council['audit']['mode']}"
            f"\tmax_rounds={council['budgets']['max_rounds']}"
        )
    print("starts-work\tfalse")
    return 0


def command_rubric_list(args: argparse.Namespace) -> int:
    inventory = rubric_inventory(args.root)
    if args.json:
        print(json.dumps(inventory, sort_keys=True))
        return 0
    rubrics = inventory["rubrics"]
    if not rubrics:
        print("dw rubric list: no rubric policies configured (ordinary mode is healthy)")
        return 0
    for rubric in rubrics:
        state = "valid" if rubric["valid"] else "invalid"
        print(
            f"{rubric.get('slug') or rubric['name']}\t{state}"
            f"\t{rubric['path']}\t{rubric.get('version') or '-'}"
            f"\t{rubric.get('semantic_hash', '-')}"
        )
    return 0


def command_rubric_validate(args: argparse.Namespace) -> int:
    path = find_rubric_path(args.root, args.rubric)
    document = validate_rubric(args.root, path)
    if args.json:
        print(json.dumps(document, sort_keys=True))
    elif document["valid"]:
        compiled = document["compiled"]
        print(
            f"valid\t{path.relative_to(args.root)}"
            f"\t{compiled['rubric']['slug']}@{compiled['rubric']['version']}"
            f"\t{compiled['semantic_hash']}"
        )
    else:
        for diagnostic in document["diagnostics"]:
            print(
                f"ERROR {diagnostic['source']}:{diagnostic['pointer']} "
                f"[{diagnostic['code']}] {diagnostic['message']}; "
                f"{diagnostic['remediation']}"
            )
    return 0 if document["valid"] else 1


def command_workflow_list(args: argparse.Namespace) -> int:
    inventory = workflow_inventory(args.root)
    if args.json:
        print(json.dumps(inventory, sort_keys=True))
        return 0
    workflows = inventory["workflows"]
    if not workflows:
        print("dw workflow list: no workflow policies configured")
        return 0
    for workflow in workflows:
        state = "valid" if workflow["valid"] else "invalid"
        print(
            f"{workflow.get('slug') or workflow['name']}\t{state}"
            f"\t{workflow['path']}\t{workflow.get('version') or '-'}"
            f"\t{workflow.get('bundle_hash', '-')}"
        )
    return 0


def command_workflow_validate(args: argparse.Namespace) -> int:
    document = validate_workflow(args.root, args.workflow)
    if args.json:
        print(json.dumps(document, sort_keys=True))
    elif document["valid"]:
        compiled = document["compiled"]
        print(
            f"valid\t{compiled['source']}\t{compiled['slug']}@{compiled['version']}"
            f"\t{compiled['bundle_hash']}"
        )
    else:
        for diagnostic in document["diagnostics"]:
            print(
                f"ERROR {diagnostic['source']}:{diagnostic['pointer']} "
                f"[{diagnostic['code']}] {diagnostic['message']}; "
                f"{diagnostic['remediation']}"
            )
    return 0 if document["valid"] else 1


def command_workflow_simulate(args: argparse.Namespace) -> int:
    find_workflow_path(args.root, args.workflow)
    simulation = simulate_workflow(args.root, args.workflow)
    if args.json:
        print(json.dumps(simulation, sort_keys=True))
        return 0
    workflow = simulation["workflow"]
    print(
        f"workflow\t{workflow['slug']}@{workflow['version']}"
        f"\t{workflow['bundle_hash']}"
    )
    for index, wave in enumerate(simulation["waves"]):
        print(f"wave\t{index}\t{','.join(wave)}")
    for child in simulation["children"]:
        print(
            f"child\t{child['address']}\t{child['slug']}@{child['version']}"
            f"\t{child['bundle_hash']}"
        )
    for loop in simulation["loops"]:
        print(
            f"loop\t{loop['address']}\t{loop['purpose']}"
            f"\tmax_rounds={loop['max_rounds']}"
        )
    for debate in simulation["debates"]:
        print(
            f"debate\t{debate['address']}\tmax_rounds={debate['max_rounds']}"
            f"\tjudge={debate['judge_role']}\tquorum={debate['quorum']}"
            f"\tartifact_tokens={debate['artifact_max_tokens']}"
            f"\ttie={debate['tie_policy']}\tdissent={debate['dissent_policy']}"
        )
    for route in simulation["routes"]:
        print(
            f"route\t{route['source']}\t{route['outcome']}"
            f"\t{route['kind']}:{route['target']}"
        )
    for name, value in simulation["envelopes"]["worst_case"].items():
        print(f"envelope\t{name}\t{value}")
    print("starts-work\tfalse")
    return 0


def command_program_list(args: argparse.Namespace) -> int:
    inventory = program_summary_inventory(args.root)
    if args.json:
        print(json.dumps(inventory, sort_keys=True))
        return 0
    programs = inventory["programs"]
    runs = inventory["runs"]
    if not programs and not runs:
        print("dw program list: no programs configured (ordinary mode is healthy)")
        return 0
    for program in programs:
        state = "valid" if program["valid"] else "invalid"
        print(
            f"{program['slug'] or program['name']}\t{state}\t{program['path']}"
            f"\t{program.get('semantic_hash', '-')}\t{program['title'] or '-'}"
        )
    for run in runs:
        print(
            f"run\t{run['run_id']}\t{run['state']}\t"
            f"{run.get('operational_state', '-')}\t"
            f"{run.get('program') or '-'}\t{run.get('stop') or '-'}"
        )
    return 0


def command_program_scaffold(args: argparse.Namespace) -> int:
    try:
        answers_text = args.answers.read_text(encoding="utf-8")
    except OSError as exc:
        raise DwError(f"cannot read scaffold answers {args.answers}: {exc}") from exc
    answers = load_scaffold_answers(answers_text)
    base_proposal = None
    if getattr(args, "proposal", None) is not None:
        from dw_pmo.setup_proposal import load_proposal

        try:
            base_proposal = load_proposal(args.proposal.read_text(encoding="utf-8"))
        except OSError as exc:
            raise DwError(f"cannot read base proposal {args.proposal}: {exc}") from exc
    proposal = scaffold_program(args.root, answers, base_proposal=base_proposal)
    if args.json:
        print(json.dumps(proposal, sort_keys=True, separators=(",", ":"), ensure_ascii=False))
    else:
        print(json.dumps(proposal, indent=2, sort_keys=True, ensure_ascii=False))
    return 0


def command_program_validate(args: argparse.Namespace) -> int:
    path = find_program_path(args.root, args.program)
    document = validate_program_path(args.root, path)
    if args.json:
        print(json.dumps(document, sort_keys=True))
    elif document["valid"]:
        print(f"valid\t{path.relative_to(args.root)}")
        for finding in document.get("findings", []):
            print(
                f"FINDING {finding['source']}:{finding['pointer']} "
                f"[{finding['code']}] {finding['message']}; "
                f"{finding['remediation']}"
            )
    else:
        for diagnostic in document["diagnostics"]:
            print(
                f"ERROR {diagnostic['source']}:{diagnostic['pointer']} "
                f"[{diagnostic['code']}] {diagnostic['message']}; "
                f"{diagnostic['remediation']}"
            )
    return 0 if document["valid"] else 1


def _render_program_selection(document: dict[str, object]) -> str:
    lines = [
        f"program\t{document['program']['slug']}\t{document['program']['semantic_hash']}",
        f"applicable\t{str(document.get('applicable', document.get('selection') is not None)).lower()}",
    ]
    selection = document.get("selection")
    if selection:
        lines.append(
            f"selected\t{selection['story']}\tphase={selection['phase']}"
            f"\t{selection['reason']}\tworkflow={selection['workflow']['slug']}"
            f"\tteam={selection['team']}"
        )
    else:
        lines.append("selected\t-")
    for candidate in document["candidates"]:
        blockers = ",".join(candidate["dependency_blockers"]) or "-"
        lines.append(
            f"candidate\t{candidate['story']}\tphase={candidate['phase']}"
            f"\tstatus={candidate['status']}\treason={candidate['reason']}"
            f"\tdependencies={blockers}"
        )
    assignment = document.get("assignment")
    if assignment:
        for role in assignment["roles"]:
            members = role.get("members", [])
            if not members:
                lines.append(
                    f"assignment\t{role['role']}\t{role['duty']}\t-\t-"
                )
            for selected in members:
                lines.append(
                    f"assignment\t{role['role']}[{selected['slot']}]"
                    f"\t{role['duty']}\t{selected['agent']}\t{selected['profile']}"
                    f"\tprincipal={selected['principal_fingerprint']}"
                )
        separation = assignment.get("separation")
        if separation:
            lines.append(
                f"separation\t{str(separation['passed']).lower()}"
                f"\t{json.dumps(separation['facts'], sort_keys=True)}"
            )
    for issue in document["issues"]:
        lines.append(f"REFUSE\t{issue['code']}\t{issue['message']}")
    lines.extend([
        "starts-work\tfalse",
        "writes-policy\tfalse",
        "writes-roadmap\tfalse",
        "writes-run-state\tfalse",
        "creates-grant\tfalse",
    ])
    return "\n".join(lines) + "\n"


def command_program_simulate(args: argparse.Namespace) -> int:
    document = simulate_program(args.root, args.program)
    if args.json:
        print(json.dumps(document, sort_keys=True))
    else:
        print(_render_program_selection(document), end="")
    return 0


def command_program_plan(args: argparse.Namespace) -> int:
    if args.mode is None:
        document = build_program_plan(args.root, args.program)
    else:
        if not args.operator or not args.reason or not args.intent:
            die("--mode requires --operator, --reason, and --intent")
        issued = _program_time(args.issued_at)
        if args.expires_at and args.expires_in is not None:
            die("pass either --expires-at or --expires-in, not both")
        expires = (
            args.expires_at
            or issued + timedelta(seconds=args.expires_in or 3600)
        )
        document = build_program_start_plan(
            args.root,
            args.program,
            mode=args.mode,
            operator=args.operator,
            approval_reason=args.reason,
            intent_id=args.intent,
            capabilities=args.capability,
            budgets=_program_budgets(args.budget),
            issued_at=issued,
            expires_at=expires,
            remote=args.remote,
            remote_ref=args.remote_ref,
        )
    if args.json:
        print(json.dumps(document, sort_keys=True))
    elif document["kind"] == "delivery-workbench-program-start-plan":
        print(
            render_presentation(build_start_presentation(document)),
            end="",
        )
    else:
        print(_render_program_selection(document), end="")
    return 0 if document.get("applicable", True) else 1


def _program_time(raw: str | None) -> datetime:
    if raw:
        try:
            value = datetime.fromisoformat(raw.replace("Z", "+00:00"))
        except ValueError as exc:
            raise DwError("program timestamp must be ISO-8601") from exc
        if value.tzinfo is None:
            raise DwError("program timestamp must include a timezone")
        return value.astimezone(timezone.utc).replace(microsecond=0)
    return datetime.now(timezone.utc).replace(microsecond=0)


def _program_budgets(values: list[str] | None) -> dict[str, int] | None:
    if values is None:
        return None
    budgets: dict[str, int] = {}
    for raw in values:
        name, separator, amount = raw.partition("=")
        if not separator or not name or not amount:
            raise DwError("program budget must use NAME=INTEGER")
        if name in budgets:
            raise DwError(f"program budget repeated: {name}")
        try:
            budgets[name] = int(amount)
        except ValueError as exc:
            raise DwError(f"program budget is not an integer: {name}") from exc
    return budgets


def command_program_start(args: argparse.Namespace) -> int:
    if not args.approve:
        raise DwError("program start requires --approve")
    plan = _read_run_plan(args.plan)
    request = plan.get("request")
    if (
        plan.get("kind") != "delivery-workbench-program-start-plan"
        or not isinstance(request, dict)
    ):
        raise DwError("program start plan file has the wrong kind")
    projection = start_program_by_id(
        args.root,
        str(request["program"]),
        mode=str(request["mode"]),
        operator=request["operator"],
        approval_reason=str(request["approval_reason"]),
        intent_id=str(request["intent_id"]),
        capabilities=list(request["capabilities"]),
        budgets=dict(request["budgets"]),
        issued_at=str(request["issued_at"]),
        expires_at=str(request["expires_at"]),
        remote=request["remote"],
        remote_ref=request["remote_ref"],
        expect=args.expect,
    )
    if args.json:
        print(json.dumps(projection, sort_keys=True))
    else:
        print(
            render_presentation(
                build_live_presentation(
                    build_program_view(args.root, projection["run_id"])
                )
            ),
            end="",
        )
    return 0


def command_program_show(args: argparse.Namespace) -> int:
    document = build_program_view(args.root, args.run_id)
    if args.json:
        print(json.dumps(document, sort_keys=True))
    else:
        print(
            render_presentation(build_live_presentation(document)),
            end="",
        )
    return 0


def command_program_preview(args: argparse.Namespace) -> int:
    preview = build_program_act_preview(
        args.root,
        args.run_id,
        args.action,
        reason=args.reason or "",
        decision=args.decision or "",
        request_id=args.request_id or "",
        max_ticks=args.max_ticks,
        max_seconds=args.max_seconds,
    )
    if args.json:
        print(json.dumps(preview, sort_keys=True))
    else:
        view = build_program_view(args.root, args.run_id)
        print(
            render_presentation(
                build_action_presentation(
                    preview,
                    view.get("bounded_actions"),
                )
            ),
            end="",
        )
    return 0 if preview["applicable"] else 1


def _apply_program_args(args: argparse.Namespace, action: str) -> dict[str, object]:
    return apply_program_act(
        args.root,
        args.run_id,
        action,
        args.expect,
        reason=getattr(args, "reason", "") or "",
        decision=getattr(args, "decision", "") or "",
        request_id=getattr(args, "request_id", "") or "",
        max_ticks=getattr(args, "max_ticks", 100),
        max_seconds=getattr(args, "max_seconds", 300),
    )


def command_program_act(args: argparse.Namespace) -> int:
    result = _apply_program_args(args, args.program_command)
    if args.json:
        print(json.dumps(result, sort_keys=True))
    else:
        print(
            render_presentation(
                build_live_presentation(
                    build_program_view(
                        args.root,
                        str(result.get("run_id") or args.run_id),
                    )
                )
            ),
            end="",
        )
    return 0


def command_program_supervise(args: argparse.Namespace) -> int:
    result = _apply_program_args(args, "supervise")
    if args.json:
        print(json.dumps(result, sort_keys=True))
    else:
        print(
            render_presentation(
                build_live_presentation(
                    build_program_view(args.root, result["run_id"])
                )
            ),
            end="",
        )
    return 0


def command_program_tail(args: argparse.Namespace) -> int:
    import time as time_module

    cursor = args.after
    if args.json:
        if args.follow:
            raise DwError("program tail --json cannot be combined with --follow")
        print(json.dumps(
            tail_program_events(args.root, args.run_id, cursor),
            sort_keys=True,
        ))
        return 0
    while True:
        tail = tail_program_events(args.root, args.run_id, cursor)
        for event in tail["events"]:
            print(json.dumps(event, sort_keys=True))
            cursor = int(event["seq"])
        sys.stdout.flush()
        if not args.follow:
            return 0
        time_module.sleep(max(0.1, args.interval))


def command_program_stream(args: argparse.Namespace) -> int:
    document = read_program_stream(
        args.root,
        args.run_id,
        args.session_id,
        args.stream,
        max_bytes=args.max_bytes,
    )
    if args.json:
        print(json.dumps(document, sort_keys=True))
    else:
        print(
            document["content"],
            end="" if str(document["content"]).endswith("\n") else "\n",
        )
        if document["truncated"]:
            print(
                f"[truncated: {document['included_bytes']}/{document['bytes']} bytes]",
                file=sys.stderr,
            )
    return 0


def _render_run_plan(plan: dict[str, object]) -> str:
    return render_presentation(build_start_presentation(plan))


def command_run_plan(args: argparse.Namespace) -> int:
    if args.issued_at:
        try:
            issued = datetime.fromisoformat(args.issued_at.replace("Z", "+00:00"))
        except ValueError as exc:
            raise DwError("--issued-at must be an ISO-8601 timestamp") from exc
        if issued.tzinfo is None:
            raise DwError("--issued-at must include a timezone")
        issued = issued.astimezone(timezone.utc).replace(microsecond=0)
    else:
        issued = datetime.now(timezone.utc).replace(microsecond=0)
    if args.expires_at and args.expires_in is not None:
        die("pass either --expires-at or --expires-in, not both")
    expires = args.expires_at or issued + timedelta(seconds=args.expires_in or 3600)
    plan = build_run_plan(
        args.root,
        args.score,
        args.project,
        args.story,
        issued_at=issued,
        expires_at=expires,
        standing_nudges=args.standing_nudge or [],
        signal_channel=args.signal_channel,
    )
    if args.json:
        print(json.dumps(plan, sort_keys=True))
    else:
        print(_render_run_plan(plan), end="")
    return 0 if plan["applicable"] else 1


def _read_run_plan(path: Path) -> dict[str, object]:
    try:
        text = sys.stdin.read() if str(path) == "-" else path.read_text(encoding="utf-8")
        value = json.loads(text)
    except (OSError, json.JSONDecodeError) as exc:
        raise DwError(f"cannot read run plan {path}: {exc}") from exc
    if not isinstance(value, dict):
        raise DwError("run plan file must contain one JSON object")
    return value


def command_run_start(args: argparse.Namespace) -> int:
    projection = start_run(
        args.root,
        _read_run_plan(args.plan),
        args.expect or "",
        approved=args.approve,
        approved_by=args.operator or "",
    )
    if args.json:
        print(json.dumps(projection, sort_keys=True))
    else:
        print(
            render_presentation(
                build_live_presentation(
                    build_run_view(args.root, projection["run_id"])
                )
            ),
            end="",
        )
    return 0


def command_run_list(args: argparse.Namespace) -> int:
    inventory = run_inventory(args.root)
    if args.json:
        print(json.dumps(inventory, sort_keys=True))
        return 0
    for item in inventory["runs"]:
        if item["valid"]:
            run = item["run"]
            print(f"{run['run_id']}\t{run['state']}\t{run['score']['slug']}\t{run['story']['id']}")
        else:
            print(f"{item['run_id']}\tinvalid\t{item['error']}")
    return 0


def command_run_show(args: argparse.Namespace) -> int:
    projection = replay_run(args.root, args.run_id)
    if args.json:
        print(json.dumps(projection, sort_keys=True))
    else:
        print(
            render_presentation(
                build_live_presentation(
                    build_run_view(args.root, args.run_id)
                )
            ),
            end="",
        )
    return 0


def command_run_view(args: argparse.Namespace) -> int:
    document = build_run_view(args.root, args.run_id)
    if args.json:
        print(json.dumps(document, sort_keys=True))
    else:
        print(
            render_presentation(build_live_presentation(document)),
            end="",
        )
    return 0


def command_run_preview(args: argparse.Namespace) -> int:
    preview = build_run_act_preview(
        args.root,
        args.run_id,
        args.action,
        reason=args.reason or "",
        decision=args.decision or "",
        correlation_id=args.correlation or "",
    )
    if args.json:
        print(json.dumps(preview, sort_keys=True))
    else:
        view = build_run_view(args.root, args.run_id)
        print(
            render_presentation(
                build_action_presentation(
                    preview,
                    view.get("bounded_actions"),
                )
            ),
            end="",
        )
    return 0 if preview["applicable"] else 1


def command_run_transition(args: argparse.Namespace) -> int:
    projection = apply_run_act(
        args.root,
        args.run_id,
        args.run_command,
        args.expect or "",
        reason=getattr(args, "reason", "") or "",
    )
    if args.json:
        print(json.dumps(projection, sort_keys=True))
    else:
        print(
            render_presentation(
                build_live_presentation(
                    build_run_view(args.root, args.run_id)
                )
            ),
            end="",
        )
    return 0


def command_run_tick(args: argparse.Namespace) -> int:
    result = apply_run_act(args.root, args.run_id, "tick", args.expect)
    if args.json:
        print(json.dumps(result, sort_keys=True))
    else:
        print(
            render_presentation(
                build_live_presentation(
                    build_run_view(args.root, args.run_id)
                )
            ),
            end="",
        )
    return 0


def command_run_supervise(args: argparse.Namespace) -> int:
    result = supervise_run(
        args.root,
        args.run_id,
        max_ticks=args.max_ticks,
        interval_seconds=args.interval,
    )
    if args.json:
        print(json.dumps(result, sort_keys=True))
    else:
        print(
            render_presentation(
                build_live_presentation(
                    build_run_view(args.root, args.run_id)
                )
            ),
            end="",
        )
    return 0


def command_run_checkpoint(args: argparse.Namespace) -> int:
    projection = apply_run_act(
        args.root,
        args.run_id,
        "checkpoint",
        args.expect,
        decision=args.decision,
        correlation_id=args.correlation or "",
    )
    if args.json:
        print(json.dumps(projection, sort_keys=True))
    else:
        print(
            render_presentation(
                build_live_presentation(
                    build_run_view(args.root, args.run_id)
                )
            ),
            end="",
        )
    return 0


def command_run_request(args: argparse.Namespace) -> int:
    projection = apply_run_act(
        args.root,
        args.run_id,
        "request",
        args.expect,
        decision=args.decision,
        correlation_id=args.correlation,
    )
    if args.json:
        print(json.dumps(projection, sort_keys=True))
    else:
        print(
            render_presentation(
                build_live_presentation(
                    build_run_view(args.root, args.run_id)
                )
            ),
            end="",
        )
    return 0


def command_run_tail(args: argparse.Namespace) -> int:
    import time

    cursor = args.after
    while True:
        tail = tail_run_events(args.root, args.run_id, cursor)
        for event in tail["events"]:
            print(json.dumps(event, sort_keys=True))
            cursor = int(event["seq"])
        sys.stdout.flush()
        if not args.follow:
            return 0
        time.sleep(max(0.1, args.interval))


def command_run_stream(args: argparse.Namespace) -> int:
    document = read_run_stream(
        args.root,
        args.run_id,
        args.executor,
        args.execution_id,
        args.stream,
        max_bytes=args.max_bytes,
    )
    if args.json:
        print(json.dumps(document, sort_keys=True))
    else:
        print(document["content"], end="" if str(document["content"]).endswith("\n") else "\n")
        if document["truncated"]:
            print(
                f"[truncated: {document['included_bytes']}/{document['bytes']} bytes]",
                file=sys.stderr,
            )
    return 0


def command_phase_list(args: argparse.Namespace) -> int:
    project = get_project(args.root, args.project)
    for phase in discover_phases(project):
        print(f"{phase.number}\t{phase.path.name}\t{phase.path.relative_to(args.root)}")
    return 0


def command_phase_show(args: argparse.Namespace) -> int:
    project = get_project(args.root, args.project)
    phase = get_phase(project, args.phase)
    status_file = phase.path / "current-phase-status.md"
    if not status_file.exists():
        die(f"phase status file missing: {status_file}")
    print(read_text(status_file), end="")
    return 0


def command_phase_create(args: argparse.Namespace) -> int:
    project = get_project(args.root, args.project)
    plan = plan_phase_create(
        args.root,
        project,
        args.number,
        args.title,
        slug=args.slug,
        status=args.status,
        goal=args.goal,
    )
    apply_plan(plan, validate_after=False)
    print(plan.summary["phase_dir"])
    return 0


def command_phase_close(args: argparse.Namespace) -> int:
    project = get_project(args.root, args.project)
    phase = get_phase(project, args.phase)
    summary_body = read_body_arg(args.summary, args.from_file)
    plan = plan_phase_close(
        args.root,
        project,
        phase,
        summary_body=summary_body,
        status=args.status,
        force=args.force,
    )
    apply_plan(plan, validate_after=False)
    print(plan.summary["summary_path"])
    return 0


def command_phase_pause(args: argparse.Namespace) -> int:
    project = get_project(args.root, args.project)
    phase = get_phase(project, args.phase)
    plan = plan_phase_pause(args.root, project, phase, args.reason or "")
    apply_plan(plan, validate_after=False)
    print(f"phase-{phase.number}\tpaused\t{plan.summary['reason']}")
    return 0


def command_phase_resume(args: argparse.Namespace) -> int:
    project = get_project(args.root, args.project)
    phase = get_phase(project, args.phase)
    plan = plan_phase_resume(args.root, project, phase)
    apply_plan(plan, validate_after=False)
    print(f"phase-{phase.number}\tin-progress\t(was: {plan.summary['previous_status']})")
    return 0


def command_story_create(args: argparse.Namespace) -> int:
    project = get_project(args.root, args.project)
    phase = get_phase(project, args.phase)
    plan = plan_story_create(
        args.root,
        project,
        phase,
        args.title,
        slug=args.slug,
        status=args.status,
    )
    apply_plan(plan, validate_after=False)
    print(plan.summary["story_path"])
    return 0


def command_story_status(args: argparse.Namespace) -> int:
    project = get_project(args.root, args.project)
    phase = get_phase(project, args.phase)
    evidence_body = read_body_arg(args.evidence_body, args.evidence_from_file)
    plan = plan_story_status(
        args.root,
        project,
        phase,
        args.story,
        args.status,
        evidence_body=evidence_body,
        force=args.force,
        reason=args.reason or "",
    )
    apply_plan(plan, validate_after=False)
    print(f"{plan.summary['story_id']}\t{plan.summary['status']}\t{plan.summary['story_path']}")
    return 0


def command_story_evidence(args: argparse.Namespace) -> int:
    project = get_project(args.root, args.project)
    phase = get_phase(project, args.phase)
    body = read_body_arg(args.body, args.from_file)
    plan = plan_story_evidence(
        args.root,
        project,
        phase,
        args.story,
        body=body,
        force=args.force,
    )
    apply_plan(plan, validate_after=False)
    print(plan.summary["evidence_path"])
    return 0


def command_story_show(args: argparse.Namespace) -> int:
    project = get_project(args.root, args.project)
    phase = get_phase(project, args.phase)
    detail = story_detail(project, phase, args.story, args.root)
    if args.json:
        print(json.dumps(detail, sort_keys=True))
        return 0
    note = f" ({detail['status_note']})" if detail["status_note"] else ""
    print(f"{detail['story_id']} — {detail['title']}")
    print(f"status\t{detail['status_token']}{note}")
    print(f"phase\t{phase.number} ({phase.path.name})")
    print(f"story\t{detail['paths']['story']}")
    evidence_state = "" if detail["evidence_exists"] else "\t(not written yet)"
    print(f"evidence\t{detail['paths']['evidence'] or '-'}{evidence_state}")
    print(f"api\t{detail['links']['story']}\t{detail['links']['trace']}")
    print("captured runs:")
    if detail["captured_runs"]:
        for run in detail["captured_runs"]:
            print(f"  - {run['timestamp']} `{run['command']}` exit {run['exit_code']}")
    else:
        print("  - none (narrative-only or no evidence yet)")
    print("--- story ---")
    print(detail["story_markdown"].rstrip() or "(missing story file)")
    print("--- evidence ---")
    print(detail["evidence_markdown"].rstrip() or "(no evidence file yet)")
    return 0


def command_story_list(args: argparse.Namespace) -> int:
    project = get_project(args.root, args.project)
    phases = [get_phase(project, args.phase)] if args.phase else discover_phases(project)
    for phase in phases:
        for row in parse_story_rows(phase.path / "current-phase-status.md"):
            if args.status and row.status != args.status:
                continue
            print(f"{row.story_id}\t{row.status}\tphase-{phase.number}\t{row.title}")
    return 0


def command_tree(args: argparse.Namespace) -> int:
    projects = [get_project(args.root, args.project)] if args.project else discover_projects(args.root)
    status_filter = "done" if args.done else args.status
    for project in projects:
        print(f"{project.slug} ({project.prefix})")
        phases = discover_phases(project)
        if args.phase:
            phases = [get_phase(project, args.phase)]
        for phase in phases:
            print(f"  phase {phase.number}: {phase.path.name}")
            rows = parse_story_rows(phase.path / "current-phase-status.md")
            if not rows:
                print("    (no stories)")
                continue
            for row in rows:
                if status_filter and row.status != status_filter:
                    continue
                story_num = story_num_from_file(row.story_file)
                evidence_file = phase.path / f"evidence-story-{story_num:02d}.md" if story_num else None
                evidence = "yes" if evidence_file and evidence_file.exists() else "no"
                print(f"    {row.story_id} [{row.status}] evidence:{evidence} {row.title}")
    return 0


def command_next(args: argparse.Namespace) -> int:
    # Exit contract: 0 = story found, 2 = nothing actionable, 1 = error
    # (errors surface as DwError via main()).
    project = get_project(args.root, args.project)
    found = next_story(project, args.root)
    if found is None:
        parked = parked_summary(project, args.root)
        if args.json:
            print(json.dumps({"next_story": None, "parked": parked}, sort_keys=True))
        else:
            headline = parked_headline(parked)
            tail = f"; parked: {headline} — see dw holds" if headline else ""
            print(f"dw next: nothing actionable (no in-progress, ready, or backlog stories){tail}", file=sys.stderr)
        return 2
    if args.json:
        print(json.dumps(found, sort_keys=True))
    else:
        print(f"{found['story_id']}\t{found['status']}\t{found['phase_path']}\t{found['title']}")
    return 0


def command_board(args: argparse.Namespace) -> int:
    project = get_project(args.root, args.project)
    model = board_model(project, args.root)
    if args.phase:
        phase = get_phase(project, args.phase)
        model["phases"] = [
            lane for lane in model["phases"] if lane["number"] == phase.number
        ]
    if args.json:
        print(json.dumps(model, sort_keys=True))
        return 0
    print(render_board(model, expand_closed=args.all))
    return 0


def command_holds(args: argparse.Namespace) -> int:
    # Exit contract mirrors dw next: 0 = holds listed, 2 = nothing
    # parked, 1 = error. One greppable line per hold.
    project = get_project(args.root, args.project)
    parked = parked_summary(project, args.root)
    if args.json:
        print(json.dumps(parked, sort_keys=True))
        return 0 if (parked["paused_phases"] or parked["parked_stories"]) else 2
    lines = parked_lines(parked)
    if not lines:
        print("dw holds: nothing parked (no blocked or on-hold stories, no paused phases)", file=sys.stderr)
        return 2
    for line in lines:
        print(line)
    return 0


def command_agent_docs(args: argparse.Namespace) -> int:
    target = args.file if args.file else None
    path, action = write_agent_docs(args.root, target)
    try:
        shown = str(path.relative_to(args.root))
    except ValueError:
        shown = str(path)
    print(f"{shown}\t{action}")
    return 0


def command_adopt(args: argparse.Namespace) -> int:
    if args.apply:
        die("dw adopt --apply is retired; create a reviewed setup proposal, then use 'dw setup preview <proposal-file>' and 'dw setup apply --proposal <id> --expect <token>'")
    result = run_adoption(
        args.root,
        args.from_report,
        slug=args.project,
        name=args.project_name,
        prefix=args.project_prefix,
        apply=args.apply,
    )
    if result["mode"] == "preview":
        print(f"dw adopt preview for project '{result['project']}' (nothing written):")
        for item in result["planned"]:
            print(f"  - {item}")
        print("Re-run with --apply to scaffold.", file=sys.stderr)
        return 0
    print(f"dw adopt applied for project '{result['project']}':")
    for item in result["applied"]:
        print(f"  - {item}")
    for issue in result["issues"]:
        print(f"ERROR {issue}")
    return 1 if result["issues"] else 0


def command_doctor(args: argparse.Namespace) -> int:
    if args.fix_hooks:
        check = fix_hooks_path(args.root)
        mark = "ok " if check.ok else "FAIL"
        print(f"{mark}  {check.name}: {check.detail}")
        return 0 if check.ok else 1
    checks = run_doctor(args.root)
    print(render_doctor(checks), end="")
    return 0 if all(check.ok for check in checks) else 1


def command_status(args: argparse.Namespace) -> int:
    briefing = build_status(args.root, args.project)
    if args.json:
        print(json.dumps(briefing, sort_keys=True))
    else:
        print(render_status(briefing), end="")
        print(
            render_delivery_setup_pointer(
                build_delivery_setup(args.root, args.project)
            ),
            end="",
        )
    return 0 if briefing["verdict"] == "ready" else 1


def command_knowledge_map(args: argparse.Namespace) -> int:
    document = read_symbol_map(args.root)
    print(json.dumps(document, sort_keys=True, separators=(",", ":")))
    return 0


def command_knowledge_refresh(args: argparse.Namespace) -> int:
    document = refresh_symbol_map(args.root)
    print(json.dumps(document, sort_keys=True, separators=(",", ":")))
    return 0


def command_knowledge_ground(args: argparse.Namespace) -> int:
    project = get_project(args.root, args.project)
    document = ground_project_story(args.root, project, args.story)
    print(json.dumps(document, sort_keys=True, separators=(",", ":")))
    return 0


def command_knowledge_lessons(args: argparse.Namespace) -> int:
    document = build_lesson_inventory(args.root)
    print(json.dumps(document, sort_keys=True, separators=(",", ":")))
    return 0


def command_setup(args: argparse.Namespace) -> int:
    setup_args = list(args.setup_args)
    if setup_args and setup_args[0] == "preview":
        if len(setup_args) != 2 or args.proposal or args.expect or args.technical:
            die("usage: dw setup preview <proposal-file>")
        preview = preview_setup(args.root, Path(setup_args[1]))
        print(json.dumps(preview, sort_keys=True, separators=(",", ":"), ensure_ascii=False))
        return 0
    if setup_args and setup_args[0] == "apply":
        if len(setup_args) != 1 or not args.proposal or not args.expect or args.technical:
            die("usage: dw setup apply --proposal <id> --expect <setup-token>")
        result = apply_setup(args.root, args.proposal, args.expect)
        print(json.dumps(result, sort_keys=True, separators=(",", ":"), ensure_ascii=False))
        return 0
    if args.proposal or args.expect:
        die("--proposal and --expect are only valid with 'dw setup apply'")
    if len(setup_args) > 1:
        die("usage: dw setup [project] [--technical]")
    project = setup_args[0] if setup_args else None
    setup = build_delivery_setup(args.root, project)
    print(render_delivery_setup(setup, technical=args.technical), end="")
    return 0 if setup["readiness"] == "ready" else 1


def command_step(args: argparse.Namespace) -> int:
    if args.apply:
        result, exit_code = apply_step(
            args.root,
            args.project,
            args.expect or "",
        )
        if args.json:
            print(json.dumps(result, sort_keys=True))
            return exit_code
        output = result["output"]
        stdout = output["stdout"]
        stderr = output["stderr"]
        if stdout:
            print(stdout, end="" if stdout.endswith("\n") else "\n")
        if stderr:
            print(
                stderr,
                end="" if stderr.endswith("\n") else "\n",
                file=sys.stderr,
            )
        rendered = render_presentation(
            build_step_result_presentation(result)
        )
        print(
            rendered,
            end="",
            file=(
                sys.stderr
                if result["outcome"] in {"refused", "interrupted", "failed"}
                else sys.stdout
            ),
        )
        return exit_code
    if args.expect:
        die("--expect is only valid with --apply")
    preview = build_step(args.root, args.project)
    if args.json:
        print(json.dumps(preview, sort_keys=True))
    else:
        print(render_step(preview), end="")
    return 0


def command_context(args: argparse.Namespace) -> int:
    projects = [get_project(args.root, args.project)] if args.project else discover_projects(args.root)
    status_filter = "done" if args.done else args.status
    payload = build_context_payload(
        args.root,
        projects,
        phase_selector=args.phase,
        status_filter=status_filter,
        include_trace=args.trace,
    )
    print(json.dumps(payload, indent=None if args.compact else 2, sort_keys=True))
    return 0


def command_evidence_capture(args: argparse.Namespace) -> int:
    project = get_project(args.root, args.project)
    phase = get_phase(project, args.phase)
    cmd = list(args.cmd or [])
    exit_code, evidence_path, timestamp = run_capture(
        args.root, project, phase, args.story, cmd, args.max_output_bytes
    )
    try:
        shown = str(evidence_path.relative_to(args.root))
    except ValueError:
        shown = str(evidence_path)
    print(f"{shown}\t{exit_code}\t{timestamp}")
    return exit_code


def command_contract_new(args: argparse.Namespace) -> int:
    story_ids = []
    for raw in args.story or []:
        story_ids.extend(s.strip() for s in raw.split(",") if s.strip())
    reasons = list(args.reasons or [])
    path = write_contract(
        args.root,
        story_ids=story_ids or None,
        consent=args.consent,
        reasons=reasons or None,
        force=args.force,
        tests_capture=args.tests_capture,
        tier=args.tier,
    )
    text = read_text(path)
    facts = parse_contract_facts(text) or {}
    print(f".tmp/CONTRACT.md\t{facts.get('index_tree', 'unknown')}\t{facts.get('story', 'none')}")
    print(
        "dw contract new: facts stamped. Verify each rule, flip every '- [ ]' to '- [x]', "
        "then commit. Restaging invalidates the contract (re-run with --force).",
        file=sys.stderr,
    )
    return 0


def command_contract_digest(args: argparse.Namespace) -> int:
    path = args.root / ".tmp" / "CONTRACT.md"
    if not path.is_file():
        die("no contract at .tmp/CONTRACT.md")
    print(contract_digest(read_text(path)))
    return 0


def command_contract_trailers(args: argparse.Namespace) -> int:
    path = args.root / ".tmp" / "CONTRACT.md"
    if not path.is_file():
        die("no contract at .tmp/CONTRACT.md")
    text = read_text(path)
    facts = parse_contract_facts(text)
    story_ids = list(facts["story_ids"]) if facts else []
    bundle = None
    bundle_path = args.root / ".tmp" / "BUNDLE-OK.md"
    if bundle_path.is_file():
        for line in read_text(bundle_path).splitlines():
            if line.strip():
                bundle = line.strip().lstrip("#").strip()
                break
    append_trailers(args.root, args.message_file, story_ids, contract_digest(text), bundle=bundle)
    return 0


def command_gate(args: argparse.Namespace) -> int:
    result = run_gate(args.root)
    if args.porcelain:
        print(render_gate_porcelain(result), end="")
    if not result.ok:
        sys.stderr.write(render_gate_failure(result))
        return 1
    if not args.porcelain:
        print(
            f"dw gate: pass ({result.checked_boxes}/{result.expected_boxes} checkboxes, "
            f"{len(result.shipped_stories)} story flip(s))"
        )
    return 0


def command_verify(args: argparse.Namespace) -> int:
    result = run_verify(
        args.root,
        range_spec=args.range,
        all_history=args.all,
        epoch=args.epoch,
    )
    if args.porcelain:
        print(render_verify_porcelain(result), end="")
    else:
        print(render_verify(result), end="")
    if result.error:
        return 2
    return 0 if result.ok else 1


def command_check(args: argparse.Namespace) -> int:
    from dw_pmo import check_project
    from dw_pmo.riderdocs import rider_docs_issues

    projects = [get_project(args.root, args.project)] if args.project else discover_projects(args.root)
    issues: list[str] = []
    warnings: list[str] = []
    for project in projects:
        issues.extend(check_project(project, args.root))
        warnings.extend(grounding_warnings(project, args.root))
    # Repo-level: rendered agent surfaces must match canon (WLA-12-04).
    issues.extend(rider_docs_issues(args.root))
    for issue in issues:
        print(f"ERROR {issue}")
    for warning in warnings:
        print(f"WARNING {warning}")
    if not issues:
        print("dw check: ok")
    return 1 if issues else 0


def command_state(args: argparse.Namespace) -> int:
    from dw_pmo.statefeed import build_state_feed, render_state_feed

    if args.json or args.write:
        rendered = render_state_feed(args.root)
        if args.write:
            args.write.parent.mkdir(parents=True, exist_ok=True)
            args.write.write_text(rendered + "\n", encoding="utf-8")
            print(f"{args.write}")
        if args.json:
            print(rendered)
        return 0
    for project in build_state_feed(args.root)["projects"]:
        phase = project["current_phase"]
        nxt = project["next_story"]
        phase_part = (
            f"phase {phase['number']} [{phase['status']}] "
            f"{phase['stories_done']}/{phase['stories_total']}"
            if phase
            else "no phases"
        )
        next_part = (
            f"next {nxt['story_id']} [{nxt['status']}]" if nxt else "nothing actionable"
        )
        print(f"{project['slug']}\t{phase_part}\t{next_part}\twarnings:{project['warnings']}")
    return 0


def command_events(args: argparse.Namespace) -> int:
    from dw_pmo.events import read_events

    entries = read_events(args.root, tail=args.tail)
    if args.json:
        print(json.dumps(entries, sort_keys=True))
        return 0
    for entry in entries:
        detail = " ".join(f"{k}={v}" for k, v in (entry.get("detail") or {}).items() if v is not None)
        print(f"{entry.get('ts')}\t{entry.get('event')}\t{entry.get('story') or '-'}\t{detail or '-'}")
    return 0


def command_hook(args: argparse.Namespace) -> int:
    from dw_pmo import agenthooks

    agents = (
        ["claude", "codex"] if args.agent == "all" else [args.agent]
    )
    if args.hook_action == "install":
        return max(agenthooks.install_agent(agent) for agent in agents)
    if args.hook_action == "uninstall":
        return max(agenthooks.uninstall_agent(agent) for agent in agents)
    if args.hook_action == "status":
        for agent in agents:
            report = agenthooks.status_agent(agent)
            marks = " ".join(
                f"{event}:{'on' if on else 'off'}"
                for event, on in report["events"].items()
            )
            print(f"{report['agent']}\t{marks}\t{report['settings']}")
        return 0
    if args.hook_action == "emit":
        stdin_text = "" if sys.stdin.isatty() else sys.stdin.read()
        return agenthooks.emit(args.agent, args.event, stdin_text)
    print(f"dw hook: unknown action {args.hook_action!r}", file=sys.stderr)
    return 1


def command_sessions(args: argparse.Namespace) -> int:
    from dw_pmo.sessions import correlate_sessions, render_sessions

    if args.json:
        print(render_sessions(args.registry))
        return 0
    doc = correlate_sessions(args.registry)
    if doc.get("registry") != "ok":
        print(f"sessions: registry {doc.get('registry')}")
        return 0 if doc.get("registry") == "absent" else 1
    for s_ in doc["sessions"]:
        where = s_["stories"][0]["story_id"] if s_["correlation"] == "on_story" else s_["correlation"]
        flags = []
        if s_["awaiting_response"]:
            flags.append("awaiting-response")
        if s_["stale"]:
            flags.append("stale")
        tmux = s_["tmux"]
        if tmux:
            flags.append(f"tmux {tmux['session']}:{tmux['window']}.{tmux['pane']}")
        print(f"{s_['agent']}\t{where}\t{s_['repo_root'] or '-'}\t{' '.join(flags) or '-'}")
    return 0


def command_rider_docs(args: argparse.Namespace) -> int:
    from dw_pmo.riderdocs import rider_docs_issues, write_rider_docs

    if args.check:
        issues = rider_docs_issues(args.root)
        for issue in issues:
            print(f"ERROR {issue}")
        if issues:
            return 1
        print("dw rider docs: all rendered surfaces match canon")
        return 0
    for path, action in write_rider_docs(args.root):
        try:
            shown = path.relative_to(args.root)
        except ValueError:
            shown = path
        print(f"{shown}\t{action}")
    return 0


def command_rider_install(args: argparse.Namespace) -> int:
    from dw_pmo.riderdocs import install_codex_rider, install_holdspeak_presence, install_pi_rider

    installer = {"codex": install_codex_rider, "pi": install_pi_rider, "holdspeak": install_holdspeak_presence}[args.surface]
    result = installer(args.root)
    for path, action in result["actions"]:
        try:
            shown = path.relative_to(args.root)
        except ValueError:
            shown = path
        print(f"{shown}\t{action}")
    if result.get("mcp_snippet"):
        print()
        print(result["mcp_snippet"])
    return 0


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(prog="dw", description=help_text("cli"))
    parser.add_argument("--version", action="version", version=f"dw {DW_VERSION}")
    parser.add_argument("--root", type=Path, default=None, help="repository root containing pm/roadmap")
    sub = parser.add_subparsers(dest="command", required=True)

    status = sub.add_parser(
        "status",
        help=help_text("status"),
        description=help_text("status"),
    )
    status.add_argument("project", nargs="?")
    status.add_argument("--json", action="store_true", help="emit the versioned status document")
    status.set_defaults(func=command_status)

    knowledge = sub.add_parser(
        "knowledge",
        help=help_text("knowledge"),
        description=help_text("knowledge"),
    )
    knowledge_sub = knowledge.add_subparsers(
        dest="knowledge_command", required=True
    )
    knowledge_map = knowledge_sub.add_parser(
        "map", help="read the fresh index-tree-bound symbol and structure map"
    )
    knowledge_map.add_argument(
        "--json", action="store_true", help="emit the versioned derived fact"
    )
    knowledge_map.set_defaults(func=command_knowledge_map)
    knowledge_refresh = knowledge_sub.add_parser(
        "refresh", help="incrementally refresh the derived map from tracked blobs"
    )
    knowledge_refresh.add_argument(
        "--json", action="store_true", help="emit the versioned derived fact"
    )
    knowledge_refresh.set_defaults(func=command_knowledge_refresh)
    knowledge_ground = knowledge_sub.add_parser(
        "ground", help="classify one story's advisory localization hints"
    )
    knowledge_ground.add_argument("project", help="roadmap project slug")
    knowledge_ground.add_argument("story", help="story ID or story filename")
    knowledge_ground.add_argument(
        "--json", action="store_true", help="emit the versioned grounding result"
    )
    knowledge_ground.set_defaults(func=command_knowledge_ground)
    knowledge_lessons = knowledge_sub.add_parser(
        "lessons", help="list earned machine lessons with provenance"
    )
    knowledge_lessons.add_argument(
        "--json", action="store_true", help="emit the versioned lesson inventory"
    )
    knowledge_lessons.set_defaults(func=command_knowledge_lessons)

    setup = sub.add_parser(
        "setup",
        help=help_text("setup"),
        description=help_text("setup"),
    )
    setup.add_argument(
        "setup_args",
        nargs="*",
        help="legacy project slug, or reserved preview/apply subverb arguments",
    )
    setup.add_argument("--proposal", help="proposal id from a setup preview")
    setup.add_argument("--expect", help="exact single-use setup token from a preview")
    setup.add_argument(
        "--technical",
        action="store_true",
        help="show exact source models and copyable inspection commands",
    )
    setup.set_defaults(func=command_setup)

    step = sub.add_parser(
        "step",
        help=help_text("step"),
        description=help_text("step"),
    )
    step.add_argument("project", nargs="?")
    step.add_argument("--json", action="store_true", help="emit the pure versioned preview document")
    step.add_argument("--apply", action="store_true", help="apply one current action after validating --expect")
    step.add_argument("--expect", help="sha256 state token from a fresh dw step preview")
    step.set_defaults(func=command_step)

    notifications = sub.add_parser(
        "notifications",
        help=help_text("notifications"),
        description=help_text("notifications"),
    )
    notifications_sub = notifications.add_subparsers(dest="notifications_command", required=True)
    notifications_list = notifications_sub.add_parser("list", help="list derived notifications with unread/delivery state")
    notifications_list.add_argument("--json", action="store_true", help="emit the stamped inventory")
    notifications_list.set_defaults(func=command_notifications_list)
    notifications_ack = notifications_sub.add_parser("ack", help="acknowledge one notification (idempotent, receipted)")
    notifications_ack.add_argument("notification_id")
    notifications_ack.set_defaults(func=command_notifications_ack)
    notifications_delivered = notifications_sub.add_parser(
        "delivered", help="record one delivery attempt outcome for a channel consumer"
    )
    notifications_delivered.add_argument("notification_id")
    notifications_delivered.add_argument("--channel", default="telegram")
    notifications_delivered.add_argument("--failed", help="content-free failure reason; omit for success")
    notifications_delivered.set_defaults(func=command_notifications_delivered, failed=None)

    signals = sub.add_parser(
        "signals",
        help=help_text("signals"),
        description=help_text("signals"),
    )
    signals_sub = signals.add_subparsers(dest="signals_command", required=True)
    signals_list = signals_sub.add_parser("list", help="list observed channels and derived status")
    signals_list.add_argument("--remote", help="filter by remote name")
    signals_list.add_argument("--branch", help="filter by branch name")
    signals_list.add_argument("--json", action="store_true", help="emit the stamped inventory")
    signals_list.set_defaults(func=command_signals_list, remote=None, branch=None)
    signals_observe = signals_sub.add_parser(
        "observe", help="one bounded observe pass: poll, diff semantically, append changed facts"
    )
    signals_observe.add_argument("--remote", help="remote name (default origin)")
    signals_observe.add_argument("--branch", help="branch name (default current branch)")
    signals_observe.add_argument(
        "--provider", choices=("github", "fixture"), default="github", help="SCM provider adapter"
    )
    signals_observe.add_argument("--fixture-file", help="snapshot JSON for --provider fixture")
    signals_observe.add_argument("--json", action="store_true", help="emit the stamped observe result")
    signals_observe.set_defaults(func=command_signals_observe, remote=None, branch=None, fixture_file=None)

    orchestration = sub.add_parser(
        "orchestration",
        help=help_text("orchestration"),
        description=help_text("orchestration"),
    )
    orchestration_sub = orchestration.add_subparsers(dest="orchestration_command", required=True)
    orchestration_list = orchestration_sub.add_parser("list", help="list scores under pm/orchestration")
    orchestration_list.add_argument("--json", action="store_true", help="emit the stamped inventory")
    orchestration_list.set_defaults(func=command_orchestration_list)
    orchestration_show = orchestration_sub.add_parser("show", help="compile and show one score")
    orchestration_show.add_argument("score", help="score slug or filename stem")
    orchestration_show.add_argument("--json", action="store_true", help="emit the compiled score")
    orchestration_show.set_defaults(func=command_orchestration_show)
    orchestration_validate = orchestration_sub.add_parser("validate", help="validate one score with JSON-pointer diagnostics")
    orchestration_validate.add_argument("score", help="score slug or filename stem")
    orchestration_validate.add_argument("--json", action="store_true", help="emit the stamped validation document")
    orchestration_validate.set_defaults(func=command_orchestration_validate)
    orchestration_simulate = orchestration_sub.add_parser("simulate", help="dry-run deterministic scheduling without events or work")
    orchestration_simulate.add_argument("score", help="score slug or filename stem")
    orchestration_simulate.add_argument("--json", action="store_true", help="emit the stamped simulation")
    orchestration_simulate.set_defaults(func=command_orchestration_simulate)

    organization = sub.add_parser(
        "organization",
        help=help_text("organization"),
        description=help_text("organization"),
    )
    organization_sub = organization.add_subparsers(
        dest="organization_command", required=True
    )
    organization_list = organization_sub.add_parser(
        "list", help="list tracked organizations; an empty inventory is healthy"
    )
    organization_list.add_argument(
        "--json", action="store_true", help="emit the stamped inventory"
    )
    organization_list.set_defaults(func=command_organization_list)
    organization_validate = organization_sub.add_parser(
        "validate", help="validate roles, visibility, independence, and finite replacement"
    )
    organization_validate.add_argument(
        "organization", help="organization slug or filename stem"
    )
    organization_validate.add_argument(
        "--json", action="store_true", help="emit source-aware diagnostics"
    )
    organization_validate.set_defaults(func=command_organization_validate)
    organization_simulate = organization_sub.add_parser(
        "simulate", help="show logical assignment proofs, councils, and concurrency waves"
    )
    organization_simulate.add_argument(
        "organization", help="organization slug or filename stem"
    )
    organization_simulate.add_argument(
        "--json", action="store_true", help="emit the pure organization simulation"
    )
    organization_simulate.set_defaults(func=command_organization_simulate)

    rubric = sub.add_parser(
        "rubric",
        help=help_text("rubric"),
        description=help_text("rubric"),
    )
    rubric_sub = rubric.add_subparsers(dest="rubric_command", required=True)
    rubric_list = rubric_sub.add_parser(
        "list", help="list tracked rubric policies; an empty library is healthy"
    )
    rubric_list.add_argument(
        "--json", action="store_true", help="emit the stamped rubric inventory"
    )
    rubric_list.set_defaults(func=command_rubric_list)
    rubric_validate = rubric_sub.add_parser(
        "validate", help="validate criteria, evidence, freshness, and aggregation"
    )
    rubric_validate.add_argument(
        "rubric", help="rubric slug or filename stem"
    )
    rubric_validate.add_argument(
        "--json", action="store_true", help="emit source-aware diagnostics"
    )
    rubric_validate.set_defaults(func=command_rubric_validate)

    workflow = sub.add_parser(
        "workflow",
        help=help_text("workflow"),
        description=help_text("workflow"),
    )
    workflow_sub = workflow.add_subparsers(dest="workflow_command", required=True)
    workflow_list = workflow_sub.add_parser(
        "list", help="list tracked workflow policies; an empty library is healthy"
    )
    workflow_list.add_argument("--json", action="store_true", help="emit the stamped inventory")
    workflow_list.set_defaults(func=command_workflow_list)
    workflow_validate = workflow_sub.add_parser(
        "validate", help="validate hierarchy, references, routes, and finite bounds"
    )
    workflow_validate.add_argument("workflow", help="workflow slug or filename stem")
    workflow_validate.add_argument("--json", action="store_true", help="emit source-aware diagnostics")
    workflow_validate.set_defaults(func=command_workflow_validate)
    workflow_simulate = workflow_sub.add_parser(
        "simulate", help="expand hierarchy, all routes, rounds, and finite envelopes"
    )
    workflow_simulate.add_argument("workflow", help="workflow slug or filename stem")
    workflow_simulate.add_argument("--json", action="store_true", help="emit the pure simulation")
    workflow_simulate.set_defaults(func=command_workflow_simulate)

    program = sub.add_parser(
        "program",
        help=help_text("program"),
        description=help_text("program"),
    )
    program_sub = program.add_subparsers(dest="program_command", required=True)
    program_list = program_sub.add_parser(
        "list", help="list optional programs; an empty inventory is healthy"
    )
    program_list.add_argument("--json", action="store_true", help="emit the stamped inventory")
    program_list.set_defaults(func=command_program_list)
    program_scaffold = program_sub.add_parser(
        "scaffold",
        help="compile typed setup answers into one inert validated proposal",
    )
    program_scaffold.add_argument(
        "--answers", required=True, type=Path,
        help="closed scaffold answers JSON file",
    )
    program_scaffold.add_argument(
        "--proposal", type=Path, default=None,
        help="base setup proposal (the conversation's draft) to scope "
        "against and embed policy into, before the roadmap exists",
    )
    program_scaffold.add_argument(
        "--json", action="store_true", help="emit canonical compact proposal JSON",
    )
    program_scaffold.set_defaults(func=command_program_scaffold)
    program_validate = program_sub.add_parser(
        "validate", help="validate one program, references, scope, and bindings"
    )
    program_validate.add_argument("program", help="program slug or filename stem")
    program_validate.add_argument("--json", action="store_true", help="emit validation diagnostics")
    program_validate.set_defaults(func=command_program_validate)
    program_simulate = program_sub.add_parser(
        "simulate", help="explain every roadmap candidate and deterministic assignment"
    )
    program_simulate.add_argument("program", help="program slug or filename stem")
    program_simulate.add_argument("--json", action="store_true", help="emit the pure simulation")
    program_simulate.set_defaults(func=command_program_simulate)
    program_plan = program_sub.add_parser(
        "plan",
        help=help_text("program_plan"),
        description=help_text("program_plan"),
    )
    program_plan.add_argument("program", help="program slug or filename stem")
    program_plan.add_argument(
        "--mode", choices=("advisory", "checkpointed", "continuous"),
        help="build an exact start plan at this policy-bounded mode",
    )
    program_plan.add_argument("--operator", help="accountable local operator id")
    program_plan.add_argument("--reason", help="bounded grant approval reason")
    program_plan.add_argument("--intent", help="idempotent grant intent id")
    program_plan.add_argument("--issued-at", help="exact ISO-8601 issuance")
    program_plan_expiry = program_plan.add_mutually_exclusive_group()
    program_plan_expiry.add_argument(
        "--expires-in", type=int, default=None,
        help="finite grant lifetime in seconds (default: 3600)",
    )
    program_plan_expiry.add_argument(
        "--expires-at", help="exact ISO-8601 expiry",
    )
    program_plan.add_argument(
        "--capability", action="append", default=None,
        help="exact granted capability (repeatable; default: tracked policy)",
    )
    program_plan.add_argument(
        "--budget", action="append", default=None, metavar="NAME=INTEGER",
        help="finite grant budget (repeatable; default: tracked policy)",
    )
    program_plan.add_argument("--remote", help="exact configured Git remote")
    program_plan.add_argument(
        "--remote-ref", help="exact observed remote-tracking ref",
    )
    program_plan.add_argument("--json", action="store_true", help="emit the pure program plan")
    program_plan.set_defaults(func=command_program_plan)

    program_start = program_sub.add_parser(
        "start",
        help=help_text("program_start"),
        description=help_text("program_start"),
    )
    program_start.add_argument(
        "--plan", type=Path, required=True,
        help="start-plan JSON file, or - for stdin",
    )
    program_start.add_argument(
        "--expect", required=True,
        help="exact start_token from that plan",
    )
    program_start.add_argument(
        "--approve", action="store_true",
        help="explicitly approve this exact local grant",
    )
    program_start.add_argument("--json", action="store_true")
    program_start.set_defaults(func=command_program_start)

    program_show = program_sub.add_parser(
        "show",
        help=help_text("program_show"),
        description=help_text("program_show"),
    )
    program_show.add_argument("run_id")
    program_show.add_argument("--json", action="store_true")
    program_show.set_defaults(func=command_program_show)

    program_preview = program_sub.add_parser(
        "preview",
        help=help_text("program_preview"),
        description=help_text("program_preview"),
    )
    program_preview.add_argument("run_id")
    program_preview.add_argument(
        "action",
        choices=(
            "tick", "supervise", "request", "pause", "resume",
            "revoke", "cancel",
        ),
    )
    program_preview.add_argument(
        "--reason", default="",
        help="bounded reason for request/control operations",
    )
    program_preview.add_argument(
        "--decision", default="",
        help="closed approve/reject request decision",
    )
    program_preview.add_argument(
        "--request-id", default="",
        help="exact outstanding program request id",
    )
    program_preview.add_argument("--max-ticks", type=int, default=100)
    program_preview.add_argument("--max-seconds", type=int, default=300)
    program_preview.add_argument("--json", action="store_true")
    program_preview.set_defaults(func=command_program_preview)

    for transition in ("pause", "resume", "revoke", "cancel"):
        command = program_sub.add_parser(
            transition,
            help=f"append one exact {transition} program transition",
        )
        command.add_argument("run_id")
        command.add_argument(
            "--reason", required=True,
            help="bounded content-safe reason",
        )
        command.add_argument(
            "--expect", required=True,
            help=f"act_token from a fresh program preview {transition}",
        )
        command.add_argument("--json", action="store_true")
        command.set_defaults(func=command_program_act)

    program_tick = program_sub.add_parser(
        "tick",
        help=help_text("program_tick"),
        description=help_text("program_tick"),
    )
    program_tick.add_argument("run_id")
    program_tick.add_argument(
        "--expect", required=True,
        help="act_token from a fresh program preview tick",
    )
    program_tick.add_argument("--json", action="store_true")
    program_tick.set_defaults(func=command_program_act)

    program_supervise = program_sub.add_parser(
        "supervise",
        help=help_text("program_supervise"),
        description=help_text("program_supervise"),
    )
    program_supervise.add_argument("run_id")
    program_supervise.add_argument("--max-ticks", type=int, default=100)
    program_supervise.add_argument("--max-seconds", type=int, default=300)
    program_supervise.add_argument(
        "--expect", required=True,
        help="act_token from a fresh matching supervise preview",
    )
    program_supervise.add_argument("--json", action="store_true")
    program_supervise.set_defaults(func=command_program_supervise)

    program_request = program_sub.add_parser(
        "request",
        help=help_text("program_request"),
        description=help_text("program_request"),
    )
    program_request.add_argument("run_id")
    program_request.add_argument("request_id")
    program_request.add_argument("decision", choices=("approve", "reject"))
    program_request.add_argument("--reason", required=True)
    program_request.add_argument(
        "--expect", required=True,
        help="act_token from a fresh matching request preview",
    )
    program_request.add_argument("--json", action="store_true")
    program_request.set_defaults(func=command_program_act)

    program_stream = program_sub.add_parser(
        "stream",
        help="explicitly open one bounded program-agent stdout or stderr log",
    )
    program_stream.add_argument("run_id")
    program_stream.add_argument("session_id")
    program_stream.add_argument("stream", choices=("stdout", "stderr"))
    program_stream.add_argument("--max-bytes", type=int, default=20_000)
    program_stream.add_argument("--json", action="store_true")
    program_stream.set_defaults(func=command_program_stream)

    program_tail = program_sub.add_parser(
        "tail",
        help=help_text("program_tail"),
        description=help_text("program_tail"),
    )
    program_tail.add_argument("run_id")
    program_tail.add_argument(
        "--after", type=int, default=0,
        help="emit events with seq greater than this cursor (default: all)",
    )
    program_tail.add_argument("--follow", action="store_true")
    program_tail.add_argument("--interval", type=float, default=1.0)
    program_tail.add_argument(
        "--json", action="store_true",
        help="emit the canonical bounded tail document (non-follow only)",
    )
    program_tail.set_defaults(func=command_program_tail)

    run = sub.add_parser(
        "run",
        help=help_text("run"),
        description=help_text("run"),
    )
    run_sub = run.add_subparsers(dest="run_command", required=True)
    run_plan = run_sub.add_parser(
        "plan", help=help_text("run_plan"), description=help_text("run_plan")
    )
    run_plan.add_argument("score", help="score slug or filename stem")
    run_plan.add_argument("--project", help="roadmap project slug (required when ambiguous)")
    run_plan.add_argument("--story", required=True, help="exact in-progress story id")
    run_plan.add_argument("--issued-at", help="exact ISO-8601 issuance (for deterministic adapter parity/replay)")
    run_plan_expiry = run_plan.add_mutually_exclusive_group()
    run_plan_expiry.add_argument("--expires-in", type=int, default=None, help="grant lifetime in seconds (default: 3600; maximum: 86400)")
    run_plan_expiry.add_argument("--expires-at", help="exact ISO-8601 expiry timestamp")
    run_plan.add_argument("--standing-nudge", action="append", help="standing nudge matcher: signal or signal=target (repeatable)")
    run_plan.add_argument("--signal-channel", help="outward signal channel to bind: remote/branch")
    run_plan.add_argument("--json", action="store_true", help="emit the exact start plan")
    run_plan.set_defaults(func=command_run_plan, standing_nudge=None, signal_channel=None)

    run_start = run_sub.add_parser(
        "start", help=help_text("run_start"), description=help_text("run_start")
    )
    run_start.add_argument("--plan", type=Path, required=True, help="plan JSON file, or - for stdin")
    run_start.add_argument("--expect", required=True, help="exact start_token from that plan")
    run_start.add_argument("--approve", action="store_true", help="explicitly approve this exact local grant")
    run_start.add_argument("--operator", required=True, help="bounded operator identity recorded in the grant")
    run_start.add_argument("--json", action="store_true", help="emit the initial run projection")
    run_start.set_defaults(func=command_run_start)

    run_list = run_sub.add_parser(
        "list", help=help_text("run_list"), description=help_text("run_list")
    )
    run_list.add_argument("--json", action="store_true")
    run_list.set_defaults(func=command_run_list)
    run_show = run_sub.add_parser(
        "show", help=help_text("run_show"), description=help_text("run_show")
    )
    run_show.add_argument("run_id")
    run_show.add_argument("--json", action="store_true")
    run_show.set_defaults(func=command_run_show)

    run_view = run_sub.add_parser(
        "view", help=help_text("run_view"), description=help_text("run_view")
    )
    run_view.add_argument("run_id")
    run_view.add_argument("--json", action="store_true")
    run_view.set_defaults(func=command_run_view)

    run_preview = run_sub.add_parser(
        "preview", help=help_text("run_preview"), description=help_text("run_preview")
    )
    run_preview.add_argument("run_id")
    run_preview.add_argument("action", choices=("tick", "pause", "resume", "revoke", "cancel", "checkpoint", "request"))
    run_preview.add_argument("--reason", default="", help="required for pause/revoke/cancel; bound into the token")
    run_preview.add_argument("--decision", default="", help="typed decision for checkpoint/request")
    run_preview.add_argument("--correlation", default="", help="outstanding request correlation id; checkpoint defaults to its pending id")
    run_preview.add_argument("--json", action="store_true")
    run_preview.set_defaults(func=command_run_preview)

    for transition in ("pause", "resume", "revoke", "cancel"):
        command = run_sub.add_parser(transition, help=f"append one exact {transition} transition")
        command.add_argument("run_id")
        command.add_argument("--expect", required=True, help=f"act_token from a fresh matching run preview {transition}")
        if transition != "resume":
            command.add_argument("--reason", required=True, help="bounded content-safe reason")
        command.add_argument("--json", action="store_true")
        command.set_defaults(func=command_run_transition)

    run_tick = run_sub.add_parser(
        "tick",
        help=help_text("run_tick"),
        description=help_text("run_tick"),
    )
    run_tick.add_argument("run_id")
    run_tick.add_argument("--expect", required=True, help="act_token from a fresh matching run preview tick")
    run_tick.add_argument("--json", action="store_true")
    run_tick.set_defaults(func=command_run_tick)

    run_supervise = run_sub.add_parser(
        "supervise",
        help=help_text("run_supervise"),
        description=help_text("run_supervise"),
    )
    run_supervise.add_argument("run_id")
    run_supervise.add_argument("--max-ticks", type=int, default=100)
    run_supervise.add_argument("--interval", type=float, default=1.0)
    run_supervise.add_argument("--json", action="store_true")
    run_supervise.set_defaults(func=command_run_supervise)

    run_checkpoint = run_sub.add_parser(
        "checkpoint",
        help=help_text("run_checkpoint"),
        description=help_text("run_checkpoint"),
    )
    run_checkpoint.add_argument("run_id")
    run_checkpoint.add_argument("decision")
    run_checkpoint.add_argument("--correlation", default="", help="exact request id; defaults to the pending checkpoint")
    run_checkpoint.add_argument("--expect", required=True, help="act_token from a fresh matching checkpoint preview")
    run_checkpoint.add_argument("--json", action="store_true")
    run_checkpoint.set_defaults(func=command_run_checkpoint)

    run_request = run_sub.add_parser(
        "request",
        help=help_text("run_request"),
        description=help_text("run_request"),
    )
    run_request.add_argument("run_id")
    run_request.add_argument("correlation")
    run_request.add_argument("decision")
    run_request.add_argument("--expect", required=True, help="act_token from a fresh matching request preview")
    run_request.add_argument("--json", action="store_true")
    run_request.set_defaults(func=command_run_request)

    run_stream = run_sub.add_parser("stream", help="explicitly open one bounded agent/check stdout or stderr log")
    run_stream.add_argument("run_id")
    run_stream.add_argument("executor", choices=("agent", "check"))
    run_stream.add_argument("execution_id")
    run_stream.add_argument("stream", choices=("stdout", "stderr"))
    run_stream.add_argument("--max-bytes", type=int, default=20_000)
    run_stream.add_argument("--json", action="store_true")
    run_stream.set_defaults(func=command_run_stream)
    run_tail = run_sub.add_parser(
        "tail", help=help_text("run_tail"), description=help_text("run_tail")
    )
    run_tail.add_argument("run_id")
    run_tail.add_argument("--after", type=int, default=-1, help="emit events with seq greater than this cursor (default: all)")
    run_tail.add_argument("--follow", action="store_true", help="keep polling the ledger until interrupted")
    run_tail.add_argument("--interval", type=float, default=1.0, help="poll interval in seconds for --follow")
    run_tail.set_defaults(func=command_run_tail)

    projects = sub.add_parser("projects", help="list roadmap projects")
    projects.set_defaults(func=command_projects)

    tree = sub.add_parser("tree", help="show project/phase/story tree")
    tree.add_argument("project", nargs="?")
    tree.add_argument("--phase")
    tree.add_argument("--status")
    tree.add_argument("--done", action="store_true")
    tree.set_defaults(func=command_tree)

    board = sub.add_parser(
        "board", help=help_text("board"), description=help_text("board")
    )
    board.add_argument("project", nargs="?")
    board.add_argument("--phase", help="show one phase's lane only")
    board.add_argument("--all", action="store_true", help="expand closed phases into full lanes")
    board.add_argument("--json", action="store_true", help="print the board model instead of drawing it")
    board.set_defaults(func=command_board)

    holds = sub.add_parser(
        "holds", help=help_text("holds"), description=help_text("holds")
    )
    holds.add_argument("project", nargs="?")
    holds.add_argument("--json", action="store_true")
    holds.set_defaults(func=command_holds)

    next_cmd = sub.add_parser(
        "next", help=help_text("next"), description=help_text("next")
    )
    next_cmd.add_argument("project", nargs="?")
    next_cmd.add_argument("--json", action="store_true", help="emit the story as a JSON object")
    next_cmd.set_defaults(func=command_next)

    doctor = sub.add_parser(
        "doctor", help=help_text("doctor"), description=help_text("doctor")
    )
    doctor.add_argument(
        "--fix-hooks",
        action="store_true",
        help="normalize a same-clone absolute core.hooksPath back to the relative .githooks form",
    )
    doctor.set_defaults(func=command_doctor)

    adopt = sub.add_parser("adopt", help="scaffold the roadmap from an adoption discovery report (preview by default)")
    adopt.add_argument("--from-report", type=Path, required=True, help="path to adoption-discovery.md")
    adopt.add_argument("--project", help="project slug (default: the report's Roadmap root line)")
    adopt.add_argument("--project-name", help="human project name for a new README")
    adopt.add_argument("--project-prefix", help="story-ID prefix for a new README (default: from the report's story IDs)")
    adopt.add_argument("--apply", action="store_true", help="write the scaffold (default is a dry-run preview)")
    adopt.set_defaults(func=command_adopt)

    state = sub.add_parser(
        "state", help=help_text("state"), description=help_text("state")
    )
    state.add_argument("--json", action="store_true", help="emit the feed_schema JSON document")
    state.add_argument("--write", type=Path, default=None, help="also write the JSON document to this path")
    state.set_defaults(func=command_state)

    events = sub.add_parser(
        "events", help=help_text("events"), description=help_text("events")
    )
    events.add_argument("--tail", type=int, default=None, help="only the last N events")
    events.add_argument("--json", action="store_true", help="emit events as a JSON array")
    events.set_defaults(func=command_events)

    sessions = sub.add_parser(
        "sessions", help=help_text("sessions"), description=help_text("sessions")
    )
    sessions.add_argument("--json", action="store_true", help="emit the sessions_schema JSON document")
    sessions.add_argument("--registry", type=Path, default=None, help="registry file to read (default: the HoldSpeak desk registry)")
    sessions.set_defaults(func=command_sessions)

    hook = sub.add_parser("hook", help="agent hook seam: instant push events (docs/absorption-ccgram.md §1)")
    hook.add_argument("hook_action", choices=["install", "uninstall", "status", "emit"])
    hook.add_argument("--agent", default="all", choices=["claude", "codex", "all"], help="which agent CLI (emit requires a specific one)")
    hook.add_argument("--event", default="", help="hook event name (emit only)")
    hook.set_defaults(func=command_hook)

    rider = sub.add_parser("rider", help="agent-surface rider tooling (canonical brief renderers)")
    rider_sub = rider.add_subparsers(dest="rider_command", required=True)
    rider_docs = rider_sub.add_parser("docs", help="regenerate rendered agent surfaces (.claude/commands, plugin/commands, .codex/skills, managed doc blocks) from canon")
    rider_docs.add_argument("--check", action="store_true", help="report drift without writing anything")
    rider_docs.set_defaults(func=command_rider_docs)

    rider_install = rider_sub.add_parser("install", help="wire an agent rider into this repo from canon")
    rider_install.add_argument("surface", choices=["codex", "pi", "holdspeak"], help="rider surface to install")
    rider_install.set_defaults(func=command_rider_install)

    agent_docs = sub.add_parser("agent-docs", help="write or refresh the managed Delivery Workbench block in CLAUDE.md/AGENTS.md")
    agent_docs.add_argument("--file", type=Path, help="explicit target file (default: existing CLAUDE.md, then AGENTS.md, else new CLAUDE.md)")
    agent_docs.set_defaults(func=command_agent_docs)

    context = sub.add_parser("context", help="print machine-readable roadmap context")
    context.add_argument("project", nargs="?")
    context.add_argument("--phase")
    context.add_argument("--status")
    context.add_argument("--done", action="store_true")
    context.add_argument("--trace", action="store_true")
    context.add_argument("--compact", action="store_true")
    context.set_defaults(func=command_context)

    check = sub.add_parser(
        "check", help=help_text("check"), description=help_text("check")
    )
    check.add_argument("project", nargs="?")
    check.set_defaults(func=command_check)

    gate = sub.add_parser(
        "gate", help=help_text("gate"), description=help_text("gate")
    )
    gate.add_argument("--hook", choices=["pre-commit"], default=None, help="hook context (reserved)")
    gate.add_argument("--porcelain", action="store_true", help="stable key=value output for machine consumers")
    gate.set_defaults(func=command_gate)

    verify = sub.add_parser(
        "verify",
        help=help_text("verify"),
        description=help_text("verify"),
    )
    verify.add_argument(
        "range", nargs="?", default=None, metavar="<base>..<head>",
        help="commit range to verify (default: merge-base of the default branch and HEAD, to HEAD)",
    )
    verify.add_argument("--all", action="store_true", help="verify the full history from the epoch to HEAD")
    verify.add_argument("--epoch", default=None, help="rev where remote rules begin (default: auto-detect first digest-trailer commit; PMO_VERIFY_EPOCH config)")
    verify.add_argument("--porcelain", action="store_true", help="stable key=value output for machine consumers")
    verify.set_defaults(func=command_verify)

    contract = sub.add_parser("contract", help="commit-contract commands")
    contract_sub = contract.add_subparsers(dest="contract_command", required=True)
    contract_new = contract_sub.add_parser("new", help="generate .tmp/CONTRACT.md with stamped, gate-verified facts")
    contract_new.add_argument("--story", action="append", help="story ID(s) this commit works under (repeatable or comma-separated)")
    contract_new.add_argument("--consent", choices=["yes", "no"], default="no", help="work-log consent")
    contract_new.add_argument("--reasons", action="append", help="work-log reason line (repeatable; used with --consent yes)")
    contract_new.add_argument(
        "--tests-capture",
        help="discharge the 'Tests ran.' rule mechanically: <staged-evidence-path>[#timestamp] of a passing captured run",
    )
    contract_new.add_argument(
        "--tier",
        choices=["auto", "full", "short"],
        default="auto",
        help="contract tier; auto picks short only for commits that do not touch the roadmap tree (full is always accepted)",
    )
    contract_new.add_argument("--force", action="store_true", help="replace an existing contract")
    contract_new.set_defaults(func=command_contract_new)
    contract_digest_cmd = contract_sub.add_parser("digest", help="print the sha256 digest of the current contract")
    contract_digest_cmd.set_defaults(func=command_contract_digest)
    contract_trailers = contract_sub.add_parser("trailers", help="stamp PMO trailers onto a commit message file (used by the commit-msg shim)")
    contract_trailers.add_argument("--message-file", type=Path, required=True)
    contract_trailers.set_defaults(func=command_contract_trailers)

    phase = sub.add_parser("phase", help="phase commands")
    phase_sub = phase.add_subparsers(dest="phase_command", required=True)
    phase_list = phase_sub.add_parser("list", help="list phases")
    phase_list.add_argument("project", nargs="?")
    phase_list.set_defaults(func=command_phase_list)
    phase_show = phase_sub.add_parser("show", help="print phase status")
    phase_show.add_argument("project")
    phase_show.add_argument("phase")
    phase_show.set_defaults(func=command_phase_show)
    phase_create = phase_sub.add_parser("create", help="create a phase")
    phase_create.add_argument("project")
    phase_create.add_argument("number", type=int)
    phase_create.add_argument("title")
    phase_create.add_argument("--goal")
    phase_create.add_argument("--slug")
    phase_create.add_argument("--status", default="not-started")
    phase_create.set_defaults(func=command_phase_create)
    phase_close = phase_sub.add_parser("close", help="close a phase with a final summary")
    phase_close.add_argument("project")
    phase_close.add_argument("phase")
    phase_close.add_argument("--summary")
    phase_close.add_argument("--from-file", type=Path)
    phase_close.add_argument("--status", default="done")
    phase_close.add_argument("--force", action="store_true")
    phase_close.set_defaults(func=command_phase_close)
    phase_pause = phase_sub.add_parser("pause", help="park a phase with a recorded reason (header + README index row)")
    phase_pause.add_argument("project")
    phase_pause.add_argument("phase")
    phase_pause.add_argument("--reason", help="why the phase is pausing (required)")
    phase_pause.set_defaults(func=command_phase_pause)
    phase_resume = phase_sub.add_parser("resume", help="release a paused phase back to in-progress")
    phase_resume.add_argument("project")
    phase_resume.add_argument("phase")
    phase_resume.set_defaults(func=command_phase_resume)

    evidence = sub.add_parser("evidence", help="evidence commands")
    evidence_sub = evidence.add_subparsers(dest="evidence_command", required=True)
    evidence_capture = evidence_sub.add_parser(
        "capture",
        help="run a command and append the captured, verifiable output block to the story's evidence file",
    )
    evidence_capture.add_argument("project")
    evidence_capture.add_argument("phase")
    evidence_capture.add_argument("story")
    evidence_capture.add_argument("--max-output-bytes", type=int, default=20000)
    evidence_capture.add_argument("cmd", nargs="*", help="command to run (everything after `--`)")
    evidence_capture.set_defaults(func=command_evidence_capture)

    story = sub.add_parser("story", help="story commands")
    story_sub = story.add_subparsers(dest="story_command", required=True)
    story_list = story_sub.add_parser("list", help="list stories")
    story_list.add_argument("project", nargs="?")
    story_list.add_argument("--phase")
    story_list.add_argument("--status")
    story_list.set_defaults(func=command_story_list)
    story_create = story_sub.add_parser("create", help="create a story")
    story_create.add_argument("project")
    story_create.add_argument("phase")
    story_create.add_argument("title")
    story_create.add_argument("--slug")
    story_create.add_argument("--status", default="backlog")
    story_create.set_defaults(func=command_story_create)
    story_show = story_sub.add_parser("show", help="browse one story whole: header, status + why, bodies, captured runs, receipts and links")
    story_show.add_argument("project")
    story_show.add_argument("phase")
    story_show.add_argument("story")
    story_show.add_argument("--json", action="store_true")
    story_show.set_defaults(func=command_story_show)
    story_status = story_sub.add_parser("status", help="transactionally update story and phase-table status")
    story_status.add_argument("project")
    story_status.add_argument("phase")
    story_status.add_argument("story")
    story_status.add_argument("status")
    story_status.add_argument("--evidence-body")
    story_status.add_argument("--evidence-from-file", type=Path)
    story_status.add_argument("--force", action="store_true")
    story_status.add_argument(
        "--reason",
        help="why this status (required for on-hold/paused; recorded in the cell as decoration)",
    )
    story_status.set_defaults(func=command_story_status)
    story_evidence = story_sub.add_parser("evidence", help="create or attach paired story evidence")
    story_evidence.add_argument("project")
    story_evidence.add_argument("phase")
    story_evidence.add_argument("story")
    story_evidence.add_argument("--body")
    story_evidence.add_argument("--from-file", type=Path)
    story_evidence.add_argument("--force", action="store_true")
    story_evidence.set_defaults(func=command_story_evidence)
    return parser


def main(argv: list[str] | None = None) -> int:
    argv = list(sys.argv[1:] if argv is None else argv)
    # Everything after a standalone `--` is an opaque passthrough command
    # (used by `evidence capture`); argparse's REMAINDER would otherwise
    # swallow options that precede it.
    passthrough: list[str] | None = None
    if "--" in argv:
        split = argv.index("--")
        passthrough = argv[split + 1:]
        argv = argv[:split]
    parser = build_parser()
    args = parser.parse_args(argv)
    if passthrough is not None:
        args.cmd = passthrough
    args.root = (args.root.resolve() if args.root else find_root(Path.cwd()))
    try:
        return args.func(args)
    except DwError as err:
        print(f"dw: {err.message}", file=sys.stderr)
        return err.code


if __name__ == "__main__":
    raise SystemExit(main())
