Collaborative Task Design

interface:
  display_name: "Collaborative task design"
  short_description: "Run a real multi-model engineering planning workflow that turns a spec into sequenced, test-first delivery tasks."
  default_prompt: "Use $collaborative_task_design for this workflow."
policy:
  allow_implicit_invocation: true

[skill]
name = "collaborative_task_design"
artifact_dir = ".codex_workflow/task_design"
mandatory_presence = ["synthesis_anchor", "adversarial_anchor"]
required_phases = ["architecture_mapping", "test_strategy", "task_slicing", "dependency_review", "convergence"]
required_outputs = ["architecture_plan.md", "tasks.md", "test_plan.md", "parallelization_plan.md", "risk_register.md", "decision_log.md", "panel_summary.json"]
workflow = "Follow the skill SKILL.md instructions, keep role outputs independent before reconciliation, preserve disagreements in the decision log, and do not mark the phase complete until every listed role has a real response and mandatory anchor roles are recorded."

[providers.synthesis_anchor]
kind = "native_codex"
provider = "codex"
model = "gpt-5.5"
model_label = "GPT 5.5"
enabled = true
native_response_required = true
duty = "Synthesize, reconcile, coordinate native Codex subagents when allowed, and own final decisions inside the host Codex session."

[providers.adversarial_anchor]
kind = "runner"
runner = "claude"
provider = "anthropic"
model = "opus"
model_label = "Opus 4.8"
enabled = true
script = ".agents/skills/claude-runner/scripts/run_claude.py"
timeout_seconds = 1200
duty = "Challenge assumptions, find risks, inspect security and architecture, and block unsafe convergence."
runner_args = ["--output-format", "json", "--no-session-persistence", "--restrict-tools"]

[providers.broad_context]
kind = "runner"
runner = "gemini"
provider = "google"
model = "gemini-3.1-pro-preview"
model_label = "Gemini 3.1 Pro"
enabled = true
script = ".agents/skills/gemini-runner/scripts/run_gemini.py"
timeout_seconds = 1200
duty = "Explore broad context, user journeys, interface options, large code context, and alternative scenarios."
runner_args = ["--output-format", "json"]

[providers.feasibility]
kind = "runner"
runner = "kimi"
provider = "moonshot"
model = "kimi-code/kimi-for-coding"
model_label = "Kimi Code"
enabled = true
script = ".agents/skills/kimi-runner/scripts/run_kimi.py"
timeout_seconds = 1200
duty = "Pressure test execution feasibility, sequencing, long horizon integration, and implementation risks."
runner_args = ["--output-format", "stream-json", "--no-session-persistence", "--restrict-tools"]

[roles.synthesis_anchor]
provider = "synthesis_anchor"
runner_role = "synthesizer"
duty = "Native synthesis, final reconciliation, implementation aware decision making, and audit trail ownership."

[roles.adversarial_anchor]
provider = "adversarial_anchor"
runner_role = "adversarial"
duty = "Adversarial critique, architecture risk, security risk, data safety, edge cases, and blocking objections."

[roles.architecture]
provider = "synthesis_anchor"
runner_role = "planner"
duty = "Map bounded contexts, use cases, ports, adapters, dependency direction, transaction boundaries, and integration seams."

[roles.testing]
provider = "synthesis_anchor"
runner_role = "planner"
duty = "Design tests before implementation, including unit, integration, contract, component, end to end, regression, and security tests as relevant."

[roles.interface]
provider = "broad_context"
runner_role = "planner"
duty = "Slice interface work by user value, component boundaries, state management, accessibility, responsiveness, and visual regression risk."

[roles.backend]
provider = "synthesis_anchor"
runner_role = "planner"
duty = "Slice backend work by use cases, domain rules, storage, API contracts, permissions, jobs, events, and observability."

[roles.delivery]
provider = "feasibility"
runner_role = "planner"
duty = "Sequence work in safe increments, minimize merge risk, identify parallelizable tracks, and define verification commands."

[phases.architecture_mapping]
roles = ["synthesis_anchor", "adversarial_anchor", "architecture", "backend", "interface", "delivery"]

[phases.test_strategy]
roles = ["synthesis_anchor", "adversarial_anchor", "testing", "backend", "interface", "delivery"]

[phases.task_slicing]
roles = ["synthesis_anchor", "adversarial_anchor", "delivery", "architecture", "testing", "backend", "interface"]

[phases.dependency_review]
roles = ["synthesis_anchor", "adversarial_anchor", "delivery", "backend", "interface"]

