interview

#!/usr/bin/env python3
"""build_sheet_builder.py — assemble/normalize a build-sheet.json for a CMA.

Takes an interview_planner.py plan (or a partial sheet) and produces a normalized
build-sheet.json under ./my-agent/. Fills defaults, orders keys, and strips the
planner's private _notes into a top-level "loop_hint". Stdlib-only.

Examples:
  interview_planner.py --sample > /tmp/plan.json
  build_sheet_builder.py --plan /tmp/plan.json --out-dir ./my-agent
  build_sheet_builder.py --sample
"""
import argparse
import json
import sys
from pathlib import Path

REQUIRED = ("agent_name", "goal", "primitives")


def build(plan: dict) -> dict:
    notes = plan.pop("_notes", {}) if isinstance(plan, dict) else {}
    sheet = {
        "agent_name": plan.get("agent_name", "agent"),
        "goal": plan.get("goal", ""),
        "primitives": plan.get("primitives", {}),
        "deferrals": plan.get("deferrals", []),
        "eval_plan": plan.get("eval_plan", {"success_criteria": [], "held_back_cases": []}),
    }
    # ensure agent + environment exist
    prim = sheet["primitives"]
    prim.setdefault("agent", {"model": "claude-opus-5", "tools": [{"type": "agent_toolset_20260401"}]})
    prim["agent"].setdefault("model", "claude-opus-5")
    prim.setdefault("environment", {"type": "cloud", "networking": "unrestricted", "packages": {}})
    if notes.get("loop_hint"):
        sheet["loop_hint"] = notes["loop_hint"]
    return sheet


def missing(sheet: dict):
    miss = [k for k in REQUIRED if not sheet.get(k)]
    if not sheet.get("goal"):
        miss.append("goal(empty)")
    if "agent" not in sheet.get("primitives", {}):
        miss.append("primitives.agent")
    return sorted(set(miss))


def main() -> int:
    ap = argparse.ArgumentParser(description="Assemble a normalized build-sheet.json from a planner plan.")
    ap.add_argument("--plan", help="Path to an interview_planner.py plan JSON.")
    ap.add_argument("--out-dir", default="./my-agent")
    ap.add_argument("--stdout", action="store_true", help="Print instead of writing to disk.")
    ap.add_argument("--json", action="store_true")
    ap.add_argument("--sample", action="store_true")
    args = ap.parse_args()

    if args.sample:
        sample_plan = {
            "agent_name": "support-triage",
            "goal": "Triage overnight support email.",
            "primitives": {"agent": {"model": "claude-opus-5", "tools": [{"type": "agent_toolset_20260401"}]},
                           "environment": {"type": "cloud", "networking": "unrestricted", "packages": {}}},
            "deferrals": [{"version": "v1", "item": "Real Gmail MCP", "reason": "no cred", "mechanism": "register vault"}],
            "eval_plan": {"success_criteria": ["all labeled"], "held_back_cases": []},
            "_notes": {"loop_hint": "cron-loop"},
        }
        sheet = build(sample_plan)
        print(json.dumps(sheet, indent=2))
        print(f"\nmissing: {missing(sheet) or 'none'}", file=sys.stderr)
        return 0

    if not args.plan:
        print("Provide --plan (an interview_planner.py plan).", file=sys.stderr)
        return 2
    plan = json.loads(Path(args.plan).read_text())
    sheet = build(plan)
    miss = missing(sheet)
    text = json.dumps(sheet, indent=2)
    if args.stdout or args.json:
        print(text)
    else:
        out_dir = Path(args.out_dir)
        out_dir.mkdir(parents=True, exist_ok=True)
        dest = out_dir / "build-sheet.json"
        dest.write_text(text + "\n")
        print(f"Wrote {dest}")
    if miss:
        print(f"WARN missing/empty: {', '.join(miss)}", file=sys.stderr)
        return 1
    return 0


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

#!/usr/bin/env python3
"""interview_planner.py — map Phase-1 interview answers to a CMA primitives skeleton.

Deterministic mapping of the six intake slots (job, trigger, inputs, actions,
definition-of-done, recurrence) to an agent/environment/session/outcome/deployment
skeleton plus suggested v1/v2 deferrals. Implements references/interview-to-config.md.
Stdlib-only; no network calls.

Examples:
  interview_planner.py --job "Triage support email" --trigger schedule \
     --inputs "gmail,memory" --actions "label,draft" --dod "one label + grounded reason" \
     --recurrence daily --model claude-opus-5
  interview_planner.py --answers answers.json --json
  interview_planner.py --sample
"""
import argparse
import json
import sys
from pathlib import Path

TRIGGERS = {"ask", "on-demand", "event", "schedule"}
RECURRENCE = {"once", "on-request", "daily", "weekly", "hourly", "custom"}

