diff --git a/autoresearch/prefill/program.md b/autoresearch/prefill/program.md index 8e31253..6050e6d 100644 --- a/autoresearch/prefill/program.md +++ b/autoresearch/prefill/program.md @@ -43,8 +43,12 @@ Use a lexicographic objective: frontier. Otherwise restore the previous candidate. 9. Append the result and repeat. -Every candidate must target one current unresolved proof obligation and contain -a falsifiable hypothesis plus distinct Generator and Critic directives. +Every candidate must target exactly one current unresolved leaf obligation and +contain a falsifiable hypothesis plus distinct Generator and Critic directives. +An unresolved Critic verdict must isolate one strictly smaller missing lemma; +the host records that lemma as a deduplicated child obligation. Completed GAN +runs, transcripts, checkpoints, and ledger updates remain durable even when the +candidate strategy is reverted or fixed evaluation fails. Do not optimize output wording, scores, prizes, or other proof-irrelevant content. Prefill performance is a tertiary objective after mathematical diff --git a/autoresearch/prefill/supervisor.py b/autoresearch/prefill/supervisor.py index 39094dc..f227cb7 100644 --- a/autoresearch/prefill/supervisor.py +++ b/autoresearch/prefill/supervisor.py @@ -107,6 +107,29 @@ def _extract_json(text: str) -> dict: def parse_research_verdict(output: str, candidate_id: str) -> dict: + event_matches = re.findall( + r"^\[autoresearch-verdict\]\s+(\{.*\})\s*$", + output, + re.MULTILINE, + ) + if event_matches: + fields = json.loads(event_matches[-1]) + if fields.get("candidate_id") != candidate_id: + raise ValueError("research verdict candidate ID mismatch") + if fields.get("outcome") not in { + "SUPPORTED", "FALSIFIED", "DECOMPOSED", "INCONCLUSIVE", + }: + raise ValueError("invalid research verdict outcome") + if ( + len(str(fields.get("evidence", ""))) < 40 + or len(str(fields.get("new_frontier", ""))) < 30 + ): + raise ValueError("research verdict lacks substantive evidence/frontier") + return { + "outcome": fields["outcome"], + "evidence": fields["evidence"], + "new_frontier": fields["new_frontier"], + } matches = list(re.finditer( r"^(?:critic>\s*)?### AUTORESEARCH_VERDICT\s*$" r"(?P.*?)(?=^### |\Z)", @@ -139,6 +162,24 @@ def parse_research_verdict(output: str, candidate_id: str) -> dict: } +def _pending_leaf_ids(ledger: dict) -> list[str]: + obligations = ledger.get("obligations", []) + unresolved = { + str(item.get("obligation_id", "")) + for item in obligations + if item.get("status") == "UNRESOLVED" + } + unresolved_parents = { + str(item.get("parent_id", "")) + for item in obligations + if ( + item.get("status") == "UNRESOLVED" + and item.get("parent_id") + ) + } + return sorted(unresolved - unresolved_parents) + + class StrategyPrefillHeartbeat: def __init__( self, @@ -232,10 +273,12 @@ def propose_candidate( "Do not weaken full context, final-only snapshots, or no-fallback rules." " The hypothesis must not repeat any hypothesis or hash in RESULTS. " "It must either construct a concrete object or attempt a concrete " - "counterexample for one smaller decomposition leaf." + "counterexample for one smaller decomposition leaf. The " + "target_obligation_id must be one of PENDING_LEAF_IDS." f"\n\nPROGRAM:\n{program}\n\nCURRENT:\n{json.dumps(current)}" f"\n\nRESULTS:\n{results_text[-12000:]}" f"\n\nLEDGER:\n{json.dumps(ledger)}" + f"\n\nPENDING_LEAF_IDS:\n{json.dumps(_pending_leaf_ids(ledger))}" ) ids = tokenizer.apply_chat_template( [{"role": "user", "content": prompt}], @@ -396,7 +439,7 @@ def run_gan_experiment( candidate_path: Path, state_path: Path, timeout_s: float, -) -> tuple[str, dict, str]: +) -> tuple[str, str]: command = [ "bash", str(repo / "scripts/run_agent_gan_repl.sh"), "--skip-ensure", "--no-auto-loop", @@ -447,12 +490,7 @@ def terminate_on_timeout() -> None: if not matches: raise RuntimeError("GAN experiment produced no benchmark run id") run_id = matches[-1] - report = _json_request( - f"http://127.0.0.1:8090/v1/network/benchmarks/{run_id}", - ) - if report.get("status") != "completed": - raise RuntimeError(f"GAN benchmark is not completed: {report.get('status')}") - return run_id, report, output + return run_id, output def read_results(path: Path) -> list[dict]: @@ -478,7 +516,9 @@ def best_kept(results: list[dict]) -> dict | None: def should_keep(result: dict, baseline: dict | None) -> bool: if not result["accepted"]: return False - if result.get("research_outcome") not in {"SUPPORTED", "FALSIFIED"}: + if result.get("research_outcome") not in { + "SUPPORTED", "FALSIFIED", "DECOMPOSED", + }: return False if not result.get("hypothesis_novel", False): return False @@ -497,7 +537,7 @@ def should_keep(result: dict, baseline: dict | None) -> bool: "proof_obligations_unresolved", "compute_chunk_tokens", "candidate_sha256", "report_path", "hypothesis_sha256", "research_outcome", "research_evidence", - "new_frontier", + "new_frontier", "transcript_path", "error", ) @@ -551,6 +591,13 @@ def run_iteration(args, iteration: int) -> dict: previous_chunk = int(current["prefill_compute_chunk_tokens"]) proposed = current + gan_completed = False + run_id = "" + experiment_id = "" + report_path = reports_dir / "not-started.json" + transcript_path = reports_dir / "not-started.log" + hypothesis_sha256 = "" + candidate_sha256 = hashlib.sha256(previous_candidate).hexdigest() try: print( f"[autoresearch] iteration={iteration} " @@ -569,6 +616,7 @@ def run_iteration(args, iteration: int) -> dict: "[autoresearch] phase=strategy-proposal real-gemma", flush=True, ) + ledger_data = json.loads(ledger_path.read_text()) proposed = propose_candidate( address=args.address, tokenizer_id=args.tokenizer_id, @@ -577,8 +625,14 @@ def run_iteration(args, iteration: int) -> dict: results_text=( results_path.read_text() if results_path.exists() else "" ), - ledger=json.loads(ledger_path.read_text()), + ledger=ledger_data, ) + if proposed["target_obligation_id"] not in _pending_leaf_ids( + ledger_data, + ): + raise ValueError( + "strategy agent targeted a non-leaf proof obligation", + ) candidate_path.write_text(render_candidate(proposed)) print( f"[autoresearch] phase=candidate-written " @@ -602,21 +656,34 @@ def run_iteration(args, iteration: int) -> dict: } if hypothesis_sha256 in seen_hypotheses: raise ValueError("strategy agent repeated a previous hypothesis") + candidate_sha256 = hashlib.sha256( + candidate_path.read_bytes(), + ).hexdigest() experiment_id = ( f"ar_{int(time.time())}_{iteration}_" f"{hashlib.sha256(candidate_path.read_bytes()).hexdigest()[:8]}" ) report_path = reports_dir / f"{experiment_id}.json" + transcript_path = reports_dir / f"{experiment_id}.log" print( f"[autoresearch] phase=gan-experiment id={experiment_id}", flush=True, ) - run_id, report, gan_output = run_gan_experiment( + run_id, gan_output = run_gan_experiment( repo=root, candidate_path=candidate_path, state_path=state_path, timeout_s=args.experiment_timeout_s, ) + gan_completed = True + transcript_path.write_text(gan_output) + report = _json_request( + f"http://127.0.0.1:8090/v1/network/benchmarks/{run_id}", + ) + if report.get("status") != "completed": + raise RuntimeError( + f"GAN benchmark is not completed: {report.get('status')}", + ) report_path.write_text(json.dumps(report, ensure_ascii=False, indent=2)) candidate_module = _load_candidate(candidate_path) result = evaluate(report, candidate_module) @@ -628,6 +695,7 @@ def run_iteration(args, iteration: int) -> dict: "research_outcome": verdict["outcome"], "research_evidence": verdict["evidence"], "new_frontier": verdict["new_frontier"], + "transcript_path": str(transcript_path), "hypothesis_novel": True, }) keep = should_keep(result, baseline) @@ -660,22 +728,23 @@ def run_iteration(args, iteration: int) -> dict: "proof_obligations_unresolved" ], "compute_chunk_tokens": result["compute_chunk_tokens"], - "candidate_sha256": hashlib.sha256( - candidate_path.read_bytes(), - ).hexdigest(), + "candidate_sha256": candidate_sha256, "report_path": str(report_path), "hypothesis_sha256": hypothesis_sha256, "research_outcome": verdict["outcome"], "research_evidence": verdict["evidence"], "new_frontier": verdict["new_frontier"], + "transcript_path": str(transcript_path), } append_result(results_path, row) if not keep: candidate_path.write_bytes(previous_candidate) - _restore(state_path, previous_state) - _restore(ledger_path, previous_ledger) deploy_candidate(args.worker_ssh, previous_chunk) - print("[autoresearch] phase=reverted", flush=True) + print( + "[autoresearch] phase=candidate-reverted " + "completed-run-preserved", + flush=True, + ) else: print("[autoresearch] phase=kept", flush=True) return row @@ -685,8 +754,9 @@ def run_iteration(args, iteration: int) -> dict: flush=True, ) candidate_path.write_bytes(previous_candidate) - _restore(state_path, previous_state) - _restore(ledger_path, previous_ledger) + if not gan_completed: + _restore(state_path, previous_state) + _restore(ledger_path, previous_ledger) try: deploy_candidate(args.worker_ssh, previous_chunk) except Exception as rollback_exc: @@ -695,7 +765,37 @@ def run_iteration(args, iteration: int) -> dict: f"error={type(rollback_exc).__name__}: {rollback_exc}", flush=True, ) - raise + if not gan_completed: + raise + row = { + "timestamp": time.time(), + "experiment_id": experiment_id, + "run_id": run_id, + "candidate_id": proposed.get("candidate_id", ""), + "target_obligation_id": proposed.get("target_obligation_id", ""), + "constraints_pass": False, + "accepted": False, + "kept": False, + "baseline_metric_s": ( + baseline["metric_cold_critic_prefill_s"] if baseline else "" + ), + "compute_chunk_tokens": proposed.get( + "prefill_compute_chunk_tokens", + "", + ), + "candidate_sha256": candidate_sha256, + "report_path": str(report_path), + "hypothesis_sha256": hypothesis_sha256, + "research_outcome": "EVALUATION_FAILED", + "transcript_path": str(transcript_path), + "error": f"{type(exc).__name__}: {exc}", + } + append_result(results_path, row) + print( + f"[autoresearch] phase=completed-run-preserved run={run_id}", + flush=True, + ) + return row def main() -> int: diff --git a/scripts/agent_gan_repl.py b/scripts/agent_gan_repl.py index 4148154..0293580 100644 --- a/scripts/agent_gan_repl.py +++ b/scripts/agent_gan_repl.py @@ -200,12 +200,22 @@ def load_proof_ledger(path: Path) -> ProofObligationLedger | None: ProofObligation(**item) for item in raw.pop("obligations", []) ] ledger = ProofObligationLedger(obligations=obligations, **raw) + obligation_ids = { + item.obligation_id for item in ledger.obligations + } if ( ledger.schema_version != 1 or not ledger.ledger_id or not ledger.obligations - or len({item.obligation_id for item in ledger.obligations}) - != len(ledger.obligations) + or len(obligation_ids) != len(ledger.obligations) + or any( + item.parent_id + and ( + item.parent_id not in obligation_ids + or item.parent_id == item.obligation_id + ) + for item in ledger.obligations + ) ): raise ValueError("invalid proof obligation ledger") return ledger @@ -216,16 +226,30 @@ def pending_obligations( ) -> list[ProofObligation]: if ledger is None: return [] + unresolved_parent_ids = { + item.parent_id + for item in ledger.obligations + if item.status == "UNRESOLVED" and item.parent_id + } return [ item for item in ledger.obligations - if item.status == "UNRESOLVED" + if ( + item.status == "UNRESOLVED" + and item.obligation_id not in unresolved_parent_ids + ) ] -def format_proof_ledger(ledger: ProofObligationLedger) -> str: +def format_proof_ledger( + ledger: ProofObligationLedger, + obligations: list[ProofObligation] | None = None, +) -> str: + selected = obligations if obligations is not None else pending_obligations(ledger) items = "\n".join( - f"- {item.obligation_id}: {item.statement}" - for item in pending_obligations(ledger) + f"- {item.obligation_id}" + f"{f' (parent={item.parent_id})' if item.parent_id else ''}: " + f"{item.statement}" + for item in selected ) return ( f"PROOF OBLIGATION LEDGER id={ledger.ledger_id} " @@ -273,17 +297,18 @@ def apply_critic_verdicts( continue body = match.group("body") status_match = re.search( - r"^Status:\s*(PROVED|DISPROVED|UNRESOLVED)\s*$", + r"^\*{0,2}Status:\*{0,2}\s*" + r"(PROVED|DISPROVED|UNRESOLVED)\s*$", body, re.MULTILINE, ) evidence_match = re.search( - r"^Evidence:\s*(.+)$", + r"^\*{0,2}Evidence:\*{0,2}\s*(.+)$", body, re.MULTILINE, ) missing_match = re.search( - r"^Missing lemma:\s*(.*)$", + r"^\*{0,2}Missing lemma:\*{0,2}\s*(.*)$", body, re.MULTILINE, ) @@ -318,6 +343,134 @@ def apply_critic_verdicts( return applied +def _normalize_obligation_statement(statement: str) -> str: + return " ".join(re.findall(r"[a-z0-9]+", statement.lower())) + + +def create_child_obligations( + ledger: ProofObligationLedger, + critic_text: str, + run_id: str, + parent_ids: set[str], +) -> list[ProofObligation]: + existing = { + (item.parent_id, _normalize_obligation_statement(item.statement)) + for item in ledger.obligations + } + created: list[ProofObligation] = [] + for match in _ISSUE_VERDICT.finditer(critic_text): + parent_id = match.group(1) + if parent_id not in parent_ids: + continue + body = match.group("body") + status_match = re.search( + r"^\*{0,2}Status:\*{0,2}\s*" + r"(PROVED|DISPROVED|UNRESOLVED)\s*$", + body, + re.MULTILINE, + ) + missing_match = re.search( + r"^\*{0,2}Missing lemma:\*{0,2}\s*(.+)$", + body, + re.MULTILINE, + ) + if ( + status_match is None + or status_match.group(1) != "UNRESOLVED" + or missing_match is None + ): + continue + statement = missing_match.group(1).strip() + normalized = _normalize_obligation_statement(statement) + parent = next( + item for item in ledger.obligations + if item.obligation_id == parent_id + ) + if ( + not normalized + or normalized in {"none", "no missing lemma"} + or normalized == _normalize_obligation_statement(parent.statement) + or (parent_id, normalized) in existing + ): + continue + suffix = hashlib.sha256( + f"{parent_id}:{normalized}".encode(), + ).hexdigest()[:10] + child = ProofObligation( + obligation_id=f"{parent_id}-{suffix}", + statement=statement, + parent_id=parent_id, + last_run_id=run_id, + last_evidence="Created from Critic missing lemma.", + ) + ledger.obligations.append(child) + existing.add((parent_id, normalized)) + created.append(child) + if created: + ledger.version += 1 + return created + + +def build_autoresearch_verdict( + candidate, + ledger: ProofObligationLedger, + applied_verdicts: dict[str, str], + created: list[ProofObligation], +) -> dict: + target_id = str(candidate.TARGET_OBLIGATION_ID) + target = next( + (item for item in ledger.obligations if item.obligation_id == target_id), + None, + ) + status = applied_verdicts.get(target_id, "UNRESOLVED") + target_children = [item for item in created if item.parent_id == target_id] + if status == "PROVED": + outcome = "SUPPORTED" + evidence = target.last_evidence if target is not None else ( + "The targeted proof obligation was closed by the Critic." + ) + frontier = ( + "Integrate the proved leaf into its parent proof obligation and " + "audit every dependency." + ) + elif status == "DISPROVED": + outcome = "FALSIFIED" + evidence = target.last_evidence if target is not None else ( + "The targeted hypothesis was disproved by the Critic." + ) + frontier = ( + "Exclude the falsified approach and construct a distinct " + "hypothesis for the same proof obligation." + ) + elif target_children: + outcome = "DECOMPOSED" + evidence = ( + "The Critic kept the target unresolved and isolated a concrete " + f"missing lemma: {target_children[0].statement}" + ) + frontier = target_children[0].statement + else: + outcome = "INCONCLUSIVE" + evidence = ( + target.last_evidence if target is not None else + "The Critic supplied no structurally valid target verdict." + ) + frontier = ( + target.statement if target is not None else + "Construct a concrete smaller proof obligation." + ) + return { + "candidate_id": str(candidate.CANDIDATE_ID), + "target_obligation_id": target_id, + "outcome": outcome, + "evidence": evidence, + "new_frontier": frontier, + "created_obligation_ids": [ + item.obligation_id for item in target_children + ], + } + + def save_critic_issue_batch(path: Path, batch: CriticIssueBatch) -> None: path.parent.mkdir(parents=True, exist_ok=True) temporary = path.with_suffix(path.suffix + ".tmp") @@ -890,17 +1043,7 @@ def get_stats(): steering, str(research_candidate.GENERATOR_DIRECTIVE), ))) - critic_strategy = str( - research_candidate.CRITIC_DIRECTIVE, - ) + ( - "\n\nAt the end, emit exactly:\n" - "### AUTORESEARCH_VERDICT\n" - f"Candidate ID: {research_candidate.CANDIDATE_ID}\n" - "Outcome: SUPPORTED|FALSIFIED|INCONCLUSIVE\n" - "Evidence: \n" - "New frontier: " - ) + critic_strategy = str(research_candidate.CRITIC_DIRECTIVE) if command.action in {"continue", "steer"} and args.auto_loop: auto_loop_active = True phase = ReplPhase.RUNNING @@ -913,8 +1056,18 @@ def get_stats(): ) proof_ledger = load_proof_ledger(proof_ledger_path) turn_obligations = pending_obligations(proof_ledger) + if research_candidate is not None and turn_obligations: + target_id = str(research_candidate.TARGET_OBLIGATION_ID) + turn_obligations = [ + item for item in turn_obligations + if item.obligation_id == target_id + ] + if not turn_obligations: + raise ValueError( + f"candidate target is not an unresolved leaf: {target_id}", + ) proof_ledger_text = ( - format_proof_ledger(proof_ledger) + format_proof_ledger(proof_ledger, turn_obligations) if turn_obligations else "" ) if proof_ledger is not None: @@ -1131,18 +1284,51 @@ def get_stats(): skip_special_tokens=True, ) applied_verdicts = {} + created_obligations = [] if proof_ledger is not None and turn_obligations: applied_verdicts = apply_critic_verdicts( proof_ledger, critic_text, run_id, ) + created_obligations = create_child_obligations( + proof_ledger, + critic_text, + run_id, + { + item.obligation_id + for item in turn_obligations + }, + ) for obligation_id, status in applied_verdicts.items(): print( f"[critic-verdict] id={obligation_id} " f"status={status}", flush=True, ) + for item in created_obligations: + print( + f"[proof-obligation-created] " + f"id={item.obligation_id} " + f"parent={item.parent_id} " + f"{item.statement}", + flush=True, + ) + if research_candidate is not None: + print( + "[autoresearch-verdict] " + + json.dumps( + build_autoresearch_verdict( + research_candidate, + proof_ledger, + applied_verdicts, + created_obligations, + ), + ensure_ascii=False, + sort_keys=True, + ), + flush=True, + ) print( f"[proof-ledger-result] id={proof_ledger.ledger_id} " f"version={proof_ledger.version} " diff --git a/tests/inference_engine/bench/test_autoresearch_supervisor.py b/tests/inference_engine/bench/test_autoresearch_supervisor.py index dda0468..a23b09a 100644 --- a/tests/inference_engine/bench/test_autoresearch_supervisor.py +++ b/tests/inference_engine/bench/test_autoresearch_supervisor.py @@ -7,6 +7,7 @@ render_candidate, should_keep, StrategyPrefillHeartbeat, + _pending_leaf_ids, validate_candidate, ) from pathlib import Path @@ -64,6 +65,13 @@ def test_keep_requires_novel_mathematical_advancement(): "proof_obligations_unresolved": 5, "metric_cold_critic_prefill_s": 900, }, baseline) + assert should_keep({ + "accepted": True, + "research_outcome": "DECOMPOSED", + "hypothesis_novel": True, + "proof_obligations_unresolved": 5, + "metric_cold_critic_prefill_s": 900, + }, baseline) assert not should_keep({ "accepted": True, "research_outcome": "INCONCLUSIVE", @@ -93,6 +101,30 @@ def test_parse_research_verdict_uses_last_complete_critic_block(): assert "admissible test functions" in verdict["new_frontier"] +def test_parse_host_generated_research_verdict_event(): + output = ( + '[autoresearch-verdict] {"candidate_id":"candidate-v4",' + '"outcome":"DECOMPOSED",' + '"evidence":"The Critic isolated a concrete missing convergence lemma.",' + '"new_frontier":"Prove locally uniform convergence on compact subsets."}' + ) + verdict = parse_research_verdict(output, "candidate-v4") + assert verdict["outcome"] == "DECOMPOSED" + + +def test_pending_leaf_ids_excludes_unresolved_parents(): + ledger = {"obligations": [ + {"obligation_id": "RH-C1", "status": "UNRESOLVED", "parent_id": ""}, + { + "obligation_id": "RH-C1-child", + "status": "UNRESOLVED", + "parent_id": "RH-C1", + }, + {"obligation_id": "RH-C2", "status": "UNRESOLVED", "parent_id": ""}, + ]} + assert _pending_leaf_ids(ledger) == ["RH-C1-child", "RH-C2"] + + def test_results_are_append_only_and_best_is_selected(tmp_path): path = tmp_path / "results.tsv" common = { @@ -201,6 +233,8 @@ def test_supervisor_predeploys_before_real_strategy_proposal(): ) assert "phase=predeploy-current" in body assert "phase=strategy-proposal real-gemma" in body + assert "if not gan_completed:" in body + assert "phase=completed-run-preserved" in body def test_gan_subprocess_output_is_streamed_not_captured(): diff --git a/tests/inference_engine/bridge/test_agent_gan_repl.py b/tests/inference_engine/bridge/test_agent_gan_repl.py index 18bc0fe..1e22890 100644 --- a/tests/inference_engine/bridge/test_agent_gan_repl.py +++ b/tests/inference_engine/bridge/test_agent_gan_repl.py @@ -23,6 +23,8 @@ is_runtime_artifact_prompt, consume_critic_issue_batch, apply_critic_verdicts, + build_autoresearch_verdict, + create_child_obligations, format_critic_issue_injection, format_proof_ledger, generator_issue_coverage, @@ -456,6 +458,62 @@ def test_proof_ledger_rejects_weak_or_missing_closure(): assert len(pending_obligations(ledger)) == 2 +def test_missing_lemma_creates_deduplicated_child_and_selects_leaf(): + ledger = ProofObligationLedger( + ledger_id="rh-ledger", + obligations=[ProofObligation("RH-C2", "Prove zero convergence.")], + ) + critic = """ +### ISSUE_VERDICT RH-C2 +**Status:** UNRESOLVED +**Evidence:** The proposed convergence argument does not control zeros on compact subsets. +**Missing lemma:** Prove locally uniform convergence on every compact subset of the critical strip. +""" + apply_critic_verdicts(ledger, critic, "br_first") + created = create_child_obligations( + ledger, + critic, + "br_first", + {"RH-C2"}, + ) + assert len(created) == 1 + assert created[0].parent_id == "RH-C2" + assert pending_obligations(ledger) == created + assert create_child_obligations( + ledger, + critic, + "br_repeat", + {"RH-C2"}, + ) == [] + + +def test_host_generated_autoresearch_verdict_uses_new_child_frontier(): + ledger = ProofObligationLedger( + ledger_id="rh-ledger", + obligations=[ + ProofObligation("RH-C2", "Prove zero convergence."), + ProofObligation( + "RH-C2-child", + "Prove locally uniform convergence on compact subsets.", + parent_id="RH-C2", + ), + ], + ) + + class Candidate: + CANDIDATE_ID = "candidate-v4" + TARGET_OBLIGATION_ID = "RH-C2" + + verdict = build_autoresearch_verdict( + Candidate, + ledger, + {"RH-C2": "UNRESOLVED"}, + [ledger.obligations[1]], + ) + assert verdict["outcome"] == "DECOMPOSED" + assert verdict["created_obligation_ids"] == ["RH-C2-child"] + + def test_recover_complete_checkpoint_from_timestamped_log(tmp_path): path = tmp_path / "agent.log" path.write_text(