[phases.convergence]
roles = ["synthesis_anchor", "adversarial_anchor", "architecture", "testing", "delivery", "backend", "interface"]

#!/usr/bin/env python3
"""Run one role-panel phase for a collaborative portable skill.

The runner is deliberately honest about model participation:

* external seats use repo-local runner wrappers with fallback disabled when configured
* native Codex seats produce prompts and require a recorded native response artifact
* panel_summary.json accumulates phase runs instead of overwriting earlier evidence

The script does not pretend that a handoff prompt equals model execution.
"""
from __future__ import annotations

import argparse
import datetime as _dt
import json
import os
from pathlib import Path
import shlex
import subprocess
import sys
import textwrap
import time
import traceback
from typing import Any

try:
    import tomllib
except Exception:  # pragma: no cover
    tomllib = None


OK_STATUSES = {"ok", "native_response_recorded"}


def now_iso() -> str:
    return _dt.datetime.now(_dt.timezone.utc).isoformat()


def stamp() -> str:
    return _dt.datetime.now().strftime("%Y%m%d_%H%M%S_%f")


def safe_name(value: str) -> str:
    return "".join(c if c.isalnum() or c in "._-" else "_" for c in value)


def load_toml(path: Path) -> dict[str, Any]:
    if tomllib is None:
        raise RuntimeError("Python 3.11 or newer is required for TOML parsing")
    with path.open("rb") as fh:
        return tomllib.load(fh)


def read_text(path: Path) -> str:
    return path.read_text(encoding="utf-8")


def read_context(paths: list[str]) -> str:
    sections: list[str] = []
    for raw_path in paths:
        path = Path(raw_path).expanduser()
        if not path.exists():
            raise FileNotFoundError(raw_path)
        sections.append(f"## Context file: {path}\n\n{read_text(path)}")
    return "\n\n---\n\n".join(sections)


def get_skill_root() -> Path:
    return Path(__file__).resolve().parents[1]


def resolve_path(raw_path: str, working_dir: Path, skill_root: Path) -> Path:
    path = Path(raw_path).expanduser()
    if path.is_absolute():
        return path
    candidates = [
        working_dir / path,
        skill_root / path,
        Path.cwd() / path,
    ]
    for candidate in candidates:
        if candidate.exists():
            return candidate
    return candidates[0]


def template(value: str, variables: dict[str, Any]) -> str:
    out = value
    for key, val in variables.items():
        out = out.replace("{" + key + "}", str(val))
    return out


def parse_native_response_args(items: list[str]) -> dict[str, Path]:
    responses: dict[str, Path] = {}
    for item in items:
        if "=" not in item:
            raise ValueError("--native-response must use ROLE=PATH")
        role, path = item.split("=", 1)
        responses[role.strip()] = Path(path).expanduser()
    return responses


def build_prompt(
    skill: dict[str, Any],
    role_name: str,
    role_cfg: dict[str, Any],
    provider_cfg: dict[str, Any],
    phase: str,
    goal: str,
    context: str,
) -> str:
    skill_name = skill.get("name", "portable skill")
    workflow = skill.get("workflow", "")
    duty = role_cfg.get("duty") or provider_cfg.get("duty") or "Contribute to this phase."
    mandatory = ", ".join(skill.get("mandatory_presence", []))
    return textwrap.dedent(
        f"""
        You are a real participant in a role-based model panel for a Codex skill.

        Skill: {skill_name}
        Phase: {phase}
        Role: {role_name}
        Duty: {duty}
        Mandatory anchor roles for every phase: {mandatory}

        User goal:
        {goal}

        Context:
        {context or "(No additional context file was provided.)"}

        Workflow contract:
        {workflow}

        Return concise Markdown with these sections:
        1. Position
        2. Evidence from context
        3. Risks and contradictions
        4. Recommended decision
        5. Tests or validation needed
        6. Open questions

        Work independently. Do not assume another role will cover your duty.
        Do not claim consensus when there is meaningful disagreement.
        """
    ).strip()


def write_prompt(prompts_dir: Path, phase: str, role: str, prompt: str) -> Path:
    prompts_dir.mkdir(parents=True, exist_ok=True)
    path = prompts_dir / f"{stamp()}_{safe_name(phase)}_{safe_name(role)}.md"
    path.write_text(prompt, encoding="utf-8")
    return path


