| |
| """Create one synthetic Meta3 response for local parser tests. No network calls.""" |
|
|
| from __future__ import annotations |
|
|
| import argparse |
| import json |
| from datetime import datetime, timezone |
| from pathlib import Path |
| from typing import Any |
|
|
|
|
| ROOT = Path(__file__).resolve().parents[1] |
|
|
|
|
| def load_json(path: Path) -> dict[str, Any]: |
| with path.open("r", encoding="utf-8") as handle: |
| data = json.load(handle) |
| if not isinstance(data, dict): |
| raise ValueError(f"Expected JSON object: {path}") |
| return data |
|
|
|
|
| def load_jsonl(path: Path) -> list[dict[str, Any]]: |
| rows: list[dict[str, Any]] = [] |
| with path.open("r", encoding="utf-8") as handle: |
| for line_no, line in enumerate(handle, start=1): |
| if not line.strip(): |
| continue |
| data = json.loads(line) |
| if not isinstance(data, dict): |
| raise ValueError(f"Expected JSON object in {path}:{line_no}") |
| rows.append(data) |
| return rows |
|
|
|
|
| def write_json(path: Path, data: dict[str, Any]) -> None: |
| path.parent.mkdir(parents=True, exist_ok=True) |
| path.write_text(json.dumps(data, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") |
|
|
|
|
| def latest_run_dir() -> Path: |
| runs_root = ROOT / "runs" |
| candidates = sorted(p for p in runs_root.iterdir() if p.is_dir()) |
| if not candidates: |
| raise FileNotFoundError("No run dirs found. Run scripts/meta3_prepare.py first.") |
| return candidates[-1] |
|
|
|
|
| def run_dir_for(run_id: str | None) -> Path: |
| return ROOT / "runs" / run_id if run_id else latest_run_dir() |
|
|
|
|
| def main() -> int: |
| parser = argparse.ArgumentParser(description="Write one mock OpenAI-compatible response.") |
| parser.add_argument("--run-id", default=None, help="Run id under runs/. Defaults to latest run.") |
| parser.add_argument("--mode", choices=["calibration", "claim_ledger"], default="calibration") |
| parser.add_argument("--packet-index", type=int, default=0) |
| args = parser.parse_args() |
|
|
| run_dir = run_dir_for(args.run_id) |
| packet_manifest = load_json(run_dir / f"outputs/workpackets/{args.mode}_packet_manifest.json") |
| packets = load_jsonl(Path(str(packet_manifest["packet_path"]))) |
| if not packets: |
| raise ValueError("No workpackets found") |
| packet = packets[args.packet_index] |
|
|
| answer = { |
| "question_id": packet["question_id"], |
| "mode": packet["mode"], |
| "canonical_thesis": "Mock thesis for parser validation only.", |
| "claims": [ |
| { |
| "claim": "Mock claim anchored to the first digest.", |
| "supporting_digest_indices": [1], |
| "supporting_source_paths": [packet["digests"][0]["source_path"]], |
| "counterevidence_or_limit": "Mock response is not evidence for the real analysis.", |
| "confidence": "low", |
| } |
| ], |
| "counterevidence_status": "mock_not_evaluated", |
| "uncertainty": "Synthetic response; validates extraction and schema only.", |
| "owner_next_step": "Run real network packets only after requirements and preflight pass.", |
| } |
| response = { |
| "packet_id": packet["packet_id"], |
| "status_code": 200, |
| "started_at": datetime.now(timezone.utc).isoformat(), |
| "finished_at": datetime.now(timezone.utc).isoformat(), |
| "request_model": "mock-model", |
| "mock": True, |
| "network_calls_made": 0, |
| "response": { |
| "choices": [ |
| { |
| "message": { |
| "role": "assistant", |
| "content": json.dumps(answer, ensure_ascii=False), |
| } |
| } |
| ] |
| }, |
| } |
| output_path = run_dir / "outputs/responses" / args.mode / f"{packet['packet_id']}.json" |
| write_json(output_path, response) |
| print(f"mock_response={output_path}") |
| print(f"packet_id={packet['packet_id']}") |
| print("network_calls_made=0") |
| return 0 |
|
|
|
|
| if __name__ == "__main__": |
| raise SystemExit(main()) |
|
|
