mirror of
https://github.com/badlogic/pi-mono.git
synced 2026-11-05 23:37:25 +00:00
fix(durable): close keeps admitted aborts, reports name the extension that ran, review nits
- Close reads the records it writes back on the line, after admitted commits, so a queued abort mark survives. - A task's report names the extension that installed the definition it ran, captured at reservation. - Terminal and completing records drop the restart count; retryTask() ignores them. - wake() counts a reconcile pass only when it has work, so a run's end sends no extra active wake. - A failed agent resolution no longer breaks the tool result path that names a tool's extension. - Spec §10.1 list, changelog sections, test cleanup of killed storages.
This commit is contained in:
Mario Zechner
parent
440a5fcdb8
commit
5ba1cbaf3d
7 changed files with 172 additions and 60 deletions
|
|
@ -9,7 +9,6 @@
|
|||
- Opening a SQLite storage migrates it to schema version 2 automatically, rebuilding its task table. A migrated database cannot be opened by 1.3.0 or earlier, so a deployment cannot roll back past this release. JSONL storage needs no migration, but 1.3.0 or earlier never resumes a sleeping task it finds (a run's conversation stays busy).
|
||||
- `TaskRuntime` has `sleepUntil()` and `setTimer()`, and `ToolExecutionApi` has `now()`: code that builds either object by hand, such as a test fake or a tool wrapper that does not spread the original, must add them.
|
||||
- A task cut off by a crash 5 times in a row without committing a new state is parked instead of run again (see the crash-loop guard under Added); set `settings.restartLimit` to `Infinity` to keep the old behavior. `TaskInspection`'s blocked reasons include `restart_limit`.
|
||||
- `onReport` gets a second argument, `info` (see below), and `TaskRuntime.report()` takes an optional `from`; an `onReport` written for one argument still works.
|
||||
|
||||
### Added
|
||||
|
||||
|
|
@ -26,6 +25,7 @@
|
|||
### Changed
|
||||
|
||||
- A task phase that faults is reported through `onReport` (with `faulted: true`), not only recorded in its outcome.
|
||||
- `onReport` gets a second argument, `info`, and `TaskRuntime.report()` takes an optional `from` (see Added); an `onReport` written for one argument still works.
|
||||
- Generation retry backoffs, deferred-response polls, and compaction retry backoffs sleep instead of waiting in a running phase, so a process can stop or hibernate during them. A record an earlier version stored mid-backoff sleeps again when it resumes before its time. A deferred poll whose model is gone after a reopen now fails `no_model` once the poll is due, not at once.
|
||||
- Code mode measures script deadlines and wall time on the Harness clock (`HarnessOptions.now`) instead of `Date.now()`.
|
||||
- SQLite drivers, portable interfaces, and the generic migration runner now come from `@earendil-works/chord/sqlite`. Existing Durable SQLite entry points keep re-exporting their public names; storage helpers and Durable schema migrations stay here.
|
||||
|
|
|
|||
|
|
@ -5323,20 +5323,26 @@ dropped, since nothing is left to report it to. A throwing `now()` is reported
|
|||
once, and `Date.now` serves instead.
|
||||
|
||||
**Every report says where it came from.** `onReport(error, info)` gets a
|
||||
`ReportInfo`: `source` is `session` for the error that failed the Session (a
|
||||
Storage error, often a platform reset, or the Session's own bookkeeping),
|
||||
`task` for a task's phase that faulted (`faulted: true`), a task's
|
||||
`runtime.report()`, a failed migration, or a task parked by the restart limit
|
||||
(`blocked: "restart_limit"`), with `taskId`, `taskKind`, and `conversationId`; `extension` for hooks,
|
||||
wrappers, system prompt sections, and a tool's invalid structured output;
|
||||
`listener`; `host` for the clock, settings, registry, environment builder, and
|
||||
model provider calls such as cancelling a deferred response; an `extension`
|
||||
report names the extension (`extension`) when the failing code is known to
|
||||
belong to one: the hook's, the wrapper's, or the last selected extension that
|
||||
provides or wraps the section or tool. A `task` report names the extension that
|
||||
installed the task's definition; built-in tasks have none; `timer` for `setTimer`; and `wake` for `onWake`. A host acts on them
|
||||
outside the scheduler, for example counting faults of one task kind toward a
|
||||
breaker.
|
||||
`ReportInfo` whose `source` is one of:
|
||||
|
||||
- `session`: the error that failed the Session (a Storage error, often a
|
||||
platform reset, or the Session's own bookkeeping);
|
||||
- `task`: a task's phase that faulted (`faulted: true`), a task's
|
||||
`runtime.report()`, a failed migration, or a task parked or faulted by the
|
||||
restart limit (`blocked: "restart_limit"`), with `taskId`, `taskKind`, and
|
||||
`conversationId`;
|
||||
- `extension`: hooks, wrappers, system prompt sections, and a tool's invalid
|
||||
structured output;
|
||||
- `listener`; `host` for the clock, settings, registry, environment builder, and
|
||||
model provider calls such as cancelling a deferred response; `timer` for
|
||||
`setTimer`; and `wake` for `onWake`.
|
||||
|
||||
`extension` names the extension whose code failed, when known. A `task` report
|
||||
names the extension that installed the definition the task ran, as of its
|
||||
reservation; built-in tasks have none. An `extension` report names the hook's or
|
||||
the wrapper's extension, or the last selected extension that provides or wraps
|
||||
the section or tool. A host acts on reports outside the scheduler, for example
|
||||
counting faults of one task kind toward a breaker.
|
||||
|
||||
## 11. Backends
|
||||
|
||||
|
|
|
|||
|
|
@ -96,6 +96,8 @@ type Fit =
|
|||
type Invocation = {
|
||||
readonly taskId: TaskId;
|
||||
readonly kind: string;
|
||||
/** The extension that installed the definition this invocation runs, as of its reservation; none for a built-in. */
|
||||
readonly extension: string | undefined;
|
||||
readonly conversationId: ConversationId;
|
||||
readonly mode: "run" | "abort";
|
||||
readonly controller: AbortController;
|
||||
|
|
@ -194,8 +196,8 @@ export type TaskSchedulerOptions = {
|
|||
readonly onWake: ((wake: Wake) => void) | undefined;
|
||||
/** The host's timers (`HarnessOptions.setTimer`). */
|
||||
readonly timers: Timers;
|
||||
/** Write records straight to Storage while closing, after admission is sealed. */
|
||||
readonly writeAtClose: (writes: readonly StorageWrite[]) => Promise<void>;
|
||||
/** While closing, on the line after admitted commits settle, write what the callback returns straight to Storage. */
|
||||
readonly writeAtClose: (writes: () => readonly StorageWrite[]) => Promise<void>;
|
||||
readonly report: (error: unknown, info: ReportInfo) => void;
|
||||
/** Harness cleanup staged in the commit that makes an outcome the scheduler wrote itself terminal. */
|
||||
readonly settleOutcome: (tx: Transaction, record: AnyTaskRecord, outcome: SchedulerOutcome) => Promise<void>;
|
||||
|
|
@ -244,7 +246,7 @@ export class TaskScheduler {
|
|||
readonly #now: () => number;
|
||||
readonly #onWake: ((wake: Wake) => void) | undefined;
|
||||
readonly #timers: Timers;
|
||||
readonly #writeAtClose: (writes: readonly StorageWrite[]) => Promise<void>;
|
||||
readonly #writeAtClose: (writes: () => readonly StorageWrite[]) => Promise<void>;
|
||||
readonly #report: (error: unknown, info: ReportInfo) => void;
|
||||
readonly #settleOutcome: TaskSchedulerOptions["settleOutcome"];
|
||||
readonly #withdrawInputs: TaskSchedulerOptions["withdrawInputs"];
|
||||
|
|
@ -313,8 +315,8 @@ export class TaskScheduler {
|
|||
/** Tasks a pass left out as parked; each pass checks them again, as `settings.restartLimit` may have risen. */
|
||||
readonly #parkedTasks = new Set<TaskId>();
|
||||
#reconcileScheduled = false;
|
||||
/** Reconcile passes underway (they may overlap): their commits finalize held outcomes and mark cascades, recovery work. */
|
||||
#reconciling = 0;
|
||||
/** Reconcile passes underway with work (they may overlap): their commits finalize held outcomes and mark cascades. */
|
||||
#reconcilingWork = 0;
|
||||
#sweepScheduled = false;
|
||||
#cascadePending = false;
|
||||
#unsubscribeRegistry: () => void = () => {};
|
||||
|
|
@ -410,7 +412,8 @@ export class TaskScheduler {
|
|||
await this.#session.commitWith(async (tx) => {
|
||||
const current = await tx.task(id);
|
||||
if (current === undefined) throw new Error(`Task ${id} does not exist`);
|
||||
if (!this.#parked(current)) return;
|
||||
const status = current.state.status;
|
||||
if (status === "terminal" || status === "completing" || !this.#parked(current)) return;
|
||||
tx.setTask(progressed(current));
|
||||
}, context);
|
||||
this.resume();
|
||||
|
|
@ -451,9 +454,9 @@ export class TaskScheduler {
|
|||
*/
|
||||
wake(): Wake {
|
||||
const now = this.#now();
|
||||
// A reconcile pass scheduled or underway is work too: it may finalize held outcomes and mark cascades, at open as
|
||||
// recovery. A process that dies before it lands must be woken again.
|
||||
let active = this.#invocations.size > 0 || this.#reconcileScheduled || this.#reconciling > 0;
|
||||
// A reconcile pass with work to do is work too: it finalizes held outcomes and marks cascades, at open as
|
||||
// recovery. A process that dies before it lands must be woken again. A pass with nothing to do is not.
|
||||
let active = this.#invocations.size > 0 || this.#reconcilingWork > 0 || this.#reconcileWork();
|
||||
let at: number | undefined;
|
||||
let snapshot: RegistrySnapshot | undefined;
|
||||
let snapshotFailed = false;
|
||||
|
|
@ -496,15 +499,18 @@ export class TaskScheduler {
|
|||
const stopped = [...this.#invocations.values()];
|
||||
for (const invocation of stopped) invocation.controller.abort();
|
||||
await Promise.allSettled(stopped.map((invocation) => invocation.done));
|
||||
// A close is not a crash: a task it stopped is `pending` again, so the next open counts no restart for it.
|
||||
const released = stopped.flatMap((invocation) => {
|
||||
const record = this.#live.get(invocation.taskId);
|
||||
if (record?.state.status !== "running") return [];
|
||||
return [withState(record, { status: "pending", checkpoint: record.state.checkpoint })];
|
||||
});
|
||||
if (released.length === 0) return;
|
||||
// A failure leaves them `running`, counted as if cut off; the Storage guard fails the Session and reports it.
|
||||
await this.#writeAtClose(released.map((value) => ({ type: "task", value }) as const)).catch(() => {});
|
||||
if (stopped.length === 0) return;
|
||||
// A close is not a crash: a task it stopped is `pending` again, so the next open counts no restart for it. Read
|
||||
// on the line after commits admitted before the seal, such as an abort mark, so the write keeps them. A failure
|
||||
// leaves them `running`, counted as if cut off; the Storage guard fails the Session and reports it.
|
||||
await this.#writeAtClose(() =>
|
||||
stopped.flatMap((invocation) => {
|
||||
const record = this.#live.get(invocation.taskId);
|
||||
if (record?.state.status !== "running") return [];
|
||||
const value = withState(record, { status: "pending", checkpoint: record.state.checkpoint });
|
||||
return [{ type: "task", value } as const];
|
||||
}),
|
||||
).catch(() => {});
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
@ -747,7 +753,8 @@ export class TaskScheduler {
|
|||
*/
|
||||
async #reconcile(): Promise<void> {
|
||||
this.#reconcileScheduled = false;
|
||||
this.#reconciling++;
|
||||
const work = this.#reconcileWork();
|
||||
if (work) this.#reconcilingWork++;
|
||||
const cascade = this.#cascadePending;
|
||||
this.#cascadePending = false;
|
||||
const checks = [...this.#failFastChecks];
|
||||
|
|
@ -794,11 +801,16 @@ export class TaskScheduler {
|
|||
// running the pass again, so it fails the Session, which reports it.
|
||||
this.#failSession(error);
|
||||
}
|
||||
this.#reconciling--;
|
||||
if (work) this.#reconcilingWork--;
|
||||
this.#scheduleWake();
|
||||
this.#settleIdle();
|
||||
}
|
||||
|
||||
/** Whether a reconcile pass would have work: a cascade to derive, held outcomes to check, or `failFast` waits. */
|
||||
#reconcileWork(): boolean {
|
||||
return this.#cascadePending || this.#finalizeChecks.size > 0 || this.#failFastChecks.size > 0;
|
||||
}
|
||||
|
||||
/** Whether any of `ids` holds or ended with an outcome other than `completed`. */
|
||||
async #anyFailed(ids: readonly TaskId[]): Promise<boolean> {
|
||||
for (const id of ids) {
|
||||
|
|
@ -1331,19 +1343,19 @@ export class TaskScheduler {
|
|||
readonly taskId: TaskId;
|
||||
readonly kind: string;
|
||||
readonly conversationId: ConversationId;
|
||||
/** The installer of the definition whose code ran (an invocation's); absent: the current one's. */
|
||||
readonly extension?: string | undefined;
|
||||
}): ReportInfo {
|
||||
const info = taskInfo(task);
|
||||
let extension: string | undefined;
|
||||
try {
|
||||
const snapshot = this.#registry.snapshot();
|
||||
const definition = snapshot.task(task.kind);
|
||||
if (definition !== undefined) {
|
||||
extension = snapshot
|
||||
.installed()
|
||||
.findLast((installed) => installed.tasks?.includes(definition) === true)?.name;
|
||||
if ("extension" in task) extension = task.extension;
|
||||
else {
|
||||
try {
|
||||
const snapshot = this.#registry.snapshot();
|
||||
extension = installerOf(snapshot, snapshot.task(task.kind));
|
||||
} catch {
|
||||
// A registry that throws is reported where it is read for scheduling; a report goes out without the name.
|
||||
}
|
||||
} catch {
|
||||
// A registry that throws is reported where it is read for scheduling; a report goes out without the name.
|
||||
}
|
||||
return extension === undefined ? info : { ...info, extension };
|
||||
}
|
||||
|
|
@ -1504,7 +1516,7 @@ export class TaskScheduler {
|
|||
// Registered on the line, so marks and later reservations see it and close joins it; `#end` makes it a
|
||||
// candidate again.
|
||||
this.#runnable.delete(id);
|
||||
const invocation = this.#createInvocation(record, mode);
|
||||
const invocation = this.#createInvocation(record, mode, installerOf(snapshot, resolution.task));
|
||||
reservations.push({ invocation, task: resolution.task, snapshot });
|
||||
}
|
||||
}, this.#context);
|
||||
|
|
@ -1559,9 +1571,10 @@ export class TaskScheduler {
|
|||
// Blocked now without a commit: the graph and the host's wake must hear it.
|
||||
queueMicrotask(() => this.#notifyBlocked());
|
||||
this.#scheduleWake();
|
||||
const extension = installerOf(snapshot, fit.task);
|
||||
this.#report(
|
||||
error,
|
||||
this.#taskInfo({ taskId: record.id, kind: record.kind, conversationId: record.conversationId }),
|
||||
this.#taskInfo({ taskId: record.id, kind: record.kind, conversationId: record.conversationId, extension }),
|
||||
);
|
||||
return { kind: "blocked", reason: "migration_failed" };
|
||||
}
|
||||
|
|
@ -1624,12 +1637,13 @@ export class TaskScheduler {
|
|||
return { kind: "ready", migrates: fit.migrates };
|
||||
}
|
||||
|
||||
#createInvocation(record: AnyTaskRecord, mode: "run" | "abort"): Invocation {
|
||||
#createInvocation(record: AnyTaskRecord, mode: "run" | "abort", extension: string | undefined): Invocation {
|
||||
const controller = new AbortController();
|
||||
const { promise: done, resolve: finish } = Promise.withResolvers<void>();
|
||||
const invocation: Invocation = {
|
||||
taskId: record.id,
|
||||
kind: record.kind,
|
||||
extension,
|
||||
conversationId: record.conversationId,
|
||||
mode,
|
||||
controller,
|
||||
|
|
@ -1668,7 +1682,9 @@ export class TaskScheduler {
|
|||
const runtime = this.#runtime(invocation, phase);
|
||||
let previous: PhaseResult | undefined;
|
||||
for (;;) {
|
||||
const current = await this.#step(invocation, (tx, current) => this.#decide(tx, current, previous, state));
|
||||
const current = await this.#step(invocation, (tx, current) =>
|
||||
this.#decide(tx, current, previous, state, invocation.extension),
|
||||
);
|
||||
// Close may seal between the decision and dispatch.
|
||||
if (current === undefined || this.#closing) return;
|
||||
const checkpoint = current.state.checkpoint;
|
||||
|
|
@ -1692,6 +1708,7 @@ export class TaskScheduler {
|
|||
current: ErasedRunningTask,
|
||||
previous: PhaseResult | undefined,
|
||||
state: { task: AnyTask; snapshot: RegistrySnapshot; reported: ReportedTask },
|
||||
extension: string | undefined,
|
||||
): Decision {
|
||||
// 3. abort mark: end; a fresh abort invocation starts once the task's ordinary owned work is gone.
|
||||
if (current.abortRequested) return false;
|
||||
|
|
@ -1722,7 +1739,12 @@ export class TaskScheduler {
|
|||
const cause = next === undefined ? "missing_task" : "incompatible_task";
|
||||
this.#report(
|
||||
new Error(`Task ${current.id} keeps running under its old ${current.kind} definition`, { cause }),
|
||||
this.#taskInfo({ taskId: current.id, kind: current.kind, conversationId: current.conversationId }),
|
||||
this.#taskInfo({
|
||||
taskId: current.id,
|
||||
kind: current.kind,
|
||||
conversationId: current.conversationId,
|
||||
extension,
|
||||
}),
|
||||
);
|
||||
}
|
||||
}
|
||||
|
|
@ -2207,6 +2229,12 @@ export class TaskScheduler {
|
|||
|
||||
const HOST: ReportInfo = { source: "host" };
|
||||
|
||||
/** The extension that installed `task` in `snapshot`, the last one that lists it; none for a built-in task. */
|
||||
function installerOf(snapshot: RegistrySnapshot, task: AnyTask | undefined): string | undefined {
|
||||
if (task === undefined) return undefined;
|
||||
return snapshot.installed().findLast((installed) => installed.tasks?.includes(task) === true)?.name;
|
||||
}
|
||||
|
||||
/** The report info of a failure of a task. */
|
||||
function taskInfo(task: {
|
||||
readonly taskId: TaskId;
|
||||
|
|
@ -2300,10 +2328,10 @@ function progressed(record: AnyTaskRecord): AnyTaskRecord {
|
|||
return rest as AnyTaskRecord;
|
||||
}
|
||||
|
||||
/** Replace a live record's state; memos disappear once an outcome is decided. */
|
||||
/** Replace a live record's state; memos and the restart count disappear once an outcome is decided. */
|
||||
function withState(record: AnyTaskRecord, state: TaskState<JsonValue, JsonValue>): AnyTaskRecord {
|
||||
if (state.status !== "terminal" && state.status !== "completing") return { ...record, state } as AnyTaskRecord;
|
||||
const { memos: _memos, ...rest } = record;
|
||||
const { memos: _memos, restarts: _restarts, ...rest } = record;
|
||||
return { ...rest, state };
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -561,7 +561,11 @@ async function finalResult(
|
|||
} else {
|
||||
// The model never sees structured output, so its call stands; the host learns of the broken tool.
|
||||
final = rest;
|
||||
const extension = toolExtension(await runtime.agent(context), tool.name);
|
||||
// Only names the extension: a failed resolution leaves the report without the name.
|
||||
const extension = await runtime.agent(context).then(
|
||||
(agent) => toolExtension(agent, tool.name),
|
||||
() => undefined,
|
||||
);
|
||||
runtime.report(new Error(broken), { source: "extension", ...(extension === undefined ? {} : { extension }) });
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -552,14 +552,15 @@ export class SessionImpl implements Session {
|
|||
}
|
||||
|
||||
/**
|
||||
* Internal, for `beforeClose()`: write records straight to Storage, on the line, after admission is sealed. Nothing
|
||||
* is published, since the Session is closing; a failed Session writes nothing.
|
||||
* Internal, for `beforeClose()`: on the line, after every commit admitted before the seal has settled, write what
|
||||
* `writes()` returns then straight to Storage. Nothing is published, since the Session is closing; a failed Session
|
||||
* writes nothing.
|
||||
*/
|
||||
protected writeAtClose(writes: readonly StorageWrite[]): Promise<void> {
|
||||
if (writes.length === 0) return Promise.resolve();
|
||||
protected writeAtClose(writes: () => readonly StorageWrite[]): Promise<void> {
|
||||
return this.#enqueue(async () => {
|
||||
if (this.#failure !== undefined) return;
|
||||
await this.#storage.commit(writes, BACKGROUND_CONTEXT);
|
||||
const batch = writes();
|
||||
if (batch.length > 0) await this.#storage.commit(batch, BACKGROUND_CONTEXT);
|
||||
});
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -6,17 +6,23 @@ import {
|
|||
defineExtension,
|
||||
defineTask,
|
||||
type Registry,
|
||||
type Storage,
|
||||
type TaskId,
|
||||
type Wake,
|
||||
} from "@earendil-works/pi-durable";
|
||||
import { afterEach, describe, expect, it } from "vitest";
|
||||
import { openNodeSqliteStorage } from "../src/storage/sqlite/node.ts";
|
||||
import { context, flush } from "./session-support.ts";
|
||||
import { completed, eventually, openTasks } from "./task-support.ts";
|
||||
import { completed, deferred, eventually, openTasks } from "./task-support.ts";
|
||||
|
||||
const directories = new Set<string>();
|
||||
|
||||
/** Storages of killed Harnesses, never closed by them: closed after each test. */
|
||||
const killed = new Set<Storage>();
|
||||
|
||||
afterEach(async () => {
|
||||
for (const storage of killed) await storage.close(context).catch(() => {});
|
||||
killed.clear();
|
||||
for (const directory of directories) await rm(directory, { recursive: true, force: true });
|
||||
directories.clear();
|
||||
});
|
||||
|
|
@ -67,6 +73,7 @@ async function openWith(path: string, registry: Registry, settings: { restartLim
|
|||
});
|
||||
const kill = (): void => {
|
||||
dead = true;
|
||||
killed.add(inner);
|
||||
};
|
||||
return { harness, wakes, reports, kill };
|
||||
}
|
||||
|
|
@ -134,7 +141,7 @@ describe("tasks that cannot run", () => {
|
|||
let opened = await openWith(path, registry, { restartLimit: 3 });
|
||||
const root = await opened.harness.root(context);
|
||||
const id = await root.commit((tx) => tx.createTask(Loop, null, { ownership: { kind: "conversation" } }), context);
|
||||
// Each open runs the phase, and each close cuts it off: three restarts.
|
||||
// Each open runs the phase, and each crash cuts it off: three restarts.
|
||||
for (let run = 1; run <= 3; run++) {
|
||||
opened.harness.resume();
|
||||
await eventually(() => state.started === run);
|
||||
|
|
@ -189,6 +196,33 @@ describe("tasks that cannot run", () => {
|
|||
await opened.harness.close(context);
|
||||
});
|
||||
|
||||
it("keeps an abort mark admitted before close in the record close writes back", async () => {
|
||||
const path = await sqlitePath();
|
||||
const state = { crashing: true, started: 0 };
|
||||
const registry = createRegistry();
|
||||
const Loop = crashLoop(state);
|
||||
registry.install(defineExtension({ name: "loop", tasks: [Loop] }));
|
||||
const { harness } = await openWith(path, registry);
|
||||
const root = await harness.root(context);
|
||||
const id = await root.commit((tx) => tx.createTask(Loop, null, { ownership: { kind: "conversation" } }), context);
|
||||
harness.resume();
|
||||
await eventually(() => state.started === 1);
|
||||
// Hold the line; the abort mark is admitted behind it, then close begins.
|
||||
const gate = deferred();
|
||||
const holding = root.commit(() => gate.promise, context);
|
||||
const aborting = harness.abortTask(id, context);
|
||||
const closing = harness.close(context);
|
||||
// The invocation ends while the line is held; its write-back waits behind the abort mark.
|
||||
await flush();
|
||||
gate.resolve();
|
||||
await holding;
|
||||
expect(await aborting).toBe("marked");
|
||||
await closing;
|
||||
const raw = await openNodeSqliteStorage(path);
|
||||
expect(await raw.task(id, context)).toMatchObject({ abortRequested: true, state: { status: "pending" } });
|
||||
await raw.close(context);
|
||||
});
|
||||
|
||||
it("faults a task whose abort handler is cut off restartLimit times", async () => {
|
||||
const path = await sqlitePath();
|
||||
let aborts = 0;
|
||||
|
|
@ -222,6 +256,8 @@ describe("tasks that cannot run", () => {
|
|||
status: "faulted",
|
||||
error: { detail: { reason: "restart_limit" } },
|
||||
});
|
||||
// The receipt keeps no restart count.
|
||||
expect(settled.restarts).toBeUndefined();
|
||||
expect(aborts).toBe(1);
|
||||
await eventually(() => opened.reports.length > 0);
|
||||
expect(opened.reports.map((error) => (error as Error).cause)).toEqual(["restart_limit"]);
|
||||
|
|
|
|||
|
|
@ -17,7 +17,7 @@ import {
|
|||
import { describe, expect, it } from "vitest";
|
||||
import { chatSetup } from "./chat-support.ts";
|
||||
import { ControlledStorage, context } from "./session-support.ts";
|
||||
import { completed, eventually } from "./task-support.ts";
|
||||
import { completed, deferred, eventually } from "./task-support.ts";
|
||||
|
||||
type Report = { readonly message: string; readonly info: ReportInfo };
|
||||
|
||||
|
|
@ -75,6 +75,43 @@ describe("reports", () => {
|
|||
await harness.close(context);
|
||||
});
|
||||
|
||||
it("name the extension whose definition ran, also after another one replaced it", async () => {
|
||||
const gate = deferred();
|
||||
const started = deferred();
|
||||
const definition = (name: string) =>
|
||||
defineTask<null, { phase: "run" }, null>({
|
||||
name: "test.replaced",
|
||||
version: 1,
|
||||
initial: () => ({ phase: "run" }),
|
||||
phases: {
|
||||
run: async () => {
|
||||
started.resolve();
|
||||
await gate.promise;
|
||||
throw new Error(`broken in ${name}`);
|
||||
},
|
||||
},
|
||||
abort: async () => {},
|
||||
});
|
||||
const Old = definition("old");
|
||||
const { harness, registry, reports } = await open([]);
|
||||
const old = defineExtension({ name: "old-extension", tasks: [Old] });
|
||||
registry.install(old);
|
||||
const root = await harness.root(context);
|
||||
const id = await root.commit((tx) => tx.createTask(Old, null, { ownership: { kind: "conversation" } }), context);
|
||||
harness.resume();
|
||||
await started.promise;
|
||||
registry.uninstall(old);
|
||||
registry.install(defineExtension({ name: "new-extension", tasks: [definition("new")] }));
|
||||
gate.resolve();
|
||||
await harness.waitForTask(id, context);
|
||||
await eventually(() => reports.some((report) => report.info.faulted === true));
|
||||
expect(reports.find((report) => report.info.faulted === true)).toMatchObject({
|
||||
message: "broken in old",
|
||||
info: { extension: "old-extension" },
|
||||
});
|
||||
await harness.close(context);
|
||||
});
|
||||
|
||||
it("name the extension of a failing section or hook, and tell them from a failing environment builder", async () => {
|
||||
const setup = chatSetup();
|
||||
setup.faux.setResponses([fauxAssistantMessage("answer")]);
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue