"""Experiment 5-13: an Agent that creates and validates another Agent.""" from __future__ import annotations import json import hashlib import os import re import shutil import time from dataclasses import dataclass from pathlib import Path, PurePosixPath from typing import Any from openai import OpenAI from validator import validate_agent ROOT = Path(__file__).resolve().parent REFERENCE = ROOT / "reference_agent" DEFAULT_PROTOCOL = ROOT / "experiment_protocol.json" ALLOWED_TEMPLATE_FILES = {"domain_spec.json"} REQUIRED_SCRATCH_FILES = { "agent.py", "domain_tools.py", "main.py", "system_prompt.md", "tools.json", "requirements.txt", "tests/test_contract.py", "tests/test_domain_tools.py", "README.md", } SCRATCH_FILE_GROUPS = ( ("domain_tools.py",), ("tools.json",), ("agent.py",), ("main.py",), ("tests/test_contract.py",), ("tests/test_domain_tools.py",), ("system_prompt.md", "requirements.txt", "README.md"), ) SCRATCH_CHECKPOINT = "scratch_generation_checkpoint.json" SCRATCH_CHECKPOINT_CALLS = "creator_calls.checkpoint.json" @dataclass(frozen=True) class ResolvedBackend: """A real OpenAI-compatible endpoint used by creator and generated Agents.""" provider: str client: OpenAI model: str api_key: str base_url: str | None def generated_agent_env(self) -> dict[str, str]: """Expose this endpoint through the standard environment contract. Scratch-mode output is model-generated and may only understand the conventional OPENAI_* variables. Aliasing a resolved compatible endpoint here lets both comparison arms use the *same* real model. Clearing OpenRouter prevents generated code from selecting a stale router key merely because one exists in the parent shell. """ return { "OPENAI_API_KEY": self.api_key, "OPENAI_BASE_URL": self.base_url or "https://api.openai.com/v1", "OPENAI_MODEL": self.model, "OPENROUTER_API_KEY": "", "OPENROUTER_MODEL": "", "AGENT_PROVIDER": "openai", } def resolve_client() -> ResolvedBackend: requested = os.getenv("AGENT_CREATOR_PROVIDER", "auto").strip().casefold() aliases = {"kimi": "moonshot", "volcengine": "ark", "doubao": "ark"} requested = aliases.get(requested, requested) if requested not in {"auto", "moonshot", "ark", "openai", "openrouter"}: raise RuntimeError( "AGENT_CREATOR_PROVIDER must be auto, moonshot/kimi, ark, openai, or openrouter" ) candidates: list[tuple[str, str | None, str | None, str | None]] = [ ( "moonshot", os.getenv("MOONSHOT_API_KEY") or os.getenv("KIMI_API_KEY"), os.getenv("MOONSHOT_BASE_URL") or os.getenv("KIMI_BASE_URL") or "https://api.moonshot.cn/v1", os.getenv("AGENT_CREATOR_MODEL") or os.getenv("KIMI_MODEL") or "kimi-k3", ), ( "ark", os.getenv("ARK_API_KEY"), os.getenv("ARK_BASE_URL") or "https://ark.cn-beijing.volces.com/api/v3", os.getenv("AGENT_CREATOR_MODEL") or os.getenv("ARK_MODEL") or os.getenv("ARK_ENDPOINT"), ), ( "openai", os.getenv("OPENAI_API_KEY"), os.getenv("OPENAI_BASE_URL") or "https://api.openai.com/v1", os.getenv("AGENT_CREATOR_MODEL") or os.getenv("OPENAI_MODEL") or "gpt-5.6-luna", ), ( "openrouter", os.getenv("OPENROUTER_API_KEY"), "https://openrouter.ai/api/v1", os.getenv("AGENT_CREATOR_MODEL") or os.getenv("OPENROUTER_MODEL") or "openai/gpt-5.6-luna", ), ] for provider, api_key, base_url, model in candidates: if requested not in {"auto", provider}: continue if api_key and model: request_timeout = float(os.getenv("AGENT_CREATOR_REQUEST_TIMEOUT", "600")) if request_timeout < 60: raise RuntimeError("AGENT_CREATOR_REQUEST_TIMEOUT must be at least 60 seconds") return ResolvedBackend( provider=provider, client=OpenAI( api_key=api_key, base_url=base_url, timeout=request_timeout, # The creator records and controls retries itself. Hidden SDK # retries would make raw request accounting and checkpoints # incomplete. max_retries=0, ), model=model, api_key=api_key, base_url=base_url, ) wanted = "a usable real endpoint" if requested == "auto" else requested raise RuntimeError(f"No credentials/model configured for {wanted}") def _json_object(text: str) -> dict[str, Any]: text = text.strip() if text.startswith("```"): text = text.split("\n", 1)[1].rsplit("```", 1)[0] try: value = json.loads(text) except json.JSONDecodeError: match = re.search(r"\{.*\}", text, re.DOTALL) if not match: raise value = json.loads(match.group(0)) if not isinstance(value, dict): raise ValueError("model response must be a JSON object") return value @dataclass class GenerationStats: strategy: str model: str prompt_tokens: int | None completion_tokens: int | None generation_s: float model_calls: int = 1 repair_attempts: int = 0 cached_prompt_tokens: int | None = 0 transport_failures: int = 0 usage_complete: bool = True def load_protocol(path: Path = DEFAULT_PROTOCOL) -> tuple[dict[str, Any], str]: raw = path.read_bytes() value = json.loads(raw) if not isinstance(value, dict) or value.get("experiment") != "5-13": raise ValueError("experiment protocol must be an Experiment 5-13 object") return value, hashlib.sha256(raw).hexdigest() def _usage_cost(usage: dict[str, Any], pricing: dict[str, Any]) -> dict[str, Any]: prompt = int(usage.get("prompt_tokens") or 0) cached = int(usage.get("cached_prompt_tokens") or 0) completion = int(usage.get("completion_tokens") or 0) cached = min(max(cached, 0), prompt) uncached = prompt - cached cost = ( uncached * float(pricing["uncached_input_per_million"]) + cached * float(pricing["cached_input_per_million"]) + completion * float(pricing["output_per_million"]) ) / 1_000_000 usage_complete = usage.get("usage_complete") is not False return { "prompt_tokens": prompt, "cached_prompt_tokens": cached, "uncached_prompt_tokens": uncached, "completion_tokens": completion, "total_tokens": prompt + completion, "requests": int(usage.get("requests") or 0), "cost": round(cost, 9), "currency": pricing["currency"], "pricing_as_of": pricing["as_of"], "pricing_source_url": pricing["source_url"], "all_usage_priced": usage_complete, "cost_qualification": ( "complete observed provider usage priced in native currency" if usage_complete else "observed-usage lower bound; at least one failed request had unknown usage" ), } class AgentCreator: def __init__(self, client: Any, model: str): self.client = client self.model = model self.call_records: list[dict[str, Any]] = [] def _ask( self, prompt: str, *, purpose: str = "creator_generation", max_tokens: int = 12000, ) -> tuple[dict[str, Any], GenerationStats]: started = time.perf_counter() messages = [ { "role": "system", "content": ( "You are a senior Agent engineer. Return one JSON object only. " "Never include credentials. Generated code must be complete, typed, tested, " "use standard role/tool-call messages, preserve tool results, cap iterations, " "and read model/API configuration from environment variables." ), }, {"role": "user", "content": prompt}, ] kwargs: dict[str, Any] = dict( model=self.model, messages=messages, response_format={"type": "json_object"}, temperature=1 if any(tag in self.model.casefold() for tag in ("kimi-", "gpt-5")) else 0, max_tokens=max_tokens, ) prompt_tokens = 0 completion_tokens = 0 cached_prompt_tokens = 0 usage_seen = False model_calls = 0 parse_errors: list[str] = [] response_format_enabled = True transport_failures = 0 for attempt in range(1, 4): response = None for transport_attempt in range(1, 4): request_started = time.perf_counter() try: response = self.client.chat.completions.create(**kwargs) break except Exception as exc: message = str(exc).casefold() unsupported_format = response_format_enabled and any( tag in message for tag in ("response_format", "json_object", "unsupported", "not support") ) retryable = type(exc).__name__ in { "APITimeoutError", "APIConnectionError", "InternalServerError", "RateLimitError", } self.call_records.append( { "call_index": len(self.call_records) + 1, "purpose": purpose, "logical_attempt": attempt, "transport_attempt": transport_attempt, "request": { "model": self.model, "messages": json.loads(json.dumps(messages, ensure_ascii=False)), "response_format": kwargs.get("response_format"), "temperature": kwargs.get("temperature"), "max_tokens": kwargs.get("max_tokens"), }, "response": None, "usage": None, "error": { "type": type(exc).__name__, "message": str(exc)[:2000], "retryable": retryable, "usage_known": False, }, "duration_s": round(time.perf_counter() - request_started, 3), "duration_from_operation_start_s": round( time.perf_counter() - started, 3 ), } ) if unsupported_format: kwargs.pop("response_format", None) response_format_enabled = False continue if not retryable or transport_attempt == 3: raise transport_failures += 1 if response is None: # pragma: no cover - defensive; exceptions raise above raise RuntimeError("creator request ended without a response") model_calls += 1 usage = getattr(response, "usage", None) prompt_details = getattr(usage, "prompt_tokens_details", None) if usage is not None: usage_seen = True prompt_tokens += getattr(usage, "prompt_tokens", 0) or 0 completion_tokens += getattr(usage, "completion_tokens", 0) or 0 cached_prompt_tokens += getattr(prompt_details, "cached_tokens", 0) or 0 content = response.choices[0].message.content or "{}" self.call_records.append( { "call_index": len(self.call_records) + 1, "purpose": purpose, "logical_attempt": attempt, "transport_attempt": transport_attempt, "request": { "model": self.model, "messages": json.loads(json.dumps(messages, ensure_ascii=False)), "response_format": kwargs.get("response_format"), "temperature": kwargs.get("temperature"), "max_tokens": kwargs.get("max_tokens"), }, "response": { "id": getattr(response, "id", None), "model": getattr(response, "model", self.model), "finish_reason": getattr(response.choices[0], "finish_reason", None), "content": content, }, "usage": { "prompt_tokens": getattr(usage, "prompt_tokens", None), "cached_prompt_tokens": getattr(prompt_details, "cached_tokens", None), "completion_tokens": getattr(usage, "completion_tokens", None), }, "error": None, "duration_s": round(time.perf_counter() - request_started, 3), "duration_from_operation_start_s": round(time.perf_counter() - started, 3), } ) try: payload = _json_object(content) return payload, GenerationStats( strategy="", model=self.model, prompt_tokens=prompt_tokens if usage_seen else None, completion_tokens=completion_tokens if usage_seen else None, generation_s=round(time.perf_counter() - started, 3), model_calls=model_calls, cached_prompt_tokens=cached_prompt_tokens if usage_seen else None, transport_failures=transport_failures, usage_complete=transport_failures == 0, ) except (json.JSONDecodeError, ValueError) as exc: parse_errors.append(f"attempt {attempt}: {type(exc).__name__}: {exc}") if attempt == 3: break # Do not echo a potentially huge/truncated response back into the # context. Ask for a fresh compact envelope and account for both # calls in the efficiency comparison. messages.append( { "role": "user", "content": ( "The prior response was not a complete parseable JSON object " f"({type(exc).__name__}). Start over; do not continue it. Return the " "same complete requested object, but keep the whole response below " "6,000 tokens by using concise code and no duplicated explanation." ), } ) raise ValueError( "model returned no parseable JSON object after 3 real calls: " + " | ".join(parse_errors) ) @staticmethod def _safe_files(payload: dict[str, Any], allowed: set[str]) -> dict[str, str]: files = payload.get("files") if not isinstance(files, dict): # Code models use a few semantically equivalent envelope labels. # Accept one explicit mapping wrapper, but never relax the path # allowlist applied below. wrappers = [ payload.get(key) for key in ("artifacts", "outputs", "generated_files") if isinstance(payload.get(key), dict) ] if len(wrappers) == 1: files = wrappers[0] # Code-oriented models sometimes honor the requested file mapping but # omit the redundant outer ``files`` key. Accept that wire-shape only # when every top-level key is itself an allowed path; the same path # traversal and allowlist checks below still apply. elif payload and set(payload).issubset(allowed): files = payload else: raise ValueError("model response must contain a files object") clean: dict[str, str] = {} for raw_path, content in files.items(): path = PurePosixPath(raw_path) if path.is_absolute() or ".." in path.parts or str(path) not in allowed: raise ValueError(f"model attempted disallowed file: {raw_path}") # A JSON file is occasionally returned as a nested JSON value instead # of a JSON-encoded string. Serializing that value is lossless and # does not weaken the executable-file content checks. if path.suffix == ".json" and isinstance(content, (dict, list)): content = json.dumps(content, ensure_ascii=False, indent=2) if not isinstance(content, str): raise ValueError(f"file content must be text: {raw_path}") clean[str(path)] = content return clean @staticmethod def _write_files(output: Path, files: dict[str, str]) -> None: for relative, content in files.items(): path = output / relative path.parent.mkdir(parents=True, exist_ok=True) path.write_text(content, encoding="utf-8") @staticmethod def _normalize_files(files: dict[str, str]) -> dict[str, str]: """Normalize harmless wire-shape variation without altering behavior.""" result = dict(files) if "tools.json" in result: value = json.loads(result["tools.json"]) if isinstance(value, list): value = {"tools": value} if not isinstance(value, dict) or not isinstance(value.get("tools"), list): raise ValueError("tools.json must be an object with a tools array") result["tools.json"] = json.dumps(value, ensure_ascii=False, indent=2) return result @staticmethod def _validate_template_spec( payload: dict[str, Any], requirements: str ) -> dict[str, Any]: """Validate the compact specialization consumed by the proven template. Template mode intentionally asks the model for domain decisions only. The standard Agent loop, tool dispatcher, CLI, and tests remain deterministic reference code, so their token cost and correctness do not depend on the model regenerating boilerplate. """ raw = payload.get("specialization") if not isinstance(raw, dict): raise ValueError("template response must contain a specialization object") required_strings = ( "name", "role", "sample_task", "tool_name", "tool_description", "record_noun", "records_argument", "identifier_field", "required_field", "status_field", "evidence_field", "approved_label", "rejected_label", ) spec: dict[str, Any] = {} for key in required_strings: value = raw.get(key) if not isinstance(value, str) or not value.strip(): raise ValueError(f"specialization.{key} must be a non-empty string") spec[key] = value.strip() identifier = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$") for key in ( "tool_name", "records_argument", "identifier_field", "required_field", "status_field", "evidence_field", ): if not identifier.fullmatch(spec[key]): raise ValueError(f"specialization.{key} must be a Python/JSON identifier") if len(spec["tool_name"]) > 64: raise ValueError("specialization.tool_name must be at most 64 characters") passing_values = raw.get("passing_values") if not isinstance(passing_values, list) or not passing_values or not all( isinstance(value, str) and value.strip() for value in passing_values ): raise ValueError("specialization.passing_values must be a non-empty string list") spec["passing_values"] = [value.strip() for value in passing_values] remediation = raw.get("remediation_by_status") if not isinstance(remediation, dict) or not remediation or not all( isinstance(key, str) and key.strip() and isinstance(value, str) and value.strip() for key, value in remediation.items() ): raise ValueError( "specialization.remediation_by_status must map statuses to actions" ) spec["remediation_by_status"] = { key.strip(): value.strip() for key, value in remediation.items() } default_remediation = raw.get("default_remediation") if not isinstance(default_remediation, str) or not default_remediation.strip(): raise ValueError("specialization.default_remediation must be non-empty") spec["default_remediation"] = default_remediation.strip() # The authoritative user requirement is copied verbatim; the model cannot # silently narrow it while producing the compact domain overlay. spec["requirements"] = requirements spec["schema_version"] = "1.0" return spec @staticmethod def _materialize_template(output: Path) -> None: spec = json.loads((output / "domain_spec.json").read_text(encoding="utf-8")) records_argument = spec["records_argument"] identifier_field = spec["identifier_field"] required_field = spec["required_field"] status_field = spec["status_field"] evidence_field = spec["evidence_field"] record_noun = spec["record_noun"] system_prompt = f"""{spec['role']} Your complete requirement is: {spec['requirements']} Mandatory workflow: 1. Read the structured records supplied by the user. 2. Call `{spec['tool_name']}` exactly once with the complete `{records_argument}` array; do not omit records. 3. Treat a required {record_noun} as passing only when the tool says it passes. 4. Base the final decision, evidence, and remediation only on the tool result and the user's supplied facts. Never invent a registration ID, facts ID, database prerequisite, hidden tool, or missing-information request when the user already supplied the records. Never approve when any required record is not in the configured passing state. In the final answer, state `{spec['approved_label']}` or `{spec['rejected_label']}`, list every non-passing required record with its exact evidence, and give a specific remediation action for each one. """ tool = { "type": "function", "function": { "name": spec["tool_name"], "description": spec["tool_description"], "parameters": { "type": "object", "properties": { records_argument: { "type": "array", "description": ( f"The complete list of {record_noun} records supplied by the user." ), "items": { "type": "object", "properties": { identifier_field: {"type": "string"}, required_field: {"type": "boolean"}, status_field: {"type": "string"}, evidence_field: {"type": "string"}, }, "required": [ identifier_field, required_field, status_field, evidence_field, ], "additionalProperties": True, }, } }, "required": [records_argument], "additionalProperties": False, }, }, } readme = f"""# {spec['name']} This Agent was produced by adapting the proven Experiment 5-13 reference Agent. Its model-generated artifact is the compact `domain_spec.json`; the bounded Agent loop, standard tool protocol, dispatcher, CLI, and contract tests are inherited. ## Purpose {spec['requirements']} ## Run ```bash pip install -r requirements.txt export OPENAI_API_KEY=... export OPENAI_MODEL=... python main.py --task '{spec['sample_task']}' ``` The tool accepts the complete `{records_argument}` array directly. It does not require facts to be pre-registered and never stores credentials. """ (output / "system_prompt.md").write_text(system_prompt, encoding="utf-8") (output / "tools.json").write_text( json.dumps({"tools": [tool]}, ensure_ascii=False, indent=2), encoding="utf-8", ) (output / "README.md").write_text(readme, encoding="utf-8") def _generate_template_spec( self, requirements: str, prompt: str ) -> tuple[dict[str, Any], GenerationStats]: total: GenerationStats | None = None feedback = "" for _attempt in range(1, 4): payload, stats = self._ask( prompt + feedback, purpose="template_specialization", max_tokens=2500, ) total = self._add_stats(total, stats) try: return self._validate_template_spec(payload, requirements), total except ValueError as exc: feedback = ( "\n\nThe previous compact specialization was invalid: " f"{exc}. Return a corrected complete object only." ) raise ValueError("model failed to return a valid template specialization") @staticmethod def _add_stats(total: GenerationStats | None, current: GenerationStats) -> GenerationStats: if total is None: return current total.prompt_tokens = ( (total.prompt_tokens or 0) + (current.prompt_tokens or 0) if total.prompt_tokens is not None or current.prompt_tokens is not None else None ) total.completion_tokens = ( (total.completion_tokens or 0) + (current.completion_tokens or 0) if total.completion_tokens is not None or current.completion_tokens is not None else None ) total.cached_prompt_tokens = ( (total.cached_prompt_tokens or 0) + (current.cached_prompt_tokens or 0) if total.cached_prompt_tokens is not None or current.cached_prompt_tokens is not None else None ) total.generation_s = round(total.generation_s + current.generation_s, 3) total.model_calls += current.model_calls total.transport_failures += current.transport_failures total.usage_complete = total.usage_complete and current.usage_complete return total def _generate_valid_envelope( self, prompt: str, allowed: set[str], required: set[str], *, purpose: str = "creator_generation", max_tokens: int = 12000, ) -> tuple[dict[str, Any], dict[str, str], GenerationStats]: total: GenerationStats | None = None feedback = "" envelope_errors: list[str] = [] for attempt in range(1, 4): payload, stats = self._ask( prompt + feedback, purpose=purpose, max_tokens=max_tokens, ) total = self._add_stats(total, stats) try: files = self._normalize_files(self._safe_files(payload, allowed)) missing = required - set(files) if missing: raise ValueError(f"response omitted required files: {sorted(missing)}") return payload, files, total except (ValueError, json.JSONDecodeError) as exc: if "disallowed file" in str(exc): raise envelope_errors.append( f"attempt {attempt}: {type(exc).__name__}: {exc}; " f"top-level keys={sorted(payload)}" ) feedback = ( "\n\nYOUR PREVIOUS RESPONSE FAILED THE ENVELOPE CHECK. " f"Error: {exc}. Top-level keys were {sorted(payload)}. " "Return a fresh complete JSON object with exactly one top-level " "`files` object. " "Do not discuss the error and do not omit any required file. Keep code concise." ) raise ValueError( "model failed to return a complete safe file envelope after 3 calls: " + " | ".join(envelope_errors) ) @staticmethod def _validate_scratch_blueprint(payload: dict[str, Any]) -> dict[str, Any]: """Validate the model-authored contract used across staged scratch calls. Generating every source file in one JSON response is unnecessarily brittle: JSON escaping expands code and some otherwise capable models truncate the envelope. The blueprint is not a reference implementation. It is the scratch model's own architecture decision, carried across smaller real generation calls so independently generated files agree on interfaces. """ name = payload.get("name") sample_task = payload.get("sample_task") design = payload.get("design") if not isinstance(name, str) or not name.strip(): raise ValueError("scratch blueprint name must be non-empty") if not isinstance(sample_task, str) or not sample_task.strip(): raise ValueError("scratch blueprint sample_task must be non-empty") if not isinstance(design, dict): raise ValueError("scratch blueprint design must be an object") required = ( "tool_name", "records_argument", "identifier_field", "required_field", "status_field", "evidence_field", "passing_value", "agent_contract", "dispatcher_contract", "cli_contract", "test_contract", ) for key in required: value = design.get(key) if not isinstance(value, str) or not value.strip(): raise ValueError(f"scratch blueprint design.{key} must be non-empty") identifier = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$") for key in ( "tool_name", "records_argument", "identifier_field", "required_field", "status_field", "evidence_field", ): if not identifier.fullmatch(str(design[key])): raise ValueError( f"scratch blueprint design.{key} must be a Python/JSON identifier" ) return { "name": name.strip(), "sample_task": sample_task.strip(), "design": {key: str(design[key]).strip() for key in required}, } @staticmethod def _renumber_calls(calls: list[dict[str, Any]]) -> list[dict[str, Any]]: result = json.loads(json.dumps(calls, ensure_ascii=False)) for index, record in enumerate(result, start=1): record["call_index"] = index return result @staticmethod def _stats_from_calls( calls: list[dict[str, Any]], *, strategy: str, repair_attempts: int = 0 ) -> GenerationStats: prompt_tokens = 0 cached_prompt_tokens = 0 completion_tokens = 0 successful = 0 failures = 0 usage_complete = True duration_s = 0.0 model = "" for record in calls: duration_s += float(record.get("duration_s") or 0) request = record.get("request") or {} model = str(request.get("model") or model) error = record.get("error") if error: failures += 1 message = str(error.get("message") or "").casefold() # A request rejected before inference solely because the endpoint # does not implement response_format has known zero token usage. if not any( marker in message for marker in ( "response_format", "json_object", "unsupported", "not support", ) ): usage_complete = False continue usage = record.get("usage") if not isinstance(usage, dict): usage_complete = False continue observed_prompt = usage.get("prompt_tokens") observed_completion = usage.get("completion_tokens") if not isinstance(observed_prompt, int) or not isinstance( observed_completion, int ): usage_complete = False continue successful += 1 prompt_tokens += observed_prompt completion_tokens += observed_completion observed_cached = usage.get("cached_prompt_tokens") if isinstance(observed_cached, int): cached_prompt_tokens += observed_cached return GenerationStats( strategy=strategy, model=model, prompt_tokens=prompt_tokens, completion_tokens=completion_tokens, generation_s=round(duration_s, 3), model_calls=successful, repair_attempts=repair_attempts, cached_prompt_tokens=cached_prompt_tokens, transport_failures=failures, usage_complete=usage_complete, ) def _generate_scratch_files( self, requirements: str, checkpoint_dir: Path, *, resume: bool = False, ) -> tuple[dict[str, Any], dict[str, str], GenerationStats]: """Generate a standalone Agent from scratch through resumable real calls.""" blueprint_prompt = f"""Design a standalone tool-using Agent from scratch for: {requirements} Do not use or assume any reference implementation. Return exactly one compact JSON object: {{ "name": "short Agent name", "sample_task": "one executable sample task", "design": {{ "tool_name": "snake_case tool name", "records_argument": "array argument in the supplied task", "identifier_field": "record identifier field", "required_field": "boolean required field", "status_field": "status field", "evidence_field": "evidence field", "passing_value": "the only passing status", "agent_contract": "precise bounded ReAct loop and standard message protocol", "dispatcher_contract": "precise validation and result contract", "cli_contract": "--task/--model input and one-JSON stdout/exit behavior", "test_contract": "dependency-injected protocol and domain cases to test" }} }} For this release-readiness task, map the real gates/id/required/outcome/evidence fields, accept only `passed`, and require REFUSED when any required gate is non-passing. Keep the entire blueprint below 900 tokens. """ checkpoint_path = checkpoint_dir / SCRATCH_CHECKPOINT calls_path = checkpoint_dir / SCRATCH_CHECKPOINT_CALLS requirements_sha256 = hashlib.sha256(requirements.encode("utf-8")).hexdigest() call_start = len(self.call_records) existing_calls: list[dict[str, Any]] = [] if resume: if not checkpoint_path.is_file() or not calls_path.is_file(): raise ValueError("scratch resume requires both generation checkpoint files") state = json.loads(checkpoint_path.read_text(encoding="utf-8")) calls_payload = json.loads(calls_path.read_text(encoding="utf-8")) existing_calls = list(calls_payload.get("calls") or []) if state.get("requirements_sha256") != requirements_sha256: raise ValueError("scratch resume rejected: requirements changed") if state.get("model") != self.model: raise ValueError("scratch resume rejected: creator model changed") else: state = { "schema_version": "1.0", "experiment": "5-13", "strategy": "scratch", "requirements_sha256": requirements_sha256, "model": self.model, "status": "in_progress", "blueprint": None, "blueprint_receipts": [], "completed_groups": [], "group_receipts": {}, "file_sha256": {}, "stats": None, "error": None, } def combined_calls() -> list[dict[str, Any]]: return self._renumber_calls( [*existing_calls, *self.call_records[call_start:]] ) def persist(status: str, error: str | None = None) -> GenerationStats: calls = combined_calls() stats = self._stats_from_calls(calls, strategy="scratch") state["status"] = status state["stats"] = stats.__dict__ state["error"] = error state["updated_at_utc"] = time.strftime( "%Y-%m-%dT%H:%M:%SZ", time.gmtime() ) _write_evidence(checkpoint_path, state, forbidden_secret="") _write_evidence( calls_path, { "schema_version": "1.0", "experiment": "5-13", "strategy": "scratch", "model": self.model, "checkpoint_status": status, "calls": calls, }, forbidden_secret="", ) return stats expected_file_sha = dict(state.get("file_sha256", {})) mismatched_files: list[str] = [] for relative, expected_sha in expected_file_sha.items(): path = checkpoint_dir / relative if not path.is_file(): raise ValueError(f"scratch checkpoint file missing: {relative}") observed = hashlib.sha256(path.read_bytes()).hexdigest() if observed != expected_sha: mismatched_files.append(relative) if mismatched_files: # A failed enclosing run may still have completed and materialized a # model-authored deterministic/live repair. Those calls are durably # retained next to the scratch directory. Replay only successful, # safe repair envelopes over the generation hashes and accept the # current files iff every byte is explained by those receipts. An # invalid/truncated repair response changes no expected hash. failed_calls_path = checkpoint_dir.parent / "scratch_failed_creator_calls.json" if failed_calls_path.is_file() and state.get("status") == "generated": failed_payload = json.loads(failed_calls_path.read_text(encoding="utf-8")) for record in failed_payload.get("calls") or []: purpose = str(record.get("purpose") or "") response = record.get("response") or {} if not purpose.startswith( ("scratch_deterministic_repair", "scratch_live_repair") ): continue try: repair_payload = _json_object(str(response.get("content") or "")) repaired = self._normalize_files( self._safe_files(repair_payload, REQUIRED_SCRATCH_FILES) ) except (json.JSONDecodeError, ValueError): continue for relative, content in repaired.items(): expected_file_sha[relative] = hashlib.sha256( content.encode("utf-8") ).hexdigest() unexplained = [] for relative in mismatched_files: observed = hashlib.sha256((checkpoint_dir / relative).read_bytes()).hexdigest() if observed != expected_file_sha.get(relative): unexplained.append(relative) if unexplained: raise ValueError( "scratch checkpoint has file changes not explained by retained model repair " f"receipts: {', '.join(sorted(unexplained))}" ) blueprint = state.get("blueprint") if blueprint is None: feedback = "" try: for _attempt in range(1, 4): before = len(self.call_records) payload, _stats = self._ask( blueprint_prompt + feedback, purpose="scratch_blueprint", max_tokens=2500, ) state["blueprint_receipts"].extend( { "response_id": (record.get("response") or {}).get("id"), "response_model": (record.get("response") or {}).get("model"), } for record in self.call_records[before:] if record.get("response") ) try: blueprint = self._validate_scratch_blueprint(payload) state["blueprint"] = blueprint persist("in_progress") break except ValueError as exc: persist("in_progress", f"invalid blueprint: {exc}") feedback = ( "\nThe prior blueprint was invalid: " f"{exc}. Return a corrected complete blueprint only." ) except Exception as exc: persist("failed", f"{type(exc).__name__}: {exc}") raise else: blueprint = self._validate_scratch_blueprint(blueprint) if blueprint is None: persist("failed", "model failed to return a valid scratch blueprint") raise ValueError("model failed to return a valid scratch blueprint") context_by_group: dict[tuple[str, ...], tuple[str, ...]] = { ("domain_tools.py",): (), ("tools.json",): (), ("agent.py",): ("domain_tools.py", "tools.json"), ("main.py",): ("agent.py",), ("tests/test_contract.py",): ("agent.py", "main.py"), ("tests/test_domain_tools.py",): ("domain_tools.py", "tools.json"), ("system_prompt.md", "requirements.txt", "README.md"): ("tools.json",), } for paths in SCRATCH_FILE_GROUPS: group_id = "|".join(paths) if group_id in state.get("completed_groups", []): continue context_paths = context_by_group[paths] context_files = { path: (checkpoint_dir / path).read_text(encoding="utf-8") for path in context_paths } group_specific = "" if paths == ("main.py",): group_specific = """ This group is deliberately tiny. `agent.py` already owns the complete CLI and exports `main`. Make `main.py` a thin executable wrapper that imports and calls that function; do not duplicate the Agent, tool, parsing, or CLI implementation. Keep it below 15 physical lines. """ elif paths == ("tests/test_contract.py",): group_specific = """ Use dependency-injected fake chat-completion objects. Test the exact standard assistant.tool_calls -> role=tool protocol, matching tool_call_id, bounded loop, history order, provider usage aggregation, and CLI result contract. """ elif paths == ("tests/test_domain_tools.py",): group_specific = """ Test all-passed approval, required failed/skipped refusal, optional failure, evidence preservation, malformed records, and unknown tool rejection. """ group_prompt = f"""Implement one file group of a standalone Agent built from scratch. Authoritative requirement: {requirements} The same model already authored this architecture contract: {json.dumps(blueprint, ensure_ascii=False)} Previously generated interface context (read-only; keep compatible): {json.dumps(context_files, ensure_ascii=False)} Return exactly one JSON object of the form {{"files": {{"path": "complete content"}}}}. Return exactly these files and no others: {', '.join(paths)}. The final Agent must define `GeneratedAgent(model=None, client=None)` and `run(task, history=None, max_iterations=...)`. It must use the current OpenAI client and `client.chat.completions.create`, standard assistant tool_calls followed by role=tool results with matching tool_call_id, a bounded for/range loop, argument validation, dependency injection in deterministic tests, and API/model configuration only from environment or `--model`. `python main.py --task "..." --model "..." --history-json '[]'` must perform a real model run, print exactly one JSON object, and exit zero only when `ok` is true. The JSON must preserve the full credential-free `messages` list and aggregate provider `usage` with prompt_tokens, cached_prompt_tokens, completion_tokens, and requests. Prior history must remain in exact order. Never hard-code temperature. Tests must assert protocol, history preservation, usage, and domain behavior, not placeholders or mocks of the complete experiment. Keep each Python file below 170 lines and this response below 4,500 tokens. {group_specific} """ before = len(self.call_records) try: _payload, group_files, _group_stats = self._generate_valid_envelope( group_prompt, set(paths), set(paths), purpose=f"scratch_group:{group_id}", max_tokens=12000, ) self._write_files(checkpoint_dir, group_files) for relative in paths: state["file_sha256"][relative] = hashlib.sha256( (checkpoint_dir / relative).read_bytes() ).hexdigest() state["completed_groups"].append(group_id) state["group_receipts"][group_id] = [ { "response_id": (record.get("response") or {}).get("id"), "response_model": (record.get("response") or {}).get("model"), "purpose": record.get("purpose"), } for record in self.call_records[before:] if record.get("response") ] persist("in_progress") except Exception as exc: persist( "failed", f"scratch file group {list(paths)} failed: {type(exc).__name__}: {exc}", ) raise ValueError( f"scratch file group {list(paths)} failed: {exc}" ) from exc files = { relative: (checkpoint_dir / relative).read_text(encoding="utf-8") for relative in REQUIRED_SCRATCH_FILES if (checkpoint_dir / relative).is_file() } missing = REQUIRED_SCRATCH_FILES - set(files) if missing: persist("failed", f"staged scratch generation omitted files: {sorted(missing)}") raise ValueError(f"staged scratch generation omitted files: {sorted(missing)}") stats = persist("generated") return blueprint, files, stats def _repair_until_deterministic( self, *, strategy: str, requirements: str, output: Path, allowed: set[str], stats: GenerationStats, ) -> GenerationStats: """Use actual compiler/pytest feedback to repair the created Agent.""" for repair in range(1, 4): report = validate_agent(output, timeout=120) if report.ok: return stats current_files = { relative: (output / relative).read_text(encoding="utf-8") for relative in sorted(allowed) if (output / relative).is_file() and relative != "README.md" } repair_prompt = f"""Repair a generated Agent for this requirement: {requirements} Generation strategy: {strategy} Compiler/test/contract errors: {json.dumps(report.errors, ensure_ascii=False)[:10000]} Current editable files: {json.dumps(current_files, ensure_ascii=False)} Return one JSON object only: {{"files": {{"path": "complete replacement content"}}}}. Return only files that need correction, but each returned file must be complete. Paths must be chosen from: {', '.join(sorted(allowed))}. For template strategy, agent.py and main.py are immutable. tools.json must be an object with a top-level "tools" array. Fix root causes and keep tests meaningful; never delete assertions merely to make pytest pass. The common Agent contract requires `run(task, history=None, max_iterations=...)`, CLI `--history-json`, raw messages, and aggregate usage. In this shared harness, `ok` means the Agent run completed its task successfully; it does not mean the business decision was approval. Therefore an evidence-backed `REFUSED` decision is a successful completion with `ok=true`, a populated `answer`, and CLI exit 0, exactly like an `APPROVED` decision. Only validation/runtime failure or an unfinished iteration-capped run has `ok=false`/nonzero exit. Generated tests must enforce this common execution contract while separately asserting the decision. """ payload, repair_stats = self._ask( repair_prompt, purpose=f"{strategy}_deterministic_repair:{repair}", max_tokens=7000, ) replacements = self._normalize_files(self._safe_files(payload, allowed)) if not replacements: raise ValueError("repair response contained no replacement files") self._write_files(output, replacements) if strategy == "template": self._materialize_template(output) stats = self._add_stats(stats, repair_stats) stats.repair_attempts = repair return stats def judge_live_completion( self, *, requirements: str, live_task: str, live_result: dict[str, Any] | None, ) -> dict[str, Any]: """Judge task completion, not merely a successful process exit.""" prompt = f"""Audit a generated Agent's real live run. Agent requirements: {requirements} Identical controlled live task given to both comparison arms: {live_task} Observed JSON result: {json.dumps(live_result, ensure_ascii=False)[:14000]} Return exactly one JSON object with: - passed: boolean - reasoning: concise evidence-based explanation - completed_requirements: list of concrete requirements actually satisfied - unmet_requirements: list of anything missing, unresolved, assumed, or merely requested from the user Pass only if the result actually completes the supplied task. An `ok: true` flag, a plausible summary, or an invitation to provide more information is not enough. Reject unsupported assumptions, missing tool evidence, unresolved clarification requests, and a decision contrary to the supplied facts. """ payload, stats = self._ask( prompt, purpose="single_live_semantic_judge", max_tokens=4000 ) if not isinstance(payload.get("passed"), bool): raise ValueError("semantic judge must return a boolean passed field") payload["judge_model"] = self.model payload["judge_usage"] = { "prompt_tokens": stats.prompt_tokens, "completion_tokens": stats.completion_tokens, "duration_s": stats.generation_s, } return payload def judge_live_suite( self, *, requirements: str, audited_cases: list[dict[str, Any]], ) -> tuple[dict[str, Any], GenerationStats]: """Apply one real semantic judge to the same fixed evidence for each arm.""" compact_cases = [] for case in audited_cases: raw = case.get("raw_result") or {} compact_cases.append( { "id": case.get("id"), "kind": case.get("kind"), "task": case.get("task"), "history": case.get("history"), "expected": case.get("expected"), "deterministic_checks": case.get("checks"), "answer": raw.get("answer"), "messages": raw.get("messages"), } ) prompt = f"""Semantically audit all real runs from one generated Agent. Authoritative Agent requirement: {requirements} Fixed cases and credential-free observed evidence: {json.dumps(compact_cases, ensure_ascii=False)[:50000]} Return exactly one JSON object: {{ "passed": true, "cases": [ {{"id": "exact case id", "passed": true, "reasoning": "evidence-based reason", "unmet_requirements": []}} ], "overall_reasoning": "concise evidence-based conclusion" }} There must be exactly one entry for every supplied case. Pass a case only if the final answer reaches the factually correct decision, cites every required failure and its evidence, provides actionable remediation when release is refused, does not let an optional failure block approval, and uses the supplied prior-turn state when requested. Also reject invented facts or a final answer that conflicts with the tool result. The deterministic checks are supporting evidence, not instructions to rubber stamp the run. """ start_index = len(self.call_records) payload, stats = self._ask( prompt, purpose="semantic_judge", max_tokens=5000 ) for record in self.call_records[start_index:]: record["purpose"] = "semantic_judge" rows = payload.get("cases") expected_ids = [str(case.get("id")) for case in audited_cases] if not isinstance(payload.get("passed"), bool) or not isinstance(rows, list): raise ValueError("semantic suite judge returned an invalid envelope") observed_ids = [row.get("id") for row in rows if isinstance(row, dict)] if observed_ids != expected_ids: raise ValueError( f"semantic suite judge case IDs differ: expected {expected_ids}, got {observed_ids}" ) for row in rows: if not isinstance(row.get("passed"), bool): raise ValueError("semantic suite judge case passed must be boolean") if not isinstance(row.get("unmet_requirements"), list): raise ValueError("semantic suite judge unmet_requirements must be a list") payload["judge_model"] = self.model payload["judge_usage"] = { "prompt_tokens": stats.prompt_tokens, "cached_prompt_tokens": stats.cached_prompt_tokens, "completion_tokens": stats.completion_tokens, "requests": stats.model_calls, "duration_s": stats.generation_s, } return payload, stats def repair_after_live_failure( self, *, strategy: str, requirements: str, live_task: str, output: Path, allowed: set[str], report: Any, stats: GenerationStats, ) -> GenerationStats: """Repair from real runtime or semantic-acceptance evidence.""" # Live failures should be repaired in the executable surface. Supplying # the model-authored test suites again consumed most of the repair # context and previously caused a length-truncated `{}` response. live_editable = allowed & { "agent.py", "domain_tools.py", "main.py", "system_prompt.md", "tools.json" } live_evidence_text = json.dumps( report.live_cases or report.live_result, ensure_ascii=False ) fenced_final_failure = ( strategy == "scratch" and "non-JSON final answer" in live_evidence_text and "```json" in live_evidence_text ) completion_contract_failure = ( strategy == "scratch" and any( marker in "\n".join(report.errors) for marker in ( "process_exit_zero", "agent_reported_ok", "answer_has_expected_decision", "answer_has_required_content", ) ) ) if fenced_final_failure: # The raw evidence localizes this failure to final-answer parsing; # keeping other generated files out of the edit context discourages # broad rewrites that can regress already-passing contracts. live_editable = {"agent.py"} elif completion_contract_failure: # The raw live checks localize this class of failure to result # packaging and CLI exit semantics. Keep domain policy and tool # evidence immutable while the model repairs only that surface. live_editable = {"agent.py", "main.py"} current_files = { relative: (output / relative).read_text(encoding="utf-8") for relative in sorted(live_editable) if (output / relative).is_file() and relative != "README.md" } repair_prompt = f"""Repair a generated Agent after a real live validation failure. Original Agent requirement: {requirements} The identical live task that both comparison arms must complete: {live_task} Generation strategy: {strategy} Validation errors: {json.dumps(report.errors, ensure_ascii=False)[:8000]} Live result: {json.dumps(report.live_cases or report.live_result, ensure_ascii=False)[:14000]} Semantic judgment: {json.dumps(report.semantic_judgment, ensure_ascii=False)[:6000]} Current editable files: {json.dumps(current_files, ensure_ascii=False)[:30000]} Return one JSON object only: {{"files": {{"path": "complete replacement content"}}}}. Return only complete replacements selected from: {', '.join(sorted(allowed))}. Do not weaken tests or print a fabricated success. Fix the real root cause. The CLI must accept `--task`, `--model`, and `--history-json`, emit one JSON object with raw messages and aggregate provider usage, and exit zero only after actual completion. `run` must accept and preserve prior history. Do not hard-code `temperature`; use the endpoint default. For template strategy, agent.py and main.py are immutable, so fix the domain prompt/tools/dispatcher instead. Preserve the existing implementation and make the smallest root-cause repair; do not rewrite unrelated files. Preserve the shared harness invariant exactly: `ok` reports task execution, not release approval. Both an evidence-backed APPROVED and an evidence-backed REFUSED final decision have `ok=true`, a populated `answer`, and CLI exit 0. Runtime/validation errors have `ok=false` and exit 1; an unfinished iteration-capped run is nonzero. Keep the business `decision` field separate. { "The raw provider final is a fenced JSON object followed by prose. Reuse the generated " "brace-balanced JSON extraction helper to parse the first complete object instead of treating " "the whole message as bare JSON. Return only a complete agent.py replacement and preserve the " "shared completed-task ok/answer/exit contract." if fenced_final_failure else "" } { "The live evidence shows a completed policy decision but the shared result wrapper/CLI " "misclassified it. Make only the minimal agent.py/main.py changes needed to emit a populated " "answer, keep decision separate, set ok=true for completed APPROVED or REFUSED tasks, and exit " "zero for either completed decision. Do not change domain policy or tool evidence." if completion_contract_failure else "" } """ total: GenerationStats | None = None envelope_error = "" replacements: dict[str, str] = {} for attempt in range(1, 4): payload, repair_stats = self._ask( repair_prompt + ( "\n\nYour prior response failed the safe replacement-envelope check: " f"{envelope_error}. Return a fresh compact object with a non-empty `files` mapping." if envelope_error else "" ), purpose=f"{strategy}_live_repair:{attempt}", max_tokens=9000, ) total = self._add_stats(total, repair_stats) try: replacements = self._normalize_files(self._safe_files(payload, allowed)) if not replacements: raise ValueError("live repair response contained no replacement files") break except (ValueError, json.JSONDecodeError) as exc: envelope_error = str(exc) if not replacements or total is None: raise ValueError( "live repair produced no safe replacement after 3 calls: " + envelope_error ) self._write_files(output, replacements) if strategy == "template": self._materialize_template(output) stats = self._add_stats(stats, total) stats.repair_attempts += 1 return stats def create_from_template(self, requirements: str, output: Path) -> GenerationStats: if output.exists(): raise FileExistsError(output) shutil.copytree(REFERENCE, output) prompt = f"""Create a compact domain specialization for a proven policy-record Agent: {requirements} The reference already supplies the bounded standard tool loop, CLI, a validated generic record-policy dispatcher, documentation materializer, and tests. Do not generate Python or Markdown. Return exactly: {{"specialization": {{ "name": "short Agent name", "role": "precise role and domain responsibility", "sample_task": "one short example task", "tool_name": "snake_case function name", "tool_description": "description saying it evaluates the complete user-supplied records", "record_noun": "singular domain record name", "records_argument": "the array argument name present in the user's structured input", "identifier_field": "record identifier field", "required_field": "boolean required/optional field", "status_field": "status/outcome field", "evidence_field": "evidence field", "passing_values": ["exact status values that mean pass"], "approved_label": "exact approval label", "rejected_label": "exact refusal label", "remediation_by_status": {{"known non-pass status": "specific corrective action"}}, "default_remediation": "safe corrective action for any other non-pass status" }}}} For the supplied release-readiness requirement, map the actual `gates` array and its `id`/`required`/`outcome`/`evidence` fields directly, accept only `passed`, use APPROVED/REFUSED, and provide actions for both failed and skipped. Never introduce a facts ID or pre-registration step. Keep the entire response below 1,200 tokens. """ spec, stats = self._generate_template_spec(requirements, prompt) (output / "domain_spec.json").write_text( json.dumps(spec, ensure_ascii=False, indent=2), encoding="utf-8" ) self._materialize_template(output) (output / "generation.json").write_text( json.dumps({"strategy": "template", "requirements": requirements, "name": spec["name"], "sample_task": spec["sample_task"], "artifact": "domain_spec.json"}, ensure_ascii=False, indent=2), encoding="utf-8", ) stats.strategy = "template" return self._repair_until_deterministic( strategy="template", requirements=requirements, output=output, allowed=ALLOWED_TEMPLATE_FILES, stats=stats, ) def create_from_scratch( self, requirements: str, output: Path, *, resume: bool = False ) -> GenerationStats: if output.exists(): if resume: if not output.is_dir(): raise FileExistsError(output) elif output.is_dir() and not any(output.iterdir()): output.rmdir() else: raise FileExistsError(output) output.mkdir(parents=True, exist_ok=resume) payload, files, stats = self._generate_scratch_files( requirements, output, resume=resume ) # Files were durably written group-by-group. Rewrite the same verified # contents only to keep this method's postcondition explicit. self._write_files(output, files) (output / "generation.json").write_text( json.dumps({"strategy": "scratch", "requirements": requirements, "name": payload.get("name"), "sample_task": payload.get("sample_task")}, ensure_ascii=False, indent=2), encoding="utf-8", ) stats.strategy = "scratch" return self._repair_until_deterministic( strategy="scratch", requirements=requirements, output=output, allowed=REQUIRED_SCRATCH_FILES, stats=stats, ) def _write_evidence(path: Path, payload: Any, *, forbidden_secret: str) -> str: text = json.dumps(payload, ensure_ascii=False, indent=2) if forbidden_secret and forbidden_secret in text: raise ValueError(f"credential leaked into {path.name}") if re.search(r"\bsk-[A-Za-z0-9_-]{12,}\b", text): raise ValueError(f"credential-shaped value leaked into {path.name}") path.parent.mkdir(parents=True, exist_ok=True) path.write_text(text, encoding="utf-8") return hashlib.sha256(path.read_bytes()).hexdigest() def _live_usage(validation: dict[str, Any]) -> dict[str, int]: totals = { "prompt_tokens": 0, "cached_prompt_tokens": 0, "completion_tokens": 0, "requests": 0, } for case in validation.get("live_cases") or []: usage = (case.get("raw_result") or {}).get("usage") or {} for key in totals: totals[key] += int(usage.get(key) or 0) return totals def run_experiment( requirements: str, output_root: Path, *, live: bool = True, live_task: str | None = None, resume: bool = False, protocol_path: Path = DEFAULT_PROTOCOL, ) -> dict[str, Any]: protocol, protocol_sha256 = load_protocol(protocol_path) backend = resolve_client() creator = AgentCreator(backend.client, backend.model) protocol_exact = requirements == protocol["requirements"] and live_task is None live_cases = list(protocol["live_cases"]) if protocol_exact else [] if live_task is not None: # A custom single task remains useful for development, but cannot satisfy # the frozen book protocol or be reported as the official experiment. live_cases = [] output_root.mkdir(parents=True, exist_ok=True) (output_root / "experiment_protocol.json").write_bytes(protocol_path.read_bytes()) previous_result: dict[str, Any] = {} comparison_path = output_root / "comparison.json" if resume and comparison_path.is_file(): previous_result = json.loads(comparison_path.read_text(encoding="utf-8")) if previous_result.get("protocol", {}).get("sha256") != protocol_sha256: raise ValueError("resume rejected: experiment protocol hash changed") backend_requirement = protocol["backend_requirement"] current_backend_ok = ( backend.provider == backend_requirement["provider"] and backend.model == backend_requirement["model"] ) runs: dict[str, Any] = {} for strategy in protocol["comparison_design"]["strategies"]: destination = output_root / strategy previous_run = previous_result.get("runs", {}).get(strategy, {}) if resume and previous_run.get("validation", {}).get("ok") is True: evidence = previous_run.get("evidence") or {} for evidence_name in ("creator_calls", "live"): item = evidence.get(evidence_name) or {} evidence_path = output_root / str(item.get("path") or "") if not evidence_path.is_file(): raise ValueError( f"resume rejected: missing accepted {strategy} {evidence_name} evidence" ) observed = hashlib.sha256(evidence_path.read_bytes()).hexdigest() if observed != item.get("sha256"): raise ValueError( f"resume rejected: changed accepted {strategy} {evidence_name} evidence" ) runs[strategy] = previous_run continue call_start = len(creator.call_records) previous_calls: list[dict[str, Any]] = [] previous_calls_path = destination / "creator_calls.json" checkpoint_calls_path = destination / SCRATCH_CHECKPOINT_CALLS failed_calls_path = output_root / f"{strategy}_failed_creator_calls.json" # Use the longest valid retained chain. A later incomplete run can have # more calls than either the staged-generation checkpoint or an older # failed-run receipt; source precedence alone would silently lose usage. if resume: call_candidates: list[list[dict[str, Any]]] = [] for calls_source in ( previous_calls_path, failed_calls_path, checkpoint_calls_path, ): if calls_source.is_file(): value = json.loads(calls_source.read_text(encoding="utf-8")) calls = value.get("calls") or [] if isinstance(calls, list): call_candidates.append(calls) if call_candidates: previous_calls = max(call_candidates, key=len) try: previous_generation = ( previous_run.get("generation") ) if resume and destination.is_dir() and previous_generation: stats = GenerationStats(**previous_generation) # A live repair may fix runtime behavior while regressing a # deterministic contract. Never skip directly back to live # calls on resume: re-establish compile/tests first using the # same real model and its actual pytest feedback. allowed = ( ALLOWED_TEMPLATE_FILES if strategy == "template" else REQUIRED_SCRATCH_FILES ) stats = creator._repair_until_deterministic( strategy=strategy, requirements=requirements, output=destination, allowed=allowed, stats=stats, ) else: stats = ( creator.create_from_template(requirements, destination) if strategy == "template" else creator.create_from_scratch( requirements, destination, resume=resume and destination.exists(), ) ) if previous_calls: # Include every retained generation/repair request in the # efficiency accounting, including a length-truncated call # that incurred real provider usage. stats = creator._stats_from_calls( [*previous_calls, *creator.call_records[call_start:]], strategy=strategy, ) allowed = ALLOWED_TEMPLATE_FILES if strategy == "template" else REQUIRED_SCRATCH_FILES validation_attempts = 0 semantic_stats: GenerationStats | None = None for live_attempt in range(3): validation_attempts += 1 report = validate_agent( destination, live_task=live_task if live and not protocol_exact else None, live_cases=live_cases if live and protocol_exact else None, model=backend.model, timeout=300, extra_env=backend.generated_agent_env(), ) if live and protocol_exact and report.live_ok: try: judgment, judge_stats = creator.judge_live_suite( requirements=requirements, audited_cases=report.live_cases, ) semantic_stats = creator._add_stats(semantic_stats, judge_stats) report.semantic_judgment = judgment report.semantic_ok = ( judgment["passed"] and all( row["passed"] and not row["unmet_requirements"] for row in judgment["cases"] ) ) if not report.semantic_ok: report.errors.append( "semantic suite acceptance failed: " + str(judgment.get("overall_reasoning", "no reasoning returned")) ) except Exception as exc: report.semantic_ok = False report.errors.append( f"semantic suite acceptance error: {type(exc).__name__}: {exc}" ) if report.ok or not live or live_attempt == 2: break if not (report.structural_ok and report.compile_ok and report.tests_ok): break stats = creator.repair_after_live_failure( strategy=strategy, requirements=requirements, live_task=json.dumps(live_cases, ensure_ascii=False), output=destination, allowed=allowed, report=report, stats=stats, ) # A runtime repair is model-authored executable code. Re-run the # full deterministic contract immediately and, if necessary, # repair from actual pytest feedback before another live case. stats = creator._repair_until_deterministic( strategy=strategy, requirements=requirements, output=destination, allowed=allowed, stats=stats, ) validation = report.to_dict() live_hash = _write_evidence( destination / "live_evidence.json", { "schema_version": "2.0", "experiment": "5-13", "strategy": strategy, "protocol_sha256": protocol_sha256, "provider": backend.provider, "model": backend.model, "validation": validation, }, forbidden_secret=backend.api_key, ) calls = creator._renumber_calls( [*previous_calls, *creator.call_records[call_start:]] ) calls_hash = _write_evidence( destination / "creator_calls.json", { "schema_version": "1.0", "experiment": "5-13", "strategy": strategy, "protocol_sha256": protocol_sha256, "calls": calls, }, forbidden_secret=backend.api_key, ) creation_usage = { "prompt_tokens": stats.prompt_tokens or 0, "cached_prompt_tokens": stats.cached_prompt_tokens or 0, "completion_tokens": stats.completion_tokens or 0, "requests": stats.model_calls, "usage_complete": stats.usage_complete, } live_usage = _live_usage(validation) semantic_usage = { "prompt_tokens": semantic_stats.prompt_tokens if semantic_stats else 0, "cached_prompt_tokens": semantic_stats.cached_prompt_tokens if semantic_stats else 0, "completion_tokens": semantic_stats.completion_tokens if semantic_stats else 0, "requests": semantic_stats.model_calls if semantic_stats else 0, "usage_complete": semantic_stats.usage_complete if semantic_stats else True, } pricing = backend_requirement["pricing"] runs[strategy] = { "generation": stats.__dict__, "validation_attempts": validation_attempts, "validation": validation, "evidence": { "creator_calls": { "path": f"{strategy}/creator_calls.json", "sha256": calls_hash, "credential_free": True, }, "live": { "path": f"{strategy}/live_evidence.json", "sha256": live_hash, "credential_free": validation.get("raw_evidence_ok") is True, }, }, "accounting": { "creation": _usage_cost(creation_usage, pricing), "live_validation": _usage_cost(live_usage, pricing), "semantic_judge": _usage_cost(semantic_usage, pricing), }, } except Exception as exc: calls = creator._renumber_calls( [*previous_calls, *creator.call_records[call_start:]] ) calls_path = output_root / f"{strategy}_failed_creator_calls.json" try: calls_hash = _write_evidence( calls_path, { "schema_version": "1.0", "experiment": "5-13", "strategy": strategy, "protocol_sha256": protocol_sha256, "calls": calls, "failure": f"{type(exc).__name__}: {exc}", }, forbidden_secret=backend.api_key, ) except Exception: calls_hash = None runs[strategy] = { "generation": None, "validation_attempts": 0, "validation": { "ok": False, "structural_ok": False, "compile_ok": False, "tests_ok": False, "live_ok": False if live else None, "protocol_ok": False if live else None, "multiturn_ok": False if live else None, "raw_evidence_ok": bool(calls_hash), "usage_ok": False if live else None, "semantic_ok": False if live else None, "duration_s": 0.0, "errors": [f"{type(exc).__name__}: {exc}"], "live_result": None, "semantic_judgment": None, "live_cases": [], "quality_score": 0, "quality_max_score": 0, }, "evidence": { "creator_calls": { "path": calls_path.name, "sha256": calls_hash, "credential_free": calls_hash is not None, } }, "accounting": None, } passing = [name for name, run in runs.items() if run["validation"]["ok"]] template = runs["template"] scratch = runs["scratch"] template_score = int(template["validation"].get("quality_score") or 0) scratch_score = int(scratch["validation"].get("quality_score") or 0) quality_noninferiority = template_score >= scratch_score quality_strict_advantage = template_score > scratch_score template_creation = (template.get("accounting") or {}).get("creation") or {} scratch_creation = (scratch.get("accounting") or {}).get("creation") or {} template_seconds = (template.get("generation") or {}).get("generation_s") scratch_seconds = (scratch.get("generation") or {}).get("generation_s") efficiency_measurable = all( value is not None for value in ( template_creation.get("total_tokens"), scratch_creation.get("total_tokens"), template_seconds, scratch_seconds, ) ) efficiency_advantage = ( efficiency_measurable and template_creation["total_tokens"] < scratch_creation["total_tokens"] and template_seconds < scratch_seconds ) book_joint_claim_supported = quality_strict_advantage and efficiency_advantage practical_tradeoff_supported = quality_noninferiority and efficiency_advantage required_case_count = len(protocol["live_cases"]) every_case_attempted = all( len(run["validation"].get("live_cases") or []) == required_case_count for run in runs.values() ) raw_evidence_complete = all( (run.get("evidence") or {}).get("creator_calls", {}).get("credential_free") is True and (run.get("evidence") or {}).get("live", {}).get("credential_free") is True for run in runs.values() ) usage_and_cost_complete = all( run["validation"].get("usage_ok") is True and (run.get("accounting") or {}).get("creation", {}).get("all_usage_priced") is True and (run.get("accounting") or {}).get("live_validation", {}).get("all_usage_priced") is True for run in runs.values() ) campaign_complete = ( live and protocol_exact and current_backend_ok and every_case_attempted and raw_evidence_complete and usage_and_cost_complete and efficiency_measurable ) manuscript_acceptance_complete = campaign_complete and len(passing) == 2 result_claim = ( "inconclusive_incomplete_campaign" if not campaign_complete else "template_quality_and_efficiency_advantage_observed" if book_joint_claim_supported else "book_joint_advantage_not_observed" ) if template_score != scratch_score: winner = "template" if template_score > scratch_score else "scratch" elif efficiency_measurable: winner = min( ("template", "scratch"), key=lambda name: ( runs[name]["accounting"]["creation"]["total_tokens"], runs[name]["generation"]["generation_s"], ), ) else: winner = None completion = { "campaign_complete": campaign_complete, "manuscript_acceptance_complete": manuscript_acceptance_complete, "protocol_hash_recorded": True, "protocol_exact": protocol_exact, "current_recommended_backend": current_backend_ok, "both_arms_generated": all(run.get("generation") is not None for run in runs.values()), "both_arms_pass_static_and_tests": all( run["validation"].get("structural_ok") and run["validation"].get("compile_ok") and run["validation"].get("tests_ok") for run in runs.values() ), "both_arms_standard_tool_protocol": all( run["validation"].get("protocol_ok") is True for run in runs.values() ), "both_arms_complete_common_tasks": len(passing) == 2, "both_arms_multiturn_state": all( run["validation"].get("multiturn_ok") is True for run in runs.values() ), "raw_evidence_credential_free": raw_evidence_complete, "usage_and_native_cost_complete": usage_and_cost_complete, "blockers": [], } if not manuscript_acceptance_complete: completion["blockers"] = [ name for name, passed_gate in completion.items() if isinstance(passed_gate, bool) and not passed_gate ] result = { "schema_version": "2.0", "experiment": "5-13", "generated_at_utc": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), "requirements": requirements, "provider": backend.provider, "model": backend.model, "api_style": backend_requirement["api_style"], "live_validation_required": live, "resumed_from_existing": resume and bool(previous_result), "protocol": { "path": "experiment_protocol.json", "sha256": protocol_sha256, "schema_version": protocol["schema_version"], "frozen_at_utc": protocol["frozen_at_utc"], }, "runs": runs, "completion": completion, "quality_comparison": { "metric": protocol["comparison_design"]["quality_metric"], "scores": {"template": template_score, "scratch": scratch_score}, "maximum_scores": { "template": template["validation"].get("quality_max_score"), "scratch": scratch["validation"].get("quality_max_score"), }, "template_noninferior": quality_noninferiority, "template_strict_advantage": quality_strict_advantage, }, "efficiency_comparison": { "metric": protocol["comparison_design"]["efficiency_metric"], "creation": { "template": { "total_tokens": template_creation.get("total_tokens"), "generation_s": template_seconds, "cost": template_creation.get("cost"), "currency": template_creation.get("currency"), }, "scratch": { "total_tokens": scratch_creation.get("total_tokens"), "generation_s": scratch_seconds, "cost": scratch_creation.get("cost"), "currency": scratch_creation.get("currency"), }, }, "template_advantage": efficiency_advantage, }, "hypothesis_result": { "book_joint_claim_supported": book_joint_claim_supported, "practical_noninferior_quality_plus_efficiency_supported": practical_tradeoff_supported, "result_claim": result_claim, "criteria_changed_after_observing_results": False, }, "winner_under_preregistered_order": winner, "official_complete": manuscript_acceptance_complete, } _write_evidence(comparison_path, result, forbidden_secret=backend.api_key) return result