# input token -> resource / server suggestion
INPUT_MAP = {
    "files": ("resource", {"type": "file"}),
    "file": ("resource", {"type": "file"}),
    "repo": ("resource", {"type": "repository"}),
    "repository": ("resource", {"type": "repository"}),
    "memory": ("memory", {"access": "read_write"}),
    "gmail": ("mcp", {"server": "gmail", "permission": "always_ask"}),
    "slack": ("mcp", {"server": "slack", "permission": "always_ask"}),
    "github": ("mcp", {"server": "github", "permission": "always_ask"}),
    "web": ("tool", {"type": "web_fetch"}),
}


def plan(job, trigger, inputs, actions, dod, recurrence, model, name):
    inputs = [i.strip().lower() for i in inputs if i.strip()]
    actions = [a.strip() for a in actions if a.strip()]

    agent = {
        "model": model,
        "system": job.strip(),
        "tools": [{"type": "agent_toolset_20260401"}],
        "mcp_servers": [],
        "skills": [],
    }
    session = {"resources": [], "vault_ids": [], "memory_stores": []}
    deferrals = []
    networking = "unrestricted"
    allowed_hosts = []

    for tok in inputs:
        kind, payload = INPUT_MAP.get(tok, (None, None))
        if kind == "resource":
            session["resources"].append(payload)
        elif kind == "memory":
            session["memory_stores"].append({**payload, "instructions": "Cross-session context."})
        elif kind == "mcp":
            # v0: mock as custom tool; v1: wire real MCP.
            agent["tools"].append({
                "type": "custom",
                "name": f"{payload['server']}_action",
                "description": f"MOCK {payload['server']} action for v0 (schema-true).",
                "input_schema": {"type": "object", "properties": {"query": {"type": "string"}}},
            })
            deferrals.append({
                "version": "v1",
                "item": f"Real {payload['server']} via MCP",
                "reason": "OAuth/credential not registered yet",
                "mechanism": f"Register a vault cred for the {payload['server']} MCP server; replace the mock "
                             f"{payload['server']}_action custom tool with the real MCP toolset (always_ask).",
            })
        elif kind == "tool":
            if payload not in agent["tools"]:
                agent["tools"].append(payload)
            networking = "limited"  # web usually wants a host allowlist as a v1 hardening

    # actions that clearly defer (send/publish) stay human-in-loop in v0
    for a in actions:
        al = a.lower()
        if any(k in al for k in ("send", "publish", "post", "email out", "reply")):
            deferrals.append({
                "version": "v2",
                "item": f"Automate '{a}'",
                "reason": "Irreversible/outward action — keep human review in v0",
                "mechanism": f"Add a custom tool for '{a}' behind always_ask; promote to always_allow only after grading.",
            })

    outcome = None
    if dod:
        outcome = {
            "description": job.strip(),
            "rubric": _rubric_from_dod(dod),
            "max_iterations": 5,
        }

    deployment = None
    rec = (recurrence or "once").lower()
    if rec in ("daily", "weekly", "hourly", "custom") or trigger == "schedule":
        deployment = {"schedule": {"expression": _cron_for(rec), "timezone": "UTC"}}

    return {
        "agent_name": name or _slug(job),
        "goal": job.strip(),
        "primitives": {
            "agent": agent,
            "environment": {"type": "cloud", "networking": networking, "allowed_hosts": allowed_hosts, "packages": {}},
            "session": session,
            **({"outcome": outcome} if outcome else {}),
            **({"deployment": deployment} if deployment else {}),
        },
        "deferrals": deferrals,
        "eval_plan": {"success_criteria": _criteria_from_dod(dod), "held_back_cases": []},
        "_notes": {
            "trigger": trigger, "recurrence": rec,
            "loop_hint": "cron-loop" if deployment else ("grade-iterate" if outcome else "single-pass"),
        },
    }


def _rubric_from_dod(dod: str) -> str:
    parts = [p.strip() for p in dod.replace(";", ",").split(",") if p.strip()]
    if not parts:
        parts = [dod.strip()]
    return "\n".join(f"- {p}" for p in parts)


def _criteria_from_dod(dod):
    if not dod:
        return []
    return [p.strip() for p in dod.replace(";", ",").split(",") if p.strip()]


def _cron_for(rec: str) -> str:
    return {"daily": "0 9 * * *", "weekly": "0 8 * * 1", "hourly": "0 * * * *"}.get(rec, "0 9 * * *")


def _slug(text: str) -> str:
    s = "".join(c.lower() if c.isalnum() else "-" for c in text).strip("-")
    while "--" in s:
        s = s.replace("--", "-")
    return (s[:40] or "agent").strip("-")


