[fix] Coalesce duplicate tool-call records and recover swallowed Pi errors - #5119
Merged
Conversation
…Pi error recovery dir
Three session-records bugs found by inspecting live record dumps:
1. Tool calls streamed as many partial-args snapshots each persisted as a
separate record (430 records for 18 calls). buildPersistingEmitter now
buffers one open tool-call slot, accumulating while a snapshot continues
the same id, and flushes once on a different step, a 3s idle TTL, or
turn drain. Flushed tool-family records (tool_call/tool_result/
interaction_request) carry a stable uuid5 id keyed on
(session, toolCallId, record_type), so a resumed/re-sent snapshot
upserts the same row instead of duplicating it.
2. A failed run (model/provider error) left no record at all — the error
only ever reached an OTel span or the terminal wire result, never an
AgentEvent. The engine now emits {type:"error"} through the run's own
sink at every point it already detects a failure, so it flows through
the persisting emitter (durable record) and the live stream uniformly.
3. That error detection for local Pi runs was reading the wrong directory:
findSwallowedPiError was called with plan.sourcePiAgentDir (the static
source login dir), but Pi's subprocess is pointed at a throwaway
per-run dir via PI_CODING_AGENT_DIR (prepareLocalPiAssets's return
value), which is where the swallowed error transcript actually lands.
Verified live: this alone was the root cause of the missing error
record for Pi runs.
Backend: record_id is now optional and upserted (ON CONFLICT DO UPDATE)
on (project_id, record_id), preserving record_index/created_at so an
overwrite never re-orders the transcript. record_id is no longer
time-ordered (uuid5/uuid4 replacing uuid7), so reads now order by
(created_at, record_index) instead.
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
|
The latest updates on your projects. Learn more about Vercel for GitHub.
|
dosubot
Bot
added
size:L
|
Important Review skippedAuto reviews are disabled on base/target branches other than the default branch. Please check the settings in the CodeRabbit UI or the ⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Pro Plus Run ID: You can disable this status message by setting the Use the checkbox below for a quick retry:
✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
Contributor
There was a problem hiding this comment.
Pull request overview
This PR improves session record durability and readability in Agenta’s runner + API by (1) coalescing streamed tool-call snapshots into a single persisted record, (2) ensuring run failures are emitted as durable error records (not only traces/terminal results), and (3) fixing local Pi “swallowed error” recovery by reading the correct per-run transcript directory.
Changes:
- Runner: buffer/coalesce
tool_callsnapshot streams (per tool-call id) into one persisted record, with a stable deterministicrecord_idfor tool-family events. - Runner: emit/persist
errorevents on swallowed Pi errors and outer run failures so the session inspector sees failures durably. - API: accept optional producer-supplied
record_id, upsert records on(project_id, record_id), and adjust read ordering away from time-ordered UUIDs.
Reviewed changes
Copilot reviewed 14 out of 14 changed files in this pull request and generated 3 comments.
Show a summary per file
| File | Description |
|---|---|
| services/runner/tests/unit/session-persist.test.ts | Adds unit tests for tool-call coalescing, TTL flush, drain flush, and stable ids. |
| services/runner/tests/unit/sandbox-agent-pi-error.test.ts | Regression test ensuring swallowed Pi errors are read from the per-run agent dir. |
| services/runner/src/tracing/otel.ts | Extends assistant span stamping to return an error message for failed assistant turns. |
| services/runner/src/sessions/record-id.ts | Introduces deterministic uuid5-based stable record ids for tool-family records. |
| services/runner/src/sessions/persist.ts | Implements tool-call snapshot coalescing + stable ids on ingest for tool-family events. |
| services/runner/src/server.ts | Persists an error record when run() throws outside the engine’s own error handling. |
| services/runner/src/engines/sandbox_agent.ts | Fixes local Pi transcript dir usage and emits error events for swallowed/outer errors. |
| api/oss/tests/pytest/unit/sessions/test_records_mapping_upsert.py | Adds API-side unit coverage for honoring stable ids, uuid4 fallback, and upsert behavior. |
| api/oss/src/dbs/postgres/sessions/records/mappings.py | Honors producer record_id when provided; otherwise mints uuid4. |
| api/oss/src/dbs/postgres/sessions/records/dbas.py | Switches record_id default to uuid4 and updates ordering semantics documentation. |
| api/oss/src/dbs/postgres/sessions/records/dao.py | Adds ON CONFLICT upsert on append and updates record ordering to (created_at, record_index). |
| api/oss/src/core/sessions/records/dtos.py | Adds optional record_id to the ingest DTO. |
| api/oss/src/apis/fastapi/sessions/router.py | Passes record_id from the ingest request into the DTO. |
| api/oss/src/apis/fastapi/sessions/models.py | Adds optional record_id to the FastAPI ingest request model. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
Comment on lines
365
to
369
| /** Fill an LLM span from a finished assistant message (model, tokens, finish, output). */ | ||
| function applyAssistant(span: Span, msg: any, capture: boolean): void { | ||
| /** Returns the error message when the assistant turn failed (stopReason/errorMessage), else | ||
| * undefined — so the caller can emit a matching `error` event, not just stamp the span. */ | ||
| function applyAssistant(span: Span, msg: any, capture: boolean): string | undefined { | ||
| if (msg.provider) span.setAttribute("gen_ai.system", msg.provider); |
| * substitute for a close signal the harness may never send (a call that streams then | ||
| * stalls without a `tool_result`). | ||
| */ | ||
| const OPEN_TOOL_TTL_MS = Number(process.env.AGENTA_RECORD_TOOL_TTL_MS ?? 3000); |
| RecordDBE.session_id == session_id, | ||
| ) | ||
| .order_by(RecordDBE.record_id.asc()) | ||
| .order_by(RecordDBE.created_at.asc(), RecordDBE.record_index.asc()) |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Context
Inspecting a live session's record dump surfaced three problems in how the runner persists agent turns:
{}and fills the args in incrementally). Each snapshot was persisted as its own record, so one session showed 430tool_callrows for 18 actual calls, with one call alone producing 39 rows.message→done, because the error only ever reached an OTel span or the terminal wire result, never anAgentEvent.findSwallowedPiError, which reads Pi's own transcript when Pi reports a failed call as a bareend_turn) was reading the wrong directory. It usedplan.sourcePiAgentDir(the static source login dir, e.g.~/.pi/agent), but Pi's subprocess is actually pointed at a throwaway per-run directory viaPI_CODING_AGENT_DIR(prepareLocalPiAssets's return value), which is where the transcript is really written. The static dir never has the transcript, so recovery silently failed on every local Pi error.Changes
Tool-call coalescing.
buildPersistingEmitternow buffers one open tool-call slot instead of persisting every snapshot. A same-id snapshot overwrites the buffered args and resets a 3s idle timer; the slot flushes to a single record when a different step arrives (a different tool, a message, the closingtool_result, etc.) or the timer fires (the safety net for a call that streams then stalls with no close signal). The live stream still receives every raw snapshot; only storage coalesces.Flushed tool-family records (
tool_call/tool_result/interaction_request) carry a stableuuid5id keyed on(session_id, toolCallId, record_type), using the sameuuid5(uuid5(NAMESPACE_DNS, "agenta"), "records")construction the meters domain already uses. A resumed or re-sent snapshot upserts the same row instead of duplicating it. Sincerecord_idis no longer time-ordered (uuid5/uuid4 replacing the previous uuid7 default), reads now order by(created_at, record_index)instead ofrecord_id.Before: 430
tool_callrecords for 18 calls (one call: 39 records, each a growing args snapshot).After: 1 record per call, upserted in place; the final row carries the last-seen args.
Error events. The engine now emits
{type: "error", message}through its own event sink at every point it already detects a run failure (the swallowed-Pi-error branch and the outer catch), beforefinish()flushes the sink. This routes the error through the same persisting emitter every other event uses, so it becomes a durable record and reaches the live stream, instead of being visible only on the OTel span or the terminal wire result.Pi transcript directory fix.
findSwallowedPiErroris now called withrunAgentDir ?? plan.sourcePiAgentDirinstead ofplan.sourcePiAgentDiralone, whererunAgentDiris the valueprepareLocalPiAssetsalready returns (the per-run dir it created and pointed Pi's subprocess at). This was the actual root cause of #2 for local Pi runs: the error existed,findSwallowedPiErrorjust couldn't see it.Backend:
record_idis now optional on ingest and upserted (ON CONFLICT DO UPDATE) on(project_id, record_id), overwriting the payload but preservingrecord_index/created_atso an overwrite never re-orders the transcript. The DBE default mint changed fromuuid7touuid4(a stable id has no reason to imply time-ordering it doesn't have).Tests / notes
session-persist.test.ts(coalesce-to-final-args, different-id flush, drain-flush for a paused call, TTL flush with fake timers, distinct stable ids for a call vs. its result) andsandbox-agent-pi-error.test.ts(regression proving the static source dir finds nothing while the actual per-run dir finds the swallowed error). Full suite:pnpm testandpnpm run typecheck, both green.test_records_mapping_upsert.py(honors a supplied stable id, uuid4 fallback, and asserts the compiled DAO statement is anON CONFLICT DO UPDATEthat never touchesrecord_index). Sessions + records unit suite green.message→donein the session inspector; after, the inspector shows theerrorrecord with the real provider message, confirmed via runner logs and a screenshot of the session inspector.(created_at, record_index)sort doesn't need one).