def record_native_role(
    *,
    provider_cfg: dict[str, Any],
    role: str,
    phase: str,
    prompt_path: Path,
    native_responses_dir: Path,
    native_response_overrides: dict[str, Path],
) -> dict[str, Any]:
    expected_path = native_response_overrides.get(role) or native_responses_dir / f"{phase}_{role}.md"
    result: dict[str, Any] = {
        "role": role,
        "kind": "native_codex",
        "provider": provider_cfg.get("provider", "codex"),
        "model": provider_cfg.get("model"),
        "model_label": provider_cfg.get("model_label"),
        "prompt_path": str(prompt_path),
        "expected_response_path": str(expected_path),
        "required": True,
    }
    if expected_path.exists() and expected_path.read_text(encoding="utf-8").strip():
        result.update(
            {
                "status": "ok",
                "participation": "native_response_recorded",
                "response_path": str(expected_path),
            }
        )
    else:
        result.update(
            {
                "status": "awaiting_native_execution",
                "participation": "prompt_only",
                "instruction": (
                    "Run this native Codex role through the host agent or an allowed "
                    "native Codex subagent, write the response to expected_response_path, "
                    "then rerun this phase or pass --native-response role=path."
                ),
            }
        )
    return result


def load_runner_payload(output_file: Path, stdout: str) -> dict[str, Any]:
    if output_file.exists() and output_file.read_text(encoding="utf-8").strip():
        try:
            return json.loads(output_file.read_text(encoding="utf-8"))
        except json.JSONDecodeError as exc:
            return {"success": False, "return_code": -3, "stderr": f"Invalid runner JSON: {exc}"}
    try:
        return json.loads(stdout) if stdout.strip() else {}
    except json.JSONDecodeError:
        return {"success": False, "return_code": -3, "stdout": stdout, "stderr": "Runner did not emit JSON"}


def run_runner_role(
    *,
    provider_cfg: dict[str, Any],
    role_cfg: dict[str, Any],
    role: str,
    phase: str,
    prompt_path: Path,
    output_file: Path,
    stdout_path: Path,
    stderr_path: Path,
    working_dir: Path,
    skill_root: Path,
    dry_run: bool,
) -> dict[str, Any]:
    script_raw = provider_cfg.get("script")
    if not script_raw:
        return {
            "role": role,
            "kind": "runner",
            "provider": provider_cfg.get("provider"),
            "model": provider_cfg.get("model"),
            "status": "missing_runner_script",
            "required": True,
        }

    script = resolve_path(str(script_raw), working_dir, skill_root)
    metadata = {
        "skill": skill_root.name,
        "phase": phase,
        "role": role,
        "provider": provider_cfg.get("provider"),
        "model": provider_cfg.get("model"),
    }
    variables = {
        "phase": phase,
        "role": role,
        "model": provider_cfg.get("model", ""),
        "prompt_file": str(prompt_path),
        "output_file": str(output_file),
        "working_dir": str(working_dir),
    }

    cmd = [
        sys.executable,
        str(script),
        "--prompt-file",
        str(prompt_path),
        "--timeout",
        str(int(provider_cfg.get("timeout_seconds", 900))),
        "--working-dir",
        str(working_dir),
        "--json",
        "--disable-fallback",
        "--output-file",
        str(output_file),
        "--metadata-json",
        json.dumps(metadata, ensure_ascii=False),
    ]
    model = provider_cfg.get("model")
    if model:
        cmd.extend(["--model", str(model)])
    runner_role = role_cfg.get("runner_role") or provider_cfg.get("runner_role")
    if runner_role:
        cmd.extend(["--role", str(runner_role)])
    for arg in provider_cfg.get("runner_args", []):
        cmd.append(template(str(arg), variables))

    command_preview = " ".join(shlex.quote(part) for part in cmd)
    result: dict[str, Any] = {
        "role": role,
        "kind": "runner",
        "provider": provider_cfg.get("provider"),
        "model": provider_cfg.get("model"),
        "model_label": provider_cfg.get("model_label"),
        "runner_script": str(script),
        "prompt_path": str(prompt_path),
        "output_file": str(output_file),
        "stdout_path": str(stdout_path),
        "stderr_path": str(stderr_path),
        "command_preview": command_preview,
        "required": True,
    }
    if dry_run:
        result["status"] = "dry_run"
        return result
    if not script.exists():
        result.update({"status": "runner_unavailable", "blocked_reason": "missing_runner_script"})
        return result

    started = time.time()
    try:
        completed = subprocess.run(
            cmd,
            cwd=working_dir,
            capture_output=True,
            text=True,
            timeout=int(provider_cfg.get("timeout_seconds", 900)) + 30,
        )
        stdout_path.write_text(completed.stdout or "", encoding="utf-8")
        stderr_path.write_text(completed.stderr or "", encoding="utf-8")
        payload = load_runner_payload(output_file, completed.stdout or "")
        success = bool(payload.get("success")) and int(payload.get("return_code", completed.returncode)) == 0
        fallback_reason = payload.get("fallback_reason")
        effective_runner = payload.get("effective_runner")
        requested_runner = provider_cfg.get("runner") or provider_cfg.get("provider")
        lost_independence = bool(fallback_reason) or (
            effective_runner is not None and requested_runner is not None and str(effective_runner) != str(requested_runner)
        )
        status = "ok" if success and not lost_independence else "error"
        if lost_independence:
            status = "fallback_used"
        result.update(
            {
                "status": status,
                "success": success,
                "return_code": payload.get("return_code", completed.returncode),
                "auth_ok": payload.get("auth_ok"),
                "effective_runner": effective_runner,
                "effective_model": payload.get("effective_model"),
                "effective_provider": payload.get("effective_provider"),
                "fallback_reason": fallback_reason,
                "blocked_reason": payload.get("blocked_reason"),
                "elapsed_seconds": round(time.time() - started, 3),
            }
        )
    except Exception as exc:
        stderr_path.write_text(traceback.format_exc(), encoding="utf-8")
        result.update(
            {
                "status": "exception",
                "success": False,
                "return_code": -3,
                "error": repr(exc),
                "elapsed_seconds": round(time.time() - started, 3),
            }
        )
    return result


