fix(ingest): fold attachment bodies in a batch onto their prompt

backend/app/main.py

@@ -959,6 +959,27 @@ async def _ingest_body(
     return body, None
 
 
+def _ingest_batch(source: str, bodies: list[dict[str, Any]]) -> list[dict[str, Any]]:
+    """Store a batch in order. Attachment bodies fold onto their prompt, as
+    they do sent alone (a replayed spool can carry one); each run of other
+    bodies between them is one write transaction."""
+    results: list[dict[str, Any]] = []
+    run: list[tuple[str, dict[str, Any]]] = []
+    for body in bodies:
+        if not body.get("_attach_to_prompt"):
+            run.append((source, body))
+            continue
+        if run:
+            results.extend(db.record_ingests(run))
+            run = []
+        sid = str(body.get("session_id") or "")
+        attached = db.attach_to_prompt(sid, body.get("text"), body.get("attachments") or [], body.get("timestamp"))
+        results.append({"ok": True, "attached": attached, "session_id": sid})
+    if run:
+        results.extend(db.record_ingests(run))
+    return results
+
+
 @app.post("/v1/ingest/{source}")
 async def ingest(source: str, request: Request) -> dict[str, Any]:
     if source not in AGENTS:
@@ -976,7 +997,7 @@ async def ingest(source: str, request: Request) -> dict[str, Any]:
         return result
 
     if isinstance(body, list):
-        results = await asyncio.to_thread(db.record_ingests, [(source, b) for b in body])
+        results = await asyncio.to_thread(_ingest_batch, source, body)
         for r in results:
             _bump_revision(r.get("session_id"))
         return {"ok": True, "results": results}

backend/tests/test_ingest_batch.py

@@ -45,3 +45,27 @@ def test_single_body_ingest_still_returns_its_result(fresh_db: Path):
     )
     assert resp.status_code == 200
     assert resp.json()["raw_status"] == "projected"
+
+
+def test_batch_folds_attachments_onto_their_prompt(fresh_db: Path):
+    """A replayed spool can carry an attachment body inside a batch; it must
+    fold onto its prompt as it does sent alone, not become an event."""
+    client = TestClient(main.app, base_url="http://127.0.0.1")
+    body = [
+        {"session_id": "att1", "hook_event_name": "UserPromptSubmit", "prompt": "see this",
+         "timestamp": "2026-07-01T00:00:00Z"},
+        {"session_id": "att1", "_attach_to_prompt": True, "text": "see this",
+         "attachments": [{"type": "image", "name": "shot.png"}], "timestamp": "2026-07-01T00:00:00Z"},
+        {"session_id": "att1", "hook_event_name": "UserPromptSubmit", "prompt": "next",
+         "timestamp": "2026-07-01T00:01:00Z"},
+    ]
+    resp = client.post("/v1/ingest/claude", json=body)
+    assert resp.status_code == 200
+    results = resp.json()["results"]
+    assert len(results) == 3 and results[1]["attached"]
+    with store.read() as conn:
+        rows = conn.execute(
+            "SELECT category, detail, attachments FROM events WHERE session_id = 'att1' ORDER BY ts"
+        ).fetchall()
+    assert [r["detail"] for r in rows if r["category"] == "prompt"] == ["see this", "next"]
+    assert "shot.png" in (rows[0]["attachments"] or "")