sorokin.oleg.edu/learnharness
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 6 Wiki Activity

fix: preserve failed responses when retrying model calls

Browse Source 
Keep failed attempts in the saved transcript and omit only trailing failures from Retry's temporary agent history. Use the matching temporary prefix length when persisting genuinely new output.

Preserve the error notice across plan/compaction markers, refuse Retry after later compaction, and avoid synthesizing results for tool calls inside failed generations while retaining Pi's native filtering.

Verified: 173 focused Retry/session/compaction-projection/practice-record tests, typecheck, changed-file lint, and diff --check. No app launch or real provider calls.
This commit is contained in:
amosblomqvist
2026-10-27 16:49:29 +02:00
parent a572675900
commit 82c6999a63
7 changed files with 559 additions and 200 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/core/input-events.ts
+42 -25
View File
@@ -16,9 +16,9 @@
* continue — run the next turn on the current record (mount
* continuations, job-settle re-syncs: the record ends at an
* answered question with no teacher reaction yet).
* retry — re-run a turn whose model call failed (issue #110): the
* trailing failed assistant message(s) leave the record and
* the turn re-runs from the input that preceded them.
* retry — retry a failed model response (issue #110), without new
* input. Failed attempts stay in the record; only the retry's
* temporary agent history omits the trailing failures.
*
* (The plan verdict input died 2026-09-26 with the verdict itself — the
* learner's post-plan reply is a natural text message.)
@@ -218,8 +218,8 @@ export function inputAppends(
case "continue":
return dismissalConclusion(record);
case "retry":
// The failed turn's own input already landed write-ahead; the
// handler trims the failure (retryDrops) — nothing new appends.
// The input already landed write-ahead. retryDrops affects only
// the temporary agent history — nothing new appends here.
return [];
case "thinking": {
const active = activeProblemOf(record.messages);
@@ -244,40 +244,57 @@ export function inputAppends(
}
}
/** Does the record end in a failed model call (a retryable turn)? */
export function endsInFailedTurn(messages: readonly LhMessage[]): boolean {
return failedTail(messages).ids.size > 0;
/** The failure owning the ephemeral banner. Plan presentation and
* compaction do not resolve an error; substantive later input/output does.
* This is deliberately broader than Retry eligibility below. */
export function failedResponseAtEnd(messages: readonly LhMessage[]): Extract<LhMessage, { role: "assistant" }> | undefined {
for (let i = messages.length - 1; i >= 0; i--) {
const m = messages[i]!;
if (m.role === "custom" && (m.customType === "plan-block" || m.customType === "compaction")) continue;
return m.role === "assistant" && m.stopReason === "error" ? m : undefined;
}
return undefined;
}
/** The ids a retry drops (issue #110, pure): the failed assistant
* message(s) ending the record. A provider failure persists as an
* assistant message with stopReason "error", and pi refuses to continue
* from an assistant — so neither re-answering (the question is no longer
* dangling) nor `continue` can resume the turn. Ids, not a cut index:
* UI-only markers may follow the failure, and they stay. Honest 400s
* when there is no failure to retry, or when dropping it leaves nothing
* to continue from. */
/** Does the record end in a failure with input that can be retried?
* Aborts are deliberate stops; compaction after a failure requires new
* learner input rather than a retry using a changed context. */
export function endsInFailedTurn(messages: readonly LhMessage[]): boolean {
const { ids, before } = failedTail(messages);
return ids.size > 0 && before !== undefined && before.role !== "assistant";
}
/** The ids omitted ONLY from the retry's temporary agent history.
* Failed attempts (including partial text and their ids) stay on disk.
* Pi's provider conversion already skips error/aborted responses, but
* continue() rejects an assistant tail BEFORE conversion. Omitting the
* trailing failures lets that continuation start from the saved input or
* completed tool results without replaying tools or resubmitting input.
* Plan blocks may follow the failure and stay; later compaction blocks
* Retry without changing compaction itself. */
export function retryDrops(messages: readonly LhMessage[]): Set<string> {
const { ids, before } = failedTail(messages);
if (ids.size === 0) throw new ChatInputError("retry input requires a failed turn to retry");
if (ids.size === 0) {
const failure = failedResponseAtEnd(messages);
if (failure) {
throw new ChatInputError(`${failure.errorMessage ?? "An error occurred."}\n\nThis response cannot be retried after compaction. Send a new message to continue.`);
}
throw new ChatInputError("retry input requires a failed turn to retry");
}
if (!before || before.role === "assistant") throw new ChatInputError("retry input leaves no input to continue from");
return ids;
}
/** Markers that can land AFTER a failed call and are never model input:
* persistTurn's plan block (plan_init succeeded, the next call failed)
* and the settle kick's compaction marker. */
const TRAILING_MARKERS: readonly string[] = ["plan-block", "compaction"];
/** Walk back from the end over failed assistants and trailing markers:
* the failures' ids, and the message the run stops at. */
/** Walk back over failed assistants and UI-only plan blocks. A compaction
* is a boundary, not a marker Retry may cross. A failure AFTER a compaction
* can still retry: that compacted context was already its starting state. */
function failedTail(messages: readonly LhMessage[]): { ids: Set<string>; before: LhMessage | undefined } {
const ids = new Set<string>();
let i = messages.length - 1;
for (; i >= 0; i--) {
const m = messages[i]!;
if (m.role === "assistant" && m.stopReason === "error") ids.add(m.id);
else if (!(m.role === "custom" && TRAILING_MARKERS.includes(m.customType))) break;
else if (!(m.role === "custom" && m.customType === "plan-block")) break;
}
return { ids, before: messages[i] };
}
src/core/native-messages.ts
+8 -6
View File
@@ -551,9 +551,10 @@ function resultIdsFrom(messages: readonly AgentMessage[]): Set<string> {
* - markers flatten to a user message carrying their content (the
* canonical text single-sourced in core/contracts); plan blocks are
* UI-only and filtered.
* - dangling tool calls (an aborted turn's unanswered questions) get the
* synthesized interrupted result — the provider protocol's hard
* invariant: a tool call must be followed by its result.
* - dangling calls from successfully generated assistant messages (for
* example, an unanswered question) get a synthesized interrupted result.
* Error/aborted generations do not: Pi drops those assistant messages,
* so manufacturing their results would leave orphan results on the wire.
*
* Contract (pi-agent-core): never throws — an unknown shape passes
* through untouched rather than killing the loop.
@@ -619,11 +620,12 @@ export function toLlmMessages(messages: readonly AgentMessage[]): Message[] {
// Strip the record id: the wire carries the native shape only.
const { id: _id, ...native } = m as LhMessage;
out.push(native as Message);
// Dangling-call synthesis: an assistant message whose toolCalls have
// no results in the remaining tail (abort mid-question) — the
// provider rejects unanswered calls outright.
// Repair unanswered calls from completed generations, not calls in
// failed generations. Pi filters error/aborted assistants itself;
// adding results for their unexecuted calls would leave orphans.
if (role === "assistant") {
const assistant = m as AssistantMessage;
if (assistant.stopReason === "error" || assistant.stopReason === "aborted") continue;
const calls = assistant.content.filter((b) => b.type === "toolCall");
if (calls.length === 0) continue;
const answered = resultIdsFrom(messages.slice(i + 1));
src/server/chat-handler.ts
+28 -25
View File
@@ -187,6 +187,17 @@ export async function handleChatInput(
return Response.json({ error: "Unknown chat" }, { status: 404 });
}
// Refuse an ineligible Retry before catalog refresh, compaction, or any
// turn/write effects. Revalidate after the awaited gates below too.
if (input.type === "retry") {
try {
retryDrops(record.messages);
} catch (err) {
if (err instanceof ChatInputError) return Response.json({ error: err.message }, { status: 400 });
throw err;
}
}
// THE UNIVERSAL TURN GATE (BYOK, owner, 2026-09-22): a turn needs the
// CONVERSATION chain — the chat's own model override over the strong
// slot — and there is no factory floor; when both are empty the turn
@@ -224,18 +235,6 @@ export async function handleChatInput(
return Response.json(envelope, { status: 503 });
}
// THE RETRY GATE (issue #110): a retry with no failure to retry
// refuses here, BEFORE compaction and beginTurn — no effects. The
// trim itself re-validates on the record compaction re-reads below.
if (input.type === "retry") {
try {
retryDrops(record.messages);
} catch (err) {
if (err instanceof ChatInputError) return Response.json({ error: err.message }, { status: 400 });
throw err;
}
}
// THE COMPACTION GATE (owner round 14, 2026-10-05 — the structural
// ruling: NOTHING model-facing for this chat interleaves with a
// running compaction): await any in-flight compaction (the settle
@@ -274,18 +273,17 @@ export async function handleChatInput(
const compactedRecord = stores.chatStore.getChat(chatId);
if (!compactedRecord) return Response.json({ error: "Unknown chat" }, { status: 404 });
record = compactedRecord;
// THE RETRY TRIM: recomputed on the current record (the awaits above
// let other writers land), still before beginTurn. The trimmed record
// reaches disk through the write-ahead.
// Recompute Retry eligibility on current truth, still before beginTurn.
// A compaction marker or another writer may have landed during awaits.
// These omissions affect ONLY the agent context, never the saved record.
let retryOmissions = new Set<string>();
if (input.type === "retry") {
let drop: Set<string>;
try {
drop = retryDrops(record.messages);
retryOmissions = retryDrops(record.messages);
} catch (err) {
if (err instanceof ChatInputError) return Response.json({ error: err.message }, { status: 400 });
throw err;
}
record = { ...record, messages: record.messages.filter((m) => !drop.has(m.id)) };
}
// The input's native appends (pure; the honest 400 for an answer with no
@@ -344,8 +342,13 @@ export async function handleChatInput(
// ride this same write-ahead - the pump de-queued them at fire, so this
// is the ONLY copy ever appended.)
const merged = [...record.messages, ...appends];
// Retry preserves every persisted attempt. Pi needs a non-assistant
// tail to start continue(), so its temporary history omits the trailing
// failures only. Count that SAME temporary prefix at settlement, or the
// retained failures would make persistTurn skip genuinely new output.
const agentMessages = input.type === "retry" ? merged.filter((m) => !retryOmissions.has(m.id)) : merged;
const stagedAutomatic = isSubscriptionProvider(conversation.provider) && input.type === "steers";
let persistBaseCount = stagedAutomatic ? record.messages.length : merged.length;
let persistBaseCount = stagedAutomatic ? record.messages.length : agentMessages.length;
try {
if (!stagedAutomatic) stores.chatStore.saveMessages(chatId, merged);
} catch (err) {
@@ -384,7 +387,7 @@ export async function handleChatInput(
// the session is still probing: build the PROBER (quiz/ask + probe_done).
// Both share this whole composition (write-ahead persist, liveness,
// steers, usage, extraction) — only the agent differs.
const teaching = hasProbeDone(merged);
const teaching = hasProbeDone(agentMessages);
// BYOK (owner, 2026-09-22): the conversation chain was resolved by the
// turn gate above (chat.model override → strong slot → refused) — the
// engine build is pure from here; thinking levels clamp natively.
@@ -403,7 +406,7 @@ export async function handleChatInput(
// sub-agent JOB tools (research/demo/visual/pdf; adjudicated with the
// user: quiz/ask/fs/plan/probe_done are always-on).
if (isSubscriptionProvider(conversation.provider)) requestCycle = new SubscriptionRequestCycle({
initialMessages: asAgentMessages(merged),
initialMessages: asAgentMessages(agentMessages),
staged: stagedAutomatic ? asAgentMessages(appends) : undefined,
foregroundOwed: !callPurpose.background,
purpose: callPurpose,
@@ -419,7 +422,7 @@ export async function handleChatInput(
engine,
callPurpose,
requestCycle,
messages: asAgentMessages(merged),
messages: asAgentMessages(agentMessages),
onUsage: collectUsage,
toolToggles: effective.agent.tools,
// Debug (owner, 2026-10-04): teacher/prober raw provider events under
@@ -515,7 +518,7 @@ export async function handleChatInput(
// Debug recorder run marker (owner, 2026-10-04): correlates the
// JSONL event stream with the record's turns; no-op unless
// LH_PROVIDER_EVENTS=1.
providerEventNote(chatId, "turn", `run start (base: ${merged.length} messages)`);
providerEventNote(chatId, "turn", `run start (base: ${agentMessages.length} messages)`);
const handle = runPiTurn({
agent: constructed,
requestCycle,
@@ -555,8 +558,8 @@ export async function handleChatInput(
settleSteers(chatId, persisted, true, constructed);
},
onFinish: (turnMessages, isAborted) => {
// Wave B: the record write is the agent's own final state —
// the new messages are the slice past the merged base.
// Wave B: append only messages past the temporary agent prefix.
// Retry's saved record still includes its earlier failed attempts.
const stamped = persistTurn(stores.chatStore, chatId, persistBaseCount, turnMessages, isAborted);
noteSteersDurable(chatId, stores.chatStore.getChat(chatId)?.messages ?? stamped);
// usage log append (idempotent per turn id) — the stats bar