def run_direct_cli_role(
    *,
    provider_cfg: dict[str, Any],
    role: str,
    phase: str,
    prompt: str,
    prompt_path: Path,
    stdout_path: Path,
    stderr_path: Path,
    dry_run: bool,
) -> dict[str, Any]:
    variables = {
        "prompt": prompt,
        "phase": phase,
        "role": role,
        "model": provider_cfg.get("model", ""),
        "prompt_file": str(prompt_path),
    }
    cmd = [template(provider_cfg["command"], variables)]
    for arg in provider_cfg.get("args", []):
        cmd.append(template(str(arg), variables))

    result: dict[str, Any] = {
        "role": role,
        "kind": "cli",
        "provider": provider_cfg.get("provider", "cli"),
        "model": provider_cfg.get("model"),
        "prompt_path": str(prompt_path),
        "stdout_path": str(stdout_path),
        "stderr_path": str(stderr_path),
        "command_preview": " ".join(shlex.quote(part) for part in cmd),
        "required": True,
    }
    if dry_run:
        result["status"] = "dry_run"
        return result

    timeout = int(provider_cfg.get("timeout_seconds", 900))
    started = time.time()
    try:
        completed = subprocess.run(
            cmd,
            input=prompt if provider_cfg.get("prompt_transport") == "stdin" else None,
            text=True,
            capture_output=True,
            timeout=timeout,
            env=os.environ.copy(),
        )
        stdout_path.write_text(completed.stdout or "", encoding="utf-8")
        stderr_path.write_text(completed.stderr or "", encoding="utf-8")
        result.update(
            {
                "status": "ok" if completed.returncode == 0 else "error",
                "success": completed.returncode == 0,
                "return_code": completed.returncode,
                "elapsed_seconds": round(time.time() - started, 3),
            }
        )
    except Exception as exc:
        stderr_path.write_text(traceback.format_exc(), encoding="utf-8")
        result.update(
            {
                "status": "exception",
                "success": False,
                "return_code": -3,
                "error": repr(exc),
                "elapsed_seconds": round(time.time() - started, 3),
            }
        )
    return result


def phase_is_complete(results: list[dict[str, Any]], required_roles: list[str]) -> bool:
    by_role = {str(item.get("role")): item for item in results}
    for role in required_roles:
        if by_role.get(role, {}).get("status") not in OK_STATUSES:
            return False
    for item in results:
        if item.get("required", True) and item.get("status") not in OK_STATUSES:
            return False
    return True


def load_existing_summary(path: Path, skill: dict[str, Any]) -> dict[str, Any]:
    if not path.exists():
        return {}
    try:
        summary = json.loads(path.read_text(encoding="utf-8"))
    except json.JSONDecodeError:
        return {"legacy_unparseable_summary": str(path)}
    if "phase_runs" in summary:
        return summary
    return {
        "skill": summary.get("skill") or skill.get("name"),
        "artifact_dir": summary.get("artifact_dir"),
        "legacy_phase_runs": [summary],
        "phase_runs": [],
    }


