fix(codex): preserve semantic chunks through B3 merge

Persist validated Codex extraction responses as chunk files so the shared B3 merge preserves semantic nodes and edges.

Fail before cache and graph updates when extraction responses are missing or invalid.

Add regression coverage for chunk merging, cached reruns, and failure handling.

Fixes #4311

Co-authored-by: OpenAI Codex <codex@openai.com>
(cherry picked from commit da5850f0ac)

graphify/skill-codex.md

@@ -279,7 +279,11 @@ After all agents are dispatched, collect results sequentially in memory:
 result = wait_agent(handle); close_agent(handle)   # repeat per handle
 ```
 
-Parse each result as JSON. Accumulate nodes/edges/hyperedges across all results and write to `graphify-out/.graphify_semantic_new.json`. Codex collects in memory, so there are no per-chunk files on disk; the disk-based success checks in Step B3 do not apply — a chunk that returns invalid JSON is the failure signal instead.
+Collect each result in memory and associate it with the chunk number used for dispatch. Validate that every result is a JSON object with list-valued `nodes` and `edges`; `hyperedges` may be omitted or a list. Empty lists are valid extraction results.
+
+If any subagent fails, returns no result or invalid JSON, or a result cannot be persisted and read back from its expected chunk file, print the affected chunk numbers and stop before Step B3. Do not merge partial results, save cache, or continue to Part C; leave the existing graph and semantic cache untouched.
+
+Only after every result validates, write each result to its matching `graphify-out/.graphify_chunk_NN.json` file, using JSON serialization and the trusted chunk number. Preserve its `nodes`, `edges`, `hyperedges`, `input_tokens`, and `output_tokens` values. Verify that every expected chunk file exists and parses as JSON before continuing. Do not aggregate results or write `graphify-out/.graphify_semantic_new.json` in B2 — shared Step B3 performs the merge and cache update.
 
 Subagent prompt template:
 

tests/test_skill_codex_semantic_merge.py

@@ -0,0 +1,118 @@
+"""Regression coverage for Codex's in-memory B2 results and shared B3 merge."""
+from __future__ import annotations
+
+import json
+import re
+import subprocess
+import sys
+from pathlib import Path
+
+from tools.skillgen import gen
+
+
+def _codex_skill() -> str:
+    artifacts = gen.render_all(gen.load_platforms(), only="codex")
+    return next(a.content for a in artifacts if a.path == "graphify/skill-codex.md")
+
+
+def _python_blocks(body: str, start: str, end: str) -> list[str]:
+    section = body.split(start, 1)[1].split(end, 1)[0]
+    sources = []
+    for block in re.findall(r"```bash\n(.*?)```", section, re.S):
+        match = re.search(r' -c "\n(.*)\n"\s*$', block, re.S)
+        if match:
+            sources.append(match.group(1).replace('\\"', '"'))
+    return sources
+
+
+def _run(source: str, *, cwd: Path, corpus: Path, spec: Path) -> None:
+    source = source.replace("INPUT_PATH", corpus.as_posix()).replace("SPEC_PATH", spec.as_posix())
+    subprocess.run([sys.executable, "-c", source], cwd=cwd, check=True,
+                   capture_output=True, text=True)
+
+
+def test_codex_b2_persists_each_result_and_fails_closed():
+    skill = _codex_skill()
+    b2 = skill.split("**Step B2", 1)[1].split("**Step B3", 1)[0]
+
+    assert "write each result to its matching" in b2
+    assert "graphify-out/.graphify_chunk_NN.json" in b2
+    assert "Do not aggregate results or write `graphify-out/.graphify_semantic_new.json` in B2" in b2
+    assert "If any subagent fails, returns no result or invalid JSON" in b2
+    assert "stop before Step B3" in b2
+    assert "save cache" in b2
+    assert "leave the existing graph and semantic cache untouched" in b2
+    assert "read back from its expected chunk file" in b2
+    assert "Empty lists are valid extraction results" in b2
+
+
+def test_codex_chunk_results_survive_b3_merge_cache_and_all_cached_rerun(tmp_path):
+    skill = _codex_skill()
+    b0 = _python_blocks(skill, "**Step B0", "**Step B1")
+    b3 = _python_blocks(skill, "**Step B3", "#### Part C")
+    part_c = _python_blocks(skill, "#### Part C", "\n### ")[:1]
+    assert len(b0) == 1 and len(b3) == 3 and len(part_c) == 1
+
+    corpus = tmp_path / "corpus"
+    corpus.mkdir()
+    docs = [corpus / "one.md", corpus / "two.md"]
+    for doc in docs:
+        doc.write_text("semantic source", encoding="utf-8")
+    spec = tmp_path / "extraction-spec.md"
+    spec.write_text("# extraction spec\n", encoding="utf-8")
+    out = tmp_path / "graphify-out"
+    out.mkdir()
+    (out / ".graphify_detect.json").write_text(
+        json.dumps({"files": {"document": [doc.as_posix() for doc in docs]}}),
+        encoding="utf-8",
+    )
+    (out / ".graphify_ast.json").write_text(
+        json.dumps({"nodes": [{"id": "ast_node", "label": "AST"}], "edges": []}),
+        encoding="utf-8",
+    )
+
+    # B0 prepares this run. Codex has each successful in-memory response written
+    # to its own chunk file before the shared B3 merge consumes the files.
+    _run(b0[0], cwd=tmp_path, corpus=corpus, spec=spec)
+    chunks = [
+        {
+            "nodes": [{"id": "semantic_one", "label": "One", "source_file": docs[0].as_posix()}],
+            "edges": [{"source": "semantic_one", "target": "semantic_two", "relation": "references"}],
+            "hyperedges": [],
+            "input_tokens": 11,
+            "output_tokens": 5,
+        },
+        {
+            "nodes": [{"id": "semantic_two", "label": "Two", "source_file": docs[1].as_posix()}],
+            "edges": [],
+            "hyperedges": [],
+            "input_tokens": 13,
+            "output_tokens": 7,
+        },
+    ]
+    for index, chunk in enumerate(chunks, start=1):
+        (out / f".graphify_chunk_{index:02}.json").write_text(
+            json.dumps(chunk), encoding="utf-8")
+
+    for source in b3 + part_c:
+        _run(source, cwd=tmp_path, corpus=corpus, spec=spec)
+
+    semantic = json.loads((out / ".graphify_semantic.json").read_text(encoding="utf-8"))
+    assert {node["id"] for node in semantic["nodes"]} == {"semantic_one", "semantic_two"}
+    assert semantic["edges"] == chunks[0]["edges"]
+    assert semantic["input_tokens"] == 24
+    assert semantic["output_tokens"] == 12
+    extraction = json.loads((out / ".graphify_extract.json").read_text(encoding="utf-8"))
+    assert {node["id"] for node in extraction["nodes"]} == {
+        "ast_node", "semantic_one", "semantic_two"
+    }
+
+    # Cache writes from B3 must serve the next run without B2, while B0 removes
+    # any chunks left by the preceding run before the empty B3 merge.
+    for source in b0 + b3 + part_c:
+        _run(source, cwd=tmp_path, corpus=corpus, spec=spec)
+    assert not list(out.glob(".graphify_chunk_*.json"))
+    extraction = json.loads((out / ".graphify_extract.json").read_text(encoding="utf-8"))
+    assert {node["id"] for node in extraction["nodes"]} == {
+        "ast_node", "semantic_one", "semantic_two"
+    }

tests/test_skillgen.py

@@ -300,18 +300,18 @@ def test_windows_frontmatter_name_and_shell_and_extra():
     assert core.index("## Troubleshooting") < core.index("## Honesty Rules")
 
 
-def test_codex_dispatch_is_agenttask_and_collects_in_memory():
-    """codex: spawn/wait/close_agent dispatch needing multi_agent = true."""
+def test_codex_dispatch_is_agenttask_and_persists_chunk_results():
+    """codex uses native agent dispatch and writes results for shared B3."""
     core, _ = _platform_artifacts("codex")
     assert "spawn_agent" in core
     assert "wait_agent" in core
     assert "close_agent" in core
     assert "multi_agent = true" in core
-    assert "Codex collects in memory" in core
     # The B2 dispatch slot itself (Codex heading -> Step B3) must not carry the
     # claude Agent-tool example. The shared Step B3 prose mentions the agent type
     # in a re-run hint, so scope the check to the dispatch block only.
     b2 = core[core.index("**Step B2"):core.index("**Step B3")]
+    assert "graphify-out/.graphify_chunk_NN.json" in b2
     assert "Concrete example for 3 chunks" not in b2
     assert "Agent tool call 1" not in b2
 

tools/skillgen/expected/graphify__skill-codex.md

@@ -279,7 +279,11 @@ After all agents are dispatched, collect results sequentially in memory:
 result = wait_agent(handle); close_agent(handle)   # repeat per handle
 ```
 
