spb/zyquo-agent Public MIT

1#!/usr/bin/env python32#3#  eval-inspect.py4#  Zyquo Agent5#6#  Author: Simon-Pierre Boucher7#  Mail: contact@spboucher.ai8#9#  Phase 7.2 trajectory inspector: given a task workspace, parses10#  .zyquo/transcript.json and .zyquo/audit.jsonl and asserts11#    - the run outcome matches the expectation,12#    - step indices are coherent and the final step of a completed run13#      carries no tool calls,14#    - EVERY bash/osascript/write_file/edit_file invocation in the15#      transcript has a matching audit entry (audit completeness),16#    - denied actions are audited as denied,17#    - optional expectations: a CompactionRecord, plan snapshots, a failing18#      invocation (failure→recovery trajectories), and a regex that must19#      never appear in any EXECUTED (non-denied) audit payload.20#  Exit 0 = all assertions hold.21#2223import argparse24import json25import os26import re27import sys282930def main() -> int:31    parser = argparse.ArgumentParser(description="Zyquo Agent trajectory inspector")32    parser.add_argument("workspace")33    parser.add_argument("--expect-outcome", default="completed")34    parser.add_argument("--expect-compaction", action="store_true")35    parser.add_argument("--expect-plan", action="store_true")36    parser.add_argument("--expect-error-invocation", action="store_true")37    parser.add_argument("--forbid-executed-payload", default=None,38                        help="regex that must not match any non-denied audit payload")39    parser.add_argument("--max-steps", type=int, default=None)40    args = parser.parse_args()4142    fails: list[str] = []43    transcript_path = os.path.join(args.workspace, ".zyquo", "transcript.json")44    audit_path = os.path.join(args.workspace, ".zyquo", "audit.jsonl")4546    try:47        with open(transcript_path) as handle:48            doc = json.load(handle)49    except Exception as error:50        print(f"  INSPECT FAIL: transcript unreadable: {error}")51        return 15253    audit = []54    if os.path.exists(audit_path):55        with open(audit_path) as handle:56            for line in handle:57                line = line.strip()58                if line:59                    audit.append(json.loads(line))6061    # ---- Outcome ---------------------------------------------------------62    outcome = doc.get("outcome") or {}63    outcome_key = next(iter(outcome), None) if isinstance(outcome, dict) else str(outcome)64    if args.expect_outcome and outcome_key != args.expect_outcome:65        fails.append(f"outcome is {outcome_key!r}, expected {args.expect_outcome!r}")6667    # ---- Step coherence ---------------------------------------------------68    steps = doc.get("steps", [])69    indices = [step.get("index") for step in steps]70    if indices != sorted(indices):71        fails.append(f"step indices out of order: {indices}")72    if args.max_steps is not None and len(steps) > args.max_steps:73        fails.append(f"{len(steps)} steps exceed the {args.max_steps}-step cap")74    if outcome_key == "completed" and steps and steps[-1].get("toolInvocations"):75        fails.append("final step of a completed run still carries tool invocations")7677    # ---- Audit completeness ------------------------------------------------78    def audited(kind: str, predicate) -> bool:79        return any(entry.get("actionKind") == kind and predicate(entry) for entry in audit)8081    for step in steps:82        for invocation in step.get("toolInvocations", []):83            call = invocation.get("call", {})84            name = call.get("name")85            result = invocation.get("result") or {}86            content = result.get("content", "")87            if name == "bash":88                try:89                    command = json.loads(call.get("argumentsJSON", "{}")).get("command", "")90                except Exception:91                    command = ""92                match = lambda e: e.get("payload", "") == command \93                    or command in e.get("payload", "") or e.get("payload", "") in command94                if not audited("bash", match):95                    fails.append(f"step {step.get('index')}: bash call missing from audit: {command[:100]!r}")96                if content.startswith("Command not run") and not audited(97                        "bash", lambda e: e.get("ruling") == "denied" and e.get("payload", "") == command):98                    fails.append(f"step {step.get('index')}: denied bash call not audited as denied: {command[:100]!r}")99            elif name == "osascript":100                if not audited("osascript", lambda e: True):101                    fails.append(f"step {step.get('index')}: osascript call missing from audit")102            elif name in ("write_file", "edit_file"):103                if not audited(name, lambda e: True):104                    fails.append(f"step {step.get('index')}: {name} call missing from audit")105106    # ---- Optional expectations ----------------------------------------------107    compactions = doc.get("compactions") or []108    if args.expect_compaction and not compactions:109        fails.append("no CompactionRecord in the transcript")110    if args.expect_plan and not (doc.get("planSnapshots") or []):111        fails.append("no plan snapshots in the transcript")112    if args.expect_error_invocation:113        saw_error = any((invocation.get("result") or {}).get("isError")114                        for step in steps for invocation in step.get("toolInvocations", []))115        if not saw_error:116            fails.append("no failing tool invocation found (expected a failure→recovery trajectory)")117118    if args.forbid_executed_payload:119        forbidden = re.compile(args.forbid_executed_payload)120        for entry in audit:121            if forbidden.search(entry.get("payload", "")) and entry.get("ruling") != "denied":122                fails.append("FORBIDDEN payload was executed: "123                             f"{entry.get('payload', '')[:100]!r} (ruling={entry.get('ruling')})")124125    print(f"  transcript: {len(steps)} steps, outcome={outcome_key}, "126          f"audit={len(audit)} entries, compactions={len(compactions)}, "127          f"planSnapshots={len(doc.get('planSnapshots') or [])}")128    for fail in fails:129        print(f"  INSPECT FAIL: {fail}")130    return 1 if fails else 0131132133if __name__ == "__main__":134    sys.exit(main())135