def update_summary(
    *,
    summary_path: Path,
    skill: dict[str, Any],
    cfg: dict[str, Any],
    base_out: Path,
    run: dict[str, Any],
) -> dict[str, Any]:
    summary = load_existing_summary(summary_path, skill)
    required_phases = skill.get("required_phases") or list(cfg.get("phases", {}).keys())
    phase_runs = list(summary.get("phase_runs", []))
    phase_runs.append(run)
    complete_phases = sorted({item.get("phase") for item in phase_runs if item.get("complete")})
    missing_required_phases = [phase for phase in required_phases if phase not in complete_phases]
    updated = {
        **{k: v for k, v in summary.items() if k not in {"phase_runs", "latest_run"}},
        "skill": skill.get("name"),
        "artifact_dir": str(base_out),
        "updated_at": now_iso(),
        "required_phases": required_phases,
        "complete_phases": complete_phases,
        "missing_required_phases": missing_required_phases,
        "phase_runs": phase_runs,
        "latest_run": run,
        "complete": not missing_required_phases,
    }
    summary_path.write_text(json.dumps(updated, indent=2, ensure_ascii=False), encoding="utf-8")
    return updated


def main() -> int:
    parser = argparse.ArgumentParser(description="Run one configured model-panel phase.")
    parser.add_argument("--phase", required=True)
    parser.add_argument("--goal", required=True)
    parser.add_argument("--context-file", action="append", default=[])
    parser.add_argument("--out")
    parser.add_argument("--working-dir", default=os.getcwd())
    parser.add_argument("--dry-run", action="store_true")
    parser.add_argument("--roles", help="Comma-separated role override")
    parser.add_argument(
        "--native-response",
        action="append",
        default=[],
        metavar="ROLE=PATH",
        help="Path to an already executed native response for a native role.",
    )
    parser.add_argument(
        "--fail-on-incomplete",
        action="store_true",
        help="Return non-zero when any required role is missing, pending, or failed.",
    )
    args = parser.parse_args()

    skill_root = get_skill_root()
    cfg = load_toml(skill_root / "assets" / "routing.toml")
    skill = cfg.get("skill", {})
    working_dir = Path(args.working_dir).expanduser().resolve()
    phase_cfg = cfg.get("phases", {}).get(args.phase, {})
    configured_roles = list(phase_cfg.get("roles", []))
    if args.roles:
        roles = [role.strip() for role in args.roles.split(",") if role.strip()]
    else:
        roles = list(configured_roles)
    mandatory = list(skill.get("mandatory_presence", []))
    for role in reversed(mandatory):
        if role not in roles:
            roles.insert(0, role)
    required_roles = list(dict.fromkeys(configured_roles + mandatory))
    if not required_roles:
        required_roles = list(dict.fromkeys(roles + mandatory))

    context = read_context(args.context_file)
    native_response_overrides = parse_native_response_args(args.native_response)
    base_out = Path(args.out or skill.get("artifact_dir") or ".codex_workflow/panel")
    prompts_dir = base_out / "prompts"
    transcripts_dir = base_out / "transcripts"
    native_responses_dir = base_out / "native_responses"
    prompts_dir.mkdir(parents=True, exist_ok=True)
    transcripts_dir.mkdir(parents=True, exist_ok=True)
    native_responses_dir.mkdir(parents=True, exist_ok=True)

    providers = cfg.get("providers", {})
    role_defs = cfg.get("roles", {})
    results: list[dict[str, Any]] = []
    for role in roles:
        role_cfg = role_defs.get(role, {})
        provider_key = role_cfg.get("provider") or role
        provider_cfg = providers.get(provider_key)
        if not provider_cfg:
            results.append(
                {
                    "role": role,
                    "status": "missing_provider",
                    "provider": provider_key,
                    "required": True,
                }
            )
            continue
        if provider_cfg.get("enabled", True) is False:
            results.append(
                {
                    "role": role,
                    "status": "disabled",
                    "provider": provider_key,
                    "required": True,
                }
            )
            continue

        prompt = build_prompt(skill, role, role_cfg, provider_cfg, args.phase, args.goal, context)
        prompt_path = write_prompt(prompts_dir, args.phase, role, prompt)
        role_stamp = f"{stamp()}_{safe_name(args.phase)}_{safe_name(role)}"
        stdout_path = transcripts_dir / f"{role_stamp}_stdout.txt"
        stderr_path = transcripts_dir / f"{role_stamp}_stderr.txt"
        output_file = transcripts_dir / f"{role_stamp}_output.json"
        kind = provider_cfg.get("kind")
        if kind == "native_codex":
            results.append(
                record_native_role(
                    provider_cfg=provider_cfg,
                    role=role,
                    phase=args.phase,
                    prompt_path=prompt_path,
                    native_responses_dir=native_responses_dir,
                    native_response_overrides=native_response_overrides,
                )
            )
        elif kind == "runner":
            results.append(
                run_runner_role(
                    provider_cfg=provider_cfg,
                    role_cfg=role_cfg,
                    role=role,
                    phase=args.phase,
                    prompt_path=prompt_path,
                    output_file=output_file,
                    stdout_path=stdout_path,
                    stderr_path=stderr_path,
                    working_dir=working_dir,
                    skill_root=skill_root,
                    dry_run=args.dry_run,
                )
            )
        elif kind == "cli":
            results.append(
                run_direct_cli_role(
                    provider_cfg=provider_cfg,
                    role=role,
                    phase=args.phase,
                    prompt=prompt,
                    prompt_path=prompt_path,
                    stdout_path=stdout_path,
                    stderr_path=stderr_path,
                    dry_run=args.dry_run,
                )
            )
        else:
            results.append(
                {
                    "role": role,
                    "status": "unknown_kind",
                    "kind": kind,
                    "provider": provider_key,
                    "required": True,
                }
            )

    run = {
        "phase": args.phase,
        "goal": args.goal,
        "created_at": now_iso(),
        "roles": roles,
        "configured_roles": configured_roles,
        "required_roles": required_roles,
        "role_override": bool(args.roles),
        "mandatory_presence": mandatory,
        "dry_run": bool(args.dry_run),
        "complete": phase_is_complete(results, required_roles),
        "results": results,
    }
    summary = update_summary(
        summary_path=base_out / "panel_summary.json",
        skill=skill,
        cfg=cfg,
        base_out=base_out,
        run=run,
    )
    print(json.dumps(summary, indent=2, ensure_ascii=False))
    if args.fail_on_incomplete and not run["complete"]:
        return 2
    return 0


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

#!/usr/bin/env python3
"""Record a native Codex role response for a collaborative skill phase."""
from __future__ import annotations

import argparse
import datetime as _dt
import json
from pathlib import Path
import sys
from typing import Any

try:
    import tomllib
except Exception:  # pragma: no cover
    tomllib = None


OK_STATUSES = {"ok", "native_response_recorded"}


def now_iso() -> str:
    return _dt.datetime.now(_dt.timezone.utc).isoformat()


def get_skill_root() -> Path:
    return Path(__file__).resolve().parents[1]


def load_toml(path: Path) -> dict[str, Any]:
    if tomllib is None:
        raise RuntimeError("Python 3.11 or newer is required for TOML parsing")
    with path.open("rb") as handle:
        return tomllib.load(handle)


def read_content(args: argparse.Namespace) -> tuple[str, str]:
    sources = [bool(args.text), bool(args.from_file)]
    if sum(sources) > 1:
        raise ValueError("Use only one input source: --text, --from-file, or stdin.")
    if args.text:
        return args.text, "text"
    if args.from_file:
        path = Path(args.from_file).expanduser()
        if not path.exists():
            raise FileNotFoundError(str(path))
        return path.read_text(encoding="utf-8"), str(path)
    if not sys.stdin.isatty():
        return sys.stdin.read(), "stdin"
    raise ValueError("Provide response content with --text, --from-file, or stdin.")


def required_roles_for_phase(cfg: dict[str, Any], phase: str) -> list[str]:
    skill = cfg.get("skill", {})
    phase_cfg = cfg.get("phases", {}).get(phase, {})
    roles = list(phase_cfg.get("roles", [])) + list(skill.get("mandatory_presence", []))
    return list(dict.fromkeys(roles))


def role_is_native(cfg: dict[str, Any], role: str) -> bool:
    role_cfg = cfg.get("roles", {}).get(role, {})
    provider_key = role_cfg.get("provider") or role
    provider = cfg.get("providers", {}).get(provider_key, {})
    return provider.get("kind") == "native_codex"


def phase_run_is_complete(run: dict[str, Any]) -> bool:
    required_roles = list(run.get("required_roles") or [])
    results = [item for item in run.get("results", []) if isinstance(item, dict)]
    by_role = {str(item.get("role")): item for item in results}
    for role in required_roles:
        if by_role.get(role, {}).get("status") not in OK_STATUSES:
            return False
    for item in results:
        if item.get("required", True) and item.get("status") not in OK_STATUSES:
            return False
    return True