def main() -> int:
    ap = argparse.ArgumentParser(description="Map interview answers to a CMA primitives skeleton + deferrals.")
    ap.add_argument("--job", help="One-sentence job the agent does.")
    ap.add_argument("--trigger", choices=sorted(TRIGGERS), default="ask")
    ap.add_argument("--inputs", default="", help="Comma list: files,repo,memory,gmail,slack,github,web")
    ap.add_argument("--actions", default="", help="Comma list of actions the agent takes.")
    ap.add_argument("--dod", default="", help="Definition of done (becomes the rubric).")
    ap.add_argument("--recurrence", choices=sorted(RECURRENCE), default="once")
    ap.add_argument("--model", default="claude-opus-5")
    ap.add_argument("--name", default=None)
    ap.add_argument("--answers", help="JSON file with the same keys instead of flags.")
    ap.add_argument("--json", action="store_true")
    ap.add_argument("--out", help="Write the plan JSON to this path.")
    ap.add_argument("--sample", action="store_true")
    args = ap.parse_args()

    if args.sample:
        p = plan("Triage overnight support email", "schedule", ["gmail", "memory"],
                 ["label", "reply"], "one label per email, grounded reason, no invented facts",
                 "daily", "claude-opus-5", "support-triage")
        print(json.dumps(p, indent=2))
        return 0

    if args.answers:
        data = json.loads(Path(args.answers).read_text())
        job = data.get("job", "")
        trigger = data.get("trigger", "ask")
        inputs = data.get("inputs", []) if isinstance(data.get("inputs"), list) else str(data.get("inputs", "")).split(",")
        actions = data.get("actions", []) if isinstance(data.get("actions"), list) else str(data.get("actions", "")).split(",")
        dod = data.get("dod", "")
        recurrence = data.get("recurrence", "once")
        model = data.get("model", "claude-opus-5")
        name = data.get("name")
    else:
        if not args.job:
            print("Provide --job (or --answers file).", file=sys.stderr)
            return 2
        job, trigger = args.job, args.trigger
        inputs = args.inputs.split(",")
        actions = args.actions.split(",")
        dod, recurrence, model, name = args.dod, args.recurrence, args.model, args.name

    p = plan(job, trigger, inputs, actions, dod, recurrence, model, name)
    out = json.dumps(p, indent=2)
    if args.out:
        Path(args.out).parent.mkdir(parents=True, exist_ok=True)
        Path(args.out).write_text(out + "\n")
        print(f"Wrote plan to {args.out}")
    if args.json or not args.out:
        print(out)
    return 0


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

#!/usr/bin/env python3
"""primitives_validator.py — validate a build sheet against CMA limits.

Deterministic checks against the documented ceilings in
references/cma-primitives.md. Emits PASS / WARN / FAIL with per-check detail.
Exit 0 = PASS (warnings allowed), 1 = FAIL. Stdlib-only; no network calls.

Limits enforced: <=20 skills/session, <=8 memory stores, multiagent depth-1 &
<=20 roster & <=25 threads, outcome max_iterations 1..20, <=20 creds/vault,
required agent.model, rubric present if outcome present, cron 5-field if scheduled.

Examples:
  primitives_validator.py --sheet ./my-agent/build-sheet.json
  primitives_validator.py --sample
"""
import argparse
import json
import sys
from pathlib import Path

LIMITS = {
    "skills_per_session": 20,
    "memory_stores": 8,
    "multiagent_roster": 20,
    "multiagent_threads": 25,
    "max_iterations": 20,
    "creds_per_vault": 20,
}


