ewilliams28/openclaw
1
0
Fork 0
mirror of https://github.com/openclaw/openclaw.git synced 2026-08-12 21:53:00 -06:00
Code Issues Packages Projects Releases Wiki Activity

feat(channels): expose the turn-adoption lifecycle seam

Browse Source 
Replace the bare onTurnAdopted callback and queuedFollowupLifecycle with one
turnAdoptionLifecycle surface (onSettled guaranteed via finally; adoption-loss
aborts queued steering turns without transcript replay), threaded through the
turn kernel and agent runner, exposed to plugins via
runtime.state.openChannelIngressDrain and the channel-outbound SDK barrel,
with surface budgets re-measured against the narrowed baselines (#108656).
This commit is contained in:
Ayaan Zaidi
2026-07-16 21:18:09 +05:30
parent fc6b9dad0b
commit 16c14e5bbf
39 changed files with 587 additions and 287 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/channels/telegram.md
+1 -1
View File
@@ -739,7 +739,7 @@ curl "https://api.telegram.org/bot<bot_token>/getUpdates"
Webhook mode validates request guards, the Telegram secret token, and the JSON body, then commits the update to its durable ingress queue before returning an empty `200`. Successful durable adoption includes `x-openclaw-delivery-accepted: durable`; health, routing, authentication, validation, and storage-error responses omit this header. Reverse proxies and host controllers can require the header to distinguish OpenClaw adoption from a generic empty `200` without inferring acceptance from response timing.
OpenClaw then processes the update asynchronously through the same per-chat/per-topic bot lanes used by long polling, so slow agent turns do not hold Telegram's delivery ACK.
After the durable write, OpenClaw claims and processes updates through the core channel-ingress drain (per-chat/per-topic lanes, complete at turn adoption, pre-adoption stall timeout). Slow agent turns do not hold Telegram's delivery ACK.
</Accordion>
docs/plugins/sdk-channel-outbound.md
+8 -4
View File
@@ -12,10 +12,14 @@ Channel plugins expose outbound message behavior from
`openclaw/plugin-sdk/channel-inbound` for receive/context/dispatch
orchestration.
Core owns queueing, durability, generic retry policy, hooks, receipts, and
the shared `message` tool. The plugin owns native send/edit/delete calls,
target normalization, platform threading, selected quotes, notification
flags, account state, and platform-specific side effects.
Core owns queueing, durability, the durable **ingress drain**
(`createChannelIngressDrain` / `openChannelIngressDrain`), generic retry
policy, turn-adoption lifecycle (`turnAdoptionLifecycle` /
`bindIngressLifecycleToReplyOptions`), hooks, receipts, and the shared
`message` tool. The plugin owns native send/edit/delete calls, target
normalization, platform threading, selected quotes, notification flags,
account state, accept-side enqueue, lane keys, non-retryable predicates,
optional supersede authorization, and platform-specific side effects.
## Adapter
docs/plugins/sdk-runtime.md
+3 -1
View File
@@ -739,8 +739,10 @@ two-party event loops that do not go through the shared inbound reply runner.
`openChannelIngressQueue<TPayload>(...)` opens a persisted ingress queue scoped to the calling plugin, for buffering inbound events that need at-least-once processing across restarts. When stale-claim recovery uses `shouldRecover`, also provide `shouldRecoverCorrupt` if corrupt claimed payloads should be quarantined: its payload-independent claim identity lets the plugin preserve live owner and lane policy before the queue tombstones the row.
`openChannelIngressDrain(...)` opens the core channel-agnostic worker over that queue (or creates a queue when none is supplied). The drain owns stale-claim recovery, per-lane claim serialization, complete-at-adoption or complete-on-dispatch-return, retry/dead-letter disposition, optional pre-adoption supersede, and claim→adoption stall timeout. Wire claim ownership into reply generation with `turnAdoptionLifecycle` (via `bindIngressLifecycleToReplyOptions` from `plugin-sdk/channel-outbound`). Channel plugins keep accept-side enqueue, lane derivation, non-retryable classification, and any supersede authorization policy.
<Warning>
`openBlobStore`, `openKeyedStore`, `openSyncKeyedStore`, and `openChannelIngressQueue` are available only to bundled plugins and trusted official plugin installations in this release.
`openBlobStore`, `openKeyedStore`, `openSyncKeyedStore`, `openChannelIngressQueue`, and `openChannelIngressDrain` are available only to bundled plugins and trusted official plugin installations in this release.
</Warning>
</Accordion>
extensions/voice-call/doctor-contract-api.test.ts
+3
View File
@@ -46,6 +46,9 @@ function installStateRuntime(): void {
openChannelIngressQueue: (() => {
throw new Error("openChannelIngressQueue is not used by voice-call doctor tests");
}) as never,
openChannelIngressDrain: (() => {
throw new Error("openChannelIngressDrain is not used by voice-call doctor tests");
}) as never,
},
});
}
extensions/voice-call/src/manager.restore.test.ts
+3
View File
@@ -28,6 +28,9 @@ function installStateRuntime(): void {
openChannelIngressQueue: (() => {
throw new Error("openChannelIngressQueue is not used by voice-call restore tests");
}) as never,
openChannelIngressDrain: (() => {
throw new Error("openChannelIngressDrain is not used by voice-call restore tests");
}) as never,
},
});
}
extensions/voice-call/src/manager.test-harness.ts
+3
View File
@@ -89,6 +89,9 @@ function createVoiceCallStateRuntimeForTests(): VoiceCallStateRuntime["state"] {
openChannelIngressQueue: (() => {
throw new Error("openChannelIngressQueue is not used by voice-call manager tests");
}) as VoiceCallStateRuntime["state"]["openChannelIngressQueue"],
openChannelIngressDrain: (() => {
throw new Error("openChannelIngressDrain is not used by voice-call manager tests");
}) as VoiceCallStateRuntime["state"]["openChannelIngressDrain"],
};
}
extensions/voice-call/src/manager/events.test.ts
+3
View File
@@ -58,6 +58,9 @@ function installStateRuntime(): void {
openChannelIngressQueue: (() => {
throw new Error("openChannelIngressQueue is not used by voice-call event tests");
}) as never,
openChannelIngressDrain: (() => {
throw new Error("openChannelIngressDrain is not used by voice-call event tests");
}) as never,
},
});
}
extensions/voice-call/src/manager/store.test.ts
+6
View File
@@ -33,6 +33,9 @@ function installStateRuntime(): void {
openChannelIngressQueue: (() => {
throw new Error("openChannelIngressQueue is not used by voice-call store tests");
}) as never,
openChannelIngressDrain: (() => {
throw new Error("openChannelIngressDrain is not used by voice-call store tests");
}) as never,
},
});
}
@@ -93,6 +96,9 @@ describe("voice-call call record store", () => {
openChannelIngressQueue: (() => {
throw new Error("openChannelIngressQueue is not used by voice-call store tests");
}) as never,
openChannelIngressDrain: (() => {
throw new Error("openChannelIngressDrain is not used by voice-call store tests");
}) as never,
},
});
extensions/voice-call/src/runtime-state.ts
+5 -1
View File
@@ -7,7 +7,11 @@ import { createPluginRuntimeStore, type PluginRuntime } from "openclaw/plugin-sd
export type VoiceCallStateRuntime = {
state: Pick<
PluginRuntime["state"],
"resolveStateDir" | "openKeyedStore" | "openSyncKeyedStore" | "openChannelIngressQueue"
| "resolveStateDir"
| "openKeyedStore"
| "openSyncKeyedStore"
| "openChannelIngressQueue"
| "openChannelIngressDrain"
>;
};
extensions/voice-call/src/webhook.hangup-once.lifecycle.test.ts
+5
View File
@@ -26,6 +26,11 @@ function installStateRuntime(): void {
"openChannelIngressQueue is not used by voice-call webhook lifecycle tests",
);
}) as never,
openChannelIngressDrain: (() => {
throw new Error(
"openChannelIngressDrain is not used by voice-call webhook lifecycle tests",
);
}) as never,
},
});
}
scripts/plugin-sdk-surface-report.mjs
+13 -5
View File
@@ -162,8 +162,9 @@ const defaultPublicDeprecatedExportsByEntrypointBudget = Object.freeze({
"channel-lifecycle": 23,
// Registry sweep: 77 packages, zero fetch failures; channel-ingress and dead aliases
// had zero consumers.
"channel-message": 230,
"channel-message-runtime": 227,
// +11 each: durable channel-ingress drain seam (drain/lifecycle/claim/retry) mirrored by compat (#108656).
"channel-message": 241,
"channel-message-runtime": 238,
"channel-pairing-paths": 1,
// Deprecated pairing/conversation exports from the SQLite pairing migration
// landed on main (#105802) without entrypoint pins; not touched by this PR.
@@ -253,7 +254,10 @@ export function readPluginSdkSurfaceBudgets(env = process.env) {
// +4: shared audio-energy stats and speech-threshold gate through realtime-voice.
// +2: supplemental sender decision and outbound text chunk sequencer.
// +2: shared realtime voice session harness through realtime-voice.
8014,
// +24: narrowed durable channel-ingress drain seam — factory, lifecycle binding,
// tuning constants, and telegram-consumed claim helpers with compat mirrors,
// after harvesting exports orphaned by the split-out WhatsApp adapter (#108656).
8038,
env,
),
publicFunctionExports: readPluginSdkSurfaceBudgetEnv(
@@ -281,7 +285,9 @@ export function readPluginSdkSurfaceBudgets(env = process.env) {
// +3: PCM16/mu-law energy readers and speech-threshold gate factory.
// +2: supplemental sender decision and outbound text chunk sequencer.
// +1: shared realtime voice session harness through realtime-voice.
4479,
// +9: narrowed drain seam functions and compat mirrors after the
// WhatsApp-split harvest (#108656).
4489,
env,
),
publicDeprecatedExports: readPluginSdkSurfaceBudgetEnv(
@@ -298,7 +304,9 @@ export function readPluginSdkSurfaceBudgets(env = process.env) {
// +8: shared channel helpers mirrored by deprecated barrels.
// +3: receipt/snapshot exports through deprecated channel barrels.
// +1: unified implicit-mention config type through deprecated config-types.
2990,
// +24: narrowed drain seam compat mirrors in the channel-message
// deprecation-window barrels (#108656).
3014,
env,
),
publicWildcardReexports: readPluginSdkSurfaceBudgetEnv(
src/auto-reply/get-reply-options.types.ts
+30 -16
View File
@@ -46,17 +46,35 @@ export type QueuedReplyDeliveryCorrelation = {
begin: () => (() => void) | void;
};
/** Lifecycle hooks for queued follow-up replies. */
export type QueuedReplyLifecycle = {
/** Stable cancellation owner used to keep collect-mode batches authorization-safe. */
ownerKey?: string;
/** Return false when the external owner rejects this queue identity. */
onEnqueued?: () => boolean | void;
/** Retires this source's cancellation ownership while retaining its live identity. */
/**
* Exclusive: each lifecycle is its own collect-admission identity.
* Cancel-only: share collect identity via ownerKey (gateway chat.send).
*/
export type TurnAdoptionAdmission = "exclusive" | "cancel-only";
/**
* Canonical turn-ownership lifecycle (adopt / defer / abandon / settle).
* Single surface for durable ingress, gateway cancel identity, and reply-lane transfer.
*/
export type TurnAdoptionLifecycle = {
/**
* Admission isolation mode (closed). Exclusive isolates collect identity per
* lifecycle; cancel-only shares via ownerKey. Never inferred from onAbandoned.
* Durable ingress sets exclusive; gateway cancel identity sets cancel-only.
*/
admission?: TurnAdoptionAdmission;
onAdopted: () => void | Promise<void>;
/** Return false to reject followup enqueue. */
onDeferred?: () => boolean | void;
/** Deferred turn finished without owning the reply lane. */
onAbandoned?: () => void;
/** Always fires when the followup ownership cycle ends (admitted or not). Gateway cleanup. */
onSettled?: () => void;
/** Retires cancellation ownership while retaining live identity. */
onCancellationRetired?: () => void;
/** Called after the queued turn owns the reply lane, before model/tool execution. */
onAdmitted?: () => void | Promise<void>;
onComplete?: () => void;
/** Stable cancellation owner for collect-mode batches. */
ownerKey?: string;
abortSignal?: AbortSignal;
};
/** Partial assistant payload emitted during streaming or replacement updates. */
@@ -94,11 +112,9 @@ export type GetReplyOptions = {
/** Notifies when an agent run actually starts (useful for webchat command handling). */
onAgentRunStart?: (runId: string) => void;
/**
* Called after the restart-recovery delivery-context persist attempt
* completes (context may be absent when source delivery is suppressed).
* Channels may complete ingress ownership here without waiting for settle.
* Canonical adoption lifecycle (adopted / deferred / abandoned / settled + pre-adoption abort).
*/
onTurnAdopted?: () => void | Promise<void>;
turnAdoptionLifecycle?: TurnAdoptionLifecycle;
/** Shared lifecycle owner for the current user-turn transcript append. */
userTurnTranscriptRecorder?: UserTurnTranscriptRecorder;
/** Current user turn is already durable; replay it without appending another copy. */
@@ -295,8 +311,6 @@ export type GetReplyOptions = {
taskSuggestionDeliveryMode?: TaskSuggestionDeliveryMode;
/** Starts delivery tracking when this turn later drains as a queued followup. */
queuedDeliveryCorrelations?: QueuedReplyDeliveryCorrelation[];
/** Tracks ownership transfer when this turn later drains as a queued followup. */
queuedFollowupLifecycle?: QueuedReplyLifecycle;
/** Called after a queued followup owns the reply lane, before its model run starts. */
onQueuedFollowupAdmitted?: () => Promise<void> | void;
/** Allow channel-owned progress UI while final/source reply delivery remains message-tool-only. */
src/auto-reply/reply/agent-runner.misc.runreplyagent.test.ts
+8 -6
View File
@@ -3402,7 +3402,7 @@ describe("runReplyAgent private message_tool_only final warning (#85714)", () =>
sendPolicyDenied?: boolean;
isHeartbeat?: boolean;
replyOperation?: ReturnType<typeof createReplyOperation>;
queuedLifecycle?: FollowupRun["queuedLifecycle"];
turnAdoptionLifecycle?: FollowupRun["turnAdoptionLifecycle"];
}) {
const tmp = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-stranded-"));
const storePath = path.join(tmp, "sessions.json");
@@ -3459,7 +3459,9 @@ describe("runReplyAgent private message_tool_only final warning (#85714)", () =>
...(params.strandedReplyRetry ? { strandedReplyRetry: true } : {}),
enqueuedAt: Date.now(),
...(params.transcriptPrompt ? { transcriptPrompt: params.transcriptPrompt } : {}),
...(params.queuedLifecycle ? { queuedLifecycle: params.queuedLifecycle } : {}),
...(params.turnAdoptionLifecycle
? { turnAdoptionLifecycle: params.turnAdoptionLifecycle }
: {}),
run: {
agentId: "main",
agentDir: "/tmp/agent",
@@ -3527,9 +3529,9 @@ describe("runReplyAgent private message_tool_only final warning (#85714)", () =>
it("enqueues a one-shot recovery retry by default for substantive stranded finals", async () => {
const parentOnComplete = vi.fn();
const parentLifecycle = { onComplete: parentOnComplete };
const parentLifecycle = { onAdopted: async () => {}, onSettled: parentOnComplete };
const { finalAssistantText } = await runPrivateFinalCase({
queuedLifecycle: parentLifecycle,
turnAdoptionLifecycle: parentLifecycle,
});
expect(warnPrivateFinalSpy).toHaveBeenCalledTimes(1);
@@ -3542,8 +3544,8 @@ describe("runReplyAgent private message_tool_only final warning (#85714)", () =>
expect(retryRun?.prompt).toContain("message(action=send)");
expect(retryRun?.prompt).toContain(finalAssistantText);
// System retry must not inherit the client turn's one-shot lifecycle identity.
expect(retryRun?.queuedLifecycle).toBeUndefined();
expect(parentLifecycle.onComplete).toBe(parentOnComplete);
expect(retryRun?.turnAdoptionLifecycle).toBeUndefined();
expect(parentLifecycle.onSettled).toBe(parentOnComplete);
expect(parentOnComplete).not.toHaveBeenCalled();
});
src/auto-reply/reply/agent-runner.runreplyagent.e2e.test.ts
+42 -34
View File
@@ -623,12 +623,12 @@ describe("runReplyAgent active steering", () => {
return true;
},
);
const onTurnAdopted = vi.fn(async () => {
const onAdopted = vi.fn(async () => {
events.push("adoption-finalizer");
throw finalizerError;
});
const { run, typing } = createMinimalRun({
opts: { onTurnAdopted },
opts: { turnAdoptionLifecycle: { onAdopted } },
isActive: true,
isStreaming: true,
shouldSteer: true,
@@ -638,7 +638,7 @@ describe("runReplyAgent active steering", () => {
await expect(run()).resolves.toBeUndefined();
expect(events).toEqual(["transcript-committed", "adoption-finalizer"]);
expect(onTurnAdopted).toHaveBeenCalledTimes(1);
expect(onAdopted).toHaveBeenCalledTimes(1);
expect(state.queueEmbeddedAgentMessageMock).toHaveBeenCalledTimes(1);
expect(vi.mocked(enqueueFollowupRun)).not.toHaveBeenCalled();
expect(state.runEmbeddedAgentMock).not.toHaveBeenCalled();
@@ -658,9 +658,9 @@ describe("runReplyAgent active steering", () => {
target: "none",
gatewayHealth: "live",
});
const onTurnAdopted = vi.fn();
const onAdopted = vi.fn();
const { run } = createMinimalRun({
opts: { onTurnAdopted },
opts: { turnAdoptionLifecycle: { onAdopted } },
isActive: true,
isStreaming: true,
shouldSteer: true,
@@ -680,7 +680,7 @@ describe("runReplyAgent active steering", () => {
expect(state.beforeAgentReplyRunMock).toHaveBeenCalledOnce();
expect(state.runEmbeddedAgentMock).toHaveBeenCalledOnce();
expect(onTurnAdopted).not.toHaveBeenCalled();
expect(onAdopted).not.toHaveBeenCalled();
});
it("admits an ordinary rejected steering turn with durable recovery state", async () => {
@@ -690,13 +690,13 @@ describe("runReplyAgent active steering", () => {
};
const sessionStore = { main: sessionEntry };
const storePath = await createSessionStoreFile(sessionEntry);
const onTurnAdopted = vi.fn(async () => {
const onAdopted = vi.fn(async () => {
expect((await readStoredMainSession(storePath)).restartRecoveryBeforeAgentReplyState).toBe(
"admitted",
);
});
const { followupRun, run, sourceTurnId } = createMinimalRun({
opts: { onTurnAdopted },
opts: { turnAdoptionLifecycle: { onAdopted } },
isActive: true,
isStreaming: true,
shouldSteer: true,
@@ -725,7 +725,7 @@ describe("runReplyAgent active steering", () => {
await expect(run()).resolves.toEqual(expect.objectContaining({ text: "final" }));
expect(state.beforeAgentReplyRunMock).not.toHaveBeenCalled();
expect(onTurnAdopted).toHaveBeenCalledOnce();
expect(onAdopted).toHaveBeenCalledOnce();
expect(state.runEmbeddedAgentMock).toHaveBeenCalledOnce();
});
});
@@ -1348,10 +1348,10 @@ describe("runReplyAgent pending final delivery capture", () => {
const completedEntry = await readStoredMainSession(storePath);
expect(completedEntry.restartRecoveryTerminalRunIds).toEqual([first.sourceTurnId]);
state.runEmbeddedAgentMock.mockClear();
const onTurnAdopted = vi.fn();
const onAdopted = vi.fn();
const completedStore = { main: completedEntry };
const duplicate = createMinimalRun({
opts: { onTurnAdopted },
opts: { turnAdoptionLifecycle: { onAdopted } },
sessionCtx,
runOverrides: { messageProvider: "discord" },
sessionEntry: completedEntry,
@@ -1370,7 +1370,7 @@ describe("runReplyAgent pending final delivery capture", () => {
await expect(duplicate.run()).resolves.toBeUndefined();
expect(onTurnAdopted).not.toHaveBeenCalled();
expect(onAdopted).not.toHaveBeenCalled();
expect(state.runEmbeddedAgentMock).not.toHaveBeenCalled();
expect((await readStoredMainSession(storePath)).restartRecoveryTerminalRunIds).toEqual([
first.sourceTurnId,
@@ -1402,11 +1402,11 @@ describe("runReplyAgent pending final delivery capture", () => {
};
const sessionStore = { main: sessionEntry };
const storePath = await createSessionStoreFile(sessionEntry);
const onTurnAdopted = vi.fn();
const onAdopted = vi.fn();
const duplicate = createMinimalRun({
isActive: true,
shouldSteer: true,
opts: { onTurnAdopted },
opts: { turnAdoptionLifecycle: { onAdopted } },
sessionCtx,
runOverrides: { messageProvider: "discord" },
sessionEntry,
@@ -1426,7 +1426,7 @@ describe("runReplyAgent pending final delivery capture", () => {
await expect(duplicate.run()).resolves.toBeUndefined();
expect(duplicate.sourceTurnId).toBe(sourceTurnId);
expect(onTurnAdopted).not.toHaveBeenCalled();
expect(onAdopted).not.toHaveBeenCalled();
expect(state.queueEmbeddedAgentMessageMock).not.toHaveBeenCalled();
expect(state.runEmbeddedAgentMock).not.toHaveBeenCalled();
expect(await readStoredMainSession(storePath)).toMatchObject({
@@ -1461,11 +1461,11 @@ describe("runReplyAgent pending final delivery capture", () => {
};
const sessionStore = { main: sessionEntry };
const storePath = await createSessionStoreFile(sessionEntry);
const onTurnAdopted = vi.fn();
const onAdopted = vi.fn();
const duplicate = createMinimalRun({
isActive: true,
shouldSteer: true,
opts: { onTurnAdopted },
opts: { turnAdoptionLifecycle: { onAdopted } },
sessionCtx,
runOverrides: { messageProvider: "discord" },
sessionEntry,
@@ -1477,7 +1477,7 @@ describe("runReplyAgent pending final delivery capture", () => {
await expect(duplicate.run()).resolves.toBeUndefined();
expect(duplicate.sourceTurnId).toBe(sourceTurnId);
expect(onTurnAdopted).not.toHaveBeenCalled();
expect(onAdopted).not.toHaveBeenCalled();
expect(state.queueEmbeddedAgentMessageMock).not.toHaveBeenCalled();
expect(state.runEmbeddedAgentMock).not.toHaveBeenCalled();
const stored = await readStoredMainSession(storePath);
@@ -1864,9 +1864,9 @@ describe("runReplyAgent pending final delivery capture", () => {
};
const sessionStore = { main: sessionEntry };
const storePath = await createSessionStoreFile(sessionEntry);
const onTurnAdopted = vi.fn();
const onAdopted = vi.fn();
const { run } = createMinimalRun({
opts: { onTurnAdopted },
opts: { turnAdoptionLifecycle: { onAdopted } },
sessionCtx: {
Provider: "webchat",
OriginatingChannel: "webchat",
@@ -1880,7 +1880,7 @@ describe("runReplyAgent pending final delivery capture", () => {
await expect(run()).rejects.toThrow("restart recovery claim changed before agent adoption");
expect(onTurnAdopted).not.toHaveBeenCalled();
expect(onAdopted).not.toHaveBeenCalled();
expect(state.runEmbeddedAgentMock).not.toHaveBeenCalled();
expect(await readStoredMainSession(storePath)).toMatchObject({
abortedLastRun: true,
@@ -1945,7 +1945,7 @@ describe("runReplyAgent pending final delivery capture", () => {
}
});
it("fires onTurnAdopted after restart recovery delivery context persist completes", async () => {
it("fires onAdopted after restart recovery delivery context persist completes", async () => {
const sessionEntry: SessionEntry = {
sessionId: "session",
updatedAt: Date.now(),
@@ -1959,7 +1959,7 @@ describe("runReplyAgent pending final delivery capture", () => {
messageId: "1503645939964055592",
});
const events: string[] = [];
const onTurnAdopted = vi.fn(async () => {
const onAdopted = vi.fn(async () => {
const storedAtAdoption = await readStoredMainSession(storePath);
expect(storedAtAdoption.restartRecoveryDeliveryContext).toEqual({
channel: "discord",
@@ -2000,7 +2000,10 @@ describe("runReplyAgent pending final delivery capture", () => {
});
const { followupRun, run, sourceTurnId } = createMinimalRun({
opts: { onTurnAdopted, sourceReplyDeliveryMode: "message_tool_only" },
opts: {
turnAdoptionLifecycle: { onAdopted },
sourceReplyDeliveryMode: "message_tool_only",
},
sessionCtx: {
Provider: "discord",
OriginatingChannel: "discord",
@@ -2028,7 +2031,7 @@ describe("runReplyAgent pending final delivery capture", () => {
await run();
expect(onTurnAdopted).toHaveBeenCalledOnce();
expect(onAdopted).toHaveBeenCalledOnce();
expect(events).toEqual(["adopted", "agent-run"]);
expect(
(await readStoredMainSession(storePath)).restartRecoverySourceReplyDeliveryMode,
@@ -2054,13 +2057,16 @@ describe("runReplyAgent pending final delivery capture", () => {
};
const sessionStore = { main: sessionEntry };
const storePath = await createSessionStoreFile(sessionEntry);
const onTurnAdopted = vi.fn(async () => {
const onAdopted = vi.fn(async () => {
expect(
(await readStoredMainSession(storePath)).restartRecoverySameChannelThreadRequired,
).toBe(true);
});
const { followupRun, run, sourceTurnId } = createMinimalRun({
opts: { onTurnAdopted, sourceReplyDeliveryMode: "message_tool_only" },
opts: {
turnAdoptionLifecycle: { onAdopted },
sourceReplyDeliveryMode: "message_tool_only",
},
sessionCtx: {
Provider: "slack",
OriginatingChannel: "slack",
@@ -2085,7 +2091,7 @@ describe("runReplyAgent pending final delivery capture", () => {
await run();
expect(onTurnAdopted).toHaveBeenCalledOnce();
expect(onAdopted).toHaveBeenCalledOnce();
expect(
(await readStoredMainSession(storePath)).restartRecoverySameChannelThreadRequired,
).toBeUndefined();
@@ -2128,8 +2134,10 @@ describe("runReplyAgent pending final delivery capture", () => {
state.runEmbeddedAgentMock.mockImplementationOnce(runHookBackedEmbeddedAgent);
const { followupRun, run, sourceTurnId } = createMinimalRun({
opts: {
onTurnAdopted: async () => {
events.push("adopted");
turnAdoptionLifecycle: {
onAdopted: async () => {
events.push("adopted");
},
},
},
sessionCtx: {
@@ -2266,7 +2274,7 @@ describe("runReplyAgent pending final delivery capture", () => {
});
});
it("fires onTurnAdopted for suppressed-delivery runs before the agent turn", async () => {
it("fires onAdopted for suppressed-delivery runs before the agent turn", async () => {
const sessionEntry: SessionEntry = {
sessionId: "session",
updatedAt: Date.now(),
@@ -2274,7 +2282,7 @@ describe("runReplyAgent pending final delivery capture", () => {
const sessionStore = { main: sessionEntry };
const storePath = await createSessionStoreFile(sessionEntry);
const events: string[] = [];
const onTurnAdopted = vi.fn(async () => {
const onAdopted = vi.fn(async () => {
const storedAtAdoption = await readStoredMainSession(storePath);
expect(storedAtAdoption.restartRecoveryDeliveryContext).toBeUndefined();
expect(storedAtAdoption.restartRecoveryDeliveryRunId).toBeUndefined();
@@ -2290,7 +2298,7 @@ describe("runReplyAgent pending final delivery capture", () => {
const { run } = createMinimalRun({
opts: {
onTurnAdopted,
turnAdoptionLifecycle: { onAdopted },
sourceReplyDeliveryMode: "message_tool_only",
},
sessionCtx: {
@@ -2311,7 +2319,7 @@ describe("runReplyAgent pending final delivery capture", () => {
await run();
expect(onTurnAdopted).toHaveBeenCalledOnce();
expect(onAdopted).toHaveBeenCalledOnce();
expect(events).toEqual(["adopted", "agent-run"]);
});
src/auto-reply/reply/agent-runner.ts
+21 -5
View File
@@ -27,6 +27,7 @@ import { resolveFastModeState } from "../../agents/fast-mode.js";
import { resolveModelAuthMode } from "../../agents/model-auth.js";
import { isCliProvider } from "../../agents/model-selection.js";
import { deriveContextPromptTokens, hasNonzeroUsage } from "../../agents/usage.js";
import { isIngressAdoptionLostError } from "../../channels/message/ingress-drain.js";
import { enqueueCommitmentExtraction } from "../../commitments/runtime.js";
import type { OpenClawConfig } from "../../config/config.js";
import {
@@ -1233,6 +1234,8 @@ export async function runReplyAgent(params: {
replyThreadingOverride,
replyOperation: providedReplyOperation,
} = params;
// One lifecycle for all adoption sites in this run.
const turnAdoptionLifecycle = opts?.turnAdoptionLifecycle;
let activeSessionEntry = sessionEntry;
const activeSessionStore = sessionStore;
let activeIsNewSession = isNewSession;
@@ -1410,7 +1413,7 @@ export async function runReplyAgent(params: {
{
steeringMode: "all",
...(followupRun.images?.length ? { images: followupRun.images } : {}),
...(opts?.onTurnAdopted ? { waitForTranscriptCommit: true } : {}),
...(turnAdoptionLifecycle ? { waitForTranscriptCommit: true } : {}),
...(resolvedQueue.debounceMs !== undefined ? { debounceMs: resolvedQueue.debounceMs } : {}),
...(followupRun.run.sourceReplyDeliveryMode
? { sourceReplyDeliveryMode: followupRun.run.sourceReplyDeliveryMode }
@@ -1424,10 +1427,23 @@ export async function runReplyAgent(params: {
if (steerOutcome.queued) {
activeReplyOperation?.recordActivity();
try {
await opts?.onTurnAdopted?.();
await turnAdoptionLifecycle?.onAdopted();
} catch (error) {
// Transcript-backed steering is already irrevocably queued here.
// Replaying ingress would duplicate the injected user turn.
if (isIngressAdoptionLostError(error)) {
// Claim was tombstoned/superseded/guillotined after transcript commit.
// Cancel the active run so steered tools do not keep executing; do not
// rethrow — replaying ingress would duplicate the injected user turn.
const abortKey = sessionKey ?? queueKey;
if (abortKey) {
replyRunRegistry.abort(abortKey);
}
logVerbose(
`queue: active session ${steerSessionId} adoption lost after transcript commit (${error.code}); aborting steered turn without ingress replay`,
);
typing.cleanup();
return undefined;
}
// Ordinary callback failures: transcript-backed steering is irrevocable.
logVerbose(
`queue: active session ${steerSessionId} adoption finalizer failed after transcript commit: ${String(
error,
@@ -1870,7 +1886,7 @@ export async function runReplyAgent(params: {
// Adoption marks run start and must never be spool-replayed (would re-run tools).
// Suppressed delivery persists only the user transcript; crashed suppressed runs die
// silently. Deliverable turns atomically persist transcript plus recovery ownership.
await opts?.onTurnAdopted?.();
await turnAdoptionLifecycle?.onAdopted();
const runOutcome = await withBeforeAgentReplyObserver(
{
beforeDispatch: async () => {
src/auto-reply/reply/dispatch-from-config.base.test-utils.ts
+4 -3
View File
@@ -1273,9 +1273,10 @@ describe("dispatchReplyFromConfig", () => {
cfg: emptyConfig,
dispatcher,
replyOptions: {
queuedFollowupLifecycle: {
onEnqueued: vi.fn(),
onComplete: vi.fn(),
turnAdoptionLifecycle: {
onAdopted: async () => {},
onDeferred: vi.fn(),
onSettled: vi.fn(),
},
},
replyResolver,
src/auto-reply/reply/dispatch-from-config.lifecycle.ts
+1 -1
View File
@@ -160,7 +160,7 @@ export function createDispatchReplyOperationCoordinator(params: {
const allowGatewayQueueResolution =
phase === "dispatch" &&
replyTurnKind === "visible" &&
params.replyOptions?.queuedFollowupLifecycle !== undefined &&
params.replyOptions?.turnAdoptionLifecycle !== undefined &&
replyRunRegistry.get(params.dispatchOperationSessionKey) !== undefined;
if (allowGatewayQueueResolution) {
// Gateway turns need to reach getReplyFromConfig while the owner is active;
src/auto-reply/reply/followup-runner.test.ts
+27 -20
View File
@@ -410,12 +410,12 @@ async function loadFreshFollowupRunnerModuleForTest() {
runCliAgent: (params: unknown) => runCliAgentMock(params),
}));
vi.doMock("./queue.js", () => ({
admitFollowupRunLifecycle: async (run: Pick<FollowupRun, "queuedLifecycle">) => {
await run.queuedLifecycle?.onAdmitted?.();
admitFollowupRunLifecycle: async (run: Pick<FollowupRun, "turnAdoptionLifecycle">) => {
await run.turnAdoptionLifecycle?.onAdopted?.();
},
clearFollowupQueue: clearFollowupQueueForFollowupTest,
completeFollowupRunLifecycle: (run: Pick<FollowupRun, "queuedLifecycle">) =>
run.queuedLifecycle?.onComplete?.(),
completeFollowupRunLifecycle: (run: Pick<FollowupRun, "turnAdoptionLifecycle">) =>
run.turnAdoptionLifecycle?.onSettled?.(),
enqueueFollowupRun: enqueueFollowupRunForFollowupTest,
isFollowupRunAborted: (run: Pick<FollowupRun, "abortSignal" | "queueAbortSignal">) =>
run.abortSignal?.aborted === true || run.queueAbortSignal?.aborted === true,
@@ -902,13 +902,15 @@ describe("createFollowupRunner reply-lane admission", () => {
const pending = runner(
createQueuedRun({
queuedLifecycle: {
onAdmitted: async () => {
turnAdoptionLifecycle: {
onAdopted: async () => {
events.push("admission-started");
await admissionBarrier;
events.push("admitted");
},
onComplete: () => events.push("complete"),
onSettled: () => events.push("complete"),
admission: "exclusive",
onAbandoned: () => {},
},
run: { provider: "anthropic", model: "claude" },
}),
@@ -942,13 +944,15 @@ describe("createFollowupRunner reply-lane admission", () => {
const pending = runner(
createQueuedRun({
abortSignal: abortController.signal,
queuedLifecycle: {
onAdmitted: async () => {
turnAdoptionLifecycle: {
onAdopted: async () => {
events.push("admission-started");
await admissionBarrier;
events.push("admitted");
},
onComplete: () => events.push("complete"),
onSettled: () => events.push("complete"),
admission: "exclusive",
onAbandoned: () => {},
},
run: { provider: "anthropic", model: "claude" },
}),
@@ -1297,7 +1301,7 @@ describe("createFollowupRunner reply-lane admission", () => {
model: "claude",
sessionKey: "main",
},
queuedLifecycle: { onComplete },
turnAdoptionLifecycle: { onAdopted: async () => {}, onSettled: onComplete },
}),
),
).rejects.toThrow("session load failed");
@@ -2564,8 +2568,9 @@ describe("createFollowupRunner runtime config", () => {
createQueuedRun({
originatingChannel: "telegram",
originatingTo: "chat-1",
queuedLifecycle: {
onComplete: () => {
turnAdoptionLifecycle: {
onAdopted: async () => {},
onSettled: () => {
operationResultDuringCompletion = replyRunRegistryForTest.get("main")?.result;
},
},
@@ -2660,8 +2665,9 @@ describe("createFollowupRunner runtime config", () => {
createQueuedRun({
originatingChannel: "telegram",
originatingTo: "chat-1",
queuedLifecycle: {
onComplete: () => {
turnAdoptionLifecycle: {
onAdopted: async () => {},
onSettled: () => {
operationResultDuringCompletion = replyRunRegistryForTest.get("main")?.result;
},
},
@@ -3061,8 +3067,9 @@ describe("createFollowupRunner runtime config", () => {
createQueuedRun({
originatingChannel: "telegram",
originatingTo: "chat-1",
queuedLifecycle: {
onComplete: () => {
turnAdoptionLifecycle: {
onAdopted: async () => {},
onSettled: () => {
operationResultDuringCompletion = replyRunRegistryForTest.get("main")?.result;
},
},
@@ -6038,7 +6045,7 @@ describe("createFollowupRunner messaging delivery and dedupe", () => {
const finalText =
"Here is the answer the queued user asked for. It includes enough detail to be a visible response, and it has another sentence so the substantive-final detector treats it as a real reply.";
const parentOnComplete = vi.fn();
const parentLifecycle = { onComplete: parentOnComplete };
const parentLifecycle = { onAdopted: async () => {}, onSettled: parentOnComplete };
const queued = baseQueuedRun("discord");
const { onBlockReply } = await runMessagingCase({
agentResult: {
@@ -6049,7 +6056,7 @@ describe("createFollowupRunner messaging delivery and dedupe", () => {
...queued,
originatingChannel: "discord",
originatingTo: "channel:C1",
queuedLifecycle: parentLifecycle,
turnAdoptionLifecycle: parentLifecycle,
run: {
...queued.run,
sourceReplyDeliveryMode: "message_tool_only",
@@ -6072,7 +6079,7 @@ describe("createFollowupRunner messaging delivery and dedupe", () => {
expect(retry?.prompt).toContain("message(action=send)");
expect(retry?.prompt).toContain(finalText);
// System retry detaches from the client turn lifecycle; parent completion owns onComplete once.
expect(retry?.queuedLifecycle).toBeUndefined();
expect(retry?.turnAdoptionLifecycle).toBeUndefined();
expect(parentOnComplete).toHaveBeenCalledTimes(1);
});
src/auto-reply/reply/get-reply-run.ts
+8 -5
View File
@@ -1414,14 +1414,17 @@ export async function runPreparedReply(
);
// Abort-signal attachment for queued followups:
// - room_event: always inherit (source admission fence / ambient cancel).
// - Gateway-owned lifecycle (chat.send): always inherit so Esc can cancel a
// turn after chat.send terminalizes while still queued.
// - Gateway-owned lifecycle (chat.send / turnAdoptionLifecycle): always inherit
// so Esc can cancel a turn after chat.send terminalizes while still queued.
// - plain user_request without lifecycle: deliberately detach from the
// source/active-lane signal so a superseded parent abort does not cancel a
// still-valid queued user turn.
const hasQueuedOwnershipLifecycle = Boolean(opts?.turnAdoptionLifecycle);
const queuedFollowupAbortSignal =
opts?.queuedFollowupLifecycle || inboundEventKind === "room_event"
? (opts?.queuedFollowupAbortSignal ?? opts?.abortSignal)
hasQueuedOwnershipLifecycle || inboundEventKind === "room_event"
? (opts?.queuedFollowupAbortSignal ??
opts?.turnAdoptionLifecycle?.abortSignal ??
opts?.abortSignal)
: undefined;
const replyRoute = resolveEffectiveReplyRoute({
ctx: {
@@ -1531,7 +1534,7 @@ export async function runPreparedReply(
currentInboundContext,
...(queuedFollowupAbortSignal ? { abortSignal: queuedFollowupAbortSignal } : {}),
deliveryCorrelations: opts?.queuedDeliveryCorrelations,
queuedLifecycle: opts?.queuedFollowupLifecycle,
turnAdoptionLifecycle: opts?.turnAdoptionLifecycle,
onReplyAdmissionWaitChange: opts?.onReplyAdmissionWaitChange,
messageId: sessionCtx.MessageSidFull ?? sessionCtx.MessageSid,
summaryLine: baseBodyTrimmedRaw,
src/auto-reply/reply/queue.collect.test.ts
+111 -51
View File
@@ -30,13 +30,52 @@ import { clearFollowupQueue, getExistingFollowupQueue } from "./queue/state.js";
installQueueRuntimeErrorSilencer();
describe("followup queue collect routing", () => {
it("marks exclusive admission without onAbandoned and isolates collect identity", () => {
// Failure window: cancel-only used to be inferred from missing onAbandoned,
// so exclusive admission without onAbandoned shared collect identity.
const exclusiveNoAbandon = createRun({ prompt: "exclusive a" });
exclusiveNoAbandon.turnAdoptionLifecycle = {
admission: "exclusive",
onAdopted: async () => {},
};
const exclusiveSibling = createRun({ prompt: "exclusive b" });
exclusiveSibling.turnAdoptionLifecycle = {
admission: "exclusive",
onAdopted: async () => {},
};
const cancelOnly = createRun({ prompt: "cancel-only" });
cancelOnly.turnAdoptionLifecycle = {
admission: "cancel-only",
ownerKey: "gw:owner",
onAdopted: async () => {},
};
const cancelOnlyShared = createRun({ prompt: "cancel-only shared" });
cancelOnlyShared.turnAdoptionLifecycle = {
admission: "cancel-only",
ownerKey: "gw:owner",
onAdopted: async () => {},
};
const exclusiveA = resolveFollowupDeliveryContextKey(exclusiveNoAbandon);
const exclusiveB = resolveFollowupDeliveryContextKey(exclusiveSibling);
expect(exclusiveA).not.toEqual(exclusiveB);
const cancelA = resolveFollowupDeliveryContextKey(cancelOnly);
const cancelB = resolveFollowupDeliveryContextKey(cancelOnlyShared);
expect(cancelA).toEqual(cancelB);
});
it("retries lifecycle admission after a callback rejection", async () => {
const onAdmitted = vi
.fn<() => Promise<void>>()
.mockRejectedValueOnce(new Error("admission failed"))
.mockResolvedValueOnce();
const run = createRun({ prompt: "retry admission" });
run.queuedLifecycle = { onAdmitted };
run.turnAdoptionLifecycle = {
onAdopted: onAdmitted,
admission: "exclusive",
onAbandoned: () => {},
};
await expect(admitFollowupRunLifecycle(run)).rejects.toThrow("admission failed");
await expect(admitFollowupRunLifecycle(run)).resolves.toBeUndefined();
@@ -61,7 +100,12 @@ describe("followup queue collect routing", () => {
events.push("complete");
});
const run = createRun({ prompt: "complete during admission" });
run.queuedLifecycle = { onAdmitted, onComplete };
run.turnAdoptionLifecycle = {
onAdopted: onAdmitted,
onSettled: onComplete,
admission: "exclusive",
onAbandoned: () => {},
};
const admission = admitFollowupRunLifecycle(run);
await admissionStarted.promise;
@@ -84,7 +128,7 @@ describe("followup queue collect routing", () => {
const key = `test-rejected-lifecycle-${Date.now()}`;
const onEnqueued = vi.fn(() => false);
const run = createRun({ prompt: "duplicate owner" });
run.queuedLifecycle = { onEnqueued };
run.turnAdoptionLifecycle = { onAdopted: async () => {}, onDeferred: onEnqueued };
const enqueued = enqueueFollowupRun(key, run, {
mode: "followup",
@@ -748,7 +792,7 @@ describe("followup queue collect routing", () => {
originatingTo: "channel:A",
originatingChatType: "channel",
}),
queuedLifecycle: { onComplete },
turnAdoptionLifecycle: { onAdopted: async () => {}, onSettled: onComplete },
},
settings,
);
@@ -783,7 +827,11 @@ describe("followup queue collect routing", () => {
key,
{
...createRun({ prompt: "rejected" }),
queuedLifecycle: { onEnqueued, onComplete },
turnAdoptionLifecycle: {
onAdopted: async () => {},
onDeferred: onEnqueued,
onSettled: onComplete,
},
},
settings,
),
@@ -1580,7 +1628,7 @@ describe("followup queue collect routing", () => {
originatingTo: "same-target",
originatingChatType: "direct",
}),
queuedLifecycle: { onComplete },
turnAdoptionLifecycle: { onAdopted: async () => {}, onSettled: onComplete },
},
settings,
);
@@ -1907,7 +1955,7 @@ describe("followup queue collect routing", () => {
};
const controller = new AbortController();
const begin = () => () => undefined;
const lifecycle = { onComplete: () => undefined };
const lifecycle = { onAdopted: async () => {}, onSettled: () => undefined };
enqueueFollowupRun(
key,
@@ -1927,7 +1975,7 @@ describe("followup queue collect routing", () => {
first.currentInboundContext = { text: "room event body" };
first.abortSignal = controller.signal;
first.deliveryCorrelations = [{ begin }];
first.queuedLifecycle = lifecycle;
first.turnAdoptionLifecycle = lifecycle;
enqueueFollowupRun(
key,
createRun({
@@ -1948,7 +1996,7 @@ describe("followup queue collect routing", () => {
expect(calls[0]?.currentInboundContext?.text).toBe("room event body");
expect(calls[0]?.abortSignal).toBe(controller.signal);
expect(calls[0]?.deliveryCorrelations?.[0]?.begin).toBe(begin);
expect(calls[0]?.queuedLifecycle).toBe(lifecycle);
expect(calls[0]?.turnAdoptionLifecycle).toBe(lifecycle);
expect(calls[1]?.prompt).toBe("second");
});
@@ -2336,7 +2384,7 @@ describe("followup queue collect routing", () => {
originatingChannel: "webchat",
originatingTo: "session:main",
}),
queuedLifecycle: { ownerKey },
turnAdoptionLifecycle: { onAdopted: async () => {}, ownerKey },
},
settings,
);
@@ -3122,7 +3170,7 @@ describe("followup queue collect routing", () => {
{
...createRun({ prompt: "aborted" }),
abortSignal: controller.signal,
queuedLifecycle: { onComplete },
turnAdoptionLifecycle: { onAdopted: async () => {}, onSettled: onComplete },
},
settings,
);
@@ -3411,7 +3459,7 @@ describe("followup queue collect routing", () => {
currentInboundContext: { text: "live context" },
abortSignal: controller.signal,
deliveryCorrelations: [{ begin }],
queuedLifecycle: { onComplete },
turnAdoptionLifecycle: { onAdopted: async () => {}, onSettled: onComplete },
},
settings,
);
@@ -3429,7 +3477,7 @@ describe("followup queue collect routing", () => {
expect(calls[1]?.currentInboundAudio).toBe(true);
expect(calls[1]?.currentInboundContext?.text).toBe("live context");
expect(calls[1]?.abortSignal).toBe(controller.signal);
expect(calls[1]?.queuedLifecycle?.onComplete).toBe(onComplete);
expect(calls[1]?.turnAdoptionLifecycle?.onSettled).toBe(onComplete);
expect(calls[1]?.deliveryCorrelations?.[0]?.begin).toBe(begin);
});
@@ -3501,7 +3549,7 @@ describe("followup queue collect routing", () => {
currentInboundEventKind: "room_event",
currentInboundContext: { text: "dropped context" },
abortSignal: controller.signal,
queuedLifecycle: { onComplete },
turnAdoptionLifecycle: { onAdopted: async () => {}, onSettled: onComplete },
},
settings,
);
@@ -3537,9 +3585,10 @@ describe("followup queue collect routing", () => {
{
...createRun({ prompt }),
abortSignal: new AbortController().signal,
queuedLifecycle: {
turnAdoptionLifecycle: {
onAdopted: async () => {},
onCancellationRetired: sourceCancellationRetirements[index],
onComplete: sourceCompletions[index],
onSettled: sourceCompletions[index],
},
},
settings,
@@ -3551,12 +3600,12 @@ describe("followup queue collect routing", () => {
calls.push(run);
if (calls.length === 1) {
expect(run.prompt).toContain("[Queue overflow] Dropped 2 messages due to cap.");
await run.queuedLifecycle?.onAdmitted?.();
await run.turnAdoptionLifecycle?.onAdopted?.();
expect(sourceCancellationRetirements[0]).toHaveBeenCalledTimes(1);
expect(sourceCancellationRetirements[1]).not.toHaveBeenCalled();
expect(sourceCompletions[0]).not.toHaveBeenCalled();
expect(sourceCompletions[1]).not.toHaveBeenCalled();
run.queuedLifecycle?.onComplete?.();
run.turnAdoptionLifecycle?.onSettled?.();
expect(sourceCompletions[0]).toHaveBeenCalledTimes(1);
expect(sourceCompletions[1]).toHaveBeenCalledTimes(1);
return;
@@ -3587,11 +3636,13 @@ describe("followup queue collect routing", () => {
key,
{
...createRun({ prompt: "dropped lifecycle source" }),
queuedLifecycle: {
onAdmitted: async () => {
turnAdoptionLifecycle: {
onAdopted: async () => {
events.push("source-admitted");
},
onComplete: sourceComplete,
onSettled: sourceComplete,
admission: "exclusive",
onAbandoned: () => {},
},
},
settings,
@@ -3601,10 +3652,10 @@ describe("followup queue collect routing", () => {
scheduleFollowupDrain(key, async (run) => {
if (run.prompt.includes("[Queue overflow]")) {
events.push("summary-started");
expect(run.queuedLifecycle?.onAdmitted).toEqual(expect.any(Function));
await run.queuedLifecycle?.onAdmitted?.();
expect(run.turnAdoptionLifecycle?.onAdopted).toEqual(expect.any(Function));
await run.turnAdoptionLifecycle?.onAdopted?.();
events.push("model");
run.queuedLifecycle?.onComplete?.();
run.turnAdoptionLifecycle?.onSettled?.();
return;
}
events.push("live-followup");
@@ -3632,7 +3683,7 @@ describe("followup queue collect routing", () => {
let attempts = 0;
const runFollowup = async (run: FollowupRun) => {
calls.push(run);
expect(run.queuedLifecycle).toBeUndefined();
expect(run.turnAdoptionLifecycle).toBeUndefined();
attempts += 1;
if (attempts === 1) {
firstAttempt.resolve();
@@ -3654,7 +3705,7 @@ describe("followup queue collect routing", () => {
...createRun({ prompt: "dropped ambient" }),
currentInboundEventKind: "room_event",
currentInboundContext: { text: "dropped context" },
queuedLifecycle: { onComplete },
turnAdoptionLifecycle: { onAdopted: async () => {}, onSettled: onComplete },
},
settings,
);
@@ -3671,7 +3722,7 @@ describe("followup queue collect routing", () => {
expect(getExistingFollowupQueue(key)?.summarySources[0]?.currentInboundEventKind).toBe(
"room_event",
);
expect(getExistingFollowupQueue(key)?.summarySources[0]?.queuedLifecycle).toBeDefined();
expect(getExistingFollowupQueue(key)?.summarySources[0]?.turnAdoptionLifecycle).toBeDefined();
expect(getExistingFollowupQueue(key)?.summarySources[0]?.currentInboundContext).toBeUndefined();
scheduleFollowupDrain(key, runFollowup);
@@ -3703,7 +3754,7 @@ describe("followup queue collect routing", () => {
{
...createRun({ prompt: "first" }),
abortSignal: new AbortController().signal,
queuedLifecycle: { onComplete: firstComplete },
turnAdoptionLifecycle: { onAdopted: async () => {}, onSettled: firstComplete },
},
settings,
);
@@ -3712,7 +3763,7 @@ describe("followup queue collect routing", () => {
{
...createRun({ prompt: "second" }),
abortSignal: new AbortController().signal,
queuedLifecycle: { onComplete: secondComplete },
turnAdoptionLifecycle: { onAdopted: async () => {}, onSettled: secondComplete },
},
settings,
);
@@ -3736,14 +3787,16 @@ describe("followup queue collect routing", () => {
const settings: QueueSettings = { mode: "collect", debounceMs: 0 };
const first = createRun({ prompt: "first" });
first.queuedLifecycle = {
onAdmitted: async () => {
first.turnAdoptionLifecycle = {
onAdopted: async () => {
events.push("first-admitted");
},
admission: "exclusive",
onAbandoned: () => {},
};
const second = createRun({ prompt: "second" });
second.queuedLifecycle = {
onAdmitted: vi
second.turnAdoptionLifecycle = {
onAdopted: vi
.fn<() => Promise<void>>()
.mockImplementationOnce(async () => {
events.push("second-rejected");
@@ -3752,6 +3805,8 @@ describe("followup queue collect routing", () => {
.mockImplementationOnce(async () => {
events.push("second-admitted");
}),
admission: "exclusive",
onAbandoned: () => {},
};
enqueueFollowupRun(key, first, settings);
@@ -3785,7 +3840,7 @@ describe("followup queue collect routing", () => {
"second-admitted",
"model:second",
]);
expect(second.queuedLifecycle.onAdmitted).toHaveBeenCalledTimes(2);
expect(second.turnAdoptionLifecycle.onAdopted).toHaveBeenCalledTimes(2);
});
it("collects transcript-owned turns under one aggregate recorder", async () => {
@@ -3819,7 +3874,7 @@ describe("followup queue collect routing", () => {
currentInboundContext: { text: "shared gateway context", promptJoiner: " " },
deliveryCorrelations: [deliveryCorrelation],
abortSignal: new AbortController().signal,
queuedLifecycle: { onComplete },
turnAdoptionLifecycle: { onAdopted: async () => {}, onSettled: onComplete },
},
settings,
);
@@ -3920,9 +3975,10 @@ describe("followup queue collect routing", () => {
transcriptPrompt: `${prompt} transcript`,
userTurnTranscriptRecorder: createRecorder(`${prompt} transcript`),
abortSignal,
queuedLifecycle: {
turnAdoptionLifecycle: {
onAdopted: async () => {},
onCancellationRetired: sourceCancellationRetirements[index],
onComplete: sourceCompletions[index],
onSettled: sourceCompletions[index],
},
},
settings,
@@ -3934,7 +3990,7 @@ describe("followup queue collect routing", () => {
if (calls.length === 1) {
expect(run.abortSignal).toBeDefined();
expect(run.abortSignal).not.toBe(survivor.signal);
await run.queuedLifecycle?.onAdmitted?.();
await run.turnAdoptionLifecycle?.onAdopted?.();
expect(sourceCancellationRetirements[0]).toHaveBeenCalledTimes(1);
expect(sourceCancellationRetirements[1]).not.toHaveBeenCalled();
expect(sourceCompletions[0]).not.toHaveBeenCalled();
@@ -3968,7 +4024,7 @@ describe("followup queue collect routing", () => {
scheduleFollowupDrain(key, async (run) => {
expect(run.abortSignal).toBeUndefined();
expect(run.queueAbortSignal?.aborted).toBe(false);
await run.queuedLifecycle?.onAdmitted?.();
await run.turnAdoptionLifecycle?.onAdopted?.();
clearFollowupQueue(key);
expect(run.queueAbortSignal?.aborted).toBe(true);
done.resolve();
@@ -3990,7 +4046,7 @@ describe("followup queue collect routing", () => {
const enqueueSource = (prompt: string, onComplete: () => void, abortSignal?: AbortSignal) => {
const source: FollowupRun = {
...createRun({ prompt }),
queuedLifecycle: { onComplete },
turnAdoptionLifecycle: { onAdopted: async () => {}, onSettled: onComplete },
};
if (abortSignal) {
source.abortSignal = abortSignal;
@@ -4052,7 +4108,7 @@ describe("followup queue collect routing", () => {
{
...createRun({ prompt: "owner A summary" }),
abortSignal: new AbortController().signal,
queuedLifecycle: { onComplete: summarizedComplete },
turnAdoptionLifecycle: { onAdopted: async () => {}, onSettled: summarizedComplete },
},
settings,
);
@@ -4061,7 +4117,7 @@ describe("followup queue collect routing", () => {
{
...createRun({ prompt: "owner B live" }),
abortSignal: aborted.signal,
queuedLifecycle: { onComplete: abortedComplete },
turnAdoptionLifecycle: { onAdopted: async () => {}, onSettled: abortedComplete },
},
settings,
);
@@ -4105,7 +4161,7 @@ describe("followup queue collect routing", () => {
{
...createRun({ prompt: "elided and cancelled" }),
abortSignal: elided.signal,
queuedLifecycle: { onComplete: elidedComplete },
turnAdoptionLifecycle: { onAdopted: async () => {}, onSettled: elidedComplete },
},
settings,
);
@@ -4139,7 +4195,7 @@ describe("followup queue collect routing", () => {
key,
{
...createRun({ prompt: "elided source" }),
queuedLifecycle: { onComplete: elidedComplete },
turnAdoptionLifecycle: { onAdopted: async () => {}, onSettled: elidedComplete },
},
settings,
);
@@ -4147,7 +4203,7 @@ describe("followup queue collect routing", () => {
key,
{
...createRun({ prompt: "retained source" }),
queuedLifecycle: { onComplete: retainedComplete },
turnAdoptionLifecycle: { onAdopted: async () => {}, onSettled: retainedComplete },
},
settings,
);
@@ -4158,7 +4214,7 @@ describe("followup queue collect routing", () => {
if (calls.length === 1) {
expect(run.prompt).toContain("Dropped 2 messages");
expect(run.prompt).toContain("retained source");
await run.queuedLifecycle?.onAdmitted?.();
await run.turnAdoptionLifecycle?.onAdopted?.();
expect(getExistingFollowupQueue(key)?.summaryElisions).toEqual([]);
expect(getExistingFollowupQueue(key)?.droppedCount).toBe(0);
throw new Error("admitted summary failure");
@@ -4186,14 +4242,16 @@ describe("followup queue collect routing", () => {
};
const first = createRun({ prompt: "first dropped" });
first.queuedLifecycle = {
onAdmitted: async () => {
first.turnAdoptionLifecycle = {
onAdopted: async () => {
events.push("first-admitted");
},
admission: "exclusive",
onAbandoned: () => {},
};
const second = createRun({ prompt: "second dropped" });
second.queuedLifecycle = {
onAdmitted: vi
second.turnAdoptionLifecycle = {
onAdopted: vi
.fn<() => Promise<void>>()
.mockImplementationOnce(async () => {
events.push("second-rejected");
@@ -4202,6 +4260,8 @@ describe("followup queue collect routing", () => {
.mockImplementationOnce(async () => {
events.push("second-admitted");
}),
admission: "exclusive",
onAbandoned: () => {},
};
enqueueFollowupRun(key, first, settings);
@@ -4238,7 +4298,7 @@ describe("followup queue collect routing", () => {
"summary-model",
"live-followup",
]);
expect(second.queuedLifecycle.onAdmitted).toHaveBeenCalledTimes(2);
expect(second.turnAdoptionLifecycle.onAdopted).toHaveBeenCalledTimes(2);
});
});
src/auto-reply/reply/queue.in-flight.test.ts
+12 -11
View File
@@ -44,11 +44,11 @@ describe("followup queue in-flight ownership", () => {
const calls: FollowupRun[] = [];
const active = {
...createRun({ prompt: "active" }),
queuedLifecycle: { onComplete: activeComplete },
turnAdoptionLifecycle: { onAdopted: async () => {}, onSettled: activeComplete },
};
const runFollowup = async (run: FollowupRun) => {
calls.push(run);
await run.queuedLifecycle?.onAdmitted?.();
await run.turnAdoptionLifecycle?.onAdopted?.();
if (run === active) {
entered.resolve();
await release.promise;
@@ -68,7 +68,7 @@ describe("followup queue in-flight ownership", () => {
key,
{
...createRun({ prompt: "pending" }),
queuedLifecycle: { onComplete: pendingComplete },
turnAdoptionLifecycle: { onAdopted: async () => {}, onSettled: pendingComplete },
},
createSettings(dropPolicy),
"none",
@@ -111,7 +111,7 @@ describe("followup queue in-flight ownership", () => {
const rejectedComplete = vi.fn();
const active = createRun({ prompt: "active" });
const runFollowup = async (run: FollowupRun) => {
await run.queuedLifecycle?.onAdmitted?.();
await run.turnAdoptionLifecycle?.onAdopted?.();
if (run === active) {
entered.resolve();
await release.promise;
@@ -134,9 +134,10 @@ describe("followup queue in-flight ownership", () => {
key,
{
...createRun({ prompt: "rejected" }),
queuedLifecycle: {
onEnqueued: rejectedEnqueued,
onComplete: rejectedComplete,
turnAdoptionLifecycle: {
onAdopted: async () => {},
onDeferred: rejectedEnqueued,
onSettled: rejectedComplete,
},
},
createSettings("new"),
@@ -179,7 +180,7 @@ describe("followup queue in-flight ownership", () => {
originatingTo: "channel:A",
originatingChatType: "channel",
}),
queuedLifecycle: { onComplete },
turnAdoptionLifecycle: { onAdopted: async () => {}, onSettled: onComplete },
}));
const runFollowup = async (run: FollowupRun) => {
if (!aggregate) {
@@ -207,7 +208,7 @@ describe("followup queue in-flight ownership", () => {
key,
{
...createRun({ prompt: "pending-old" }),
queuedLifecycle: { onComplete: pendingComplete },
turnAdoptionLifecycle: { onAdopted: async () => {}, onSettled: pendingComplete },
},
oldSettings,
"none",
@@ -221,7 +222,7 @@ describe("followup queue in-flight ownership", () => {
expect(pendingComplete).toHaveBeenCalledOnce();
expect(groupCompletions.map((complete) => complete.mock.calls.length)).toEqual([0, 0]);
await aggregate?.queuedLifecycle?.onAdmitted?.();
await aggregate?.turnAdoptionLifecycle?.onAdopted?.();
expect(queue?.items.map((item) => item.prompt)).toEqual(["survivor"]);
expect(queue?.inFlight.size).toBe(2);
expect(getFollowupQueueDepth(key)).toBe(1);
@@ -231,7 +232,7 @@ describe("followup queue in-flight ownership", () => {
key,
{
...createRun({ prompt: "rejected-new" }),
queuedLifecycle: { onComplete: rejectedComplete },
turnAdoptionLifecycle: { onAdopted: async () => {}, onSettled: rejectedComplete },
},
{ ...initialSettings, cap: 1, dropPolicy: "new" },
"none",
src/auto-reply/reply/queue/drain.ts
+39 -19
View File
@@ -49,16 +49,28 @@ const FOLLOWUP_RUN_CALLBACKS = resolveGlobalMap<string, (run: FollowupRun) => Pr
const QUEUED_ADMISSION_OWNER_STATE_KEY = Symbol.for("openclaw.queuedAdmissionOwnerState");
const queuedAdmissionOwnerState = resolveGlobalSingleton(QUEUED_ADMISSION_OWNER_STATE_KEY, () => ({
keys: new WeakMap<NonNullable<FollowupRun["queuedLifecycle"]>, string>(),
keys: new WeakMap<NonNullable<FollowupRun["turnAdoptionLifecycle"]>, string>(),
nextId: 1,
}));
function resolveQueuedLifecycleDeliveryKey(lifecycle: FollowupRun["queuedLifecycle"]): string {
function hasExclusiveTurnAdmission(
lifecycle: FollowupRun["turnAdoptionLifecycle"],
): lifecycle is NonNullable<FollowupRun["turnAdoptionLifecycle"]> & {
admission: "exclusive";
} {
return lifecycle?.admission === "exclusive";
}
function resolveTurnAdoptionLifecycleDeliveryKey(
lifecycle: FollowupRun["turnAdoptionLifecycle"],
): string {
if (!lifecycle) {
return "";
}
const explicitOwnerKey = lifecycle.ownerKey ?? "";
if (!lifecycle.onAdmitted) {
// Closed admission marker — never infer exclusive from onAbandoned presence.
// Cancel-only owners share collect identity via ownerKey alone.
if (!hasExclusiveTurnAdmission(lifecycle)) {
return explicitOwnerKey;
}
let admissionOwnerKey = queuedAdmissionOwnerState.keys.get(lifecycle);
@@ -73,7 +85,9 @@ function resolveQueuedLifecycleDeliveryKey(lifecycle: FollowupRun["queuedLifecyc
function assertSingleAdmissionOwner(items: readonly FollowupRun[]): void {
const owners = new Set(
items.flatMap((item) => (item.queuedLifecycle?.onAdmitted ? [item.queuedLifecycle] : [])),
items.flatMap((item) =>
hasExclusiveTurnAdmission(item.turnAdoptionLifecycle) ? [item.turnAdoptionLifecycle] : [],
),
);
if (owners.size > 1) {
throw new Error("followup queue cannot aggregate distinct admission lifecycles");
@@ -180,7 +194,7 @@ export function resolveFollowupDeliveryContextKey(run: FollowupRun): string {
run.originatingReplyToMode ?? "",
normalizeChatType(run.originatingChatType) ?? "",
resolveFollowupAuthorizationKey(execution),
run.queuedLifecycle?.ownerKey ?? "",
run.turnAdoptionLifecycle?.ownerKey ?? "",
normalizeOptionalString(execution.runtimePolicySessionKey ?? execution.sessionKey) ?? "",
execution.messageProvider ?? "",
JSON.stringify([...new Set(execution.clientCaps ?? [])].toSorted()),
@@ -209,7 +223,7 @@ export function resolveFollowupDeliveryContextKey(run: FollowupRun): string {
execution.suppressNextUserMessagePersistence === true,
execution.suppressTranscriptOnlyAssistantPersistence === true,
execution.blockReplyBreak,
resolveQueuedLifecycleDeliveryKey(run.queuedLifecycle),
resolveTurnAdoptionLifecycleDeliveryKey(run.turnAdoptionLifecycle),
]);
}
@@ -297,7 +311,7 @@ type FollowupRuntimeMetadata = Pick<
| "abortSignal"
| "queueAbortSignal"
| "deliveryCorrelations"
| "queuedLifecycle"
| "turnAdoptionLifecycle"
| "onReplyAdmissionWaitChange"
>;
@@ -409,7 +423,7 @@ function resolveAggregateOwner(items: readonly FollowupRun[]): FollowupRun | und
// later transport-only source has no cancellation identity.
return (
items.findLast((item) => item.abortSignal) ??
items.findLast((item) => item.queuedLifecycle) ??
items.findLast((item) => item.turnAdoptionLifecycle) ??
items.at(-1)
);
}
@@ -542,7 +556,7 @@ function collectRuntimeMetadata(
abortSignal,
queueAbortSignal: items.find((item) => item.queueAbortSignal)?.queueAbortSignal,
deliveryCorrelations: deliveryCorrelations.length > 0 ? deliveryCorrelations : undefined,
queuedLifecycle: items.length === 1 ? items[0]?.queuedLifecycle : undefined,
turnAdoptionLifecycle: items.length === 1 ? items[0]?.turnAdoptionLifecycle : undefined,
onReplyAdmissionWaitChange:
admissionWaitCallbacks.size > 0
? (waiting) => {
@@ -655,7 +669,7 @@ function releaseQueueSummaryDeliveryForRetry(
if (sourceIndex >= 0) {
queue.summarySources[sourceIndex] = createOverflowSummaryRetrySource(source);
}
if (!source.queuedLifecycle) {
if (!source.turnAdoptionLifecycle) {
completeFollowupRunLifecycle(source);
}
}
@@ -698,7 +712,7 @@ async function runQueueSummaryDelivery(
const cancellation = createAggregateCancellation(protectedSources);
const needsAdmission =
protectedSources.length > 1 ||
protectedSources.some((source) => source.queuedLifecycle?.onAdmitted);
protectedSources.some((source) => hasExclusiveTurnAdmission(source.turnAdoptionLifecycle));
const onAdmitted = needsAdmission
? async () => {
if (admitted) {
@@ -884,7 +898,7 @@ export function createOverflowSummaryRetrySource(source: FollowupRun): FollowupR
originatingReplyToMode: source.originatingReplyToMode,
originatingChatType: source.originatingChatType,
abortSignal: source.abortSignal,
queuedLifecycle: source.queuedLifecycle,
turnAdoptionLifecycle: source.turnAdoptionLifecycle,
onReplyAdmissionWaitChange: source.onReplyAdmissionWaitChange,
...(source.currentInboundEventKind === "room_event"
? { currentInboundEventKind: "room_event" }
@@ -948,12 +962,14 @@ async function runSyntheticOverflowSummary(params: {
onReplyAdmissionWaitChange: collectRuntimeMetadata(params.sources).onReplyAdmissionWaitChange,
...(params.onAdmitted
? {
queuedLifecycle: {
onAdmitted: async () => {
turnAdoptionLifecycle: {
// Synthetic aggregate owner — not a durable exclusive ingress identity.
admission: "cancel-only" as const,
onAdopted: async () => {
await params.onAdmitted?.();
admitted = true;
},
onComplete: () => {
onSettled: () => {
if (admitted) {
for (const source of params.sources) {
completeFollowupRunLifecycle(source);
@@ -1229,7 +1245,9 @@ export function scheduleFollowupDrain(
};
const needsGroupAdmission =
activeGroupItems.length > 1 ||
activeGroupItems.some((item) => item.queuedLifecycle?.onAdmitted);
activeGroupItems.some((item) =>
hasExclusiveTurnAdmission(item.turnAdoptionLifecycle),
);
const consumeAdmittedGroup = () => {
cancellation.admit();
admitted = true;
@@ -1264,9 +1282,11 @@ export function scheduleFollowupDrain(
...collectRuntimeMetadata(activeGroupItems, cancellation.signal),
...(needsGroupAdmission
? {
queuedLifecycle: {
onAdmitted: admitGroupSources,
onComplete: () => {
turnAdoptionLifecycle: {
// Synthetic aggregate owner — sources keep their own admission.
admission: "cancel-only" as const,
onAdopted: admitGroupSources,
onSettled: () => {
if (admitted) {
completeGroup();
}
src/auto-reply/reply/queue/types.ts
+65 -53
View File
@@ -17,9 +17,9 @@ import type { UserTurnTranscriptRecorder } from "../../../sessions/user-turn-tra
import type { SkillSnapshot } from "../../../skills/types.js";
import type {
QueuedReplyDeliveryCorrelation,
QueuedReplyLifecycle,
SourceReplyDeliveryMode,
TaskSuggestionDeliveryMode,
TurnAdoptionLifecycle,
} from "../../get-reply-options.types.js";
import type { OriginatingChannelType } from "../../templating.js";
import type { ElevatedLevel, ReasoningLevel, ThinkLevel, VerboseLevel } from "../directives.js";
@@ -35,6 +35,15 @@ export type QueueSettings = {
dropPolicy?: QueueDropPolicy;
};
export type ResolveQueueSettingsParams = {
cfg: OpenClawConfig;
channel?: string;
sessionEntry?: SessionEntry;
inlineMode?: QueueMode;
inlineOptions?: Partial<QueueSettings>;
pluginDebounceMs?: number;
};
export type QueueDedupeMode = "message-id" | "prompt" | "none";
type QueueInsertPosition = "tail" | "front";
@@ -72,7 +81,8 @@ export type FollowupRun = {
/** Queue-owned cancellation fence used when lifecycle cleanup invalidates pending work. */
queueAbortSignal?: AbortSignal;
deliveryCorrelations?: QueuedReplyDeliveryCorrelation[];
queuedLifecycle?: QueuedReplyLifecycle;
/** Canonical ownership lifecycle for durable ingress / reply-lane transfer. */
turnAdoptionLifecycle?: TurnAdoptionLifecycle;
/** Dispatch-scoped freshness owner for a queued delivery-barrier wait. */
onReplyAdmissionWaitChange?: (waiting: boolean) => void;
/** Provider message ID, when available (for deduplication). */
@@ -198,76 +208,87 @@ export function resolveFollowupAbortSignal(
return signals.length > 1 ? AbortSignal.any(signals) : signals[0];
}
const enqueuedFollowupLifecycles = new WeakSet<QueuedReplyLifecycle>();
const admittedFollowupLifecycles = new WeakSet<QueuedReplyLifecycle>();
const admittingFollowupLifecycles = new WeakMap<QueuedReplyLifecycle, Promise<void>>();
const retiredFollowupCancellationLifecycles = new WeakSet<QueuedReplyLifecycle>();
const completedFollowupLifecycles = new WeakSet<QueuedReplyLifecycle>();
const completedFollowupLifecycleCallbacks = new WeakSet<QueuedReplyLifecycle>();
const enqueuedTurnAdoptionLifecycles = new WeakSet<TurnAdoptionLifecycle>();
const admittedTurnAdoptionLifecycles = new WeakSet<TurnAdoptionLifecycle>();
const admittingTurnAdoptionLifecycles = new WeakMap<TurnAdoptionLifecycle, Promise<void>>();
const retiredTurnAdoptionCancellationLifecycles = new WeakSet<TurnAdoptionLifecycle>();
const completedTurnAdoptionLifecycles = new WeakSet<TurnAdoptionLifecycle>();
const completedTurnAdoptionLifecycleCallbacks = new WeakSet<TurnAdoptionLifecycle>();
export function markFollowupRunEnqueued(run: Pick<FollowupRun, "queuedLifecycle">): boolean {
const lifecycle = run.queuedLifecycle;
if (!lifecycle || enqueuedFollowupLifecycles.has(lifecycle)) {
return true;
type FollowupLifecycleRun = Pick<FollowupRun, "turnAdoptionLifecycle">;
export function markFollowupRunEnqueued(run: FollowupLifecycleRun): boolean {
const lifecycle = run.turnAdoptionLifecycle;
if (lifecycle && !enqueuedTurnAdoptionLifecycles.has(lifecycle)) {
if (lifecycle.onDeferred?.() === false) {
return false;
}
enqueuedTurnAdoptionLifecycles.add(lifecycle);
}
if (lifecycle.onEnqueued?.() === false) {
return false;
}
enqueuedFollowupLifecycles.add(lifecycle);
return true;
}
export function retireFollowupRunCancellation(run: Pick<FollowupRun, "queuedLifecycle">): void {
const lifecycle = run.queuedLifecycle;
if (!lifecycle || retiredFollowupCancellationLifecycles.has(lifecycle)) {
export function retireFollowupRunCancellation(run: FollowupLifecycleRun): void {
const lifecycle = run.turnAdoptionLifecycle;
if (!lifecycle || retiredTurnAdoptionCancellationLifecycles.has(lifecycle)) {
return;
}
retiredFollowupCancellationLifecycles.add(lifecycle);
retiredTurnAdoptionCancellationLifecycles.add(lifecycle);
lifecycle.onCancellationRetired?.();
}
export async function admitFollowupRunLifecycle(
run: Pick<FollowupRun, "queuedLifecycle">,
): Promise<void> {
const lifecycle = run.queuedLifecycle;
if (!lifecycle || admittedFollowupLifecycles.has(lifecycle)) {
export async function admitFollowupRunLifecycle(run: FollowupLifecycleRun): Promise<void> {
const lifecycle = run.turnAdoptionLifecycle;
if (!lifecycle || admittedTurnAdoptionLifecycles.has(lifecycle)) {
return;
}
const existing = admittingFollowupLifecycles.get(lifecycle);
const existing = admittingTurnAdoptionLifecycles.get(lifecycle);
if (existing) {
await existing;
return;
}
if (completedFollowupLifecycles.has(lifecycle)) {
if (completedTurnAdoptionLifecycles.has(lifecycle)) {
throw new Error("followup run lifecycle completed before admission");
}
const admission = Promise.resolve()
.then(async () => await lifecycle.onAdmitted?.())
.then(() => {
admittedFollowupLifecycles.add(lifecycle);
});
admittingFollowupLifecycles.set(lifecycle, admission);
const admission = Promise.resolve().then(async () => {
if (!admittedTurnAdoptionLifecycles.has(lifecycle)) {
await lifecycle.onAdopted();
admittedTurnAdoptionLifecycles.add(lifecycle);
}
});
admittingTurnAdoptionLifecycles.set(lifecycle, admission);
try {
await admission;
} finally {
admittingFollowupLifecycles.delete(lifecycle);
admittingTurnAdoptionLifecycles.delete(lifecycle);
}
}
export function completeFollowupRunLifecycle(run: Pick<FollowupRun, "queuedLifecycle">): void {
const lifecycle = run.queuedLifecycle;
if (!lifecycle || completedFollowupLifecycles.has(lifecycle)) {
return;
}
completedFollowupLifecycles.add(lifecycle);
export function completeFollowupRunLifecycle(run: FollowupLifecycleRun): void {
const lifecycle = run.turnAdoptionLifecycle;
const finish = () => {
if (completedFollowupLifecycleCallbacks.has(lifecycle)) {
if (!lifecycle || completedTurnAdoptionLifecycleCallbacks.has(lifecycle)) {
return;
}
completedFollowupLifecycleCallbacks.add(lifecycle);
lifecycle.onComplete?.();
completedTurnAdoptionLifecycleCallbacks.add(lifecycle);
// onSettled must run even when onAbandoned throws (gateway/plugin cleanup).
try {
if (!admittedTurnAdoptionLifecycles.has(lifecycle)) {
lifecycle.onAbandoned?.();
}
} finally {
lifecycle.onSettled?.();
}
};
const admission = admittingFollowupLifecycles.get(lifecycle);
if (lifecycle && !completedTurnAdoptionLifecycles.has(lifecycle)) {
completedTurnAdoptionLifecycles.add(lifecycle);
}
const admission = lifecycle ? admittingTurnAdoptionLifecycles.get(lifecycle) : undefined;
if (!admission) {
finish();
return;
@@ -276,12 +297,3 @@ export function completeFollowupRunLifecycle(run: Pick<FollowupRun, "queuedLifec
// the in-flight admission attempt so adoption and abandonment cannot race.
void admission.then(finish, finish).catch(() => {});
}
export type ResolveQueueSettingsParams = {
cfg: OpenClawConfig;
channel?: string;
sessionEntry?: SessionEntry;
inlineMode?: QueueMode;
inlineOptions?: Partial<QueueSettings>;
pluginDebounceMs?: number;
};
src/auto-reply/reply/stranded-reply-recovery.test.ts
+7 -3
View File
@@ -6,13 +6,17 @@ import { createMockFollowupRun } from "./test-helpers.js";
const STRANDED_REPLY_RETRY_MARKER = "stranded-reply-retry";
describe("buildStrandedReplyRetryFollowupRun lifecycle ownership", () => {
it("does not share the client turn's queuedLifecycle with the system retry", () => {
it("does not share the client turn's turnAdoptionLifecycle with the system retry", () => {
const onComplete = vi.fn();
const onEnqueued = vi.fn(() => true);
const parent = createMockFollowupRun({
prompt: "user question",
transcriptPrompt: "user question",
queuedLifecycle: { onComplete, onEnqueued },
turnAdoptionLifecycle: {
onAdopted: async () => {},
onSettled: onComplete,
onDeferred: onEnqueued,
},
admissionSessionId: "sess-rotated",
onReplyAdmissionWaitChange: vi.fn(),
});
@@ -22,7 +26,7 @@ describe("buildStrandedReplyRetryFollowupRun lifecycle ownership", () => {
sourceReplyDeliveryMode: "message_tool_only",
});
expect(retry.queuedLifecycle).toBeUndefined();
expect(retry.turnAdoptionLifecycle).toBeUndefined();
expect(retry.strandedReplyRetry).toBe(true);
expect(retry.summaryLine).toBe(STRANDED_REPLY_RETRY_MARKER);
// Session routing stays; only the client-turn lifecycle identity is detached.
src/auto-reply/reply/stranded-reply-recovery.ts
+4 -3
View File
@@ -43,9 +43,10 @@ export function buildStrandedReplyRetryFollowupRun(
userTurnTranscriptRecorder: undefined,
currentInboundContext: undefined,
// Internally generated system turn: the client turn's lifecycle (gateway cancel
// identity) completes with the parent run. queuedLifecycle is one-shot WeakSet-tracked,
// so a shared object would be double-owned and free cancel while the retry still runs.
queuedLifecycle: undefined,
// identity) completes with the parent run. turnAdoptionLifecycle is one-shot
// WeakSet-tracked, so a shared object would be double-owned and free cancel
// while the retry still runs.
turnAdoptionLifecycle: undefined,
run: {
...base.run,
sourceReplyDeliveryMode: params.sourceReplyDeliveryMode,
src/channels/message/index.ts
+16
View File
@@ -4,6 +4,16 @@ export { deriveDurableFinalDeliveryRequirements } from "./capabilities.js";
export { defineChannelMessageAdapter } from "./adapter.js";
export { createChannelMessageAdapterFromOutbound } from "./outbound-bridge.js";
export { createDurableInboundReceiveJournalFromQueue } from "./durable-receive.js";
export { INGRESS_CLAIM_PROCESS_ID, processPidFromOwnerId } from "./ingress-claim-owner.js";
export {
bindIngressLifecycleToReplyOptions,
createChannelIngressDrain,
DEFAULT_INGRESS_ADOPTION_STALL_MS,
} from "./ingress-drain.js";
export {
DEFAULT_INGRESS_RETRY_DEAD_LETTER_MIN_AGE_MS,
DEFAULT_INGRESS_RETRY_MAX_ATTEMPTS,
} from "./ingress-retry-policy.js";
export {
verifyChannelMessageAdapterCapabilityProofs,
@@ -30,6 +40,11 @@ export {
createTypingCallbacks,
resolveChannelSourceReplyDeliveryMode,
} from "./reply-pipeline.js";
export type {
ChannelIngressDispatchLifecycle,
ChannelIngressDrain,
ChannelIngressDrainDispatchResult,
} from "./ingress-drain.js";
export type {
ChannelIngressQueue,
ChannelIngressQueueClaim,
@@ -38,6 +53,7 @@ export type {
ChannelIngressQueueRecord,
} from "./ingress-queue.js";
export type { MessageAckPolicy, MessageReceiveContext } from "./receive.js";
export type { IngressNonRetryableFailure } from "./ingress-retry-policy.js";
export type {
ChannelMessageAdapterShape,
ChannelMessageDurableFinalAdapter,
src/channels/turn/kernel.test.ts
+8 -6
View File
@@ -1083,9 +1083,9 @@ describe("channel turn kernel", () => {
expect(events).toEqual(["record", "afterRecord", "dispatch"]);
});
it("threads onTurnAdopted into assembled reply options and fires after recovery persist attempt", async () => {
it("threads turnAdoptionLifecycle into assembled reply options and fires after recovery persist attempt", async () => {
const events: string[] = [];
const onTurnAdopted = vi.fn(async () => {
const onAdopted = vi.fn(async () => {
events.push("adopted");
});
const dispatchReplyWithBufferedBlockDispatcher = vi.fn(
@@ -1093,7 +1093,7 @@ describe("channel turn kernel", () => {
events.push("dispatch-start");
// Persist attempt completes before adoption (agent-runner contract).
events.push("recovery-persist");
await params.replyOptions?.onTurnAdopted?.();
await params.replyOptions?.turnAdoptionLifecycle?.onAdopted();
events.push("settle");
return {
queuedFinal: true,
@@ -1114,14 +1114,16 @@ describe("channel turn kernel", () => {
delivery: {
deliver: vi.fn(async () => undefined),
},
onTurnAdopted,
turnAdoptionLifecycle: { onAdopted },
});
expect(onTurnAdopted).toHaveBeenCalledOnce();
expect(onAdopted).toHaveBeenCalledOnce();
expect(events).toEqual(["record", "dispatch-start", "recovery-persist", "adopted", "settle"]);
expect(dispatchReplyWithBufferedBlockDispatcher).toHaveBeenCalledWith(
expect.objectContaining({
replyOptions: expect.objectContaining({ onTurnAdopted }),
replyOptions: expect.objectContaining({
turnAdoptionLifecycle: expect.objectContaining({ onAdopted }),
}),
}),
);
});
src/channels/turn/kernel.ts
+10 -5
View File
@@ -190,11 +190,14 @@ export const recordDroppedChannelInboundHistory = recordDroppedChannelTurnHistor
function resolveAssembledReplyPipeline(
params: AssembledChannelTurn,
): Pick<AssembledChannelTurn, "dispatcherOptions" | "replyOptions"> {
const onTurnAdopted = params.onTurnAdopted ?? params.replyOptions?.onTurnAdopted;
const turnAdoptionLifecycle =
params.turnAdoptionLifecycle ?? params.replyOptions?.turnAdoptionLifecycle;
if (!params.replyPipeline) {
return {
dispatcherOptions: params.dispatcherOptions,
replyOptions: onTurnAdopted ? { ...params.replyOptions, onTurnAdopted } : params.replyOptions,
replyOptions: turnAdoptionLifecycle
? { ...params.replyOptions, turnAdoptionLifecycle }
: params.replyOptions,
};
}
const { onModelSelected, ...replyPipeline } = createChannelReplyPipeline({
@@ -212,7 +215,7 @@ function resolveAssembledReplyPipeline(
replyOptions: {
onModelSelected,
...params.replyOptions,
...(onTurnAdopted ? { onTurnAdopted } : {}),
...(turnAdoptionLifecycle ? { turnAdoptionLifecycle } : {}),
},
};
}
@@ -709,7 +712,7 @@ async function runChannelTurn<
const admission = resolved.admission ?? preflightAdmission ?? ({ kind: "dispatch" } as const);
let result: ChannelTurnResult<TDispatchResult>;
try {
// Prepared runDispatch was assembled earlier and ignores late options (including onTurnAdopted).
// Prepared runDispatch was assembled earlier and ignores late options.
const dispatchResult = await dispatchResolvedChannelTurn(
"runDispatch" in resolved
? {
@@ -729,7 +732,9 @@ async function runChannelTurn<
admission,
log: params.log,
messageId: input.id,
...(params.onTurnAdopted ? { onTurnAdopted: params.onTurnAdopted } : {}),
...(params.turnAdoptionLifecycle
? { turnAdoptionLifecycle: params.turnAdoptionLifecycle }
: {}),
},
);
result = dispatchResult.dispatched ? { ...dispatchResult, admission } : dispatchResult;
src/channels/turn/types.ts
+8 -12
View File
@@ -1,6 +1,9 @@
// Type contracts for channel turn normalization, admission, dispatch, and delivery.
import type { CommandTurnKind } from "../../auto-reply/command-turn-context.js";
import type { GetReplyOptions } from "../../auto-reply/get-reply-options.types.js";
import type {
GetReplyOptions,
TurnAdoptionLifecycle,
} from "../../auto-reply/get-reply-options.types.js";
import type { ReplyPayload } from "../../auto-reply/reply-payload.js";
import type { DispatchFromConfigResult } from "../../auto-reply/reply/dispatch-from-config.types.js";
import type { GetReplyFromConfig } from "../../auto-reply/reply/get-reply.types.js";
@@ -275,11 +278,8 @@ export type AssembledChannelTurn = {
botLoopProtection?: ChannelBotLoopProtectionFacts;
log?: (event: ChannelTurnLogEvent) => void;
messageId?: string;
/**
* Observes turn adoption without waiting for settle. Threaded into
* replyOptions for the agent runner (after recovery persist attempt).
*/
onTurnAdopted?: () => void | Promise<void>;
/** Canonical adoption lifecycle threaded into replyOptions. */
turnAdoptionLifecycle?: TurnAdoptionLifecycle;
};
/** Channel turn with dispatch runner already prepared. */
@@ -383,10 +383,6 @@ export type RunChannelTurnParams<TRaw, TDispatchResult = DispatchFromConfigResul
raw: TRaw;
adapter: ChannelTurnAdapter<TRaw, TDispatchResult>;
log?: (event: ChannelTurnLogEvent) => void;
/**
* Observes turn adoption without waiting for settle. Fired after the
* recovery-context persist attempt (context may be absent when source
* delivery is suppressed). Default callers still await full settle.
*/
onTurnAdopted?: () => void | Promise<void>;
/** Canonical adoption lifecycle for this turn. */
turnAdoptionLifecycle?: TurnAdoptionLifecycle;
};
src/gateway/server-methods/chat.ts
+6 -3
View File
@@ -1303,9 +1303,12 @@ export const chatHandlers: GatewayRequestHandlers = {
abortSignal: activeRunAbort.controller.signal,
// Keep a Gateway-owned cancel identity after this chat.send
// terminalizes while the prompt waits in followup/collect queue.
queuedFollowupLifecycle: {
turnAdoptionLifecycle: {
// Gateway cancel identity only — share collect key via ownerKey.
admission: "cancel-only",
ownerKey: queuedFollowupOwnerKey,
onEnqueued: () => {
onAdopted: async () => {},
onDeferred: () => {
queuedFollowupEnqueued = registerQueuedChatTurn({
chatQueuedTurns: ensureChatQueuedTurns(context),
runId: clientRunId,
@@ -1325,7 +1328,7 @@ export const chatHandlers: GatewayRequestHandlers = {
activeRunAbort.controller,
);
},
onComplete: () => {
onSettled: () => {
completeQueuedChatTurn(
ensureChatQueuedTurns(context),
clientRunId,
src/gateway/server-restart-sentinel.test.ts
+2 -2
View File
@@ -684,7 +684,7 @@ describe("scheduleRestartSentinelWake", () => {
},
} as Awaited<ReturnType<typeof mocks.readRestartSentinel>>);
mocks.recordInboundSessionAndDispatchReply.mockImplementationOnce(async (params) => {
await params.onTurnAdopted?.();
await params.turnAdoptionLifecycle?.onAdopted();
await params.deliver({
text: "done",
replyToId: "restart-sentinel:agent:main:main:agentTurn:123",
@@ -809,7 +809,7 @@ describe("scheduleRestartSentinelWake", () => {
it("fences an adopted generic turn in its explicit queue state directory", async () => {
mocks.recordInboundSessionAndDispatchReply.mockImplementationOnce(async (params) => {
await params.onTurnAdopted?.();
await params.turnAdoptionLifecycle?.onAdopted();
});
await deliverQueuedSessionDelivery({
src/gateway/server-restart-sentinel.ts
+8 -5
View File
@@ -361,11 +361,14 @@ export async function deliverQueuedSessionDelivery(params: {
},
// Preflight remains retryable. Ownership starts only after the agent runner
// has durably adopted the turn and before it can execute tools or reply.
onTurnAdopted: () =>
markSessionDeliveryAttemptStarted(
params.entry,
...sessionDeliveryStateDirArgs(params.stateDir),
),
turnAdoptionLifecycle: {
admission: "cancel-only",
onAdopted: () =>
markSessionDeliveryAttemptStarted(
params.entry,
...sessionDeliveryStateDirArgs(params.stateDir),
),
},
delivery: {
preparePayload: (payload) => {
if (isRestartContinuationBusyPayload(payload)) {
src/gateway/server.chat.gateway-server-chat-b.test.ts
+11 -11
View File
@@ -3918,12 +3918,12 @@ describe("gateway server chat", () => {
getRuntimeConfig: () => ({}),
dedupe: new Map(),
} as unknown as GatewayRequestContext;
let queuedLifecycle: GetReplyOptions["queuedFollowupLifecycle"];
let turnAdoptionLifecycle: GetReplyOptions["turnAdoptionLifecycle"];
const dispatchRelease = createDeferred();
dispatchInboundMessageMock.mockImplementationOnce(async (args: unknown) => {
queuedLifecycle = (args as { replyOptions?: GetReplyOptions }).replyOptions
?.queuedFollowupLifecycle;
queuedLifecycle?.onEnqueued?.();
turnAdoptionLifecycle = (args as { replyOptions?: GetReplyOptions }).replyOptions
?.turnAdoptionLifecycle;
turnAdoptionLifecycle?.onDeferred?.();
await dispatchRelease.promise;
return {};
});
@@ -3963,8 +3963,8 @@ describe("gateway server chat", () => {
context,
});
await vi.waitFor(() => expect(queuedLifecycle).toBeDefined(), FAST_WAIT_OPTS);
expect(queuedLifecycle?.ownerKey).toBe("connection:conn-tui");
await vi.waitFor(() => expect(turnAdoptionLifecycle).toBeDefined(), FAST_WAIT_OPTS);
expect(turnAdoptionLifecycle?.ownerKey).toBe("connection:conn-tui");
expect(broadcast).not.toHaveBeenCalledWith(
"chat",
expect.objectContaining({ runId: "idem-queued-followup", state: "final" }),
@@ -4034,18 +4034,18 @@ describe("gateway server chat", () => {
queuedEntry?.controller.abort();
expect(context.chatQueuedTurns.has("idem-queued-followup")).toBe(false);
queuedLifecycle?.onComplete?.();
turnAdoptionLifecycle?.onSettled?.();
expect(context.chatQueuedTurns.has("idem-queued-followup")).toBe(false);
await vi.waitFor(
() => expect(context.removeChatRun).toHaveBeenCalledTimes(1),
FAST_WAIT_OPTS,
);
let failedDispatchLifecycle: GetReplyOptions["queuedFollowupLifecycle"];
let failedDispatchLifecycle: GetReplyOptions["turnAdoptionLifecycle"];
dispatchInboundMessageMock.mockImplementationOnce(async (args: unknown) => {
failedDispatchLifecycle = (args as { replyOptions?: GetReplyOptions }).replyOptions
?.queuedFollowupLifecycle;
failedDispatchLifecycle?.onEnqueued?.();
?.turnAdoptionLifecycle;
failedDispatchLifecycle?.onDeferred?.();
throw new Error("post-enqueue bookkeeping failed");
});
await expectDefined(
@@ -4098,7 +4098,7 @@ describe("gateway server chat", () => {
payload: { status: "ok" },
});
expect(context.chatQueuedTurns.has("idem-queued-followup-post-error")).toBe(true);
failedDispatchLifecycle?.onComplete?.();
failedDispatchLifecycle?.onSettled?.();
expect(context.chatQueuedTurns.has("idem-queued-followup-post-error")).toBe(false);
} finally {
dispatchInboundMessageMock.mockReset();
src/plugin-sdk/channel-outbound.ts
+12
View File
@@ -15,10 +15,21 @@ const loadChannelMessageRuntimeModule = createLazyRuntimeModule(
export type { DurableMessageBatchSendResult } from "../channels/message/runtime.js";
export {
bindIngressLifecycleToReplyOptions,
createChannelIngressDrain,
createReplyPrefixContext,
createReplyPrefixOptions,
createTypingCallbacks,
createChannelReplyPipeline as createChannelMessageReplyPipeline,
// Narrow drain seam by maintainer decision (#108924): factory, lifecycle binding,
// tuning constants, and processPidFromOwnerId (telegram transport display). All other
// claim/retry/adoption internals stay core-owned; test helpers live on the
// private-local plugin-state-test-runtime subpath.
DEFAULT_INGRESS_ADOPTION_STALL_MS,
DEFAULT_INGRESS_RETRY_MAX_ATTEMPTS,
DEFAULT_INGRESS_RETRY_DEAD_LETTER_MIN_AGE_MS,
INGRESS_CLAIM_PROCESS_ID,
processPidFromOwnerId,
resolveChannelSourceReplyDeliveryMode as resolveChannelMessageSourceReplyDeliveryMode,
} from "../channels/message/index.js";
// Bare interval/stop orchestration for channels that own their typing renewal
@@ -136,6 +147,7 @@ export type {
ChannelMessageSendTextContext,
ChannelMessageUnknownSendContext,
ChannelMessageUnknownSendReconciliationResult,
ChannelIngressDrain,
ChannelIngressQueue,
ChannelIngressQueueClaim,
ChannelIngressQueueClaimRef,
src/plugin-sdk/plugin-state-test-runtime.ts
+13
View File
@@ -17,3 +17,16 @@ export {
closeOpenClawStateDatabaseForTest,
openOpenClawStateDatabase,
} from "../state/openclaw-state-db.js";
// Test-only ingress reliability helpers: core predicates polling/webhook tests
// assert directly; excluded from the public SDK surface (private-local subpath).
export {
INGRESS_CLAIM_LEASE_MS,
isIngressClaimOwnedByOtherLiveProcess,
} from "../channels/message/ingress-claim-owner.js";
export {
resolveIngressRetryDelayMs,
shouldDeadLetterRetryableIngressEvent,
} from "../channels/message/ingress-retry-policy.js";
// Test-only pairing-store seeding so channel tests exercise the real
// store-backed authorization path instead of injecting fake readers.
export { addChannelAllowFromStoreEntry } from "../pairing/pairing-store.js";
src/plugin-sdk/test-helpers/plugin-runtime-mock.ts
+3
View File
@@ -822,6 +822,9 @@ export function createPluginRuntimeMock(overrides: DeepPartial<PluginRuntime> =
openChannelIngressQueue: vi.fn(() => {
throw new Error("openChannelIngressQueue mock is not configured");
}) as unknown as PluginRuntime["state"]["openChannelIngressQueue"],
openChannelIngressDrain: vi.fn(() => {
throw new Error("openChannelIngressDrain mock is not configured");
}) as unknown as PluginRuntime["state"]["openChannelIngressDrain"],
},
tasks: {
runs: {
src/plugins/registry-runtime.ts
+38 -1
View File
@@ -1,5 +1,6 @@
import { normalizeOptionalString } from "@openclaw/normalization-core/string-coerce";
import { normalizeOptionalAgentRuntimeId } from "../agents/agent-runtime-id.js";
import { createChannelIngressDrain } from "../channels/message/ingress-drain.js";
import { createChannelIngressQueue } from "../channels/message/ingress-queue.js";
import type { SessionEntry } from "../config/sessions/types.js";
import {
@@ -467,7 +468,9 @@ export function createPluginRuntimeResolver(state: PluginRegistryState) {
};
if (prop === "state") {
const baseState = getRuntimeProperty();
const assertPluginStateAllowed = (methodName: "openBlobStore" | "openKeyedStore") => {
const assertPluginStateAllowed = (
methodName: "openBlobStore" | "openKeyedStore" | "openChannelIngressDrain",
) => {
const record =
pluginRuntimeRecordById.get(pluginId) ??
registry.plugins.find((entry) => entry.id === pluginId);
@@ -506,6 +509,40 @@ export function createPluginRuntimeResolver(state: PluginRegistryState) {
stateDir,
});
},
openChannelIngressDrain: <TPayload, TMetadata = unknown, TCompletedMetadata = unknown>(
options: Omit<
Parameters<
typeof createChannelIngressDrain<TPayload, TMetadata, TCompletedMetadata>
>[0],
"queue"
> & {
queue?: ReturnType<
typeof createChannelIngressQueue<TPayload, TMetadata, TCompletedMetadata>
>;
accountId?: string;
stateDir?: string;
},
) => {
assertPluginStateAllowed("openChannelIngressDrain");
const stateDir = options.stateDir ?? baseState.resolveStateDir();
const queue =
options.queue ??
createChannelIngressQueue<TPayload, TMetadata, TCompletedMetadata>({
channelId: pluginId,
accountId: options.accountId,
stateDir,
});
const {
queue: _queue,
accountId: _accountId,
stateDir: _stateDir,
...drainOptions
} = options;
return createChannelIngressDrain<TPayload, TMetadata, TCompletedMetadata>({
...drainOptions,
queue,
});
},
} satisfies PluginRuntime["state"];
}
if (prop === "config") {
src/plugins/runtime/index.ts
+5
View File
@@ -347,6 +347,11 @@ export function createPluginRuntime(_options: CreatePluginRuntimeOptions = {}):
"openChannelIngressQueue is only available through the plugin runtime proxy.",
);
},
openChannelIngressDrain: () => {
throw new Error(
"openChannelIngressDrain is only available through the plugin runtime proxy.",
);
},
},
tasks,
taskFlow,
src/plugins/runtime/types-core.ts
+15
View File
@@ -1,4 +1,5 @@
// Core runtime types define system, config, and task helper contracts for plugins.
import type { CreateChannelIngressDrainOptions } from "../../channels/message/ingress-drain.js";
import type { CreateChannelIngressQueueOptions } from "../../channels/message/ingress-queue.js";
import type { ConfigMutationBase } from "../../config/mutation-types.js";
import type { SessionPluginJsonValue } from "../../config/sessions/types.js";
@@ -437,6 +438,20 @@ export type PluginRuntimeCore = {
TMetadata,
TCompletedMetadata
>;
openChannelIngressDrain: <TPayload, TMetadata = unknown, TCompletedMetadata = unknown>(
options: Omit<
CreateChannelIngressDrainOptions<TPayload, TMetadata, TCompletedMetadata>,
"queue"
> & {
queue?: import("../../channels/message/ingress-queue.js").ChannelIngressQueue<
TPayload,
TMetadata,
TCompletedMetadata
>;
accountId?: string;
stateDir?: string;
},
) => import("../../channels/message/ingress-drain.js").ChannelIngressDrain;
};
tasks: {
runs: PluginRuntimeTaskRuns;