-Parse each result as JSON. Accumulate nodes/edges/hyperedges across all results and write to `graphify-out/.graphify_semantic_new.json`. Codex collects in memory, so there are no per-chunk files on disk; the disk-based success checks in Step B3 do not apply — a chunk that returns invalid JSON is the failure signal instead.
+Collect each result in memory and associate it with the chunk number used for dispatch. Validate that every result is a JSON object with list-valued `nodes` and `edges`; `hyperedges` may be omitted or a list. Empty lists are valid extraction results.
+
+If any subagent fails, returns no result or invalid JSON, or a result cannot be persisted and read back from its expected chunk file, print the affected chunk numbers and stop before Step B3. Do not merge partial results, save cache, or continue to Part C; leave the existing graph and semantic cache untouched.
+
+Only after every result validates, write each result to its matching `graphify-out/.graphify_chunk_NN.json` file, using JSON serialization and the trusted chunk number. Preserve its `nodes`, `edges`, `hyperedges`, `input_tokens`, and `output_tokens` values. Verify that every expected chunk file exists and parses as JSON before continuing. Do not aggregate results or write `graphify-out/.graphify_semantic_new.json` in B2 — shared Step B3 performs the merge and cache update.
 
 Subagent prompt template:
 

tools/skillgen/fragments/dispatch/codex-agenttask.md