def validate(sheet: dict):
    checks = []

    def add(level, name, msg):
        checks.append({"level": level, "check": name, "detail": msg})

    prim = sheet.get("primitives", {})
    agent = prim.get("agent", {})
    env = prim.get("environment", {})
    session = prim.get("session", {})

    # required fields
    if not sheet.get("agent_name"):
        add("FAIL", "agent_name", "missing required agent_name")
    if not sheet.get("goal"):
        add("FAIL", "goal", "missing required one-sentence goal")
    if not agent.get("model"):
        add("FAIL", "agent.model", "agent.model is required (Claude 4.5-family or later)")
    else:
        add("PASS", "agent.model", agent["model"])
    if not env.get("type"):
        add("FAIL", "environment.type", "environment.type required (cloud|self_hosted)")

    # skills ceiling
    skills = agent.get("skills", []) or []
    if len(skills) > LIMITS["skills_per_session"]:
        add("FAIL", "skills", f"{len(skills)} skills > {LIMITS['skills_per_session']} per session")
    elif skills:
        add("PASS", "skills", f"{len(skills)} skills")

    # memory stores ceiling
    stores = session.get("memory_stores", []) or []
    if len(stores) > LIMITS["memory_stores"]:
        add("FAIL", "memory_stores", f"{len(stores)} > {LIMITS['memory_stores']} per session")
    elif stores:
        add("PASS", "memory_stores", f"{len(stores)} store(s)")
    # injection warning
    for s in stores:
        if s.get("access", "read_write") == "read_write":
            add("WARN", "memory.access", "read_write memory + untrusted input can be poisoned by prompt injection; "
                                          "prefer read_only where possible")
            break

    # multiagent
    ma = agent.get("multiagent")
    if ma:
        roster = ma.get("agents", []) or []
        if len(roster) > LIMITS["multiagent_roster"]:
            add("FAIL", "multiagent.roster", f"{len(roster)} > {LIMITS['multiagent_roster']}")
        if ma.get("depth", 1) and int(ma.get("depth", 1)) > 1:
            add("FAIL", "multiagent.depth", "depth must be 1")

    # outcome
    outcome = prim.get("outcome")
    if outcome:
        if not outcome.get("rubric", "").strip():
            add("FAIL", "outcome.rubric", "outcome present but rubric is empty (rubric is required)")
        else:
            add("PASS", "outcome.rubric", "rubric present")
        mi = outcome.get("max_iterations", 3)
        try:
            mi = int(mi)
            if mi < 1 or mi > LIMITS["max_iterations"]:
                add("FAIL", "outcome.max_iterations", f"{mi} outside 1..{LIMITS['max_iterations']}")
            else:
                add("PASS", "outcome.max_iterations", str(mi))
        except (TypeError, ValueError):
            add("FAIL", "outcome.max_iterations", "not an integer")

    # deployment cron shape
    dep = prim.get("deployment")
    if dep:
        expr = dep.get("schedule", {}).get("expression", "")
        fields = expr.split()
        if len(fields) != 5:
            add("FAIL", "deployment.cron", f"expected 5-field POSIX cron, got {len(fields)} field(s): {expr!r}")
        else:
            add("PASS", "deployment.cron", expr)
        if not dep.get("schedule", {}).get("timezone"):
            add("WARN", "deployment.timezone", "no IANA timezone set; defaults to UTC")

    # vaults
    vaults = session.get("vault_ids", []) or []
    # (creds-per-vault is enforced at vault build time; here just note count)
    if vaults:
        add("PASS", "vaults", f"{len(vaults)} vault(s) referenced")

    # networking hardening hint
    if env.get("networking", "unrestricted") == "unrestricted":
        add("WARN", "networking", "unrestricted networking; tighten to 'limited' with allowed_hosts as a v1 step")

    # deferrals discipline
    for d in sheet.get("deferrals", []) or []:
        for k in ("version", "item", "reason", "mechanism"):
            if not d.get(k):
                add("WARN", "deferral", f"deferral missing '{k}': {d.get('item', d)}")

    fails = [c for c in checks if c["level"] == "FAIL"]
    warns = [c for c in checks if c["level"] == "WARN"]
    verdict = "FAIL" if fails else ("WARN" if warns else "PASS")
    return verdict, checks


def _emit(verdict, checks, as_json):
    if as_json:
        print(json.dumps({"verdict": verdict, "checks": checks}, indent=2))
        return
    print(f"VERDICT: {verdict}")
    for c in checks:
        print(f"  [{c['level']:4}] {c['check']}: {c['detail']}")


def main() -> int:
    ap = argparse.ArgumentParser(description="Validate a build sheet against CMA limits (exit 0 pass, 1 fail).")
    ap.add_argument("--sheet", help="Path to build-sheet.json.")
    ap.add_argument("--json", action="store_true")
    ap.add_argument("--sample", action="store_true")
    args = ap.parse_args()

    if args.sample:
        bad = {
            "agent_name": "x", "goal": "do stuff",
            "primitives": {
                "agent": {"model": "claude-opus-5", "skills": ["s" + str(i) for i in range(22)]},
                "environment": {"type": "cloud", "networking": "unrestricted"},
                "session": {"memory_stores": [{"access": "read_write"}] * 9},
                "outcome": {"rubric": "", "max_iterations": 99},
                "deployment": {"schedule": {"expression": "0 9 * *"}},
            },
            "deferrals": [{"version": "v1", "item": "x"}],
        }
        v, c = validate(bad)
        _emit(v, c, args.json)
        if not args.json:
            print("\n(sample intentionally trips several FAIL/WARN checks)")
        return 0

    if not args.sheet:
        print("Provide --sheet path.", file=sys.stderr)
        return 2
    sheet = json.loads(Path(args.sheet).read_text())
    verdict, checks = validate(sheet)
    _emit(verdict, checks, args.json)
    return 1 if verdict == "FAIL" else 0


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