[fix] Coalesce duplicate tool-call records and recover swallowed Pi errors

api/oss/src/apis/fastapi/sessions/models.py

@@ -1,5 +1,6 @@
 from datetime import datetime
 from typing import Any, Dict, List, Optional
+from uuid import UUID
 
 from pydantic import BaseModel, Field
 
@@ -186,6 +187,8 @@ class SessionMountsResponse(BaseModel):
 class SessionRecordIngestRequest(BaseModel):
     # project scope comes from the caller's credential, never the body
     session_id: str
+    # Optional stable id (uuid5) from the producer; absent when it has no stable key.
+    record_id: Optional[UUID] = None
     record_index: Optional[int] = None
     timestamp: Optional[datetime] = None
     record_type: Optional[str] = None

api/oss/src/apis/fastapi/sessions/router.py

@@ -576,8 +576,9 @@ async def ingest_record_event(
             organization_id=UUID(request.state.organization_id),
             project_id=UUID(project_id),
             record_event=SessionRecordEvent(
-                session_id=body.session_id,
                 project_id=UUID(project_id),
+                session_id=body.session_id,
+                record_id=body.record_id,
                 record_index=body.record_index,
                 timestamp=body.timestamp,
                 record_type=body.record_type,

api/oss/src/core/sessions/records/dtos.py

@@ -6,9 +6,10 @@
 
 
 class SessionRecordEvent(BaseModel):
-    session_id: str
     project_id: UUID
+    session_id: str
 
+    record_id: Optional[UUID] = None
     record_index: Optional[int] = None
     timestamp: Optional[datetime] = None
     record_type: Optional[str] = None

api/oss/src/dbs/postgres/sessions/records/dao.py

@@ -37,7 +37,16 @@ async def append(
                 if not (getattr(dbe, c.name) is None and c.server_default is not None)
             }
 
-            stmt = insert(RecordDBE).values(**values).returning(RecordDBE)
+            stmt = insert(RecordDBE).values(**values)
+            stmt = stmt.on_conflict_do_update(
+                index_elements=["project_id", "record_id"],
+                set_={
+                    "record_type": stmt.excluded.record_type,
+                    "record_source": stmt.excluded.record_source,
+                    "timestamp": stmt.excluded.timestamp,
+                    "attributes": stmt.excluded.attributes,
+                },
+            ).returning(RecordDBE)
             result = await session.execute(stmt)
             await session.commit()
 
@@ -59,7 +68,7 @@ async def get_records(
                     RecordDBE.project_id == project_id,
                     RecordDBE.session_id == session_id,
                 )
-                .order_by(RecordDBE.record_id.asc())
+                .order_by(RecordDBE.created_at.asc(), RecordDBE.record_index.asc())
             )
 
             dbes = (await session.execute(stmt)).scalars().all()

api/oss/src/dbs/postgres/sessions/records/dbas.py

@@ -7,21 +7,22 @@
 class RecordDBA:
     __abstract__ = True
 
-    # DB-minted uuid7 — records have no upstream id (unlike span_id/event_id).
-    # Time-ordered, so it doubles as the ordering key.
+    # Producer-supplied stable id (uuid5) where one exists, else a minted uuid4 fallback.
+    # Not time-ordered — ordering rides on record_index (see get_records), not this id.
     record_id = Column(
         UUID(as_uuid=True),
         nullable=False,
-        default=uuid.uuid7,
+        default=uuid.uuid4,
     )
 
     session_id = Column(
         String,
         nullable=False,
     )
 
-    # Producer-stamped per-session ordinal; not the ordering key (that is record_id),
-    # kept as a stable human-readable sequence from the producer.
+    # Producer-stamped per-turn ordinal and the in-session ordering key (record_id is
+    # no longer time-ordered). Restarts at 0 each cold turn, so reads tiebreak with
+    # created_at (ingest time) ahead of it — see get_records.
     record_index = Column(
         Integer,
         nullable=True,

api/oss/src/dbs/postgres/sessions/records/mappings.py

@@ -12,11 +12,12 @@ def map_record_event_to_dbe(
     event: SessionRecordEvent,
 ) -> RecordDBE:
     # The DAO inserts via an explicit insert().values(...), which bypasses the column's
-    # ORM-side default=uuid7; mint the pk here so it is never null at insert.
+    # ORM-side default; mint the pk here so it is never null at insert. Honor a producer
+    # stable id (uuid5) when supplied so retries/resumes upsert onto one row; else uuid4.
     return RecordDBE(
-        record_id=uuid.uuid7(),
         project_id=event.project_id,
         session_id=event.session_id,
+        record_id=event.record_id or uuid.uuid4(),
         record_index=event.record_index,
         timestamp=event.timestamp,
         record_type=event.record_type,

api/oss/tests/pytest/unit/sessions/test_records_mapping_upsert.py

@@ -0,0 +1,90 @@
+"""Unit tests for record id handling on ingest.
+
+Verifies:
+  - a producer-supplied stable record_id (uuid5) is honored verbatim so retries/resumes
+    upsert onto one row;
+  - an absent record_id falls back to a minted uuid4 (not uuid7 — record_id is no longer
+    the time-ordered key);
+  - the append DAO issues an ON CONFLICT DO UPDATE keyed on (project_id, record_id) that
+    overwrites the payload but preserves record_index.
+"""
+
+from uuid import UUID, uuid5, NAMESPACE_DNS
+
+from oss.src.core.sessions.records.dtos import SessionRecordEvent
+from oss.src.dbs.postgres.sessions.records.mappings import map_record_event_to_dbe
+
+
+_RECORDS_NS = uuid5(uuid5(NAMESPACE_DNS, "agenta"), "records")
+
+
+def _event(**over):
+    base = dict(
+        session_id="sess-1",
+        project_id=UUID("00000000-0000-0000-0000-0000000000aa"),
+        record_index=3,
+        record_type="tool_call",
+        record_source="agent",
+        attributes={"type": "tool_call", "input": {}},
+    )
+    base.update(over)
+    return SessionRecordEvent(**base)
+
+
+def test_supplied_record_id_is_honored():
+    stable = uuid5(_RECORDS_NS, "sess-1:call_1:tool_call")
+    dbe = map_record_event_to_dbe(event=_event(record_id=stable))
+    assert dbe.record_id == stable
+
+
+def test_absent_record_id_falls_back_to_uuid4():
+    dbe = map_record_event_to_dbe(event=_event())
+    assert isinstance(dbe.record_id, UUID)
+    # uuid4 — random, version 4 (not uuid7, which would imply time-ordering).
+    assert dbe.record_id.version == 4
+
+
+def test_append_upserts_preserving_index():
+    """The append statement must be an ON CONFLICT DO UPDATE on (project_id, record_id)
+    that overwrites attributes but does not touch record_index."""
+    from oss.src.dbs.postgres.sessions.records.dao import RecordsDAO
+
+    captured = {}
+
+    class _FakeResult:
+        def scalars(self):
+            class _S:
+                def first(_self):
+                    return None
+
+            return _S()
+
+    class _FakeSession:
+        async def execute(self, stmt):
+            captured["stmt"] = stmt
+            return _FakeResult()
+
+        async def commit(self):
+            pass
+
+    class _FakeEngine:
+        def session(self):
+            from contextlib import asynccontextmanager
+
+            @asynccontextmanager
+            async def _cm():
+                yield _FakeSession()
+
+            return _cm()
+
+    import asyncio
+
+    dao = RecordsDAO(engine=_FakeEngine())
+    asyncio.run(dao.append(event=_event()))
+
+    compiled = str(captured["stmt"]).lower()
+    assert "on conflict" in compiled
+    assert "do update" in compiled
+    # payload columns are overwritten; the ordinal is not in the update set.
+    assert "attributes" in compiled
+    assert "set record_index" not in compiled

services/runner/src/engines/sandbox_agent.ts

@@ -917,16 +917,18 @@ export async function runSandboxAgent(
     // (out-of-quota, bad key, rate limit, unknown model, ...), Pi's pi-acp bridge reports the
     // turn as a plain `end_turn` with NO content, so without this the run would return an
     // `ok:true` empty turn and the user would see a silent "No response" instead of the real
-    // failure. On the LOCAL Pi path the error is recoverable from Pi's own session transcript.
-    // Only checked when the turn produced no output and ran no tools (a real tool-only turn
-    // legitimately has empty text), and never on Daytona (the transcript lives in the remote
-    // sandbox).
+    // failure. On the LOCAL Pi path the error is recoverable from Pi's own session transcript —
+    // which lives under `runAgentDir` (the per-run throwaway dir Pi was actually pointed at via
+    // PI_CODING_AGENT_DIR), NOT `plan.sourcePiAgentDir` (the static source login dir, which has
+    // no transcripts). Only checked when the turn produced no output and ran no tools (a real
+    // tool-only turn legitimately has empty text), and never on Daytona (the transcript lives in
+    // the remote sandbox).
     const swallowedPiError =
       plan.isPi &&
       !plan.isDaytona &&
       !run.output().trim() &&
       !run.events().some((e) => e.type === "tool_call")
-        ? findSwallowedPiError(plan.sourcePiAgentDir, plan.cwd)
+        ? findSwallowedPiError(runAgentDir ?? plan.sourcePiAgentDir, plan.cwd)
         : undefined;
     let swallowedError: string | undefined;
     if (swallowedPiError) {
@@ -936,12 +938,15 @@ export async function runSandboxAgent(
         request.provider,
       );
       run.recordError(swallowedError, request.provider);
+      // Emit it as an event too (before finish() flushes the sink), so it reaches the live
+      // stream and the durable record, not only the trace span.
+      run.emitEvent({ type: "error", message: swallowedError });
     }
 
     const output = run.finish();
     await run.flush();
 
-    // Fail loud on the swallowed error detected above (A7 / "fail loud, not silent").
+    // Fail loud on the error detected above (A7 / "fail loud, not silent").
     if (swallowedError) {
       return { ok: false, error: swallowedError };
     }
@@ -970,6 +975,9 @@ export async function runSandboxAgent(
     // Stamp the error message + provider on the agent span before finishing it (F-030), so a
     // trace carries the same diagnostic the response does (it previously held only a count).
     otel?.recordError(error, request.provider);
+    // Also surface it as an event (before finish flushes the sink) so the error reaches the
+    // live stream and the durable record, not only the trace.
+    otel?.emitEvent({ type: "error", message: error });
     otel?.finish();
     await otel?.flush().catch(() => {});
     return {

services/runner/src/server.ts

@@ -239,6 +239,7 @@ async function runAndStream(
   // producer-side, independent of whether the client is still connected.
   let emitFn: EmitEvent = liveEmit;
   let flushPersist: (() => Promise<void>) | undefined;
+  let persistError: ((message: string) => void) | undefined;
   let aliveWatchdog: { release: () => Promise<void> } | undefined;
 
   if (sessionOwned) {
@@ -279,17 +280,24 @@ async function runAndStream(
     if (promptText) persist({ type: "message", text: promptText }, "user");
     emitFn = persistingEmit;
     flushPersist = flush;
+    persistError = (message) => persist({ type: "error", message }, "agent");
   }
 
   let result: AgentRunResult;
   try {
     result = await run(request, emitFn, controller.signal);
+    // A failed engine run ({ok:false}) already emitted its own error EVENT through the
+    // persisting emitter (see sandbox_agent.ts), so no extra persist here (it would
+    // duplicate the record).
     // Drain all queued persists before the sandbox tears down.
     if (flushPersist) await flushPersist();
   } catch (err) {
-    if (flushPersist) await flushPersist().catch(() => {});
     const message =
       err instanceof Error ? err.stack ?? err.message : String(err);
+    // A throw escaping run() itself (outside the engine's own try/catch) emitted no error
+    // event — persist it here as the backstop.
+    if (persistError) persistError(message);
+    if (flushPersist) await flushPersist().catch(() => {});
     result = { ok: false, error: message };
   } finally {
     // Release the alive lock and mark the stream row ended.