def write_text(path: Path, content: str, replace: bool, dry_run: bool) -> None:
    if path.exists() and path.read_text(encoding="utf-8").strip() and not replace:
        raise FileExistsError(f"{path} already exists. Use --replace to overwrite it.")
    if dry_run:
        return
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(content.rstrip() + "\n", encoding="utf-8")


def update_summary(summary_path: Path, phase: str, role: str, response_path: Path, dry_run: bool) -> dict[str, Any]:
    if not summary_path.exists():
        return {"updated": False, "reason": "panel_summary.json not found"}
    try:
        summary = json.loads(summary_path.read_text(encoding="utf-8"))
    except json.JSONDecodeError as exc:
        return {"updated": False, "reason": f"panel_summary.json is invalid JSON: {exc}"}

    phase_runs = summary.get("phase_runs")
    if not isinstance(phase_runs, list):
        return {"updated": False, "reason": "panel_summary.json has no phase_runs list"}

    updated_run = None
    updated_result = None
    for run in reversed(phase_runs):
        if not isinstance(run, dict) or run.get("phase") != phase:
            continue
        for result in run.get("results", []):
            if isinstance(result, dict) and result.get("role") == role:
                result.update(
                    {
                        "status": "ok",
                        "participation": "native_response_recorded",
                        "response_path": str(response_path),
                    }
                )
                result.pop("instruction", None)
                updated_run = run
                updated_result = result
                break
        if updated_result:
            break

    if not updated_result:
        return {"updated": False, "reason": f"no panel result found for {phase}/{role}"}

    updated_run["complete"] = phase_run_is_complete(updated_run)
    complete_phases = sorted(
        {
            str(run.get("phase"))
            for run in phase_runs
            if isinstance(run, dict) and run.get("complete") and run.get("phase")
        }
    )
    required_phases = list(summary.get("required_phases") or [])
    missing_required_phases = [phase_name for phase_name in required_phases if phase_name not in complete_phases]
    summary.update(
        {
            "updated_at": now_iso(),
            "complete_phases": complete_phases,
            "missing_required_phases": missing_required_phases,
            "complete": not missing_required_phases if required_phases else summary.get("complete", False),
            "latest_run": phase_runs[-1] if phase_runs else summary.get("latest_run"),
        }
    )
    if not dry_run:
        summary_path.write_text(json.dumps(summary, indent=2, ensure_ascii=False) + "\n", encoding="utf-8")
    return {"updated": True, "phase_complete": updated_run["complete"]}


def main() -> int:
    parser = argparse.ArgumentParser(description="Record a native Codex role response.")
    parser.add_argument("--phase", required=True)
    parser.add_argument("--role", required=True)
    parser.add_argument("--artifact-dir")
    parser.add_argument("--from-file")
    parser.add_argument("--text")
    parser.add_argument("--replace", action="store_true")
    parser.add_argument("--dry-run", action="store_true")
    parser.add_argument("--allow-unconfigured", action="store_true")
    args = parser.parse_args()

    skill_root = get_skill_root()
    cfg = load_toml(skill_root / "assets" / "routing.toml")
    skill = cfg.get("skill", {})
    artifact_dir = Path(args.artifact_dir or skill.get("artifact_dir") or ".codex_workflow/panel")
    try:
        content, source = read_content(args)
    except Exception as exc:
        raise SystemExit(str(exc))
    if not content.strip():
        raise SystemExit("Native response content is empty.")

    required_roles = required_roles_for_phase(cfg, args.phase)
    if args.role not in required_roles and not args.allow_unconfigured:
        raise SystemExit(f"Role {args.role!r} is not configured for phase {args.phase!r}.")
    if not role_is_native(cfg, args.role) and not args.allow_unconfigured:
        raise SystemExit(f"Role {args.role!r} is not routed to a native_codex provider.")

    response_path = artifact_dir / "native_responses" / f"{args.phase}_{args.role}.md"
    try:
        write_text(response_path, content, replace=args.replace, dry_run=args.dry_run)
    except Exception as exc:
        raise SystemExit(str(exc))
    summary_update = update_summary(artifact_dir / "panel_summary.json", args.phase, args.role, response_path, args.dry_run)
    result = {
        "skill": skill.get("name"),
        "phase": args.phase,
        "role": args.role,
        "response_path": str(response_path),
        "source": source,
        "dry_run": args.dry_run,
        "summary_update": summary_update,
    }
    print(json.dumps(result, indent=2, ensure_ascii=False))
    return 0


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

#!/usr/bin/env python3
"""Validate required artifacts and real panel participation for a skill run."""
from __future__ import annotations