@@ -15,7 +15,11 @@ After all agents are dispatched, collect results sequentially in memory:
 result = wait_agent(handle); close_agent(handle)   # repeat per handle
 ```
 
-Parse each result as JSON. Accumulate nodes/edges/hyperedges across all results and write to `graphify-out/.graphify_semantic_new.json`. Codex collects in memory, so there are no per-chunk files on disk; the disk-based success checks in Step B3 do not apply — a chunk that returns invalid JSON is the failure signal instead.
+Collect each result in memory and associate it with the chunk number used for dispatch. Validate that every result is a JSON object with list-valued `nodes` and `edges`; `hyperedges` may be omitted or a list. Empty lists are valid extraction results.
+
+If any subagent fails, returns no result or invalid JSON, or a result cannot be persisted and read back from its expected chunk file, print the affected chunk numbers and stop before Step B3. Do not merge partial results, save cache, or continue to Part C; leave the existing graph and semantic cache untouched.
+
+Only after every result validates, write each result to its matching `graphify-out/.graphify_chunk_NN.json` file, using JSON serialization and the trusted chunk number. Preserve its `nodes`, `edges`, `hyperedges`, `input_tokens`, and `output_tokens` values. Verify that every expected chunk file exists and parses as JSON before continuing. Do not aggregate results or write `graphify-out/.graphify_semantic_new.json` in B2 — shared Step B3 performs the merge and cache update.
 
 Subagent prompt template: