mirror of
https://github.com/badlogic/pi-mono.git
synced 2026-11-05 23:37:25 +00:00
fix(ai): separate deferred request options
This commit is contained in:
Mario Zechner
parent
f24ab6e14c
commit
686f193e51
8 changed files with 358 additions and 192 deletions
|
|
@ -178,7 +178,7 @@ Part II is backend-neutral. It defines the records a lane writes, when it writes
|
|||
|
||||
> Before an effect: write an intent record that names what will happen and the ids it will produce. After the effect: append the result as an entry with exactly those ids.
|
||||
|
||||
There is no multi-record atomicity and none is needed. Each record and each entry is durable alone. A crash between intent and result leaves the intent unfulfilled; recovery decides per intent type: complete it, retry it, or close it with a synthetic result. An intent is fulfilled if and only if an entry with its provisioned id exists. The entry can itself name the next durable state: an assistant entry with `stopReason: "deferred"` fulfills its attempt's provisioned append but suspends the step rather than closing it. A provisioned id that exists with different content is corruption.
|
||||
There is no multi-record atomicity and none is needed. Each record and each entry is durable alone. A crash between intent and result leaves the intent unfulfilled; recovery decides per intent type: complete it, retry it, or close it with a synthetic result. An intent is fulfilled if and only if an entry with its provisioned id exists. The entry can itself name the next durable state: an assistant entry with `stopReason: "deferred"` fulfills its attempt's provisioned append and closes the step; what stays outstanding is the operation — the persisted handle awaits redemption (section 6). A provisioned id that exists with different content is corruption.
|
||||
|
||||
### Provisioned ids
|
||||
|
||||
|
|
@ -194,7 +194,7 @@ type ProvisionedEntry<T extends Entry = Entry> =
|
|||
|
||||
### Record catalog
|
||||
|
||||
Every record belongs to one lane's operation log. Records that belong to an operation carry `runId`: the id of that operation's `operation_started` record. `queue_enqueued` for the next-run queue is the one record without `runId`; it is consumed by the lane's next run.
|
||||
Every record belongs to one lane's operation log. Records that belong to an operation carry `runId`: the id of that operation's `operation_started` record. Next-run queue records (`queue_enqueued` and their `queue_cancelled`) and standalone `adjustment` usage records carry no `runId`.
|
||||
|
||||
```ts
|
||||
interface RecordBase {
|
||||
|
|
@ -249,7 +249,6 @@ interface OperationStartedRecord extends RecordBase {
|
|||
interface AbortRequestedRecord extends RecordBase {
|
||||
type: "abort_requested";
|
||||
runId: string;
|
||||
reason: "user" | "shutdown";
|
||||
}
|
||||
|
||||
// Closes the operation. failed = orderly durable failure (for example,
|
||||
|
|
@ -266,8 +265,8 @@ interface OperationFinishedRecord extends RecordBase {
|
|||
// do this, for the n-th time. Steps are logged only because they are
|
||||
// retryable: the durable count caps retries across restarts — a
|
||||
// crash-restart loop cannot reset it. One record per attempt; one attempt
|
||||
// may make zero or several provider requests (hook-supplied summaries make
|
||||
// none, split-turn compaction makes two). Deferred results need no extra
|
||||
// may make zero or several provider requests (split-turn compaction
|
||||
// makes two). Deferred results need no extra
|
||||
// record: the handle lives in the persisted assistant entry (section 1).
|
||||
interface StepAttemptRecord extends RecordBase {
|
||||
type: "step_attempt";
|
||||
|
|
@ -432,7 +431,7 @@ R usage
|
|||
E assistant message
|
||||
```
|
||||
|
||||
Every provider request settles with a `usage` record (section 5); the other traces omit them for brevity.
|
||||
Every provider request settles with a `usage` record (section 5); the other traces omit them for brevity. Per-request hooks (`transform_context`, `before_request`, `after_response`) run inside every request and are omitted everywhere; Tier B records them (section 20).
|
||||
|
||||
Crash during backoff: restore counts two attempts; resume starts attempt 3. The count never resets. Retryable errors below the cap are never appended as entries. Attempts exhausted — or a non-retryable terminal error — appends an assistant message with the error, then `operation_finished` failed:
|
||||
|
||||
|
|
@ -474,7 +473,7 @@ R step_attempt step assistant, attempt 1 — new step
|
|||
E assistant message
|
||||
```
|
||||
|
||||
**One recovery per conversational input.** An overflow compaction may start only when no overflow-reason compaction `step_attempt` is newer than this run's newest consumed conversational message (prompt, steering, or follow-up). A second recoverable response inside that window appends the give-up error entry and fails the run through the drain path — a `length` response never resets the guard; only consumed conversational input does. This bounds the compact-and-retry loop at one attempt per user action. A `before_compaction` decline or an empty compaction preparation for reason `overflow` is equally terminal: without compaction the request cannot fit.
|
||||
**One recovery per conversational input.** An overflow compaction may start only when no overflow-reason compaction `step_attempt` is newer than this run's newest consumed conversational message (prompt, steering, or follow-up). A second recoverable response inside that window appends the give-up error entry and fails the run through the drain path — a `length` response never resets the guard; only consumed conversational input does. This bounds the compact-and-retry loop at one attempt per user action. A `before_compaction` decline or an empty compaction preparation for reason `overflow` is equally terminal: without compaction the request cannot fit. A hook-supplied overflow compaction writes its compaction `step_attempt` before the entry so the guard counts it — the one hook-supplied summary that writes an attempt record.
|
||||
|
||||
Per crash site:
|
||||
|
||||
|
|
@ -623,7 +622,7 @@ E assistant message stop reason deferred, carries the handle
|
|||
... hours pass, maybe a different process ...
|
||||
resume() newest entry on the lane's path is a deferred
|
||||
assistant message with no successor
|
||||
→ the attempt is outstanding, redeem it
|
||||
→ the handle is unredeemed, redeem it
|
||||
fetchDeferred(model, handle) model and handle from that entry
|
||||
E assistant message the real result
|
||||
run continues normally
|
||||
|
|
@ -633,7 +632,7 @@ The suspended lane is indistinguishable from a crashed one in storage: an open o
|
|||
|
||||
Each `resume()` performs one fetch. Three outcomes:
|
||||
|
||||
- **pending** — the provider returns stop reason `deferred` again. Nothing is written; the lane re-suspends. Poll cadence is application policy.
|
||||
- **pending** — the provider returns stop reason `deferred` again. Nothing but a possible `usage` record is written (section 15); the lane re-suspends. Poll cadence is application policy.
|
||||
- **ready** — a normal assistant message. It is appended as the successor and the run continues.
|
||||
- **terminal** — the provider returns stop reason `error` (expired, unknown, consumed), or the fetch itself rejects; the harness converts a rejection to the same error-message form. The message is appended and the run finishes failed. Redemption failure never starts an automatic replacement request; steering or follow-up input already accepted for this run can still start a later turn.
|
||||
|
||||
|
|
@ -659,10 +658,11 @@ Both reads are bounded by the size of the open operation, not by the size of the
|
|||
From those two reads, the lane's state:
|
||||
|
||||
- **aborting** — an `abort_requested` record exists.
|
||||
- **attempts used** — `step_attempt` records whose `resultEntryId` has no entry. A step is closed exactly when its provisioned result exists — a point lookup, not adjacency inference; attempts whose result landed belong to finished work.
|
||||
- **attempts used** — the newest `step_attempt` whose `resultEntryId` has no entry is the unfinished step; its `attempt` field is the durable count, its kind and `compactionReason` select the resume path. Closure is a point lookup, not adjacency inference: a step is closed exactly when the newest attempt's provisioned result exists. Earlier attempts' unfulfilled ids belong to finished work and need no inspection.
|
||||
- **overflow recovery used** — a compaction `step_attempt` with reason `overflow` is newer than the newest consumed conversational message of this run (section 6, overflow guard).
|
||||
- **tool batch** — the newest assistant entry with tool calls, each call matched against `tool_started` records and result entries (section 6, crash-site table). The assistant stop reason is retained: a `length` batch is truncated and never executes on recovery. Persisted `terminate` values on result entries decide whether the completed batch forces another turn.
|
||||
- **deferred handle** — the newest own entry is a deferred assistant message with no successor.
|
||||
- **newest own entry** — the last entry of the second read; the pure predicates (`needsAssistant()`, terminal failure, abort closure) read it.
|
||||
- **pending queue items** — `queue_enqueued` records whose provisioned entry does not exist, excluding items retracted by `queue_cancelled` and steer/follow-up items killed by this run's `abort_requested`.
|
||||
- **pending writes** — `write_deferred` records whose provisioned entry does not exist.
|
||||
- **missing initial messages** — provisioned ids from the run intent without entries.
|
||||
|
|
@ -834,6 +834,9 @@ interface AgentHarnessOptions {
|
|||
compaction?: CompactionSettings;
|
||||
steeringMode?: QueueMode;
|
||||
followUpMode?: QueueMode;
|
||||
/** Batch default; a called tool declaring executionMode "sequential"
|
||||
forces sequential regardless (section 14). */
|
||||
toolExecution?: "sequential" | "parallel"; // default parallel
|
||||
/** automatic: operation methods drive their procedures to completion.
|
||||
manual: the operation's effects park at the gate; peekAction() /
|
||||
executeAction() / runToCompletion() drive them. Deterministic tests
|
||||
|
|
@ -1034,6 +1037,8 @@ Calls on a faulted harness reject with the same `HarnessFault` instance until th
|
|||
|
||||
`finalMessage` is the run's newest entry that projects to an assistant message; `finalEntryId` is that entry's id. `leafId` is the lane's leaf when the operation finished — the race-free anchor for branch queries (`findEntriesOnBranch({ start: leafId })`). The two differ when a deferred write was applied after the final assistant message. Full transcripts are not duplicated into results; they are in the session and were delivered as events.
|
||||
|
||||
**Type provenance.** Types this document uses but does not redefine — `QueueMode`, `RetryPolicy`, `CompactionSettings`, `CompactionPreparation`, `NavigationPreparation`, `CompactResult`, `ToolResultPatch`, `SessionStats`, `SessionMetadata`, `NavigateOptions`, `EntryCursor`, `LogItem`, `StreamOptionsPatch` — keep their existing `harness/types.ts` shapes. Lowercase helpers in section 15 pseudocode without a definition (`preparation`, `runToolBatchForSingleCall`, request/option bags such as `AssistantRequest` and `FactWrite`) are constructive implementation detail, not contract.
|
||||
|
||||
### Suspended operations
|
||||
|
||||
```ts
|
||||
|
|
@ -1170,14 +1175,14 @@ Guarantees:
|
|||
- Events that report durable facts fire after the fact is committed; what an event announces is already queryable.
|
||||
- Events report final values, after hook transformation.
|
||||
- Payloads are JSON-serializable and secret-free; a server can proxy them verbatim. Live objects (models, tools) are referenced by name, never embedded.
|
||||
- Lane-scoped events carry `lane: string` (omitted below); harness-global events such as `fault` omit it. Operation-scoped events carry `runId`; turn-scoped events carry `turnId`; recovered work carries `recovery: true`.
|
||||
- Lane-scoped events carry `lane: string` (omitted below); harness-global events omit it — except `usage`, which is delivered harness-globally and carries the record's lane in its payload. Operation-scoped events carry `runId`; turn-scoped events carry `turnId`; recovered work carries `recovery: true`.
|
||||
|
||||
### Catalog
|
||||
|
||||
```ts
|
||||
// Run lifecycle
|
||||
{ type: "run_start"; runId }
|
||||
{ type: "run_resume"; runId } // resume() entered
|
||||
{ type: "run_resume"; runId } // resume() entered (any operation kind)
|
||||
{ type: "run_suspend"; runId; deferred: DeferredHandle } // lane parked
|
||||
{ type: "run_abort"; runId; steer: AgentMessage[]; followUp: AgentMessage[] } // abort accepted; cleared payloads
|
||||
{ type: "run_end"; runId; outcome: "completed" | "aborted" | "failed";
|
||||
|
|
@ -1377,12 +1382,7 @@ before_compaction: {
|
|||
|
||||
before_navigation: {
|
||||
event: { targetId; preparation: NavigationPreparation };
|
||||
result: {
|
||||
decline?: boolean;
|
||||
summary?: { summary: string; details?; usage? };
|
||||
customInstructions?: string;
|
||||
label?: string;
|
||||
} | undefined;
|
||||
result: { decline?: boolean; summary?: { summary: string; details?; usage? } } | undefined;
|
||||
}
|
||||
```
|
||||
|
||||
|
|
@ -1416,7 +1416,8 @@ interface EntryBase {
|
|||
timestamp: number; // Unix ms, storage-assigned
|
||||
}
|
||||
|
||||
interface MessageEntry extends EntryBase { type: "message"; message: AgentMessage }
|
||||
interface MessageEntry extends EntryBase { type: "message"; message: AgentMessage;
|
||||
terminate?: true }
|
||||
interface ModelChangeEntry extends EntryBase { type: "model_change"; provider: string; modelId: string }
|
||||
interface ThinkingLevelEntry extends EntryBase { type: "thinking_level_change"; thinkingLevel: string }
|
||||
interface ActiveToolsEntry extends EntryBase { type: "active_tools_change"; activeToolNames: string[] }
|
||||
|
|
@ -1804,7 +1805,7 @@ export interface ToolCallbacks {
|
|||
args?: Record<string, unknown>;
|
||||
block?: { reason: string };
|
||||
} | undefined>;
|
||||
afterToolCall?(call, result, isError, signal): Promise<ToolResultPatch | undefined>;
|
||||
afterToolCall?(call, args, result, isError, signal): Promise<ToolResultPatch | undefined>;
|
||||
/** Between phases 1 and 2: the durability point. The harness writes its
|
||||
tool_started record here. Called in source order in both modes —
|
||||
preparation is always sequential. */
|
||||
|
|
@ -1906,7 +1907,7 @@ The jobs, by caller:
|
|||
- **Lane surface** (ungated, enqueue directly):
|
||||
- *Operation acceptance* — validate idle, capture the pending `nextRun` items into `initialMessages`, write `operation_started`, set `state.operation`. The second of two concurrent acceptances sees the first and rejects `busy` with no write. `before_run` ran before this job, outside the line, on the prompt only.
|
||||
- *Queue acceptance* (`steer`, `followUp`) — validate an active, non-aborting run; write `queue_enqueued`. `nextRun` validates nothing and always accepts.
|
||||
- *Queue cancellation* (`cancelQueued`) — target entry exists: `already_consumed`; not pending (abort-drained or already cancelled): `already_cleared`; else write `queue_cancelled` and remove the item from its pending set.
|
||||
- *Queue cancellation* (`cancelQueued`) — no `queue_enqueued` for the id: `Err(UnknownQueueItem)`; target entry exists: `already_consumed`; not pending (abort-drained or already cancelled): `already_cleared`; else write `queue_cancelled` and remove the item from its pending set.
|
||||
- *Deferred-write acceptance* (lane-view writes, config setters) — run open: write `write_deferred`; structural operation open: wait for it to end, then re-enter; idle: append the entry directly.
|
||||
- *Abort* — write `abort_requested`, set `aborting`, drain `pendingSteer`/`pendingFollowUp` (payloads return to the abort caller and in the `run_abort` event), signal the active effect's `AbortController`.
|
||||
- *Resume admission* — reserve the lane's single execution slot; no write.
|
||||
|
|
@ -2013,6 +2014,7 @@ Semantics that make tests deterministic:
|
|||
records and own entries (section 7): live commits update it; restore
|
||||
recomputes it. */
|
||||
interface LaneState {
|
||||
lane: string;
|
||||
leafId: string | null;
|
||||
operation: null | {
|
||||
id: string;
|
||||
|
|
@ -2076,14 +2078,17 @@ async function appendIfMissing(target: ProvisionedEntry): Promise<void> {
|
|||
```ts
|
||||
async function resume(): Promise<ResumeResult> {
|
||||
if (missing.tools.length || missing.models.length) {
|
||||
return Result.err(new MissingIdentities({ lane: laneName(state), ...missing,
|
||||
return Result.err(new MissingIdentities({ lane: state.lane, ...missing,
|
||||
message: "Missing tools or models" }));
|
||||
}
|
||||
await fx.runHook("before_resume", beforeResumeEvent(state)); // per registration id (section 11)
|
||||
emit({ type: "run_resume", runId: op.id, recovery: true });
|
||||
// tagResume re-tags an operation Result as a ResumeResult: Ok gains
|
||||
// { operation }, Err passes through unchanged.
|
||||
switch (op.kind) {
|
||||
case "run": return { kind: "run", ...await runProcedure() };
|
||||
case "compaction": return { kind: "compaction", ...await compactionProcedure() };
|
||||
case "navigation": return { kind: "navigation", ...await navigationProcedure() };
|
||||
case "run": return tagResume("run", await runProcedure());
|
||||
case "compaction": return tagResume("compaction", await compactionProcedure());
|
||||
case "navigation": return tagResume("navigation", await navigationProcedure());
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -2126,6 +2131,9 @@ async function handleRunSignal(e: unknown): Promise<RunResult> {
|
|||
}
|
||||
```
|
||||
|
||||
|
||||
**Fixed-point self-check.** When `resume()` completes, parks, or closes its operation, the harness recomputes the section 7 reduction from storage and compares it to the live `LaneState`. A mismatch is corruption and faults the harness — writer/reducer drift is caught the moment it happens instead of one crash later. The check is cheap (the same two bounded reads restore performs) and runs in production, not only under test.
|
||||
|
||||
### The loop
|
||||
|
||||
```ts
|
||||
|
|
@ -2212,7 +2220,7 @@ async function handleRunFailed(error: OperationError): Promise<RunResult> {
|
|||
|
||||
### Steps
|
||||
|
||||
A failed attempt appends nothing; only a deferred handle, a terminal message, or the final give-up error enters the tree (section 6, retry trace).
|
||||
A failed attempt appends nothing. Besides the successful response, only a deferred handle, a terminal message, or the final give-up error enters the tree (section 6, retry trace).
|
||||
|
||||
```ts
|
||||
async function assistantStep(): Promise<AssistantMessage> {
|
||||
|
|
@ -2220,9 +2228,10 @@ async function assistantStep(): Promise<AssistantMessage> {
|
|||
if (op.aborting) throw new Aborted();
|
||||
const attempt = (op.step?.kind === "assistant" ? op.step.attempts : 0) + 1;
|
||||
if (attempt > retry.maxAttempts) {
|
||||
const error = retriesExhausted();
|
||||
// The give-up entry fulfills the last attempt's provisioned id.
|
||||
await fx.appendEntry(giveUpAssistantEntry(lastAttemptResultId(op), state));
|
||||
throw new RunFailed(retriesExhausted());
|
||||
await fx.appendEntry(giveUpAssistantEntry(lastAttemptResultId(op), state, error));
|
||||
throw new RunFailed(error);
|
||||
}
|
||||
|
||||
const options = await fx.runHook("before_request",
|
||||
|
|
@ -2255,7 +2264,7 @@ async function assistantStep(): Promise<AssistantMessage> {
|
|||
|
||||
`isRecoverableOverflow(final, state)` is `isContextOverflow(final)` — overflow-pattern errors and silent overflow — or `isRecoverableLength(final, desiredMaxOutput(state))` from section 6, where `desiredMaxOutput(state)` is the caller-supplied `maxTokens` when set, else the lane model's `maxTokens`. The check runs before the retryable-error branch: an overflow-form error compacts instead of retrying the same oversized request.
|
||||
|
||||
`summaryStep(step, reason, resultEntryId)` has the same shape: `step_attempt` before each attempt (`compactionReason` for compaction steps) carrying the step's single result id, `before_request`, one or two non-deferred requests — each followed by its `usage` record bound to that id — durable cap. It returns the summary value; the caller appends the result entry under that id. A hook-supplied summary makes no request and no request record; if it carries usage the hook measured itself, the appending procedure writes a `hook` usage record beside the entry.
|
||||
`summaryStep(step, reason, resultEntryId)` has the same shape: `step_attempt` before each attempt (`compactionReason` for compaction steps) carrying the step's single result id, `before_request`, one or two non-deferred requests — each followed by its `usage` record bound to that id — durable cap. It returns the summary value; the caller appends the result entry under that id. A hook-supplied summary makes no request and no request record; if it carries usage the hook measured itself, the appending procedure writes a `hook` usage record beside the entry. For reason `overflow` the appending procedure also writes the compaction `step_attempt`, so the once-per-input guard counts the recovery (section 6).
|
||||
|
||||
### Deferred redemption
|
||||
|
||||
|
|
@ -2267,7 +2276,10 @@ async function redeemDeferred(): Promise<AssistantMessage> {
|
|||
await fx.appendRecord(usageRecord("deferred_fetch", op.id, resultEntryId, 1, final));
|
||||
}
|
||||
if (op.aborting) throw new Aborted();
|
||||
if (final.stopReason === "deferred") throw new Park(op.deferred!); // pending; no other write
|
||||
if (final.stopReason === "deferred") {
|
||||
requireSameHandle(final.deferred, op.deferred!); // mismatch is a defect (section 16)
|
||||
throw new Park(op.deferred!); // pending; no other write
|
||||
}
|
||||
if (final.stopReason === "aborted") throw new Aborted();
|
||||
|
||||
await fx.appendEntry(assistantEntry(resultEntryId, final)); // ready or terminal
|
||||
|
|
@ -2278,8 +2290,6 @@ async function redeemDeferred(): Promise<AssistantMessage> {
|
|||
|
||||
One fetch per `resume()`. Pending re-parks without a write. A terminal answer — returned or converted from a rejected fetch — lands as the error entry and fails the run through the normal drain path, which still honors input accepted before the failure (section 6).
|
||||
|
||||
**Fixed-point self-check.** When `resume()` completes, parks, or closes its operation, the harness recomputes the section 7 reduction from storage and compares it to the live `LaneState`. A mismatch is corruption and faults the harness — writer/reducer drift is caught the moment it happens instead of one crash later. The check is cheap (the same two bounded reads restore performs) and runs in production, not only under test.
|
||||
|
||||
### Tools
|
||||
|
||||
The live path is section 14 `executeToolBatch`; the durability callbacks route through `fx`, so the gate and the traces see every write in order:
|
||||
|
|
@ -2304,8 +2314,8 @@ async function runToolBatch(assistant: AssistantMessage): Promise<void> {
|
|||
replay: declaredReplay(call),
|
||||
}));
|
||||
},
|
||||
afterToolCall: (call, result, isError) =>
|
||||
fx.runHook("after_tool", { toolCallId: call.id, toolName: call.name, ...result, isError }),
|
||||
afterToolCall: (call, args, result, isError) =>
|
||||
fx.runHook("after_tool", { toolCallId: call.id, toolName: call.name, args, ...result, isError }),
|
||||
onToolResult: async (message, terminate) => {
|
||||
// Blocked/invalid calls have no tool_started and no provisioned id;
|
||||
// their error result entry gets a fresh id (section 5).
|
||||
|
|
@ -2389,9 +2399,13 @@ async function compactionProcedure(): Promise<CompactionResult> {
|
|||
let result: CompactResult | undefined;
|
||||
if (!op.step) { // no attempt yet: the decision hook may still run
|
||||
const hook = await fx.runHook("before_compaction",
|
||||
{ reason: "manual", preparation, customInstructions: op.intent.customInstructions });
|
||||
{ reason: "manual", preparation: preparation(state),
|
||||
customInstructions: op.intent.customInstructions });
|
||||
if (hook?.decline) return await finishStructural("declined");
|
||||
result = hook?.compaction;
|
||||
if (result?.usage) {
|
||||
await fx.appendRecord(hookUsageRecord(op.id, op.intent.resultEntryId, result.usage));
|
||||
}
|
||||
}
|
||||
result ??= await summaryStep("compaction", "manual", op.intent.resultEntryId);
|
||||
await appendIfMissing(compactionEntry(op.intent.resultEntryId, result));
|
||||
|
|
@ -2408,14 +2422,25 @@ async function compactionProcedure(): Promise<CompactionResult> {
|
|||
RunFailed: without compaction the request cannot fit (section 6). */
|
||||
async function autoCompact(reason: "threshold" | "overflow"): Promise<void> {
|
||||
const resultEntryId = op.step?.kind === "compaction" ? op.step.resultEntryId : newId();
|
||||
if (!op.step) {
|
||||
const hook = await fx.runHook("before_compaction",
|
||||
{ reason, preparation: preparation(state) });
|
||||
if (op.step?.kind !== "compaction") { // no durable compaction decision yet; on the overflow
|
||||
// path op.step is the abandoned assistant step
|
||||
const prep = preparation(state);
|
||||
if (prep.nothingToCompact) {
|
||||
if (reason === "overflow") throw new RunFailed(truncationError());
|
||||
return;
|
||||
}
|
||||
const hook = await fx.runHook("before_compaction", { reason, preparation: prep });
|
||||
if (hook?.decline) {
|
||||
if (reason === "overflow") throw new RunFailed(truncationError());
|
||||
return;
|
||||
}
|
||||
if (hook?.compaction) {
|
||||
if (reason === "overflow") { // the once-per-input guard counts this attempt
|
||||
await fx.appendRecord(stepAttempt(op.id, "compaction", 1, resultEntryId, reason));
|
||||
}
|
||||
if (hook.compaction.usage) {
|
||||
await fx.appendRecord(hookUsageRecord(op.id, resultEntryId, hook.compaction.usage));
|
||||
}
|
||||
await appendIfMissing(compactionEntry(resultEntryId, hook.compaction));
|
||||
return;
|
||||
}
|
||||
|
|
@ -2433,10 +2458,14 @@ async function navigationProcedure(): Promise<NavigationResult> {
|
|||
if (op.intent.summarize && !op.targets.summary) {
|
||||
if (!moved && !op.step) { // decision hook: once, pre-move
|
||||
const hook = await fx.runHook("before_navigation",
|
||||
{ targetId: op.intent.targetId, preparation }); // preparation derives from
|
||||
{ targetId: op.intent.targetId,
|
||||
preparation: preparation(state) }); // preparation derives from
|
||||
// intent.sourceLeafId — valid pre- and post-move
|
||||
if (hook?.decline) return await finishStructural("declined");
|
||||
summary = hook?.summary;
|
||||
if (summary?.usage) {
|
||||
await fx.appendRecord(hookUsageRecord(op.id, op.intent.summaryEntryId!, summary.usage));
|
||||
}
|
||||
}
|
||||
summary ??= await summaryStep("branch_summary", undefined,
|
||||
op.intent.summaryEntryId!); // regenerates after a post-move crash
|
||||
|
|
@ -2487,6 +2516,7 @@ Hook-to-block wiring, in one table:
|
|||
| `before_tool` | `ToolCallbacks.beforeToolCall` (phase 1) |
|
||||
| `after_tool` | `ToolCallbacks.afterToolCall` (phase 3) |
|
||||
| `before_run_end` | `driverLoop` finish boundary; result committed via `fx.commitRunEndFollowUp` |
|
||||
| `before_resume` | `resume()` dispatch, before any effect |
|
||||
| — (record/entry writes) | `ToolCallbacks.onToolStart` / `onToolResult` via `fx` |
|
||||
|
||||
Notes:
|
||||
|
|
@ -2531,6 +2561,30 @@ interface AssistantMessage {
|
|||
deferred?: DeferredHandle; // present iff stopReason === "deferred"
|
||||
}
|
||||
|
||||
// Authenticated HTTP request plumbing shared by stream, image, and deferred
|
||||
// provider operations. Generation and streaming-transport controls are not
|
||||
// part of this interface.
|
||||
interface ProviderRequestOptions {
|
||||
signal?: AbortSignal;
|
||||
apiKey?: string;
|
||||
fetch?: FetchFunction;
|
||||
env?: ProviderEnv;
|
||||
onPayload?: (payload: unknown, model: Model<Api>) =>
|
||||
unknown | undefined | Promise<unknown | undefined>;
|
||||
onResponse?: (response: ProviderResponse, model: Model<Api>) => void | Promise<void>;
|
||||
headers?: ProviderHeaders;
|
||||
timeoutMs?: number;
|
||||
maxRetries?: number;
|
||||
maxRetryDelayMs?: number;
|
||||
}
|
||||
|
||||
interface DeferredFetchOptions extends ProviderRequestOptions {
|
||||
/** Maximum provider long-poll duration. Omitted or zero checks once. */
|
||||
wait?: number;
|
||||
}
|
||||
|
||||
type DeferredCancelOptions = ProviderRequestOptions;
|
||||
|
||||
// Redemption lives on the provider. The two methods are optional: their
|
||||
// presence is the capability signal. A provider without them never returns
|
||||
// stopReason "deferred" and ignores the deferred request option.
|
||||
|
|
@ -2546,28 +2600,30 @@ export interface ProviderStreams {
|
|||
`wait` expires; wait: 0 checks once)
|
||||
- terminal: stopReason "error" (expired, unknown, consumed) */
|
||||
fetchDeferred?(model: Model<Api>, handle: DeferredHandle,
|
||||
options?: { wait?: number; signal?: AbortSignal }): AssistantMessageEventStream;
|
||||
options?: DeferredFetchOptions): AssistantMessageEventStream;
|
||||
|
||||
/** Best effort; providers without cancellation omit it. */
|
||||
cancelDeferred?(model: Model<Api>, handle: DeferredHandle): Promise<void>;
|
||||
cancelDeferred?(model: Model<Api>, handle: DeferredHandle,
|
||||
options?: DeferredCancelOptions): Promise<void>;
|
||||
}
|
||||
```
|
||||
|
||||
The harness never talks to a provider object directly; it uses the same authenticated dispatch surface as ordinary requests:
|
||||
|
||||
```ts
|
||||
type ModelsDeferredOptions = StreamOptions & ModelsStreamTransforms;
|
||||
type ModelsDeferredFetchOptions = DeferredFetchOptions & ModelsRequestTransforms;
|
||||
type ModelsDeferredCancelOptions = DeferredCancelOptions & ModelsRequestTransforms;
|
||||
|
||||
interface Models {
|
||||
// existing methods
|
||||
fetchDeferred(model: Model<Api>, handle: DeferredHandle,
|
||||
options?: ModelsDeferredOptions): Promise<AssistantMessage>;
|
||||
options?: ModelsDeferredFetchOptions): Promise<AssistantMessage>;
|
||||
cancelDeferred(model: Model<Api>, handle: DeferredHandle,
|
||||
options?: ModelsDeferredOptions): Promise<void>;
|
||||
options?: ModelsDeferredCancelOptions): Promise<void>;
|
||||
}
|
||||
```
|
||||
|
||||
`Models.fetchDeferred` and `Models.cancelDeferred` delegate to the provider methods with normal model resolution and authentication (credential store, expiring tokens, header merge); `ModelsDeferredOptions` carries the normal `AbortSignal`, transport, response callbacks, and model transforms. A provider that returns `stopReason: "deferred"` must implement fetch; cancellation is best effort.
|
||||
`Models.fetchDeferred` and `Models.cancelDeferred` delegate to the provider methods with normal model resolution and authentication (credential store, expiring tokens, header merge). Their options carry the normal HTTP request settings, lifecycle callbacks, and model transforms; fetch options additionally carry the provider long-poll duration. A provider that returns `stopReason: "deferred"` must implement fetch; cancellation is best effort.
|
||||
|
||||
A terminal fetch answer is final for the run: the harness appends the error message and fails the operation (section 6). It never starts an automatic replacement request. The executor converts a rejected fetch promise into the same `stopReason: "error"` message form, so expected provider and authentication failures stay in-band. On a returned pending message the harness requires the complete handle to equal the persisted handle: a provider cannot replace durable handle data without a write, so a mismatch is a defect.
|
||||
|
||||
|
|
@ -2686,7 +2742,6 @@ pi.harness.run runId, lane, recovery
|
|||
|
||||
pi.harness.compaction manual operation
|
||||
pi.harness.navigation
|
||||
pi.harness.resume
|
||||
```
|
||||
|
||||
The harness owns the operation and turn spans. The `fx` implementation owns checkpoint, step, request, tool, hook, and append spans: `fx.streamAssistant` creates the retryable step span and its request children, and each tool step gets its own tool span. This split follows ownership of the corresponding work.
|
||||
|
|
@ -2735,7 +2790,7 @@ The in-memory backend is the reference. The parity suite runs the same setups ag
|
|||
|
||||
### Tier B — writer conformance
|
||||
|
||||
Tier A assumes live execution writes the correct prefix; Tier B verifies it. Run the public harness against an instrumented `Session` recording every entry (`E`), record (`R`), fact (`G`), and hook (`H`). Assert exact order against the section 6 traces: one-tool run, retry, terminal failure, steering during a tool, finish-boundary orders, deferred write mid-turn, abort during a tool, auto-compaction, manual compaction, navigation (move-first), deferred suspension and every fetch outcome. This tier catches the critical regression class: an effect starting before its intent record.
|
||||
Tier A assumes live execution writes the correct prefix; Tier B verifies it. Run the public harness against an instrumented `Session` recording every entry (`E`), record (`R`), lane move (`L`), fact (`G`), and hook (`H`). Assert exact order against the section 6 traces: one-tool run, retry, terminal failure, steering during a tool, queue cancellation, finish-boundary orders, deferred write mid-turn, abort during a tool, auto-compaction, context overflow (discard, guard, hook-supplied), manual compaction, navigation (move-first), deferred suspension and every fetch outcome. This tier catches the critical regression class: an effect starting before its intent record.
|
||||
|
||||
Tier B also asserts the append-only-context invariant (section 4) executably: within a run, every faux-provider request's message list extends the previous request's as an exact prefix — except across a compaction entry, the one sanctioned invalidation. This turns the KV-cache discipline from prose into a failing test whenever a write path inserts before the tail.
|
||||
|
||||
|
|
@ -2762,8 +2817,7 @@ Crash simulation is `close()` at a chosen boundary, then reopening the same back
|
|||
|
||||
Gate invariants, asserted across Tier C:
|
||||
|
||||
- after every `resume()` outcome, the recomputed reduction equals live `LaneState` (the section 15 fixed-point self-check fired and passed);
|
||||
|
||||
- After every `resume()` outcome, the recomputed reduction equals live `LaneState` (the section 15 fixed-point self-check fired and passed).
|
||||
- `peekAction()` has no side effect and is stable until `executeAction()`.
|
||||
- `executeAction()` releases exactly the peeked action, never a later one.
|
||||
- Stopping before an action leaves exactly the preceding durable prefix.
|
||||
|
|
|
|||
|
|
@ -4,6 +4,7 @@
|
|||
|
||||
### Breaking Changes
|
||||
|
||||
- Renamed the exported `ModelsStreamTransforms` interface to `ModelsRequestTransforms` because its header transformation now applies to all authenticated provider requests.
|
||||
- Required dynamic model providers to accept a concrete `RefreshModelsContext.signal`; `Models.refresh()` remains unbounded when callers omit its optional signal.
|
||||
- Required provider login, API-key check/resolution, and OAuth refresh implementations to accept a concrete abort signal; public auth and credential operations remain unbounded when callers omit their optional signal.
|
||||
- Replaced raw `RefreshModelsContext.store` access with the read-only `context.stored` snapshot and generation-checked `context.publish()` transaction.
|
||||
|
|
@ -67,6 +68,7 @@
|
|||
|
||||
### Added
|
||||
|
||||
- Added deferred provider request contracts, durable response handles, authenticated fetch/cancel dispatch, and faux-provider support for pending, ready, failed, and cancelled responses ([#7339](https://github.com/earendil-works/pi/pull/7339) by [@davidbrai](https://github.com/davidbrai)).
|
||||
- Added Baseten as a built-in OpenAI-compatible provider with models.dev catalog generation and native `chat_template_args` reasoning controls.
|
||||
|
||||
### Changed
|
||||
|
|
|
|||
|
|
@ -20,15 +20,16 @@ import type {
|
|||
AssistantMessage,
|
||||
AssistantMessageEventStream,
|
||||
Context,
|
||||
DeferredCancelOptions,
|
||||
DeferredFetchOptions,
|
||||
DeferredHandle,
|
||||
Model,
|
||||
ModelCostRates,
|
||||
ModelThinkingLevel,
|
||||
ProviderHeaders,
|
||||
ProviderRequestOptions,
|
||||
ProviderStreams,
|
||||
SimpleStreamOptions,
|
||||
StreamOptions,
|
||||
Usage,
|
||||
} from "./types.ts";
|
||||
import { operationSignal, raceWithAbortSignal } from "./utils/abort.ts";
|
||||
|
|
@ -74,14 +75,15 @@ export interface ModelsRefreshResult {
|
|||
errors: ReadonlyMap<string, Error>;
|
||||
}
|
||||
|
||||
export interface ModelsStreamTransforms {
|
||||
export interface ModelsRequestTransforms {
|
||||
/** Transform fully assembled model/auth/request headers before provider dispatch. */
|
||||
transformHeaders?: (headers: ProviderHeaders) => ProviderHeaders | Promise<ProviderHeaders>;
|
||||
}
|
||||
|
||||
export type ModelsApiStreamOptions<TApi extends Api> = ApiStreamOptions<TApi> & ModelsStreamTransforms;
|
||||
export type ModelsSimpleStreamOptions = SimpleStreamOptions & ModelsStreamTransforms;
|
||||
export type ModelsDeferredOptions = DeferredFetchOptions & ModelsStreamTransforms;
|
||||
export type ModelsApiStreamOptions<TApi extends Api> = ApiStreamOptions<TApi> & ModelsRequestTransforms;
|
||||
export type ModelsSimpleStreamOptions = SimpleStreamOptions & ModelsRequestTransforms;
|
||||
export type ModelsDeferredFetchOptions = DeferredFetchOptions & ModelsRequestTransforms;
|
||||
export type ModelsDeferredCancelOptions = DeferredCancelOptions & ModelsRequestTransforms;
|
||||
|
||||
/**
|
||||
* A provider is the concrete runtime unit. It owns id/name/base metadata,
|
||||
|
|
@ -143,7 +145,7 @@ export interface Provider<TApi extends Api = Api> {
|
|||
handle: DeferredHandle,
|
||||
options?: DeferredFetchOptions,
|
||||
): AssistantMessageEventStream;
|
||||
cancelDeferred?(model: Model<TApi>, handle: DeferredHandle, options?: StreamOptions): Promise<void>;
|
||||
cancelDeferred?(model: Model<TApi>, handle: DeferredHandle, options?: DeferredCancelOptions): Promise<void>;
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
@ -212,8 +214,12 @@ export interface Models {
|
|||
|
||||
streamSimple(model: Model<Api>, context: Context, options?: ModelsSimpleStreamOptions): AssistantMessageEventStream;
|
||||
completeSimple(model: Model<Api>, context: Context, options?: ModelsSimpleStreamOptions): Promise<AssistantMessage>;
|
||||
fetchDeferred(model: Model<Api>, handle: DeferredHandle, options?: ModelsDeferredOptions): Promise<AssistantMessage>;
|
||||
cancelDeferred(model: Model<Api>, handle: DeferredHandle, options?: ModelsDeferredOptions): Promise<void>;
|
||||
fetchDeferred(
|
||||
model: Model<Api>,
|
||||
handle: DeferredHandle,
|
||||
options?: ModelsDeferredFetchOptions,
|
||||
): Promise<AssistantMessage>;
|
||||
cancelDeferred(model: Model<Api>, handle: DeferredHandle, options?: ModelsDeferredCancelOptions): Promise<void>;
|
||||
}
|
||||
|
||||
export interface MutableModels extends Models {
|
||||
|
|
@ -627,10 +633,13 @@ class ModelsImpl implements MutableModels {
|
|||
return provider;
|
||||
}
|
||||
|
||||
private async applyAuth<TOptions extends StreamOptions & ModelsStreamTransforms>(
|
||||
private async applyAuth<TOptions extends ProviderRequestOptions & ModelsRequestTransforms>(
|
||||
model: Model<Api>,
|
||||
options: TOptions | undefined,
|
||||
): Promise<{ requestModel: Model<Api>; requestOptions: StreamOptions | undefined }> {
|
||||
): Promise<{
|
||||
requestModel: Model<Api>;
|
||||
requestOptions: Omit<TOptions, "transformHeaders"> & ProviderRequestOptions;
|
||||
}> {
|
||||
this.requireProvider(model);
|
||||
const resolution = await this.getAuth(model, {
|
||||
apiKey: options?.apiKey,
|
||||
|
|
@ -649,7 +658,8 @@ class ModelsImpl implements MutableModels {
|
|||
const env = resolution.env || options?.env ? { ...(resolution.env ?? {}), ...(options?.env ?? {}) } : undefined;
|
||||
const requestModel = auth.baseUrl ? { ...model, baseUrl: auth.baseUrl } : model;
|
||||
const { transformHeaders: _transformHeaders, ...providerOptions } = options ?? {};
|
||||
const requestOptions = { ...providerOptions, apiKey, headers, env } as StreamOptions;
|
||||
const requestOptions = { ...providerOptions, apiKey, headers, env } as Omit<TOptions, "transformHeaders"> &
|
||||
ProviderRequestOptions;
|
||||
|
||||
return { requestModel, requestOptions };
|
||||
}
|
||||
|
|
@ -696,7 +706,7 @@ class ModelsImpl implements MutableModels {
|
|||
async fetchDeferred(
|
||||
model: Model<Api>,
|
||||
handle: DeferredHandle,
|
||||
options?: ModelsDeferredOptions,
|
||||
options?: ModelsDeferredFetchOptions,
|
||||
): Promise<AssistantMessage> {
|
||||
return lazyStream(model, async () => {
|
||||
const provider = this.requireProvider(model);
|
||||
|
|
@ -708,7 +718,11 @@ class ModelsImpl implements MutableModels {
|
|||
}).result();
|
||||
}
|
||||
|
||||
async cancelDeferred(model: Model<Api>, handle: DeferredHandle, options?: ModelsDeferredOptions): Promise<void> {
|
||||
async cancelDeferred(
|
||||
model: Model<Api>,
|
||||
handle: DeferredHandle,
|
||||
options?: ModelsDeferredCancelOptions,
|
||||
): Promise<void> {
|
||||
const provider = this.requireProvider(model);
|
||||
if (!provider.cancelDeferred) {
|
||||
throw new ModelsError("provider", `Provider ${model.provider} does not support deferred responses`);
|
||||
|
|
|
|||
|
|
@ -3,6 +3,7 @@ import type {
|
|||
AssistantMessage,
|
||||
AssistantMessageEventStream,
|
||||
Context,
|
||||
DeferredCancelOptions,
|
||||
DeferredFetchOptions,
|
||||
DeferredHandle,
|
||||
ImageContent,
|
||||
|
|
@ -606,12 +607,7 @@ export function createFauxCore(options: RegisterFauxProviderOptions) {
|
|||
...submissionOptions
|
||||
} = entry.options ?? {};
|
||||
try {
|
||||
entry.final = await resolveResponse(
|
||||
entry.step,
|
||||
entry.context,
|
||||
{ ...submissionOptions, ...fetchOptions },
|
||||
entry.model,
|
||||
);
|
||||
entry.final = await resolveResponse(entry.step, entry.context, submissionOptions, entry.model);
|
||||
} catch (error) {
|
||||
entry.final = createErrorMessage(error, api, provider, entry.model.id);
|
||||
}
|
||||
|
|
@ -637,7 +633,7 @@ export function createFauxCore(options: RegisterFauxProviderOptions) {
|
|||
const cancelDeferred = async (
|
||||
requestModel: Model<string>,
|
||||
handle: DeferredHandle,
|
||||
cancelOptions?: StreamOptions,
|
||||
cancelOptions?: DeferredCancelOptions,
|
||||
): Promise<void> => {
|
||||
state.cancelledDeferred.push(structuredClone(handle));
|
||||
const entry = deferredResponses.get(handle.id);
|
||||
|
|
|
|||
|
|
@ -114,7 +114,66 @@ export interface ProviderResponse {
|
|||
headers: Record<string, string>;
|
||||
}
|
||||
|
||||
export interface StreamOptions {
|
||||
/** Authentication, HTTP transport, and lifecycle callbacks shared by provider requests. */
|
||||
export interface ProviderRequestOptions<TModel = Model<Api>> {
|
||||
signal?: AbortSignal;
|
||||
apiKey?: string;
|
||||
/**
|
||||
* Optional fetch implementation for provider HTTP requests.
|
||||
* Defaults to `globalThis.fetch`. Provider adapters that cannot inject a custom implementation may reject it.
|
||||
* This does not affect WebSocket transports.
|
||||
*/
|
||||
fetch?: FetchFunction;
|
||||
/**
|
||||
* Provider-scoped environment values. These take precedence over process.env for
|
||||
* provider configuration such as regional settings, endpoint placeholders, and
|
||||
* proxy variables.
|
||||
*/
|
||||
env?: ProviderEnv;
|
||||
/**
|
||||
* Optional callback for inspecting or replacing provider payloads before sending.
|
||||
* Return undefined to keep the payload unchanged.
|
||||
*/
|
||||
onPayload?: (payload: unknown, model: TModel) => unknown | undefined | Promise<unknown | undefined>;
|
||||
/**
|
||||
* Optional callback invoked after an HTTP response is received.
|
||||
*/
|
||||
onResponse?: (response: ProviderResponse, model: TModel) => void | Promise<void>;
|
||||
/**
|
||||
* Optional custom HTTP headers to include in API requests.
|
||||
* Merged with provider defaults; caller values override default headers.
|
||||
* On AWS Bedrock these are injected via a Smithy `build`-step middleware so
|
||||
* they are covered by SigV4 signing; reserved headers (`x-amz-*`,
|
||||
* `authorization`, `host`) are silently ignored to preserve SigV4 / bearer auth.
|
||||
* A null value suppresses a provider/API default header with the same name.
|
||||
*/
|
||||
headers?: ProviderHeaders;
|
||||
/**
|
||||
* HTTP request timeout in milliseconds for providers/SDKs that support it.
|
||||
* For example, OpenAI and Anthropic SDK clients default to 10 minutes.
|
||||
*/
|
||||
timeoutMs?: number;
|
||||
/**
|
||||
* Maximum retry attempts for providers/SDKs that support client-side retries.
|
||||
* For example, OpenAI and Anthropic SDK clients default to 2.
|
||||
*/
|
||||
maxRetries?: number;
|
||||
/**
|
||||
* Maximum delay in milliseconds to wait for a retry when the server requests a long wait.
|
||||
* If the server's requested delay exceeds this value, the request fails immediately
|
||||
* with an error containing the requested delay, allowing higher-level retry logic
|
||||
* to handle it with user visibility.
|
||||
* Default: 60000 (60 seconds). Set to 0 to disable the cap.
|
||||
*/
|
||||
maxRetryDelayMs?: number;
|
||||
}
|
||||
|
||||
export interface StreamOptions extends ProviderRequestOptions<Model<Api>> {
|
||||
/**
|
||||
* Optional callback invoked after an HTTP response is received and before
|
||||
* its body stream is consumed.
|
||||
*/
|
||||
onResponse?: (response: ProviderResponse, model: Model<Api>) => void | Promise<void>;
|
||||
temperature?: number;
|
||||
/**
|
||||
* Arbitrary sampling parameters merged into the request body as-is, after the named request
|
||||
|
|
@ -125,14 +184,6 @@ export interface StreamOptions {
|
|||
*/
|
||||
samplingParams?: Record<string, unknown>;
|
||||
maxTokens?: number;
|
||||
signal?: AbortSignal;
|
||||
apiKey?: string;
|
||||
/**
|
||||
* Optional fetch implementation for provider HTTP requests.
|
||||
* Defaults to `globalThis.fetch`. Provider adapters that cannot inject a custom implementation may reject it.
|
||||
* This does not affect WebSocket transports.
|
||||
*/
|
||||
fetch?: FetchFunction;
|
||||
/**
|
||||
* Preferred transport for providers that support multiple transports.
|
||||
* Providers that do not support this option ignore it.
|
||||
|
|
@ -149,70 +200,33 @@ export interface StreamOptions {
|
|||
* session-aware features. Ignored by providers that don't support it.
|
||||
*/
|
||||
sessionId?: string;
|
||||
/**
|
||||
* Optional callback for inspecting or replacing provider payloads before sending.
|
||||
* Return undefined to keep the payload unchanged.
|
||||
*/
|
||||
onPayload?: (payload: unknown, model: Model<Api>) => unknown | undefined | Promise<unknown | undefined>;
|
||||
/**
|
||||
* Optional callback invoked after an HTTP response is received and before
|
||||
* its body stream is consumed.
|
||||
*/
|
||||
onResponse?: (response: ProviderResponse, model: Model<Api>) => void | Promise<void>;
|
||||
/**
|
||||
* Optional custom HTTP headers to include in API requests.
|
||||
* Merged with provider defaults; caller values override default headers.
|
||||
* On AWS Bedrock these are injected via a Smithy `build`-step middleware so
|
||||
* they are covered by SigV4 signing; reserved headers (`x-amz-*`,
|
||||
* `authorization`, `host`) are silently ignored to preserve SigV4 / bearer auth.
|
||||
* A null value suppresses a provider/API default header with the same name.
|
||||
*/
|
||||
headers?: ProviderHeaders;
|
||||
/**
|
||||
* HTTP request timeout in milliseconds for providers/SDKs that support it.
|
||||
* For example, OpenAI and Anthropic SDK clients default to 10 minutes.
|
||||
*/
|
||||
timeoutMs?: number;
|
||||
/**
|
||||
* WebSocket connect timeout in milliseconds for providers that support
|
||||
* WebSocket transports. This covers the connection/open handshake only;
|
||||
* stream idleness after connection uses timeoutMs.
|
||||
*/
|
||||
websocketConnectTimeoutMs?: number;
|
||||
/**
|
||||
* Maximum retry attempts for providers/SDKs that support client-side retries.
|
||||
* For example, OpenAI and Anthropic SDK clients default to 2.
|
||||
*/
|
||||
maxRetries?: number;
|
||||
/**
|
||||
* Maximum delay in milliseconds to wait for a retry when the server requests a long wait.
|
||||
* If the server's requested delay exceeds this value, the request fails immediately
|
||||
* with an error containing the requested delay, allowing higher-level retry logic
|
||||
* to handle it with user visibility.
|
||||
* Default: 60000 (60 seconds). Set to 0 to disable the cap.
|
||||
*/
|
||||
maxRetryDelayMs?: number;
|
||||
/**
|
||||
* Optional metadata to include in API requests.
|
||||
* Providers extract the fields they understand and ignore the rest.
|
||||
* For example, Anthropic uses `user_id` for abuse tracking and rate limiting.
|
||||
*/
|
||||
metadata?: Record<string, unknown>;
|
||||
/**
|
||||
* Provider-scoped environment values. These take precedence over process.env for
|
||||
* provider configuration such as regional settings, endpoint placeholders, and
|
||||
* proxy variables.
|
||||
*/
|
||||
env?: ProviderEnv;
|
||||
}
|
||||
|
||||
export type ProviderStreamOptions = StreamOptions & Record<string, unknown>;
|
||||
|
||||
export interface DeferredFetchOptions extends StreamOptions {
|
||||
/** Maximum time in milliseconds to wait for a terminal response. Zero checks once. */
|
||||
export interface DeferredFetchOptions extends ProviderRequestOptions<Model<Api>> {
|
||||
/**
|
||||
* Maximum provider long-poll duration in milliseconds.
|
||||
* Defaults to 0, which performs one status check.
|
||||
*/
|
||||
wait?: number;
|
||||
}
|
||||
|
||||
/** Request options for best-effort deferred-response cancellation. */
|
||||
export type DeferredCancelOptions = ProviderRequestOptions<Model<Api>>;
|
||||
|
||||
/**
|
||||
* Maps known APIs to their full provider-specific stream option types.
|
||||
* Type-only imports from API implementation modules are erased at emit, so
|
||||
|
|
@ -255,7 +269,7 @@ export interface ProviderStreams {
|
|||
handle: DeferredHandle,
|
||||
options?: DeferredFetchOptions,
|
||||
): AssistantMessageEventStream;
|
||||
cancelDeferred?(model: Model<Api>, handle: DeferredHandle, options?: StreamOptions): Promise<void>;
|
||||
cancelDeferred?(model: Model<Api>, handle: DeferredHandle, options?: DeferredCancelOptions): Promise<void>;
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
@ -272,47 +286,7 @@ export interface ProviderImages {
|
|||
): Promise<AssistantImages>;
|
||||
}
|
||||
|
||||
export interface ImagesOptions {
|
||||
signal?: AbortSignal;
|
||||
apiKey?: string;
|
||||
/** Optional fetch implementation for provider HTTP requests. Defaults to `globalThis.fetch`. */
|
||||
fetch?: FetchFunction;
|
||||
/**
|
||||
* Provider-scoped environment values. These take precedence over process.env for
|
||||
* provider configuration such as endpoint placeholders and proxy variables.
|
||||
*/
|
||||
env?: ProviderEnv;
|
||||
/**
|
||||
* Optional callback for inspecting or replacing provider payloads before sending.
|
||||
* Return undefined to keep the payload unchanged.
|
||||
*/
|
||||
onPayload?: (payload: unknown, model: ImagesModel<ImagesApi>) => unknown | undefined | Promise<unknown | undefined>;
|
||||
/**
|
||||
* Optional callback invoked after an HTTP response is received.
|
||||
*/
|
||||
onResponse?: (response: ProviderResponse, model: ImagesModel<ImagesApi>) => void | Promise<void>;
|
||||
/**
|
||||
* Optional custom HTTP headers to include in API requests.
|
||||
* Merged with provider defaults; can override default headers.
|
||||
* A null value suppresses a provider/API default header with the same name.
|
||||
*/
|
||||
headers?: ProviderHeaders;
|
||||
/**
|
||||
* HTTP request timeout in milliseconds for providers/SDKs that support it.
|
||||
*/
|
||||
timeoutMs?: number;
|
||||
/**
|
||||
* Maximum retry attempts for providers/SDKs that support client-side retries.
|
||||
*/
|
||||
maxRetries?: number;
|
||||
/**
|
||||
* Maximum delay in milliseconds to wait for a retry when the server requests a long wait.
|
||||
* If the server's requested delay exceeds this value, the request fails immediately
|
||||
* with an error containing the requested delay, allowing higher-level retry logic
|
||||
* to handle it with user visibility.
|
||||
* Default: 60000 (60 seconds). Set to 0 to disable the cap.
|
||||
*/
|
||||
maxRetryDelayMs?: number;
|
||||
export interface ImagesOptions extends ProviderRequestOptions<ImagesModel<ImagesApi>> {
|
||||
/**
|
||||
* Optional metadata to include in API requests.
|
||||
* Providers extract the fields they understand and ignore the rest.
|
||||
|
|
|
|||
|
|
@ -11,7 +11,15 @@ import { cloudflareAIGatewayProvider } from "../src/providers/cloudflare-ai-gate
|
|||
import { cloudflareWorkersAIProvider } from "../src/providers/cloudflare-workers-ai.ts";
|
||||
import { fauxAssistantMessage, fauxProvider } from "../src/providers/faux.ts";
|
||||
import { googleVertexProvider } from "../src/providers/google-vertex.ts";
|
||||
import type { Api, Context, DeferredHandle, Model, ProviderStreams } from "../src/types.ts";
|
||||
import type {
|
||||
Api,
|
||||
Context,
|
||||
DeferredCancelOptions,
|
||||
DeferredFetchOptions,
|
||||
DeferredHandle,
|
||||
Model,
|
||||
ProviderStreams,
|
||||
} from "../src/types.ts";
|
||||
import { AssistantMessageEventStream } from "../src/utils/event-stream.ts";
|
||||
|
||||
function fakeAuthContext(env: Record<string, string>, files: string[] = []): AuthContext {
|
||||
|
|
@ -415,6 +423,81 @@ describe("createProvider", () => {
|
|||
expect(capturedEnv).toEqual({ PROVIDER_ONLY: "provider", REQUEST_ONLY: "request", SHARED: "request" });
|
||||
});
|
||||
|
||||
it("applies resolved request options to deferred fetch and cancellation", async () => {
|
||||
let fetchedModel: Model<Api> | undefined;
|
||||
let fetchedOptions: DeferredFetchOptions | undefined;
|
||||
let cancelledOptions: DeferredCancelOptions | undefined;
|
||||
const deferredModel = { ...testModel("api-a", "model-a"), provider: "deferred-provider" };
|
||||
const streams = recordingStreams("deferred", []);
|
||||
streams.fetchDeferred = (model, _handle, options) => {
|
||||
fetchedModel = model;
|
||||
fetchedOptions = options;
|
||||
return streams.streamSimple(model, context);
|
||||
};
|
||||
streams.cancelDeferred = async (_model, _handle, options) => {
|
||||
cancelledOptions = options;
|
||||
};
|
||||
const provider = createProvider({
|
||||
id: "deferred-provider",
|
||||
auth: {
|
||||
apiKey: {
|
||||
name: "Test",
|
||||
resolve: async () => ({
|
||||
auth: {
|
||||
apiKey: "provider-key",
|
||||
baseUrl: "https://resolved.test/v1",
|
||||
headers: { Authorization: "Bearer provider", "X-Shared": "provider" },
|
||||
},
|
||||
env: { PROVIDER_ONLY: "provider", SHARED: "provider" },
|
||||
}),
|
||||
},
|
||||
},
|
||||
models: [deferredModel],
|
||||
api: streams,
|
||||
});
|
||||
const models = createModels();
|
||||
models.setProvider(provider);
|
||||
const handle: DeferredHandle = {
|
||||
provider: deferredModel.provider,
|
||||
modelId: deferredModel.id,
|
||||
api: deferredModel.api,
|
||||
id: "response-1",
|
||||
};
|
||||
|
||||
await models.fetchDeferred(deferredModel, handle, {
|
||||
wait: 50,
|
||||
timeoutMs: 100,
|
||||
apiKey: "request-key",
|
||||
headers: { "X-Request": "request", "x-shared": "request" },
|
||||
env: { REQUEST_ONLY: "request", SHARED: "request" },
|
||||
transformHeaders: (headers) => ({ ...headers, "X-Transformed": "yes" }),
|
||||
});
|
||||
await models.cancelDeferred(deferredModel, handle, {
|
||||
timeoutMs: 200,
|
||||
transformHeaders: (headers) => ({ ...headers, "X-Cancel": "yes" }),
|
||||
});
|
||||
|
||||
expect(fetchedModel?.baseUrl).toBe("https://resolved.test/v1");
|
||||
expect(fetchedOptions).toMatchObject({
|
||||
wait: 50,
|
||||
timeoutMs: 100,
|
||||
apiKey: "request-key",
|
||||
headers: {
|
||||
Authorization: "Bearer provider",
|
||||
"X-Request": "request",
|
||||
"x-shared": "request",
|
||||
"X-Transformed": "yes",
|
||||
},
|
||||
env: { PROVIDER_ONLY: "provider", REQUEST_ONLY: "request", SHARED: "request" },
|
||||
});
|
||||
expect(cancelledOptions).toMatchObject({
|
||||
timeoutMs: 200,
|
||||
apiKey: "provider-key",
|
||||
headers: { Authorization: "Bearer provider", "X-Shared": "provider", "X-Cancel": "yes" },
|
||||
env: { PROVIDER_ONLY: "provider", SHARED: "provider" },
|
||||
});
|
||||
});
|
||||
|
||||
it("produces a stream error for a model whose api has no implementation", async () => {
|
||||
const provider = createProvider({
|
||||
id: "mixed",
|
||||
|
|
@ -510,7 +593,7 @@ describe("fauxProvider", () => {
|
|||
});
|
||||
if (!deferred.deferred) throw new Error("Faux response did not include a deferred handle");
|
||||
|
||||
const pending = await models.fetchDeferred(model, deferred.deferred, { wait: 0 });
|
||||
const pending = await models.fetchDeferred(model, deferred.deferred);
|
||||
expect(pending.stopReason).toBe("deferred");
|
||||
expect(pending.deferred).toEqual(deferred.deferred);
|
||||
|
||||
|
|
|
|||
|
|
@ -14,22 +14,25 @@ import {
|
|||
type CredentialInfo,
|
||||
type CredentialStore,
|
||||
createModels,
|
||||
type DeferredCancelOptions,
|
||||
type DeferredFetchOptions,
|
||||
type DeferredHandle,
|
||||
lazyStream,
|
||||
type Model,
|
||||
type Models,
|
||||
type ModelsApiStreamOptions,
|
||||
type ModelsDeferredOptions,
|
||||
type ModelsDeferredCancelOptions,
|
||||
type ModelsDeferredFetchOptions,
|
||||
ModelsError,
|
||||
type ModelsRefreshOptions,
|
||||
type ModelsRefreshResult,
|
||||
type ModelsRequestTransforms,
|
||||
type ModelsSimpleStreamOptions,
|
||||
type ModelsStore,
|
||||
type ModelsStreamTransforms,
|
||||
type MutableModels,
|
||||
type Provider,
|
||||
type ProviderHeaders,
|
||||
type ProviderRequestOptions,
|
||||
type SimpleStreamOptions,
|
||||
type StreamOptions,
|
||||
} from "@earendil-works/pi-ai";
|
||||
|
|
@ -559,10 +562,14 @@ export class ModelRuntime implements Models {
|
|||
return check ? { configured: true, source: "environment", label: check.source } : { configured: false };
|
||||
}
|
||||
|
||||
private async prepareRequest(
|
||||
private async prepareRequest<TOptions extends ProviderRequestOptions & ModelsRequestTransforms>(
|
||||
model: Model<Api>,
|
||||
options: (StreamOptions & ModelsStreamTransforms) | undefined,
|
||||
): Promise<{ provider: Provider; model: Model<Api>; options: StreamOptions }> {
|
||||
options: TOptions | undefined,
|
||||
): Promise<{
|
||||
provider: Provider;
|
||||
model: Model<Api>;
|
||||
options: Omit<TOptions, "transformHeaders"> & ProviderRequestOptions;
|
||||
}> {
|
||||
const provider = this.models.getProvider(model.provider);
|
||||
if (!provider) throw new ModelsError("provider", `Unknown provider: ${model.provider}`);
|
||||
const resolution = await this.getAuth(model, {
|
||||
|
|
@ -572,7 +579,8 @@ export class ModelRuntime implements Models {
|
|||
});
|
||||
if (!resolution) throw new ModelsError("auth", `Provider is not configured: ${model.provider}`);
|
||||
|
||||
const { transformHeaders, ...providerOptions } = options ?? {};
|
||||
const { transformHeaders, ...rawProviderOptions } = options ?? {};
|
||||
const providerOptions = rawProviderOptions as Omit<TOptions, "transformHeaders"> & ProviderRequestOptions;
|
||||
let headers = mergeHeaders(resolution.auth.headers, providerOptions.headers);
|
||||
if (transformHeaders) headers = await transformHeaders(headers ?? {});
|
||||
const env =
|
||||
|
|
@ -587,7 +595,7 @@ export class ModelRuntime implements Models {
|
|||
apiKey: providerOptions.apiKey ?? resolution.auth.apiKey,
|
||||
headers,
|
||||
env,
|
||||
},
|
||||
} as Omit<TOptions, "transformHeaders"> & ProviderRequestOptions,
|
||||
};
|
||||
}
|
||||
|
||||
|
|
@ -599,7 +607,7 @@ export class ModelRuntime implements Models {
|
|||
return lazyStream(model, async () => {
|
||||
const prepared = await this.prepareRequest(
|
||||
model,
|
||||
options as (StreamOptions & ModelsStreamTransforms) | undefined,
|
||||
options as (StreamOptions & ModelsRequestTransforms) | undefined,
|
||||
);
|
||||
return prepared.provider.stream(
|
||||
prepared.model as Model<TApi>,
|
||||
|
|
@ -631,7 +639,7 @@ export class ModelRuntime implements Models {
|
|||
async fetchDeferred(
|
||||
model: Model<Api>,
|
||||
handle: DeferredHandle,
|
||||
options?: ModelsDeferredOptions,
|
||||
options?: ModelsDeferredFetchOptions,
|
||||
): Promise<AssistantMessage> {
|
||||
return lazyStream(model, async () => {
|
||||
const prepared = await this.prepareRequest(model, options);
|
||||
|
|
@ -642,12 +650,16 @@ export class ModelRuntime implements Models {
|
|||
}).result();
|
||||
}
|
||||
|
||||
async cancelDeferred(model: Model<Api>, handle: DeferredHandle, options?: ModelsDeferredOptions): Promise<void> {
|
||||
async cancelDeferred(
|
||||
model: Model<Api>,
|
||||
handle: DeferredHandle,
|
||||
options?: ModelsDeferredCancelOptions,
|
||||
): Promise<void> {
|
||||
const prepared = await this.prepareRequest(model, options);
|
||||
if (!prepared.provider.cancelDeferred) {
|
||||
throw new ModelsError("provider", `Provider ${model.provider} does not support deferred responses`);
|
||||
}
|
||||
await prepared.provider.cancelDeferred(prepared.model, handle, prepared.options);
|
||||
await prepared.provider.cancelDeferred(prepared.model, handle, prepared.options as DeferredCancelOptions);
|
||||
}
|
||||
|
||||
login(providerId: string, type: AuthType, interaction: AuthInteraction): Promise<Credential> {
|
||||
|
|
|
|||
|
|
@ -3,6 +3,8 @@ import { tmpdir } from "node:os";
|
|||
import { join } from "node:path";
|
||||
import {
|
||||
createAssistantMessageEventStream,
|
||||
type DeferredCancelOptions,
|
||||
type DeferredFetchOptions,
|
||||
InMemoryModelsStore,
|
||||
type Model,
|
||||
type Provider,
|
||||
|
|
@ -113,7 +115,9 @@ describe("extension provider model lifecycle", () => {
|
|||
baseUrl: "https://native.test/v1",
|
||||
};
|
||||
let fetchedBaseUrl: string | undefined;
|
||||
let fetchedOptions: DeferredFetchOptions | undefined;
|
||||
let cancelledId: string | undefined;
|
||||
let cancelledOptions: DeferredCancelOptions | undefined;
|
||||
const provider: Provider = {
|
||||
id: "extension-native-deferred",
|
||||
name: "Extension Native Deferred",
|
||||
|
|
@ -130,8 +134,9 @@ describe("extension provider model lifecycle", () => {
|
|||
streamSimple: () => {
|
||||
throw new Error("unused");
|
||||
},
|
||||
fetchDeferred: (requestModel) => {
|
||||
fetchDeferred: (requestModel, _handle, options) => {
|
||||
fetchedBaseUrl = requestModel.baseUrl;
|
||||
fetchedOptions = options;
|
||||
const message = {
|
||||
role: "assistant" as const,
|
||||
content: [],
|
||||
|
|
@ -155,8 +160,9 @@ describe("extension provider model lifecycle", () => {
|
|||
stream.end(message);
|
||||
return stream;
|
||||
},
|
||||
cancelDeferred: async (_requestModel, handle) => {
|
||||
cancelDeferred: async (_requestModel, handle, options) => {
|
||||
cancelledId = handle.id;
|
||||
cancelledOptions = options;
|
||||
},
|
||||
};
|
||||
|
||||
|
|
@ -164,21 +170,46 @@ describe("extension provider model lifecycle", () => {
|
|||
const composedModel = runtime.getModel(provider.id, nativeModel.id);
|
||||
expect(composedModel).toBeDefined();
|
||||
|
||||
await runtime.fetchDeferred(composedModel!, {
|
||||
provider: provider.id,
|
||||
modelId: nativeModel.id,
|
||||
api: nativeModel.api,
|
||||
id: "fetch-id",
|
||||
});
|
||||
await runtime.cancelDeferred(composedModel!, {
|
||||
provider: provider.id,
|
||||
modelId: nativeModel.id,
|
||||
api: nativeModel.api,
|
||||
id: "cancel-id",
|
||||
});
|
||||
await runtime.fetchDeferred(
|
||||
composedModel!,
|
||||
{
|
||||
provider: provider.id,
|
||||
modelId: nativeModel.id,
|
||||
api: nativeModel.api,
|
||||
id: "fetch-id",
|
||||
},
|
||||
{
|
||||
wait: 25,
|
||||
headers: { "X-Fetch": "fetch" },
|
||||
transformHeaders: (headers) => ({ ...headers, "X-Transformed": "fetch" }),
|
||||
},
|
||||
);
|
||||
await runtime.cancelDeferred(
|
||||
composedModel!,
|
||||
{
|
||||
provider: provider.id,
|
||||
modelId: nativeModel.id,
|
||||
api: nativeModel.api,
|
||||
id: "cancel-id",
|
||||
},
|
||||
{
|
||||
timeoutMs: 100,
|
||||
transformHeaders: (headers) => ({ ...headers, "X-Transformed": "cancel" }),
|
||||
},
|
||||
);
|
||||
|
||||
expect(fetchedBaseUrl).toBe("https://overlay.test/v1");
|
||||
expect(fetchedOptions).toMatchObject({
|
||||
apiKey: "key",
|
||||
wait: 25,
|
||||
headers: { "X-Fetch": "fetch", "X-Transformed": "fetch" },
|
||||
});
|
||||
expect(cancelledId).toBe("cancel-id");
|
||||
expect(cancelledOptions).toMatchObject({
|
||||
apiKey: "key",
|
||||
timeoutMs: 100,
|
||||
headers: { "X-Transformed": "cancel" },
|
||||
});
|
||||
} finally {
|
||||
rmSync(tempDir, { recursive: true, force: true });
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue