mirror of
https://github.com/badlogic/pi-mono.git
synced 2026-11-05 23:37:25 +00:00
fix(durable): fail a run whose blocking compaction failed; keep compaction failures visible
A generation waiting on a blocking threshold compaction sent its request whatever the outcome. When summarization failed, the context was still above the blocking threshold: a session at 996K of a 1M window got an empty length answer on every turn and nothing showed the error. The generation now fails with model_error (Compaction failed: <error>) when its compaction failed, faulted, was orphaned, or was aborted directly; a compaction that found nothing to compact or was declined still lets the request go ahead. pi.live.compactionFailure keeps the newest failure (task, reason, error) until a compaction produces a summary. The agent events' snapshot carries it, and compaction_end carries error for a failed compaction.
This commit is contained in:
Mario Zechner
parent
15e7bdc14c
commit
89e5c2c075
9 changed files with 215 additions and 47 deletions
|
|
@ -22,9 +22,12 @@
|
|||
- `TaskRuntime.abortOwned(id, context)`: abort a task this task owns and wait until it is terminal.
|
||||
- `TaskRuntime.ownedTasks(context)`: the live tasks a task owns directly, whichever invocation created them, as copies. A tool call settling finds the nested calls it left running with it.
|
||||
- `read` returns images as one image block instead of an `unsupported_image` error: PNG, JPEG, GIF, and WebP within the model's image limits (by default 2000x2000 pixels and 4.5 MB of base64) as they are, or, with `createCodingTools({ images })`, oriented, converted, and shrunk to fit. `@earendil-works/pi-durable/images` provides a Photon (WebAssembly) image processor, with `/images/node` and `/images/cloudflare` loaders; `/images/node` prepares images in a worker thread, so a large photo never blocks the event loop, and the thread exits after 30 s without images; nothing loads it unless imported. A model without image input gets a diagnostic saying it sees a placeholder.
|
||||
- `pi.live.compactionFailure` keeps the newest compaction failure (task, reason, error) until a compaction produces a summary; the agent events' snapshot carries it as `compactionFailure`, and `compaction_end` carries `error` for a compaction that failed. A failed background or manual compaction was invisible once its status was removed.
|
||||
|
||||
### Fixed
|
||||
|
||||
- A generation whose blocking threshold compaction failed, faulted, was orphaned, or was aborted directly now fails with `model_error` (`Compaction failed: <error>`) instead of sending its request anyway. The context was still above the blocking threshold, so the request at best left no room for the answer: a session at 996K of a 1M window got an empty `length` answer on every turn, with no visible error.
|
||||
|
||||
- Compaction no longer includes thinking in the summarization request. Replayed thinking is often not in the live context, so a request that included it could be larger than the context it summarized and fail with "prompt is too long" on every attempt.
|
||||
- A throwing `onReport` no longer replaces the error being reported or becomes an unhandled rejection, a throwing `now()` is reported once and `Date.now` used, and a watch listener's error is reported as well as ending the watch with `listener_error`.
|
||||
- A task whose fault write failed no longer runs again in the same process.
|
||||
|
|
|
|||
|
|
@ -457,7 +457,9 @@ settings: {
|
|||
|
||||
When a provider rejects a request because the context is too long, generation compacts and retries once. A summary that would cut before the start of the current context settles as `stale` when it is placed, so when several are in flight, the furthest cut stays in effect. Summarization spend counts in `pi.usage`. A `beforeCompact` hook on `CompactionTask` can decline or supply its own summary.
|
||||
|
||||
Running compactions are listed in `docs["pi.live"].compactions` with their reason, attempt, and retry backoff. The agent events add `compaction_start` and `compaction_end`, and a `compactions` field in the snapshot.
|
||||
When a compaction that a generation waits for at the blocking threshold fails, the generation fails too, without sending a request: its inputs settle `unanswered` with `model_error` and detail `Compaction failed: <error>`. One that finds nothing to compact or is declined lets the request go ahead.
|
||||
|
||||
Running compactions are listed in `docs["pi.live"].compactions` with their reason, attempt, and retry backoff. The newest failure stays in `docs["pi.live"].compactionFailure` (task, reason, error) until a compaction produces a summary. The agent events add `compaction_start` and `compaction_end` (with `error` when the compaction failed), and `compactions` and `compactionFailure` fields in the snapshot.
|
||||
|
||||
## Agent Events (Experimental)
|
||||
|
||||
|
|
|
|||
|
|
@ -3660,6 +3660,18 @@ type LiveState = {
|
|||
}[];
|
||||
/** Live compaction tasks in task ID order; absent when none (section 8.7). */
|
||||
compactions?: CompactionStatus[];
|
||||
/**
|
||||
* The newest compaction failure since a compaction last produced a summary (section 8.7), so a client can show it
|
||||
* after the task ended.
|
||||
*/
|
||||
compactionFailure?: CompactionFailure;
|
||||
};
|
||||
|
||||
type CompactionFailure = {
|
||||
taskId: TaskId<CompactionResult>;
|
||||
reason: CompactionReason;
|
||||
/** The failed or faulted outcome's error message, or why the scheduler orphaned the task. */
|
||||
error: string;
|
||||
};
|
||||
|
||||
type CompactionStatus = {
|
||||
|
|
@ -3738,6 +3750,8 @@ type GenerationCheckpoint =
|
|||
compacted?: TaskId<CompactionResult>;
|
||||
/** Error text of the overflow that started `compacted`; checked once when `prepare` resumes. */
|
||||
overflow?: string;
|
||||
/** Set when the blocking threshold started `compacted`; its outcome is checked once when `prepare` resumes. */
|
||||
threshold?: true;
|
||||
}
|
||||
| {
|
||||
phase: "request";
|
||||
|
|
@ -3795,8 +3809,15 @@ The run's inputs live in `pi.live.run`, not in the task input.
|
|||
- Above `contextWindow - reserveTokens`, the blocking threshold: instead of
|
||||
appending, one commit creates a compaction owned by the generation with reason
|
||||
`threshold`, adds its status, and commits `waiting` on it with `allSettled`
|
||||
and checkpoint `{ phase: "prepare", attempt, compacted }`. Whatever its
|
||||
outcome, `prepare` then runs again and sends the request.
|
||||
and checkpoint `{ phase: "prepare", attempt, compacted, threshold: true }`.
|
||||
When `prepare` resumes, it reads the compaction's outcome once. If the
|
||||
compaction completed, with or without a summary (nothing to compact, or
|
||||
declined), `prepare` continues and sends the request. If it failed,
|
||||
faulted, was orphaned, or was aborted directly, the generation fails with
|
||||
`model_error` without a request, and its inputs settle `unanswered` with
|
||||
detail `Compaction failed: <error>` (`Compaction was aborted` for an
|
||||
abort). A failed compaction leaves a context above the blocking threshold,
|
||||
where a request would at best leave no room for the answer.
|
||||
- Above `contextWindow - reserveTokens - backgroundTokens` with
|
||||
`backgroundTokens > 0` and no compaction status in `pi.live`: the commit that
|
||||
moves to `request` also creates a background compaction, owned by the
|
||||
|
|
@ -4146,8 +4167,8 @@ the conversation.
|
|||
|
||||
where `wrapped` is `"The conversation history before this point was compacted
|
||||
into the following summary:\n\n<summary>\n"`, the summary, and
|
||||
`"\n</summary>"`. The commit that places it also removes the task's status and
|
||||
commits the task's `completed` outcome:
|
||||
`"\n</summary>"`. The commit that places it also removes the task's status,
|
||||
removes `pi.live.compactionFailure`, and commits the task's `completed` outcome:
|
||||
|
||||
- A blocking compaction appends the entry directly and completes with
|
||||
`{ entryId }`. Its generation holds the run and waits, so nothing else writes
|
||||
|
|
@ -4173,10 +4194,18 @@ aborted with its generation, before the generation's abort handler runs
|
|||
(section 5.5). A summary that is already queued is a write submission:
|
||||
conversation abort keeps it, and `Submission.abort()` withdraws it.
|
||||
|
||||
When a blocking compaction ends without an `entryId`, because it found nothing
|
||||
to compact, was declined, failed, faulted, or was aborted directly, a threshold
|
||||
generation sends its request anyway and an overflow generation fails with
|
||||
`model_error` (section 8.3).
|
||||
**Failure.** The commit that fails a compaction (`model_error`, `no_model`)
|
||||
removes its status and sets `pi.live.compactionFailure` to its task, reason,
|
||||
and error message, replacing an older failure; the Harness cleanup does the
|
||||
same for a compaction the scheduler faults or orphans, with the fault's
|
||||
message or the orphan reason. Only a compaction that produces a summary
|
||||
removes it; one that is aborted or finds nothing to compact leaves it.
|
||||
|
||||
When a blocking compaction ends without an `entryId`, an overflow generation
|
||||
fails with `model_error`. A threshold generation sends its request anyway
|
||||
when the compaction completed (it found nothing to compact or was declined),
|
||||
and fails with `model_error` when it failed, faulted, was orphaned, or was
|
||||
aborted directly (section 8.3).
|
||||
|
||||
|
||||
## 9. Document observation and Chord
|
||||
|
|
@ -4424,6 +4453,8 @@ type AgentEvent =
|
|||
nestedTools: readonly NestedToolSlot[];
|
||||
/** `pi.live.compactions` (section 8.2). */
|
||||
compactions: readonly CompactionStatus[];
|
||||
/** `pi.live.compactionFailure` (section 8.2). */
|
||||
compactionFailure?: CompactionFailure;
|
||||
inbox: readonly { id: SubmissionId; mode: InboxItem["mode"] }[];
|
||||
/** `pi.agent`; `{}` when absent. */
|
||||
agent: AgentState;
|
||||
|
|
@ -4457,7 +4488,7 @@ type AgentEvent =
|
|||
| { type: "task_failed"; taskId: TaskId; kind: string; message: string }
|
||||
| { type: "compaction_start"; taskId: TaskId; reason: CompactionReason; blocking: boolean }
|
||||
/** The task's receipt tells whether it produced a summary; the summary entry has its own events. */
|
||||
| { type: "compaction_end"; taskId: TaskId; reason: CompactionReason };
|
||||
| { type: "compaction_end"; taskId: TaskId; reason: CompactionReason; error?: string };
|
||||
|
||||
/** One change to the in-flight assistant message, relative to that message. */
|
||||
type MessageChange =
|
||||
|
|
@ -4531,8 +4562,9 @@ Events derive from committed changes:
|
|||
gains or drops `retry`, or gains `deferred` or moves its `pollAt`.
|
||||
- `task_failed`: a task of the conversation settles `faulted` or `orphaned`.
|
||||
- `compaction_start`/`compaction_end`: a status appears in or disappears from
|
||||
`pi.live.compactions`. A retry backoff shows only in the snapshot's
|
||||
`compactions`. A queued summary is placed later with its own
|
||||
`pi.live.compactions`. `compaction_end` carries `error` when the same commit
|
||||
recorded the task in `pi.live.compactionFailure`. A retry backoff shows only
|
||||
in the snapshot's `compactions`. A queued summary is placed later with its own
|
||||
`message_start`/`message_end` and `submission` events.
|
||||
|
||||
One commit produces one batch, in this order: `tool_execution_start`,
|
||||
|
|
|
|||
|
|
@ -15,7 +15,14 @@ import type {
|
|||
Tx,
|
||||
} from "../types.ts";
|
||||
import { orderToolResults } from "./context.ts";
|
||||
import { addCompactionStatus, compactionStatus, LiveDoc, type LiveState, removeCompactionStatus } from "./live.ts";
|
||||
import {
|
||||
addCompactionStatus,
|
||||
compactionStatus,
|
||||
failCompactionStatus,
|
||||
LiveDoc,
|
||||
type LiveState,
|
||||
removeCompactionStatus,
|
||||
} from "./live.ts";
|
||||
import { ensureProviderSessionId } from "./provider.ts";
|
||||
import { admitSubmission } from "./submissions.ts";
|
||||
import type {
|
||||
|
|
@ -197,8 +204,8 @@ export const CompactionTask = defineTask<CompactionInput, CompactionCheckpoint,
|
|||
if (status !== undefined) status.retry = { at: until, error: message.errorMessage ?? "" };
|
||||
return { status: "running", checkpoint: { phase: "retry", ...request, until } };
|
||||
}
|
||||
removeCompactionStatus(live, runtime.taskId);
|
||||
const text = summaryFailure(message);
|
||||
failCompactionStatus(live, runtime.taskId, text);
|
||||
return {
|
||||
status: "terminal",
|
||||
outcome: { status: "failed", error: { message: text, detail: { reason: "model_error" } } },
|
||||
|
|
@ -416,6 +423,7 @@ async function placeSummary(
|
|||
summary: string,
|
||||
): Promise<Next> {
|
||||
removeCompactionStatus(live, runtime.taskId);
|
||||
delete live.compactionFailure;
|
||||
const text = `${SUMMARY_PREFIX}${summary}${SUMMARY_SUFFIX}`;
|
||||
const entry = {
|
||||
kind: CompactionEntry.kind,
|
||||
|
|
@ -450,7 +458,7 @@ async function failNoModel(runtime: Runtime, ref: ModelRef | undefined, context:
|
|||
const message =
|
||||
ref === undefined ? "No model is configured" : `Model ${ref.provider}/${ref.modelId} is not available`;
|
||||
await runtime.commit(async (tx) => {
|
||||
removeCompactionStatus(await tx.doc(LiveDoc, runtime.conversationId), runtime.taskId);
|
||||
failCompactionStatus(await tx.doc(LiveDoc, runtime.conversationId), runtime.taskId, message);
|
||||
return { status: "terminal", outcome: { status: "failed", error: { message, detail: { reason: "no_model" } } } };
|
||||
}, context);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -15,7 +15,7 @@ import type {
|
|||
} from "../types.ts";
|
||||
import { AgentDoc } from "./agent.ts";
|
||||
import type { InboxItem, InboxState } from "./inbox.ts";
|
||||
import type { CompactionStatus, LiveState, NestedToolSlot, ToolSlot } from "./live.ts";
|
||||
import type { CompactionFailure, CompactionStatus, LiveState, NestedToolSlot, ToolSlot } from "./live.ts";
|
||||
import { NestedCallDoc, type NestedCallState } from "./tool.ts";
|
||||
import type { AgentState, CompactionReason, Harness, NestedToolExecutionResult, ToolDiagnostic } from "./types.ts";
|
||||
import { UsageDoc, type UsageState } from "./usage.ts";
|
||||
|
|
@ -49,6 +49,8 @@ export type SnapshotEvent = {
|
|||
nestedTools: readonly NestedToolSlot[];
|
||||
/** `pi.live.compactions`: live compactions with their attempt and retry backoff. */
|
||||
compactions: readonly CompactionStatus[];
|
||||
/** `pi.live.compactionFailure`: the newest compaction failure since a compaction last produced a summary. */
|
||||
compactionFailure?: CompactionFailure;
|
||||
inbox: readonly QueuedItem[];
|
||||
/** `pi.agent`; `{}` when absent. */
|
||||
agent: AgentState;
|
||||
|
|
@ -90,8 +92,11 @@ export type AgentEvent =
|
|||
| { type: "usage_changed"; usage: UsageState }
|
||||
| { type: "task_failed"; taskId: TaskId; kind: string; message: string }
|
||||
| { type: "compaction_start"; taskId: TaskId; reason: CompactionReason; blocking: boolean }
|
||||
/** The task's receipt tells whether it produced a summary; the summary entry has its own events. */
|
||||
| { type: "compaction_end"; taskId: TaskId; reason: CompactionReason };
|
||||
/**
|
||||
* The task's receipt tells whether it produced a summary; the summary entry has its own events. `error` is set when
|
||||
* the compaction failed, faulted, or was orphaned, as in `pi.live.compactionFailure`.
|
||||
*/
|
||||
| { type: "compaction_end"; taskId: TaskId; reason: CompactionReason; error?: string };
|
||||
|
||||
/**
|
||||
* Which call a tool event is about. `toolCallId` is the call's ID in the transcript and in tool events: the provider's
|
||||
|
|
@ -143,6 +148,7 @@ function snapshotOf(view: ConversationView): SnapshotEvent {
|
|||
tools: live.tools ?? [],
|
||||
nestedTools: live.nestedTools ?? [],
|
||||
compactions: live.compactions ?? [],
|
||||
...(live.compactionFailure === undefined ? {} : { compactionFailure: live.compactionFailure }),
|
||||
inbox: queued(inbox),
|
||||
agent: agent ?? AgentDoc.definition.initial(),
|
||||
usage: usage ?? UsageDoc.definition.initial(),
|
||||
|
|
@ -360,9 +366,11 @@ function translate(
|
|||
// Compaction ends, task failures, then turn and run ends.
|
||||
const compactionsBefore = was.live.compactions ?? [];
|
||||
const compactions = now.live.compactions ?? [];
|
||||
const failure = now.live.compactionFailure;
|
||||
for (const { taskId, reason } of compactionsBefore) {
|
||||
if (!compactions.some((status) => status.taskId === taskId)) {
|
||||
events.push({ type: "compaction_end", taskId, reason });
|
||||
const error = failure?.taskId === taskId ? failure.error : undefined;
|
||||
events.push({ type: "compaction_end", taskId, reason, ...(error === undefined ? {} : { error }) });
|
||||
}
|
||||
}
|
||||
// A generation's turn ends when its outcome is committed: at a `completing` hold or at terminal, whichever comes
|
||||
|
|
|
|||
|
|
@ -21,6 +21,7 @@ import type {
|
|||
NextTaskState,
|
||||
SubmissionId,
|
||||
TaskId,
|
||||
TaskOutcome,
|
||||
TaskRuntime,
|
||||
Tx,
|
||||
TypedEntry,
|
||||
|
|
@ -56,6 +57,8 @@ export type GenerationCheckpoint =
|
|||
compacted?: TaskId<CompactionResult>;
|
||||
/** Error text of the overflow that started `compacted`; checked once when `prepare` resumes. */
|
||||
overflow?: string;
|
||||
/** Set when the blocking threshold started `compacted`; its outcome is checked once when `prepare` resumes. */
|
||||
threshold?: true;
|
||||
}
|
||||
| {
|
||||
phase: "request";
|
||||
|
|
@ -130,13 +133,21 @@ export const GenerationTask = defineTask<GenerationInput, GenerationCheckpoint,
|
|||
const { model, thinkingLevel } = agent;
|
||||
const resolved = model === undefined ? undefined : runtime.models.getModel(model.provider, model.modelId);
|
||||
if (model === undefined || resolved === undefined) return failNoModel(runtime, model, context);
|
||||
const { attempt, compacted, overflow } = task.state.checkpoint;
|
||||
const { attempt, compacted, overflow, threshold: blocked } = task.state.checkpoint;
|
||||
if (compacted !== undefined && overflow !== undefined) {
|
||||
const [outcome] = await runtime.outcomes([compacted], context);
|
||||
if (outcome?.status !== "completed" || outcome.result.entryId === undefined) {
|
||||
return failModelError(runtime, overflow, context);
|
||||
}
|
||||
}
|
||||
// A threshold compaction that found nothing to compact or was declined leaves a request that may still fit. One
|
||||
// that failed left a context it judged too large: sending it would at best leave no room for the answer.
|
||||
if (compacted !== undefined && blocked === true) {
|
||||
const [outcome] = await runtime.outcomes([compacted], context);
|
||||
if (outcome !== undefined && outcome.status !== "completed") {
|
||||
return failModelError(runtime, compactionFailure(outcome), context);
|
||||
}
|
||||
}
|
||||
const view = await runtime.context(conversationId, context);
|
||||
const shown = replaySections(view.messages);
|
||||
const report = (error: unknown) => runtime.report(error);
|
||||
|
|
@ -158,7 +169,7 @@ export const GenerationTask = defineTask<GenerationInput, GenerationCheckpoint,
|
|||
// Compact first and prepare again; the transcript is unchanged until the compaction appends.
|
||||
await runtime.commit(async (tx): Promise<Next> => {
|
||||
const child = await createCompaction(tx, conversationId, { reason: "threshold" }, runtime.taskId);
|
||||
const checkpoint = { phase: "prepare", attempt, compacted: child } as const;
|
||||
const checkpoint = { phase: "prepare", attempt, compacted: child, threshold: true } as const;
|
||||
return { status: "waiting", checkpoint, on: [child], policy: "allSettled" };
|
||||
}, context);
|
||||
return;
|
||||
|
|
@ -327,6 +338,19 @@ function thresholdCompaction(
|
|||
return over;
|
||||
}
|
||||
|
||||
/** Error text for a generation whose blocking compaction ended without completing. */
|
||||
function compactionFailure(outcome: Exclude<TaskOutcome<CompactionResult>, { status: "completed" }>): string {
|
||||
switch (outcome.status) {
|
||||
case "failed":
|
||||
case "faulted":
|
||||
return `Compaction failed: ${outcome.error.message}`;
|
||||
case "orphaned":
|
||||
return `Compaction failed: ${outcome.reason}`;
|
||||
case "aborted":
|
||||
return "Compaction was aborted";
|
||||
}
|
||||
}
|
||||
|
||||
/** Settle the run's inputs `unanswered` with `model_error` and fail with `text`. */
|
||||
async function failModelError(runtime: Runtime, text: string, context: Context): Promise<void> {
|
||||
await runtime.commit(async (tx) => {
|
||||
|
|
|
|||
|
|
@ -70,6 +70,14 @@ export type CompactionStatus = {
|
|||
retry?: { at: number; error: string };
|
||||
};
|
||||
|
||||
/** The newest compaction that failed since a compaction last produced a summary (spec §8.7). */
|
||||
export type CompactionFailure = {
|
||||
taskId: TaskId<CompactionResult>;
|
||||
reason: CompactionReason;
|
||||
/** The failed or faulted outcome's error message, or why the scheduler orphaned the task. */
|
||||
error: string;
|
||||
};
|
||||
|
||||
/** Built-in live conversation state: run control and presentation of the current generation and tool round. */
|
||||
export type LiveState = {
|
||||
/** Run control: the task that settles the run's inputs, and those inputs; present exactly while busy. */
|
||||
|
|
@ -93,6 +101,11 @@ export type LiveState = {
|
|||
nestedTools?: NestedToolSlot[];
|
||||
/** Live compaction tasks in task ID order; absent when none. */
|
||||
compactions?: CompactionStatus[];
|
||||
/**
|
||||
* The newest compaction failure, so a client can show it after the task ended. A compaction that produces a summary
|
||||
* removes it; one that is aborted or finds nothing to compact leaves it.
|
||||
*/
|
||||
compactionFailure?: CompactionFailure;
|
||||
};
|
||||
|
||||
export const LiveDoc = defineDoc<LiveState>({
|
||||
|
|
@ -152,6 +165,13 @@ export function removeCompactionStatus(live: Draft<LiveState>, taskId: TaskId):
|
|||
if (statuses.length === 0) delete live.compactions;
|
||||
}
|
||||
|
||||
/** Remove the status of compaction task `taskId` and record that it failed with `error`. */
|
||||
export function failCompactionStatus(live: Draft<LiveState>, taskId: TaskId, error: string): void {
|
||||
const status = compactionStatus(live, taskId);
|
||||
if (status !== undefined) live.compactionFailure = { taskId: status.taskId, reason: status.reason, error };
|
||||
removeCompactionStatus(live, taskId);
|
||||
}
|
||||
|
||||
/**
|
||||
* The slot of tool task `taskId`, model-issued or nested, if `pi.live` still lists it. Nested slots are found by binary
|
||||
* search: a tool may make thousands of nested calls, each of which looks up its slot in several commits, and every
|
||||
|
|
@ -207,7 +227,7 @@ export function clearProgress(slot: Draft<SlotProgress>): void {
|
|||
/**
|
||||
* Harness cleanup for a terminal outcome the scheduler writes itself (`faulted` or `orphaned`). A run task ends its
|
||||
* run; a tool task's slot is marked done without an entry, and context derivation synthesizes the missing result; a
|
||||
* compaction task's status is removed.
|
||||
* compaction task's status is removed and its failure recorded.
|
||||
* Ignores other kinds so it never creates `pi.live` elsewhere. The scheduler calls this without knowing task kinds;
|
||||
* the Harness passes it in (spec §5.4).
|
||||
* REMINDER: a committed generation partial becomes an aborted assistant entry here, exactly as in the generation abort
|
||||
|
|
@ -229,7 +249,8 @@ export async function settleSchedulerOutcome(
|
|||
return;
|
||||
}
|
||||
if (record.kind === COMPACTION_TASK_KIND) {
|
||||
removeCompactionStatus(await tx.doc(LiveDoc, record.conversationId), record.id);
|
||||
const error = outcome.status === "faulted" ? outcome.error.message : outcome.reason;
|
||||
failCompactionStatus(await tx.doc(LiveDoc, record.conversationId), record.id, error);
|
||||
return;
|
||||
}
|
||||
if (!RUN_TASK_KINDS.has(record.kind)) return;
|
||||
|
|
|
|||
|
|
@ -40,6 +40,7 @@ export {
|
|||
export { Harness } from "./harness/harness.ts";
|
||||
export { InboxDoc, type InboxItem, type InboxState } from "./harness/inbox.ts";
|
||||
export {
|
||||
type CompactionFailure,
|
||||
type CompactionStatus,
|
||||
LiveDoc,
|
||||
type LiveState,
|
||||
|
|
|
|||
|
|
@ -1004,6 +1004,43 @@ describe("background threshold compaction", () => {
|
|||
await chat.harness.close(context);
|
||||
});
|
||||
|
||||
it("records a failure in pi.live, the snapshot, and compaction_end until a summary is produced", async () => {
|
||||
const chat = await open({ contextWindow: 2000 });
|
||||
await history(chat);
|
||||
chat.setup.settings.compaction = BACKGROUND;
|
||||
const events: AgentEvent[] = [];
|
||||
const stream = await watchEvents(chat.harness, chat.root.id, context);
|
||||
stream.start(async (batch) => {
|
||||
events.push(...batch);
|
||||
});
|
||||
chat.faux.summaries.push(failure("bad request"));
|
||||
await turn(chat, text("u4", 100), text("a4", 100));
|
||||
await waitFor(() => events.some((event) => event.type === "compaction_end"));
|
||||
const recorded = (await live(chat)).compactionFailure;
|
||||
expect(recorded).toMatchObject({ reason: "threshold", error: "Summarization failed: bad request" });
|
||||
expect(events.find((event) => event.type === "compaction_end")).toEqual({
|
||||
type: "compaction_end",
|
||||
taskId: recorded!.taskId,
|
||||
reason: "threshold",
|
||||
error: "Summarization failed: bad request",
|
||||
});
|
||||
const reopened = await watchEvents(chat.harness, chat.root.id, context);
|
||||
expect(reopened.snapshot.compactionFailure).toEqual(recorded);
|
||||
await reopened.stop();
|
||||
// A compaction that produces a summary removes the failure, and ends without an error.
|
||||
chat.setup.settings.compaction = MANUAL;
|
||||
chat.faux.summaries.push(summary());
|
||||
const manual = await chat.root.compact(undefined, context);
|
||||
expect((await result(chat, manual)).status).toBe("completed");
|
||||
expect((await live(chat)).compactionFailure).toBeUndefined();
|
||||
await waitFor(() => events.some((event) => event.type === "compaction_end" && event.taskId === manual));
|
||||
expect(events.find((event) => event.type === "compaction_end" && event.taskId === manual)).not.toHaveProperty(
|
||||
"error",
|
||||
);
|
||||
await stream.stop();
|
||||
await chat.harness.close(context);
|
||||
});
|
||||
|
||||
it("stops through abortTask() and Conversation.abort() with background", async () => {
|
||||
for (const stop of ["task", "conversation"] as const) {
|
||||
const chat = await open({ contextWindow: 2000 });
|
||||
|
|
@ -1074,38 +1111,63 @@ describe("blocking threshold compaction", () => {
|
|||
await chat.harness.close(context);
|
||||
});
|
||||
|
||||
for (const [name, prepare] of [
|
||||
[
|
||||
"declines",
|
||||
(chat: Chat) => addHooks(chat.setup.registry, CompactionTask, { beforeCompact: () => ({ decline: true }) }),
|
||||
],
|
||||
["fails", (chat: Chat) => chat.faux.summaries.push(failure("bad request"))],
|
||||
] as const) {
|
||||
it(`sends the request anyway when its compaction ${name}`, async () => {
|
||||
const chat = await open({ contextWindow: 1000 });
|
||||
await history(chat);
|
||||
chat.setup.settings.compaction = BLOCKING;
|
||||
prepare(chat);
|
||||
await turn(chat, text("u4", 200), "a4");
|
||||
expect((await kinds(chat.root)).includes("pi.compaction")).toBe(false);
|
||||
expect(firstUserText(chat.faux.agentRequests.at(-1)!.messages)).toBe(text("u1", 100));
|
||||
await chat.harness.close(context);
|
||||
});
|
||||
}
|
||||
it("sends the request anyway when its compaction declines", async () => {
|
||||
const chat = await open({ contextWindow: 1000 });
|
||||
await history(chat);
|
||||
chat.setup.settings.compaction = BLOCKING;
|
||||
addHooks(chat.setup.registry, CompactionTask, { beforeCompact: () => ({ decline: true }) });
|
||||
await turn(chat, text("u4", 200), "a4");
|
||||
expect((await kinds(chat.root)).includes("pi.compaction")).toBe(false);
|
||||
expect(firstUserText(chat.faux.agentRequests.at(-1)!.messages)).toBe(text("u1", 100));
|
||||
expect((await live(chat)).compactionFailure).toBeUndefined();
|
||||
await chat.harness.close(context);
|
||||
});
|
||||
|
||||
it("sends the request anyway when its compaction is aborted directly", async () => {
|
||||
// A failed compaction leaves a context it judged too large; sending it once filled a 1M window and got an empty
|
||||
// `length` answer on every turn while the failure stayed invisible.
|
||||
it("fails the run with the compaction's error, without a request, when its compaction fails", async () => {
|
||||
const chat = await open({ contextWindow: 1000 });
|
||||
await history(chat);
|
||||
chat.setup.settings.compaction = BLOCKING;
|
||||
chat.faux.summaries.push(failure("bad request"));
|
||||
const requests = chat.faux.agentRequests.length;
|
||||
const input = await chat.root.submit({ type: "input", content: text("u4", 200) }, context);
|
||||
expect(await input.wait(context)).toMatchObject({
|
||||
status: "unanswered",
|
||||
reason: "model_error",
|
||||
detail: "Compaction failed: Summarization failed: bad request",
|
||||
});
|
||||
expect(chat.faux.agentRequests).toHaveLength(requests);
|
||||
expect((await kinds(chat.root)).at(-1)).toBe("pi.user");
|
||||
const state = await live(chat);
|
||||
expect(state.compactions).toBeUndefined();
|
||||
expect(state.compactionFailure).toMatchObject({
|
||||
reason: "threshold",
|
||||
error: "Summarization failed: bad request",
|
||||
});
|
||||
await chat.harness.close(context);
|
||||
});
|
||||
|
||||
it("fails the run when its compaction is aborted directly", async () => {
|
||||
const chat = await open({ contextWindow: 1000 });
|
||||
await history(chat);
|
||||
chat.setup.settings.compaction = BLOCKING;
|
||||
const reached = deferred();
|
||||
chat.faux.summaries.push(gated(deferred(), summary(), reached));
|
||||
chat.faux.agent.push(answer("a4"));
|
||||
const requests = chat.faux.agentRequests.length;
|
||||
const input = await chat.root.submit({ type: "input", content: text("u4", 200) }, context);
|
||||
await reached.promise;
|
||||
const [child] = await compactionTasks(chat);
|
||||
await chat.harness.abortTask(child!.id, context);
|
||||
expect((await input.wait(context)).status).toBe("done");
|
||||
expect(await input.wait(context)).toMatchObject({
|
||||
status: "unanswered",
|
||||
reason: "model_error",
|
||||
detail: "Compaction was aborted",
|
||||
});
|
||||
expect(chat.faux.agentRequests).toHaveLength(requests);
|
||||
expect((await kinds(chat.root)).includes("pi.compaction")).toBe(false);
|
||||
// Aborting is not a failure.
|
||||
expect((await live(chat)).compactionFailure).toBeUndefined();
|
||||
await chat.harness.close(context);
|
||||
});
|
||||
|
||||
|
|
@ -1889,7 +1951,7 @@ describe("compaction edge cases", () => {
|
|||
await chat.harness.close(context);
|
||||
});
|
||||
|
||||
it("sends the request anyway when its blocking compaction faults", async () => {
|
||||
it("fails the run when its blocking compaction faults, and records the fault", async () => {
|
||||
const chat = await open({ contextWindow: 1000 });
|
||||
await history(chat);
|
||||
chat.setup.settings.compaction = BLOCKING;
|
||||
|
|
@ -1897,10 +1959,17 @@ describe("compaction edge cases", () => {
|
|||
chat.setup.models.completeSimple = () => {
|
||||
throw new Error("no credentials");
|
||||
};
|
||||
await turn(chat, text("u4", 200), "a4");
|
||||
const input = await chat.root.submit({ type: "input", content: text("u4", 200) }, context);
|
||||
expect(await input.wait(context)).toMatchObject({
|
||||
status: "unanswered",
|
||||
reason: "model_error",
|
||||
detail: "Compaction failed: no credentials",
|
||||
});
|
||||
chat.setup.models.completeSimple = completeSimple;
|
||||
expect((await kinds(chat.root)).includes("pi.compaction")).toBe(false);
|
||||
expect((await live(chat)).compactions).toBeUndefined();
|
||||
const state = await live(chat);
|
||||
expect(state.compactions).toBeUndefined();
|
||||
expect(state.compactionFailure).toMatchObject({ reason: "threshold", error: "no credentials" });
|
||||
await chat.harness.close(context);
|
||||
});
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue