#!/usr/bin/env python3 """OKSTRA error log helper — CLI adapter. The writer itself lives in `okstra_ctl.error_log_write`, which the deterministic dispatcher also calls in-process. This file is the argparse surface over it and re-exports the writer's names so the contract tests keep reading one module. """ from __future__ import annotations import argparse from pathlib import Path from okstra_ctl.error_log_write import ( ALLOWED_AGENT_ROLES, ALLOWED_AGENTS, ALLOWED_CAUSES, ALLOWED_ERROR_TYPES, CAUSE_EVIDENCE_FIELDS, CAUSE_PROBE_MAX_BYTES, PIPE_BUF_BYTES, STDERR_EXCERPT_MAX_BYTES, SUPPORTED_SIDECAR_SCHEMA_VERSIONS, TRUNCATION_SUFFIX, append_jsonl_line, append_observed, dump_from_worker_sidecar, normalize_cause_context, truncate_stderr, ) __all__ = [ "ALLOWED_AGENT_ROLES", "ALLOWED_AGENTS", "ALLOWED_CAUSES", "ALLOWED_ERROR_TYPES", "CAUSE_EVIDENCE_FIELDS", "CAUSE_PROBE_MAX_BYTES", "PIPE_BUF_BYTES", "STDERR_EXCERPT_MAX_BYTES", "SUPPORTED_SIDECAR_SCHEMA_VERSIONS", "TRUNCATION_SUFFIX", "append_jsonl_line", "append_observed", "dump_from_worker_sidecar", "main", "normalize_cause_context", "truncate_stderr", ] def _build_parser(): p = argparse.ArgumentParser(description="OKSTRA error log helper") sub = p.add_subparsers(dest="cmd", required=True) obs = sub.add_parser("append-observed", help="Append a lead-observed error event") obs.add_argument("--out", required=True) obs.add_argument("--task-key", required=True) obs.add_argument("--phase", required=True) obs.add_argument("--agent", required=True, choices=sorted(ALLOWED_AGENTS)) obs.add_argument("--agent-role", required=True, choices=sorted(ALLOWED_AGENT_ROLES)) obs.add_argument("--model", required=True) obs.add_argument("--error-type", required=True, choices=sorted(ALLOWED_ERROR_TYPES)) command = obs.add_mutually_exclusive_group(required=True) command.add_argument("--command") command.add_argument("--command-file", type=Path) obs.add_argument("--command-kind", required=True) obs.add_argument("--exit-code", type=int, default=None) obs.add_argument("--duration-ms", type=int, default=None) message = obs.add_mutually_exclusive_group(required=True) message.add_argument("--message") message.add_argument("--message-file", type=Path) grp = obs.add_mutually_exclusive_group() grp.add_argument("--stderr-excerpt", default=None) grp.add_argument("--stderr-excerpt-file", default=None) obs.add_argument("--cause", choices=sorted(ALLOWED_CAUSES), default=None) obs.add_argument( "--evidence", action="append", default=[], metavar="", help="cause evidence; repeat for targetProbe and controlProbe", ) obs.add_argument("--evidence-file", action="append", default=[], metavar="") _add_identity_flags(obs) dump = sub.add_parser("append-from-worker", help="Dump worker sidecar errors[] into errors.jsonl") dump.add_argument("--sidecar", required=True) dump.add_argument("--out", required=True) dump.add_argument("--task-key", required=True) dump.add_argument("--agent", required=True, choices=sorted(ALLOWED_AGENTS)) dump.add_argument("--agent-role", required=True, choices=sorted(ALLOWED_AGENT_ROLES)) dump.add_argument("--model", required=True) _add_identity_flags(dump) return p def _add_identity_flags(parser): parser.add_argument("--participant-ref", default="") parser.add_argument("--role-execution-ref", default="") parser.add_argument("--invocation-ref", default="") parser.add_argument("--attempt", type=int, default=None) parser.add_argument("--execution-label", default="") def _identity_from_args(args): payload = {} if args.participant_ref: payload["participantRef"] = args.participant_ref if args.role_execution_ref: payload["roleExecutionRef"] = args.role_execution_ref if args.invocation_ref: payload["invocationRef"] = args.invocation_ref if args.attempt is not None: payload["attempt"] = args.attempt if args.execution_label: payload["executionLabel"] = args.execution_label return payload or None def _read_stderr(args): if args.stderr_excerpt_file: return Path(args.stderr_excerpt_file).read_text() return args.stderr_excerpt def _read_message(args): return args.message_file.read_text(encoding="utf-8").strip() if args.message_file else args.message def _read_command(args): return args.command_file.read_text(encoding="utf-8").strip() if args.command_file else args.command def _read_context(args): if args.cause is None and not args.evidence and not args.evidence_file: return None if args.cause is None: raise ValueError("--evidence requires --cause") evidence: dict[str, str] = {} evidence_files = [ f"{kind}={Path(filename).read_text(encoding='utf-8').strip()}" for kind, _, filename in (raw.partition("=") for raw in args.evidence_file) ] for raw in [*args.evidence, *evidence_files]: kind, separator, value = raw.partition("=") if not separator or kind not in CAUSE_EVIDENCE_FIELDS or not value.strip(): raise ValueError( "--evidence must be one of " f"{', '.join(f'{field}=' for field in CAUSE_EVIDENCE_FIELDS)}" ) if kind in evidence: raise ValueError(f"--evidence cannot repeat {kind}") evidence[kind] = value.strip() context: dict[str, object] = {"cause": args.cause} if evidence: context["causeEvidence"] = evidence return context def main(argv=None): parser = _build_parser() args = parser.parse_args(argv) if args.cmd == "append-observed": try: context = _read_context(args) except ValueError as exc: parser.error(str(exc)) append_observed( out_path=args.out, task_key=args.task_key, phase=args.phase, agent=args.agent, agent_role=args.agent_role, model=args.model, error_type=args.error_type, command=_read_command(args), command_kind=args.command_kind, exit_code=args.exit_code, duration_ms=args.duration_ms, message=_read_message(args), stderr_excerpt=_read_stderr(args), context=context, identity=_identity_from_args(args), ) return 0 if args.cmd == "append-from-worker": n = dump_from_worker_sidecar( sidecar_path=args.sidecar, out_path=args.out, task_key=args.task_key, agent=args.agent, agent_role=args.agent_role, model=args.model, identity=_identity_from_args(args), ) print(n) return 0 # Unreachable: argparse `required=True` on subparsers raises SystemExit # before this function runs if the subcommand is missing or invalid. if __name__ == "__main__": raise SystemExit(main())