You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@maka.apache.org by wy...@apache.org on 2026/10/20 02:28:28 UTC
(maka) 01/04: fix(assistant): sync Bot model and report execution failures
This is an automated email from the ASF dual-hosted git repository.
M4n5ter pushed a commit to branch fix/assistant-bot-model-sync
in repository https://gitbox.apache.org/repos/asf/maka.git
commit 99cbff05d90d1b69b37037ac9ed623a1629238d3
Author: M4n5ter <m4...@gmail.com>
AuthorDate: Tue Oct 20 09:48:52 2026 +0800
fix(assistant): sync Bot model and report execution failures
Reconcile idle Bot workers with the conversation connection, model and thinking level through the existing configuration authority. Preserve worker history and pending interactions. Publish provider failures through the durable conversation outbox without consuming retries on publication failures.
Generated-by: OpenAI Codex
---
.../src/__tests__/assistant-bot.test.ts | 168 +++++++++++++++-
.../src/__tests__/execution-composition.test.ts | 215 +++++++++++++++++++++
.../__tests__/session-catalog-coordinator.test.ts | 113 +++++++++++
.../runtime-host/src/assistant/bot/controller.ts | 56 +++++-
packages/runtime-host/src/assistant/service.ts | 3 +
.../src/server/execution-composition.ts | 15 +-
.../src/server/session-catalog-coordinator.ts | 95 ++++++++-
7 files changed, 654 insertions(+), 11 deletions(-)
diff --git a/packages/runtime-host/src/__tests__/assistant-bot.test.ts b/packages/runtime-host/src/__tests__/assistant-bot.test.ts
index 5a82220d3f..539daffd87 100644
--- a/packages/runtime-host/src/__tests__/assistant-bot.test.ts
+++ b/packages/runtime-host/src/__tests__/assistant-bot.test.ts
@@ -636,7 +636,7 @@ test('mail arriving during a failed call resumes the blocked worker without anot
assert.equal(f.visible.filter((m) => m.text === '已恢复处理').length, 1);
});
-test('persistent model failures report error without a polling retry storm; fresh mail can retry', async (t) => {
+test('persistent model failures reach the conversation without a polling retry storm; fresh mail can retry', async (t) => {
const f = await fixture();
t.after(f.close);
f.runner(async () => {
@@ -644,13 +644,179 @@ test('persistent model failures report error without a polling retry storm; fres
});
await f.send('first-failure', '你好');
await until(async () => ((await f.service.call('assistant.binding', {})) as any).bot.error);
+ await until(() => f.visible.some((message) => message.type === 'assistant'));
+ const first = f.visible.filter((message) => message.type === 'assistant');
+ assert.equal(first.length, 1);
+ assert.match(first[0].text, /model unavailable/);
+ assert.match(first[0].text, /消息已保留.*发送新消息可重试/);
await new Promise((resolve) => setTimeout(resolve, 2200));
assert.equal(f.starts(), 1);
+ assert.equal(f.visible.filter((message) => message.type === 'assistant').length, 1);
await f.send('second-attempt', '再试一次');
await until(() => f.starts() === 2);
await until(async () => ((await f.service.call('assistant.binding', {})) as any).bot.error);
await new Promise((resolve) => setTimeout(resolve, 1200));
assert.equal(f.starts(), 2);
+ const failures = f.visible.filter((message) => message.type === 'assistant');
+ assert.equal(failures.length, 2);
+ assert.notEqual(failures[0].id, failures[1].id);
+});
+
+test('failure bubbles survive an ambiguous publication and restart without losing the execution error', async (t) => {
+ const f = await fixture();
+ t.after(f.close);
+ f.runner(async () => {
+ f.failAppend();
+ throw Error('provider rejected the selected model');
+ });
+ await f.send('publication-failure', '你好');
+ await until(() => f.visible.some((message) => message.type === 'assistant'));
+ const failure = f.visible.find((message) => message.type === 'assistant');
+ assert.match(failure.text, /provider rejected the selected model/);
+ const delivered = () => {
+ const mailbox = new DatabaseSync(
+ join(f.directory, 'native', 'index-initiative', 'bot.sqlite'),
+ {
+ readOnly: true,
+ },
+ );
+ try {
+ return mailbox.prepare('SELECT delivered FROM outbox WHERE id=?').get(failure.id)?.delivered;
+ } finally {
+ mailbox.close();
+ }
+ };
+ assert.equal(delivered(), 0);
+ // The first publication after restart can lose its response too.
+ f.failAppend();
+ await f.restart();
+ await until(() => delivered() === 1);
+ const status: any = await f.service.call('assistant.binding', {});
+ assert.match(status.bot.error, /provider rejected the selected model/);
+ assert.doesNotMatch(status.bot.error, /connection lost after commit/);
+ assert.deepEqual(
+ f.visible.filter((message) => message.type === 'assistant'),
+ [failure],
+ );
+ assert.equal(f.starts(), 1);
+});
+
+test('fresh mail still executes after a pending failure bubble repeatedly loses its publication response', async (t) => {
+ const f = await fixture();
+ t.after(f.close);
+ f.runner(async () => {
+ f.failAppend();
+ throw Error('provider unavailable before fresh mail');
+ });
+ await f.send('original', '第一条');
+ await until(() => f.visible.some((message) => message.type === 'assistant'));
+ const failure = f.visible.find((message) => message.type === 'assistant');
+ let recovered = false;
+ f.runner(async ({ invoke }) => {
+ assert.deepEqual(
+ (await invoke('InboxCheck')).items.map((message: any) => message.id),
+ ['original', 'fresh'],
+ );
+ await invoke('SendUser', {
+ messageId: 'publication-recovered',
+ messages: [{ text: '已恢复处理' }],
+ });
+ await invoke('Sleep');
+ recovered = true;
+ });
+ // Fresh input is durably accepted even when receive cannot flush the old bubble.
+ f.failAppend();
+ await assert.rejects(f.send('fresh', '继续'), /connection lost after commit/);
+ // The scheduler must retry publication without latching this unattempted input.
+ f.failAppend();
+ await until(async () => {
+ const status: any = await f.service.call('assistant.binding', {});
+ return status.bot.wakeError?.includes('connection lost after commit');
+ });
+ const pending: any = await f.service.call('assistant.binding', {});
+ assert.equal(pending.bot.error, null);
+ assert.equal(f.starts(), 1);
+ assert.equal(f.visible.filter((message) => message.type === 'assistant').length, 1);
+ await until(() => recovered);
+ await until(async () => ((await f.service.call('assistant.binding', {})) as any).bot.sleeping);
+ assert.equal(f.starts(), 2);
+ assert.deepEqual(
+ f.visible.filter((message) => message.type === 'assistant').map((message) => message.text),
+ [failure.text, '已恢复处理'],
+ );
+});
+
+test('a contained Bot failure reports the current wake provider error instead of an older turn failure', async (t) => {
+ const directory = await mkdtemp(join(tmpdir(), 'maka-bot-provider-error-'));
+ const store = new BotStore(directory);
+ store.set('worker', 'worker');
+ store.post({ id: 'mail', type: 'user_message', source: 'user', content: 'hello' });
+ const visible: any[] = [];
+ let status = 'idle';
+ let submitted = false;
+ let transcriptReads = 0;
+ const agent = {
+ snapshot: async () => ({ agent: { status }, pendingInteractions: 0 }),
+ messageExecution: async () =>
+ submitted ? { state: 'owned', turnId: 'failed-turn' } : { state: 'not_admitted' },
+ followup: async () => {
+ submitted = true;
+ status = 'running';
+ return { disposition: 'turn_started', turnId: 'failed-turn' };
+ },
+ whenIdle: async () => {
+ status = 'blocked';
+ },
+ transcript: async () => {
+ transcriptReads++;
+ return [
+ {
+ type: 'turn_state',
+ turnId: 'old-turn',
+ errorClass: 'unknown',
+ failureMessage: 'old failure',
+ },
+ {
+ type: 'turn_state',
+ turnId: 'failed-turn',
+ errorClass: 'provider_unavailable',
+ failureMessage: 'provider 404: model has no eligible provider',
+ },
+ {
+ type: 'turn_state',
+ turnId: 'unrelated-turn',
+ errorClass: 'unknown',
+ failureMessage: 'unrelated failure',
+ },
+ ];
+ },
+ cancel: async () => {},
+ };
+ const controller = new BotController(
+ {
+ agents: {
+ withInvocation: async (_: unknown, fn: () => unknown) => fn(),
+ resume: async () => agent,
+ },
+ appendConversation: async (_: string, message: any) => {
+ visible.push(message);
+ },
+ } as any,
+ { fence() {}, get: () => ({ sessionId: 'owner', cwd: directory }) } as any,
+ store,
+ );
+ t.after(async () => {
+ await controller.close();
+ store.close();
+ await rm(directory, { recursive: true, force: true });
+ });
+ controller.start();
+ await until(() => visible.length === 1);
+ assert.match(store.get<string>('error')!, /provider 404: model has no eligible provider/);
+ assert.match(visible[0].text, /provider 404: model has no eligible provider/);
+ assert.doesNotMatch(visible[0].text, /old failure|unrelated failure/);
+ assert.equal(transcriptReads, 1);
+ assert.equal(store.batch(0).items.length, 1, 'the failed wake retains its Inbox input');
});
test('blocked worker with a genuine pending interaction is not restarted by new mail', async (t) => {
diff --git a/packages/runtime-host/src/__tests__/execution-composition.test.ts b/packages/runtime-host/src/__tests__/execution-composition.test.ts
index 8f3896ebbe..d677172240 100644
--- a/packages/runtime-host/src/__tests__/execution-composition.test.ts
+++ b/packages/runtime-host/src/__tests__/execution-composition.test.ts
@@ -370,6 +370,221 @@ test('assistant messages enter a durable Inbox through real Host admission witho
});
});
+test('assistant Bot reuses its history with the visible model and connection after switching and restarting', {
+ timeout: 30000,
+}, async (t) => {
+ let assistant!: PersonalAssistantService;
+ const recover = PersonalAssistantService.prototype.recover;
+ t.mock.method(
+ PersonalAssistantService.prototype,
+ 'recover',
+ function (this: PersonalAssistantService, ...args: Parameters<typeof recover>) {
+ assistant = this;
+ return recover.apply(this, args);
+ },
+ );
+ await withCompositionRoot(async ({ root, owner }) => {
+ const originalConnectionId = await configureFakeDefaultTarget(owner, [
+ 'fake-model',
+ 'replacement-model',
+ ]);
+ let policy = await openInteractiveRuntimePolicyStoresForWrite(owner.lease);
+ const catalog = await policy.connectionCatalog.getSnapshot();
+ const created = await policy.connectionCatalog.create({
+ expectedCatalogRevision: catalog.revision,
+ connection: {
+ slug: 'replacement',
+ name: 'Replacement',
+ providerType: 'ollama',
+ enabled: true,
+ enabledModelIds: ['replacement-model'],
+ },
+ });
+ assert.equal(created.kind, 'committed');
+ if (created.kind !== 'committed') throw Error('Replacement connection was not committed');
+ const replacement = created.snapshot.connections.find((entry) => entry.slug === 'replacement')!;
+ const fetch = await policy.operations.beginModelFetch(replacement.connectionId);
+ assert.equal(fetch.kind, 'ready');
+ if (fetch.kind !== 'ready') throw Error('Replacement model fetch did not start');
+ assert.equal(
+ (
+ await policy.operations.completeModelFetch(fetch.ticket, {
+ models: [{ id: 'replacement-model' }],
+ source: 'fetched',
+ fetchedAt: Date.now(),
+ })
+ ).kind,
+ 'committed',
+ );
+ const executions: { sessionId: string; connectionId?: string; model: string }[] = [];
+ const primaryBackendFactory: BackendFactory = (context) =>
+ new (class extends FakeBackend {
+ override async *send(input: BackendSendInput): AsyncIterable<SessionEvent> {
+ executions.push({
+ sessionId: context.sessionId,
+ connectionId: context.header.llmConnectionId,
+ model: context.header.model,
+ });
+ if (context.header.model === 'fake-model') {
+ yield* super.send({ ...input, text: '__e2e_error__:network' });
+ return;
+ }
+ const invoke = (name: string, raw: unknown = {}) => {
+ const tool = assistant.tools.find((entry) => entry.name === name)!;
+ return tool.impl(raw, {
+ sessionId: context.sessionId,
+ turnId: input.turnId,
+ cwd: root,
+ toolCallId: randomUUID(),
+ abortSignal: new AbortController().signal,
+ emitOutput() {},
+ });
+ };
+ await invoke('InboxCheck');
+ await invoke('SendUser', {
+ messageId: `reply-${input.turnId}`,
+ messages: [{ text: `已通过 ${context.header.llmConnectionSlug} 恢复处理` }],
+ });
+ await invoke('Sleep');
+ assert.deepEqual(
+ await assistant.beforeFinish({
+ sessionId: context.sessionId,
+ turnId: input.turnId,
+ signal: new AbortController().signal,
+ }),
+ { allow: true },
+ );
+ yield* super.send({ ...input, text: 'done' });
+ }
+ })(context);
+ let { composition, manager } = await createCapturedExecutionComposition(owner, {
+ primaryBackendFactory,
+ });
+ const caller: ConnectionContext = {
+ hostEpoch: 'execution-composition-test',
+ connectionId: 'assistant-model-test',
+ principal: 'local_os_user',
+ acquireResidency: () => ({ release() {} }),
+ };
+ let stores = await openInteractiveExecutionStoresForWrite(owner.lease);
+ const call = async (method: 'assistant.bind' | 'assistant.status', input: unknown = {}) => {
+ const result = await composition.handlers['personal-assistant.call'](
+ { method, input },
+ caller,
+ );
+ assert.equal(result.ok, true, JSON.stringify(result));
+ return result.result.value as any;
+ };
+ try {
+ const chat = await manager.createSession({
+ cwd: root,
+ llmConnectionId: originalConnectionId,
+ llmConnectionSlug: 'fake',
+ model: 'fake-model',
+ permissionMode: 'ask',
+ labels: ['personal-assistant'],
+ });
+ await call('assistant.bind', { sessionId: chat.id });
+ const send = async (number: number) => {
+ const result = await composition.handlers['turn.message.submit'](
+ {
+ originHostEpoch: caller.hostEpoch,
+ sessionId: chat.id,
+ messageId: `model-attempt-${number}`,
+ content: { text: `尝试 ${number}` },
+ placement: 'current_turn',
+ },
+ caller,
+ );
+ assert.equal(result.ok, true, JSON.stringify(result));
+ await waitFor(async () => {
+ const status = await call('assistant.status', { includeMemory: false });
+ const visible = await readLedgerMessages(stores.runtimeEventStore, chat.id);
+ return (
+ executions.length === number &&
+ (number === 1 ? !!status.bot.error : status.bot.sleeping && !status.bot.error) &&
+ visible.filter((row) => row.type === 'assistant').length === number
+ );
+ }, 7000).catch(async (error) => {
+ throw new Error(
+ JSON.stringify({
+ attempt: number,
+ executions,
+ bot: (await call('assistant.status', { includeMemory: false })).bot,
+ visible: await readLedgerMessages(stores.runtimeEventStore, chat.id),
+ }),
+ { cause: error },
+ );
+ });
+ };
+ await send(1);
+ assert.match(
+ (await readLedgerMessages(stores.runtimeEventStore, chat.id))
+ .filter((row) => row.type === 'assistant')
+ .map((row) => row.text)
+ .join('\n'),
+ /Deterministic network failure/,
+ );
+ const workerId = executions[0].sessionId;
+ const workerBefore = await stores.sessionStore.readHeaderSnapshot(workerId);
+ const select = async (connectionId: string, connectionSlug: string) => {
+ const current = await stores.sessionStore.readHeaderRecordSnapshot(chat.id);
+ const changed = await composition.handlers['session.configuration.update'](
+ {
+ sessionId: chat.id,
+ expectedRevision: current.revision,
+ patch: {
+ modelTarget: {
+ kind: 'explicit',
+ connectionId,
+ connectionSlug,
+ model: 'replacement-model',
+ },
+ },
+ },
+ caller,
+ );
+ assert.equal(changed.ok, true, JSON.stringify(changed));
+ if (changed.ok) assert.equal(changed.result.kind, 'committed');
+ };
+ await select(originalConnectionId, 'fake');
+ await send(2);
+ await select(replacement.connectionId, 'replacement');
+ await composition.close();
+ ({ composition, manager } = await createCapturedExecutionComposition(owner, {
+ primaryBackendFactory,
+ }));
+ stores = await openInteractiveExecutionStoresForWrite(owner.lease);
+ policy = await openInteractiveRuntimePolicyStoresForWrite(owner.lease);
+ await send(3);
+ assert.deepEqual(
+ (await readLedgerMessages(stores.runtimeEventStore, chat.id))
+ .filter((row) => row.type === 'assistant')
+ .slice(1)
+ .map((row) => row.text),
+ ['已通过 fake 恢复处理', '已通过 replacement 恢复处理'],
+ );
+ assert.deepEqual(executions, [
+ { sessionId: workerId, connectionId: originalConnectionId, model: 'fake-model' },
+ { sessionId: workerId, connectionId: originalConnectionId, model: 'replacement-model' },
+ { sessionId: workerId, connectionId: replacement.connectionId, model: 'replacement-model' },
+ ]);
+ const workerAfter = await stores.sessionStore.readHeaderSnapshot(workerId);
+ assert.deepEqual(workerAfter.labels, workerBefore.labels);
+ assert.equal(workerAfter.permissionMode, workerBefore.permissionMode);
+ assert.equal(workerAfter.parentSessionId, chat.id);
+ const workerHistory = await readLedgerMessages(stores.runtimeEventStore, workerId);
+ assert.equal(workerHistory.filter((row) => row.type === 'user').length, 3);
+ assert.equal(
+ (await policy.connectionCatalog.getSnapshot()).defaultTarget?.connectionId,
+ originalConnectionId,
+ );
+ } finally {
+ await composition.close();
+ }
+ });
+});
+
test('assistant master switch gates real Host execution and stops only assistant sessions', {
timeout: 20000,
}, async () => {
diff --git a/packages/runtime-host/src/__tests__/session-catalog-coordinator.test.ts b/packages/runtime-host/src/__tests__/session-catalog-coordinator.test.ts
index 4964932dbe..d1afcfb741 100644
--- a/packages/runtime-host/src/__tests__/session-catalog-coordinator.test.ts
+++ b/packages/runtime-host/src/__tests__/session-catalog-coordinator.test.ts
@@ -724,6 +724,119 @@ test('ordinary configuration rejects the WorkHub Coordination Session identity',
assert.equal(fixture.drainRequests(), 0);
});
+test('Assistant Bot resolves a legacy owner selection by slug and preserves worker policy', async () => {
+ const { llmConnectionId: _connectionId, ...owner } = sessionHeader('assistant', []);
+ const resolutions: Parameters<RuntimePolicy['operations']['resolveExecutionConnection']>[0][] =
+ [];
+ const fixture = createFixture({
+ header: {
+ llmConnectionId: 'previous-connection',
+ llmConnectionSlug: 'previous',
+ model: 'previous-model',
+ permissionMode: 'bypass',
+ },
+ stores: {
+ readHeaderRecordSnapshot: async (id) =>
+ id === owner.id
+ ? headerSnapshot(owner, 1)
+ : headerSnapshot(fixture.header(), fixture.revision()),
+ },
+ connection: { onResolve: (ref) => resolutions.push(ref) },
+ });
+
+ await fixture.coordinator.synchronizeAssistantBotModel(owner.id, fixture.sessionId);
+
+ assert.deepEqual(resolutions, [
+ { kind: 'catalog_slug', connectionSlug: 'test' },
+ { kind: 'bound', connectionId: 'connection-1', connectionSlug: 'test' },
+ ]);
+ assert.equal(fixture.header().llmConnectionId, 'connection-1');
+ assert.equal(fixture.header().model, 'model-1');
+ assert.equal(fixture.header().permissionMode, 'bypass');
+ assert.deepEqual(fixture.header().labels, ['user-label']);
+ assert.equal(fixture.revision(), 4);
+ assert.equal(fixture.drainRequests(), 0);
+});
+
+test('Assistant Bot leaves an unchanged legacy route intact without resolving a default', async () => {
+ const { llmConnectionId: _connectionId, ...owner } = sessionHeader('assistant', []);
+ const fixture = createFixture({
+ legacyConnectionIdentity: true,
+ stores: {
+ readHeaderRecordSnapshot: async (id) =>
+ id === owner.id
+ ? headerSnapshot(owner, 1)
+ : headerSnapshot(fixture.header(), fixture.revision()),
+ },
+ connection: { onResolve: () => assert.fail('An unchanged route does not need resolution') },
+ });
+
+ await fixture.coordinator.synchronizeAssistantBotModel(owner.id, fixture.sessionId);
+
+ assert.equal(fixture.header().llmConnectionId, undefined);
+ assert.equal(fixture.revision(), 3);
+});
+
+for (const mode of ['running', 'waiting_for_user', 'queued', 'claimed'] as const)
+ test(`Assistant Bot defers a ${mode} worker model change until its next idle wake`, async () => {
+ const owner = sessionHeader('assistant', []);
+ let busy = true;
+ const fixture = createFixture({
+ header: { model: 'previous-model' },
+ stores: {
+ readHeaderRecordSnapshot: async (id) =>
+ id === owner.id
+ ? headerSnapshot(owner, 1)
+ : headerSnapshot(
+ {
+ ...fixture.header(),
+ status:
+ busy && (mode === 'running' || mode === 'waiting_for_user') ? mode : 'active',
+ },
+ fixture.revision(),
+ ),
+ },
+ isTurnBusy: () => busy && mode === 'queued',
+ manager: { runningTurnIds: () => (busy && mode === 'claimed' ? ['turn'] : []) },
+ });
+
+ await fixture.coordinator.synchronizeAssistantBotModel(owner.id, fixture.sessionId);
+ assert.equal(fixture.header().model, 'previous-model');
+ assert.equal(fixture.revision(), 3);
+
+ busy = false;
+ await fixture.coordinator.synchronizeAssistantBotModel(owner.id, fixture.sessionId);
+ assert.equal(fixture.header().model, 'model-1');
+ assert.equal(fixture.revision(), 4);
+ assert.equal(fixture.drainRequests(), 0);
+ });
+
+test('Assistant Bot clears stale thinking and follows the owner thinking level', async () => {
+ let owner = sessionHeader('assistant', []);
+ const fixture = createFixture({
+ header: { model: 'previous-model', thinkingLevel: 'high' },
+ stores: {
+ readHeaderRecordSnapshot: async (id) =>
+ id === owner.id
+ ? headerSnapshot(owner, 1)
+ : headerSnapshot(fixture.header(), fixture.revision()),
+ },
+ connection: {
+ providerType: 'custom',
+ modelOverrides: { 'model-1': { thinkingLevels: ['low', 'high'] } },
+ },
+ });
+
+ await fixture.coordinator.synchronizeAssistantBotModel(owner.id, fixture.sessionId);
+ assert.equal(fixture.header().model, 'model-1');
+ assert.equal(fixture.header().thinkingLevel, undefined);
+
+ owner = { ...owner, thinkingLevel: 'low' };
+ await fixture.coordinator.synchronizeAssistantBotModel(owner.id, fixture.sessionId);
+ assert.equal(fixture.header().thinkingLevel, 'low');
+ assert.equal(fixture.revision(), 5);
+});
+
test('WorkHub model authority preserves its execution policy and uses versioned runtime configuration', async () => {
const fixture = createFixture({
header: {
diff --git a/packages/runtime-host/src/assistant/bot/controller.ts b/packages/runtime-host/src/assistant/bot/controller.ts
index feedee87b2..5e36f8576c 100644
--- a/packages/runtime-host/src/assistant/bot/controller.ts
+++ b/packages/runtime-host/src/assistant/bot/controller.ts
@@ -24,6 +24,7 @@ import { ASSISTANT_BACKGROUND_SESSION_LABEL } from '@maka/core/session';
import type { MessageContent } from '@maka/core/events';
import type { AssistantModuleServices } from '../service.js';
import type { InitiativeStore } from '../index-initiative/store.js';
+import { executionFailure } from '../execution-outcome.js';
import { BotStore, type InboxInput } from './store.js';
const UNREAD_SLEEP_FEEDBACK =
@@ -168,14 +169,33 @@ export class BotController {
}
const attemptedThrough = this.store.head();
this.running = this.run()
- .catch((error) => {
+ .catch(async (error) => {
if (!this.abort.signal.aborted) {
this.store.transaction(() => {
+ // Retrying an undelivered error bubble must retain the original
+ // execution failure and its durable publication identity.
+ if (
+ this.store.get('error') &&
+ this.store.get<number>('errorThrough') === attemptedThrough
+ )
+ return;
this.store.set('error', String(error));
// Mail arriving during a failed execution must still trigger a new
// attempt. Do not latch the failure against that newer input.
this.store.set('errorThrough', attemptedThrough);
+ const id = randomUUID();
+ this.store.queueVisible({
+ type: 'assistant',
+ id: `bot-error:${id}`,
+ turnId: `assistant-output:error:${id}`,
+ ts: Date.now(),
+ text: `助手消息处理暂时停止：${String(error)}。消息已保留；请检查当前模型和连接，发送新消息可重试。`,
+ modelId: 'assistant-bot',
+ });
});
+ // A publication may have committed before its response was lost.
+ // Keep the outbox entry for the next tick without replacing the failure.
+ await this.flush().catch(() => {});
}
})
.finally(() => {
@@ -183,7 +203,15 @@ export class BotController {
});
}
private async run() {
- await this.flush();
+ try {
+ await this.flush();
+ } catch (error) {
+ // Pending publication is retried independently of model execution. A
+ // delivery failure cannot consume the retry triggered by fresh Inbox mail.
+ if (!this.abort.signal.aborted) this.store.set('wakeError', String(error));
+ return;
+ }
+ this.store.set('wakeError', null);
if (this.store.get('error')) {
const errorThrough = this.store.get<number>('errorThrough');
if (errorThrough !== undefined && this.store.head() <= errorThrough) return;
@@ -288,10 +316,26 @@ export class BotController {
}
const finalStatus = ((await agent.snapshot()) as any)?.agent?.status;
this.store.set('wake', null);
- if (['aborted', 'blocked', 'waiting_for_user'].includes(finalStatus))
- throw Error('Bot stopped: ' + finalStatus);
- if (!this.store.get('sleepTurn'))
- throw Error('Bot ended without Sleep; send a new message to retry');
+ if (
+ ['aborted', 'blocked', 'waiting_for_user'].includes(finalStatus) ||
+ !this.store.get('sleepTurn')
+ ) {
+ let message = ['aborted', 'blocked', 'waiting_for_user'].includes(finalStatus)
+ ? 'Bot stopped: ' + finalStatus
+ : 'Bot ended without Sleep; send a new message to retry';
+ // A contained provider failure can settle whenIdle normally. Resolve
+ // this wake's latest execution rather than reporting a stale turn error.
+ if (wake) {
+ const execution: any = await agent
+ .messageExecution(wake.messageId)
+ .catch(() => undefined);
+ if (execution?.state === 'owned' && execution.turnId) {
+ const failure = await executionFailure(agent, execution.turnId).catch(() => undefined);
+ if (failure) message = failure.message;
+ }
+ }
+ throw Error(message);
+ }
} finally {
this.activeAgent = undefined;
}
diff --git a/packages/runtime-host/src/assistant/service.ts b/packages/runtime-host/src/assistant/service.ts
index 14a46c411c..ec2b0a7509 100644
--- a/packages/runtime-host/src/assistant/service.ts
+++ b/packages/runtime-host/src/assistant/service.ts
@@ -463,6 +463,9 @@ export class PersonalAssistantService {
isConversation(sessionId: string) {
return this.enabled && this.runtime!.bot.isConversation(sessionId);
}
+ botOwnerFor(sessionId: string): string | undefined {
+ return this.enabled ? this.runtime!.bot.ownerFor(sessionId) : undefined;
+ }
hasInboxMessage(sessionId: string, id: string) {
return this.enabled && this.runtime!.bot.hasUserMessage(sessionId, id);
}
diff --git a/packages/runtime-host/src/server/execution-composition.ts b/packages/runtime-host/src/server/execution-composition.ts
index c8b94a4e88..963a003b56 100644
--- a/packages/runtime-host/src/server/execution-composition.ts
+++ b/packages/runtime-host/src/server/execution-composition.ts
@@ -2272,12 +2272,23 @@ export async function createExecutionRuntimeHostComposition(
});
},
resume: async (options, initiator) => {
- if (options.prompt)
- await submitAgentMessage(options.sessionId, options.prompt, 'next_turn', initiator);
const session = (await visibleAgentSessions(initiator)).find(
(item) => item.id === options.sessionId,
);
if (!session) throw new Error('Agent was not found');
+ if (personalAssistant.botOwnerFor(options.sessionId) === initiator.sessionId) {
+ const [pending, boundaries] = await Promise.all([
+ stores.interactionStore.listPending({ sessionId: options.sessionId }),
+ stores.sessionStore.listPendingSandboxBoundaryRequests(options.sessionId),
+ ]);
+ if (!pending.length && !boundaries.length)
+ await sessionCatalog.synchronizeAssistantBotModel(
+ initiator.sessionId,
+ options.sessionId,
+ );
+ }
+ if (options.prompt)
+ await submitAgentMessage(options.sessionId, options.prompt, 'next_turn', initiator);
return describeAgent(session);
},
get: async (id, initiator) => {
diff --git a/packages/runtime-host/src/server/session-catalog-coordinator.ts b/packages/runtime-host/src/server/session-catalog-coordinator.ts
index 41ab8e223c..243703886a 100644
--- a/packages/runtime-host/src/server/session-catalog-coordinator.ts
+++ b/packages/runtime-host/src/server/session-catalog-coordinator.ts
@@ -420,6 +420,91 @@ export class HostSessionCatalogCoordinator {
if (!outcome.ok) throw new Error(outcome.error.message);
}
+ /** Keep the hidden Bot on its owner's selected model without replacing its history. */
+ async synchronizeAssistantBotModel(ownerId: string, workerId: string): Promise<void> {
+ await this.#admission.runMany([ownerId, workerId], async (lease) => {
+ const [owner, worker] = await Promise.all([
+ this.#stores.readHeaderRecordSnapshot(ownerId),
+ this.#stores.readHeaderRecordSnapshot(workerId),
+ ]);
+ // A wake may observe a still-running worker. Leave that execution and its
+ // pending interactions intact; the next idle wake will synchronize it.
+ if (
+ this.#manager.runningTurnIds(workerId).length ||
+ this.#isTurnBusy?.(workerId) ||
+ worker.header.status === 'running' ||
+ worker.header.status === 'waiting_for_user'
+ )
+ return;
+ if (
+ !owner.header.executorId &&
+ !worker.header.executorId &&
+ owner.header.llmConnectionId === worker.header.llmConnectionId &&
+ owner.header.llmConnectionSlug === worker.header.llmConnectionSlug &&
+ owner.header.model === worker.header.model &&
+ owner.header.thinkingLevel === worker.header.thinkingLevel
+ )
+ return;
+ if (owner.header.executorId) {
+ if (
+ owner.header.executorId === worker.header.executorId &&
+ owner.header.model === worker.header.model &&
+ owner.header.executorConfig?.model === worker.header.executorConfig?.model &&
+ owner.header.executorConfig?.mode === worker.header.executorConfig?.mode &&
+ owner.header.thinkingLevel === worker.header.thinkingLevel
+ )
+ return;
+ throw new SessionOperationFailure(
+ 'operation_unavailable',
+ 'Assistant Bot cannot synchronize the selected external agent',
+ );
+ }
+ let connectionId = owner.header.llmConnectionId;
+ if (connectionId === undefined) {
+ // Older headers identify their selected connection by slug alone. Do
+ // not substitute the Host default for that explicit user selection.
+ const selected = await this.#runtimePolicy.operations.resolveExecutionConnection({
+ kind: 'catalog_slug',
+ connectionSlug: owner.header.llmConnectionSlug,
+ });
+ if (selected.kind !== 'ready')
+ throw new SessionOperationFailure(
+ 'operation_unavailable',
+ 'Assistant model connection is unavailable',
+ );
+ connectionId = selected.connection.connectionId;
+ }
+ const outcome = await this.#updateConfiguration(
+ {
+ sessionId: workerId,
+ expectedRevision: worker.revision,
+ patch: {
+ modelTarget: {
+ kind: 'explicit',
+ connectionId,
+ connectionSlug: owner.header.llmConnectionSlug,
+ model: owner.header.model,
+ },
+ thinkingLevel: owner.header.thinkingLevel ?? null,
+ },
+ },
+ 'ordinary',
+ lease,
+ );
+ if (!outcome.ok) {
+ // Execution claims can race the idle observation. The transition
+ // authority refuses them without disposing a live backend.
+ if (outcome.error.code === 'session_busy') return;
+ throw new Error(outcome.error.message);
+ }
+ if (outcome.result.kind !== 'committed')
+ throw new SessionOperationFailure(
+ 'operation_conflict',
+ 'Assistant Bot configuration changed during model synchronization',
+ );
+ });
+ }
+
/** WorkHub Action Gate path; callers cannot bypass the typed operation outcome. */
createForWorkHub(input: SessionCreateInput): Promise<OperationOutcome<'session.create'>> {
return this.#create(input);
@@ -801,6 +886,7 @@ export class HostSessionCatalogCoordinator {
async #updateConfiguration(
input: SessionConfigurationUpdateInput,
authority: 'ordinary' | 'workhub' = 'ordinary',
+ admittedLease?: SessionAdmissionLease,
): Promise<OperationOutcome<'session.configuration.update'>> {
if (authority === 'ordinary' && isWorkHubCoordinationSessionId(input.sessionId)) {
return configurationFailure(
@@ -808,7 +894,9 @@ export class HostSessionCatalogCoordinator {
'WorkHub Coordination Session configuration requires WorkHub authority',
);
}
- return this.#admission.run(input.sessionId, async (lease) => {
+ const update = async (
+ lease: SessionAdmissionLease,
+ ): Promise<OperationOutcome<'session.configuration.update'>> => {
let commitAttempted = false;
let confirmedExecutor: SessionHeader | undefined;
try {
@@ -945,7 +1033,10 @@ export class HostSessionCatalogCoordinator {
'Session configuration update outcome is unknown',
);
}
- });
+ };
+ return admittedLease
+ ? this.#admission.runAdmitted(input.sessionId, admittedLease, () => update(admittedLease))
+ : this.#admission.run(input.sessionId, update);
}
async #reconcileExecutorAfterFailedCommit(previous: SessionHeader): Promise<void> {
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@maka.apache.org
For additional commands, e-mail: commits-help@maka.apache.org