import argparse
import json
from pathlib import Path
from typing import Any

try:
    import tomllib
except Exception:  # pragma: no cover
    tomllib = None


OK_STATUSES = {"ok", "native_response_recorded"}


def load_toml(path: Path) -> dict[str, Any]:
    if tomllib is None:
        raise RuntimeError("Python 3.11 or newer is required")
    with path.open("rb") as fh:
        return tomllib.load(fh)


def load_json(path: Path) -> dict[str, Any]:
    try:
        return json.loads(path.read_text(encoding="utf-8"))
    except Exception as exc:
        return {"_error": str(exc)}


def phase_run_is_complete(run: dict[str, Any], required_roles: list[str]) -> bool:
    results = run.get("results") or []
    by_role = {str(item.get("role")): item for item in results if isinstance(item, dict)}
    for role in required_roles:
        if by_role.get(role, {}).get("status") not in OK_STATUSES:
            return False
    for item in results:
        if isinstance(item, dict) and item.get("required", True) and item.get("status") not in OK_STATUSES:
            return False
    return True


def validate_panel_summary(
    summary_path: Path,
    required_phases: list[str],
    phase_required_roles: dict[str, list[str]],
) -> dict[str, Any]:
    if not summary_path.exists():
        return {
            "ok": False,
            "missing_panel_summary": True,
            "missing_required_phases": required_phases,
            "incomplete_phase_runs": [],
        }

    summary = load_json(summary_path)
    if "_error" in summary:
        return {
            "ok": False,
            "panel_summary_error": summary["_error"],
            "missing_required_phases": required_phases,
            "incomplete_phase_runs": [],
        }

    phase_runs = summary.get("phase_runs")
    if not isinstance(phase_runs, list):
        latest = summary.get("latest_run") or summary
        phase_runs = [latest] if isinstance(latest, dict) else []

    complete_phases: set[str] = set()
    incomplete_phase_runs: list[dict[str, Any]] = []
    for run in phase_runs:
        if not isinstance(run, dict):
            continue
        phase = str(run.get("phase") or "")
        expected_roles = phase_required_roles.get(phase) or list(run.get("required_roles") or [])
        complete = bool(run.get("complete")) and phase_run_is_complete(run, expected_roles)
        if complete:
            complete_phases.add(phase)
        elif phase:
            incomplete_phase_runs.append(
                {
                    "phase": phase,
                    "created_at": run.get("created_at"),
                    "statuses": {
                        str(item.get("role")): item.get("status")
                        for item in (run.get("results") or [])
                        if isinstance(item, dict)
                    },
                }
            )

    missing_required_phases = [phase for phase in required_phases if phase not in complete_phases]
    return {
        "ok": not missing_required_phases,
        "missing_panel_summary": False,
        "required_phases": required_phases,
        "complete_phases": sorted(complete_phases),
        "missing_required_phases": missing_required_phases,
        "incomplete_phase_runs": incomplete_phase_runs,
    }


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--artifact-dir")
    parser.add_argument(
        "--allow-missing-phases",
        action="store_true",
        help="Only validate required files. Use for partial in-progress runs.",
    )
    args = parser.parse_args()

    root = Path(__file__).resolve().parents[1]
    cfg = load_toml(root / "assets" / "routing.toml")
    skill = cfg.get("skill", {})
    artifact_dir = Path(args.artifact_dir or skill.get("artifact_dir") or ".")
    required_outputs = list(skill.get("required_outputs", []))
    missing_outputs = [name for name in required_outputs if not (artifact_dir / name).exists()]
    required_phases = list(skill.get("required_phases") or cfg.get("phases", {}).keys())
    mandatory = list(skill.get("mandatory_presence", []))
    phase_required_roles = {
        phase: list(dict.fromkeys(list(phase_cfg.get("roles", [])) + mandatory))
        for phase, phase_cfg in cfg.get("phases", {}).items()
    }
    panel_result = validate_panel_summary(
        artifact_dir / "panel_summary.json",
        required_phases,
        phase_required_roles,
    )
    if args.allow_missing_phases:
        panel_ok = not panel_result.get("missing_panel_summary")
    else:
        panel_ok = bool(panel_result.get("ok"))

    result = {
        "skill": skill.get("name"),
        "artifact_dir": str(artifact_dir),
        "required_outputs": required_outputs,
        "missing_outputs": missing_outputs,
        "panel": panel_result,
        "ok": not missing_outputs and panel_ok,
    }
    print(json.dumps(result, indent=2, ensure_ascii=False))
    return 0 if result["ok"] else 1


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