From 82414f5ad7d34d4db5bc1c6a6e9d9ea0456a6514 Mon Sep 17 00:00:00 2001 From: Juan Pablo Vega Date: Tue, 7 Jul 2026 14:03:19 +0200 Subject: [PATCH] fix(sessions): coalesce tool-call records, persist error events, fix Pi error recovery dir MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- api/oss/src/apis/fastapi/sessions/models.py | 3 + api/oss/src/apis/fastapi/sessions/router.py | 3 +- api/oss/src/core/sessions/records/dtos.py | 3 +- .../src/dbs/postgres/sessions/records/dao.py | 13 ++- .../src/dbs/postgres/sessions/records/dbas.py | 11 ++- .../dbs/postgres/sessions/records/mappings.py | 5 +- .../sessions/test_records_mapping_upsert.py | 90 +++++++++++++++++ services/runner/src/engines/sandbox_agent.ts | 20 ++-- services/runner/src/server.ts | 10 +- services/runner/src/sessions/persist.ts | 98 +++++++++++++++++-- services/runner/src/sessions/record-id.ts | 41 ++++++++ services/runner/src/tracing/otel.ts | 6 +- .../tests/unit/sandbox-agent-pi-error.test.ts | 26 +++++ .../runner/tests/unit/session-persist.test.ts | 85 ++++++++++++++++ 14 files changed, 389 insertions(+), 25 deletions(-) create mode 100644 api/oss/tests/pytest/unit/sessions/test_records_mapping_upsert.py create mode 100644 services/runner/src/sessions/record-id.ts diff --git a/api/oss/src/apis/fastapi/sessions/models.py b/api/oss/src/apis/fastapi/sessions/models.py index 1cc5ca59a0..06f3642b61 100644 --- a/api/oss/src/apis/fastapi/sessions/models.py +++ b/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 diff --git a/api/oss/src/apis/fastapi/sessions/router.py b/api/oss/src/apis/fastapi/sessions/router.py index f5716cc939..d022ee4c12 100644 --- a/api/oss/src/apis/fastapi/sessions/router.py +++ b/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, diff --git a/api/oss/src/core/sessions/records/dtos.py b/api/oss/src/core/sessions/records/dtos.py index 3c17d919c3..b725b03185 100644 --- a/api/oss/src/core/sessions/records/dtos.py +++ b/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 diff --git a/api/oss/src/dbs/postgres/sessions/records/dao.py b/api/oss/src/dbs/postgres/sessions/records/dao.py index 7de476ff6c..1449d5d194 100644 --- a/api/oss/src/dbs/postgres/sessions/records/dao.py +++ b/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() diff --git a/api/oss/src/dbs/postgres/sessions/records/dbas.py b/api/oss/src/dbs/postgres/sessions/records/dbas.py index 396e28151c..1345bd2bde 100644 --- a/api/oss/src/dbs/postgres/sessions/records/dbas.py +++ b/api/oss/src/dbs/postgres/sessions/records/dbas.py @@ -7,12 +7,12 @@ 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( @@ -20,8 +20,9 @@ class RecordDBA: 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, diff --git a/api/oss/src/dbs/postgres/sessions/records/mappings.py b/api/oss/src/dbs/postgres/sessions/records/mappings.py index 6ac28ab1fe..e38a1570de 100644 --- a/api/oss/src/dbs/postgres/sessions/records/mappings.py +++ b/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, diff --git a/api/oss/tests/pytest/unit/sessions/test_records_mapping_upsert.py b/api/oss/tests/pytest/unit/sessions/test_records_mapping_upsert.py new file mode 100644 index 0000000000..b4e2c7bccd --- /dev/null +++ b/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 diff --git a/services/runner/src/engines/sandbox_agent.ts b/services/runner/src/engines/sandbox_agent.ts index cd8d9048af..5198023ee3 100644 --- a/services/runner/src/engines/sandbox_agent.ts +++ b/services/runner/src/engines/sandbox_agent.ts @@ -906,16 +906,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) { @@ -925,12 +927,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 }; } @@ -959,6 +964,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 { diff --git a/services/runner/src/server.ts b/services/runner/src/server.ts index 4485e8c0d8..b74244213d 100644 --- a/services/runner/src/server.ts +++ b/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) | undefined; + let persistError: ((message: string) => void) | undefined; let aliveWatchdog: { release: () => Promise } | 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. diff --git a/services/runner/src/sessions/persist.ts b/services/runner/src/sessions/persist.ts index fea91344f5..5ae8a2cf67 100644 --- a/services/runner/src/sessions/persist.ts +++ b/services/runner/src/sessions/persist.ts @@ -21,6 +21,7 @@ import { apiBase } from "../apiBase.ts"; import type { AgentEvent } from "../protocol.ts"; +import { stableRecordId } from "./record-id.ts"; const INGEST_MAX_RETRIES = 3; const INGEST_RETRY_BASE_MS = 100; @@ -41,6 +42,7 @@ async function postEvent( event: AgentEvent, eventIndex: number, sender: string, + recordId?: string, ): Promise { const url = `${apiBase()}/sessions/records/ingest`; let lastErr: unknown; @@ -54,6 +56,9 @@ async function postEvent( }, body: JSON.stringify({ session_id: sessionId, + // Present only for tool-family records (stable uuid5); the backend mints a + // uuid4 when omitted. A re-sent id upserts the same row. + ...(recordId ? { record_id: recordId } : {}), record_index: eventIndex, timestamp: new Date().toISOString(), record_source: sender, @@ -85,9 +90,10 @@ export function persistEvent( event: AgentEvent, eventIndex: number, sender: string = "agent", + recordId?: string, ): void { const tail = (persistChains.get(sessionId) ?? Promise.resolve()).then(() => - postEvent(sessionId, auth, event, eventIndex, sender), + postEvent(sessionId, auth, event, eventIndex, sender, recordId), ); persistChains.set(sessionId, tail); } @@ -107,14 +113,26 @@ export async function drainPersist(sessionId: string): Promise { } } +/** + * A tool call streams as many `tool_call` events with a growing partial-args snapshot for + * one id. Idle window after which an open, un-closed tool call is flushed as-is — the + * 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); + /** * Build an emitter that persists every event via the ingest chain AND calls the * original emitter (for live streaming). Returns a stateful counter so record_index - * increments monotonically per session (a stable ordinal; the DB orders by record_id). + * increments per turn (the in-session ordering key; the DB tiebreaks with ingest time). * - * The `stripReplay` filter coalesces the delta family (message_start / message_delta - * / message_end) into a single `message` event for storage; the raw deltas are - * forwarded to the live emitter unchanged. This mirrors the PoC's coalescing logic. + * Coalescing keeps one durable record per streamed family, while the live stream gets + * every raw event unchanged: + * - message_start/delta/end and thought_* accumulate text, persisted once on *_end. + * - tool_call snapshots for one id accumulate (latest args win) into a single open slot, + * persisted once when a non-continuation event arrives, the TTL fires, or the turn + * drains. The record carries a stable uuid5 id so a re-sent snapshot (or a resume) + * upserts the same row rather than appending. */ export function buildPersistingEmitter( sessionId: string, @@ -134,10 +152,55 @@ export function buildPersistingEmitter( { id: string; text: string } >(); + // At most one open tool call at a time: its index is claimed when the call first + // appears (so it sorts ahead of whatever flushes it), args are overwritten in place + // while snapshots for the same id keep arriving, and it is persisted exactly once. + let openTool: + | { id: string; index: number; event: AgentEvent; timer: NodeJS.Timeout } + | null = null; + + const flushOpenTool = (): void => { + if (!openTool) return; + const { id, index, event, timer } = openTool; + clearTimeout(timer); + openTool = null; + persistEvent( + sessionId, + auth, + event, + index, + "agent", + stableRecordId(sessionId, id, "tool_call"), + ); + }; + const emit = (event: AgentEvent): void => { // Always forward to the live stream (if any). liveEmit?.(event); + // Accumulate tool_call snapshots for one id; flush on any non-continuation below. + if (event.type === "tool_call" && event.id) { + if (openTool && openTool.id === event.id) { + // Continuation: latest args win, push the idle deadline out. + openTool.event = event; + clearTimeout(openTool.timer); + openTool.timer = setTimeout(flushOpenTool, OPEN_TOOL_TTL_MS); + return; + } + // A different call: flush the previous open slot, then open this one. + flushOpenTool(); + openTool = { + id: event.id, + index: eventIndex++, + event, + timer: setTimeout(flushOpenTool, OPEN_TOOL_TTL_MS), + }; + return; + } + // Any other event is a "different step": close the open tool call before it, so the + // tool_call record lands (with its earlier index) ahead of this event. + flushOpenTool(); + // Coalesce delta families: accumulate text; persist only on *_end. if (event.type === "message_start") { coalescedMessages.set(event.id, { id: event.id, text: "" }); @@ -190,15 +253,38 @@ export function buildPersistingEmitter( } } + // A tool_result / interaction_request carries the same tool-call id as its tool_call; + // give it its own stable id (keyed on the record type) so it lands on a distinct row. + if ( + (event.type === "tool_result" || event.type === "interaction_request") && + event.id + ) { + persistEvent( + sessionId, + auth, + event, + eventIndex++, + "agent", + stableRecordId(sessionId, event.id, event.type), + ); + return; + } + // All other events persist as-is. persistEvent(sessionId, auth, event, eventIndex++); }; const persist = (event: AgentEvent, sender: string): void => { + // Out-of-band records (the inbound user turn) still respect open-tool ordering. + flushOpenTool(); persistEvent(sessionId, auth, event, eventIndex++, sender); }; - const flush = (): Promise => drainPersist(sessionId); + const flush = (): Promise => { + // A paused call ends the turn with its slot still open — persist it before draining. + flushOpenTool(); + return drainPersist(sessionId); + }; return { emit, persist, flush }; } diff --git a/services/runner/src/sessions/record-id.ts b/services/runner/src/sessions/record-id.ts new file mode 100644 index 0000000000..6ea5451be5 --- /dev/null +++ b/services/runner/src/sessions/record-id.ts @@ -0,0 +1,41 @@ +import { createHash } from "node:crypto"; + +// RFC 4122 DNS namespace — the root the Python side also starts from (uuid.NAMESPACE_DNS). +const NAMESPACE_DNS = "6ba7b810-9dad-11d1-80b4-00c04fd430c8"; + +function uuidToBytes(uuid: string): Buffer { + return Buffer.from(uuid.replace(/-/g, ""), "hex"); +} + +/** RFC 4122 uuid5 (SHA-1) of `name` under `namespace`. */ +function uuid5(name: string, namespace: string): string { + const hash = createHash("sha1") + .update(uuidToBytes(namespace)) + .update(name, "utf8") + .digest(); + const bytes = hash.subarray(0, 16); + bytes[6] = (bytes[6] & 0x0f) | 0x50; // version 5 + bytes[8] = (bytes[8] & 0x3f) | 0x80; // RFC 4122 variant + const hex = bytes.toString("hex"); + return `${hex.slice(0, 8)}-${hex.slice(8, 12)}-${hex.slice(12, 16)}-${hex.slice(16, 20)}-${hex.slice(20)}`; +} + +// Project-wide root uuid5(NAMESPACE_DNS, "agenta"), sub-namespaced under "records" — the +// same construction the API uses for other domains (e.g. meters). Deriving a record's id as +// uuid5(this, key) makes it deterministic, so every streamed snapshot of one tool call — and +// a resume that re-announces it — maps to one id and upserts onto one row. Non-stable records +// send no id; the backend mints a uuid4 fallback. +const RECORD_NAMESPACE = uuid5("records", uuid5("agenta", NAMESPACE_DNS)); + +/** + * Stable record id for a tool-family record, keyed on (session, tool-call id, type). The + * `type` is part of the key so a `tool_call` and its closing `tool_result` — which share a + * tool-call id — land on two distinct rows instead of overwriting each other. + */ +export function stableRecordId( + sessionId: string, + toolCallId: string, + recordType: string, +): string { + return uuid5(`${sessionId}:${toolCallId}:${recordType}`, RECORD_NAMESPACE); +} diff --git a/services/runner/src/tracing/otel.ts b/services/runner/src/tracing/otel.ts index 515c2f03ef..1a4ed868b1 100644 --- a/services/runner/src/tracing/otel.ts +++ b/services/runner/src/tracing/otel.ts @@ -363,7 +363,9 @@ function lastAssistantText(messages: any): string { } /** 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); if (msg.model) span.setAttribute("gen_ai.request.model", msg.model); if (msg.responseModel || msg.model) @@ -401,7 +403,9 @@ function applyAssistant(span: Span, msg: any, capture: boolean): void { emitMessages(span, "llm.output_messages", [msg], capture); if (msg.stopReason === "error" || msg.errorMessage) { span.setStatus({ code: SpanStatusCode.ERROR, message: msg.errorMessage }); + return String(msg.errorMessage || "agent run failed"); } + return undefined; } // --------------------------------------------------------------------------- diff --git a/services/runner/tests/unit/sandbox-agent-pi-error.test.ts b/services/runner/tests/unit/sandbox-agent-pi-error.test.ts index d9cf4c9410..530fbd74e5 100644 --- a/services/runner/tests/unit/sandbox-agent-pi-error.test.ts +++ b/services/runner/tests/unit/sandbox-agent-pi-error.test.ts @@ -121,4 +121,30 @@ describe("findSwallowedPiError", () => { const piAgentDir = tempDir(); assert.equal(findSwallowedPiError(piAgentDir, "/tmp/whatever"), undefined); }); + + it("finds the error in the per-run agent dir Pi was actually pointed at, not the static source dir", () => { + // Regression: a run that materializes skills/system-prompt gets a throwaway per-run Pi + // agent dir (prepareLocalAgentDir's return value) — the engine must read the swallowed + // error from THAT dir (where Pi, pointed there via PI_CODING_AGENT_DIR, wrote its + // transcript), not from the static source login dir, which never has the transcript. + const sourceAgentDir = tempDir(); // e.g. ~/.pi/agent — never receives transcripts + const runAgentDir = tempDir(); // the throwaway dir prepareLocalAgentDir returns + const cwd = "/tmp/agenta-sandbox-agent-run1"; + writeTranscript(runAgentDir, "--tmp-agenta-sandbox-agent-run1--", cwd, [ + { + type: "message", + message: { + role: "assistant", + content: [], + stopReason: "error", + errorMessage: "insufficient credit", + }, + }, + ]); + + // The bug: passing the static source dir finds nothing. + assert.equal(findSwallowedPiError(sourceAgentDir, cwd), undefined); + // The fix: passing the actual per-run dir Pi wrote to finds the error. + assert.equal(findSwallowedPiError(runAgentDir, cwd), "insufficient credit"); + }); }); diff --git a/services/runner/tests/unit/session-persist.test.ts b/services/runner/tests/unit/session-persist.test.ts index 72217175dd..5db1789fc7 100644 --- a/services/runner/tests/unit/session-persist.test.ts +++ b/services/runner/tests/unit/session-persist.test.ts @@ -136,6 +136,91 @@ describe("buildPersistingEmitter", () => { ); assert.deepEqual(indices, [0, 1, 2]); }); + + it("coalesces tool_call snapshots for one id into a single record with final args", async () => { + const live: unknown[] = []; + const { emit, flush } = buildPersistingEmitter( + "sess-tc", + () => "Secret t", + (e) => live.push(e), + ); + + // A tool call streams a growing partial-args snapshot for one id. + emit({ type: "tool_call", id: "call_1", name: "bash", input: {} }); + emit({ type: "tool_call", id: "call_1", name: "bash", input: { command: "fi" } }); + emit({ type: "tool_call", id: "call_1", name: "bash", input: { command: "find ." } }); + // A different step closes the open call. + emit({ type: "tool_result", id: "call_1", output: "ok" }); + emit({ type: "done" }); + await flush(); + + // Live stream sees every raw snapshot. + assert.equal(live.length, 5); + // Storage sees one tool_call (final args) + one tool_result + done. + const bodies = postedBodies as Array>; + const types = bodies.map( + (b) => (b["attributes"] as Record)["type"], + ); + assert.deepEqual(types, ["tool_call", "tool_result", "done"]); + const call = bodies[0]["attributes"] as Record; + assert.deepEqual(call["input"], { command: "find ." }); + // tool_call is stamped with a stable id and keeps the earlier index (ahead of result). + assert.ok(typeof bodies[0]["record_id"] === "string"); + assert.equal(bodies[0]["record_index"], 0); + assert.equal(bodies[1]["record_index"], 1); + // tool_call and its tool_result get distinct stable ids (keyed on the record type). + assert.notEqual(bodies[0]["record_id"], bodies[1]["record_id"]); + }); + + it("a different tool id flushes the previous open call", async () => { + const { emit, flush } = buildPersistingEmitter("sess-tc2", () => "Secret t"); + + emit({ type: "tool_call", id: "call_a", name: "read", input: { path: "/x" } }); + emit({ type: "tool_call", id: "call_b", name: "read", input: { path: "/y" } }); + await flush(); + + const bodies = postedBodies as Array>; + const inputs = bodies.map( + (b) => (b["attributes"] as Record)["input"], + ); + assert.deepEqual(inputs, [{ path: "/x" }, { path: "/y" }]); + assert.deepEqual(bodies.map((b) => b["record_index"]), [0, 1]); + }); + + it("flushes an open (paused) tool_call on drain", async () => { + const { emit, flush } = buildPersistingEmitter("sess-tc3", () => "Secret t"); + + // A paused call ends the turn with its slot still open. + emit({ type: "tool_call", id: "call_p", name: "bash", input: { command: "ls" } }); + await flush(); + + const bodies = postedBodies as Array>; + assert.equal(bodies.length, 1); + assert.equal( + (bodies[0]["attributes"] as Record)["type"], + "tool_call", + ); + }); + + it("flushes an open tool_call when the idle TTL fires", async () => { + vi.useFakeTimers(); + try { + const { emit, flush } = buildPersistingEmitter("sess-tc4", () => "Secret t"); + + emit({ type: "tool_call", id: "call_ttl", name: "bash", input: { command: "x" } }); + // Nothing follows; only the TTL can close it. + assert.equal(postedBodies.length, 0); + await vi.advanceTimersByTimeAsync(3000); + assert.equal(postedBodies.length, 1); + assert.equal( + ((postedBodies[0] as Record)["attributes"] as Record)["type"], + "tool_call", + ); + await flush(); + } finally { + vi.useRealTimers(); + } + }); }); describe("drainPersist", () => {