From f4252e77ec1da2a881f719b7294e934617508d0d Mon Sep 17 00:00:00 2001 From: testikun Date: Mon, 31 Aug 2026 15:35:23 +0800 Subject: [PATCH 1/2] feat(runtime-host): bind ScheduledTasks to Connection identity Generated-by: Codex --- .../core/src/__tests__/scheduled-task.test.ts | 43 +++ packages/core/src/agent-run.ts | 7 +- packages/core/src/scheduled-task.ts | 6 + .../src/__tests__/execution-host.test.ts | 125 +++++++- ...cheduled-task-coordinator-recovery.test.ts | 279 +++++++++++++++++- .../__tests__/scheduled-task-protocol.test.ts | 36 +++ packages/runtime-host/src/protocol/index.ts | 5 +- .../src/protocol/scheduled-task.ts | 34 ++- .../src/server/scheduled-task-coordinator.ts | 164 +++++++--- packages/runtime/src/scheduled-task-tools.ts | 3 + .../__tests__/sqlite-workflow-store.test.ts | 7 + packages/storage/src/agent-run-store.ts | 14 +- .../ui/src/scheduled-task-form-dialog.tsx | 16 +- 13 files changed, 681 insertions(+), 58 deletions(-) diff --git a/packages/core/src/__tests__/scheduled-task.test.ts b/packages/core/src/__tests__/scheduled-task.test.ts index e2082e3223..4bec466709 100644 --- a/packages/core/src/__tests__/scheduled-task.test.ts +++ b/packages/core/src/__tests__/scheduled-task.test.ts @@ -175,6 +175,7 @@ describe('scheduled-task catalog', () => { const now = Date.UTC(2026, 0, 5, 8, 0, 0); const execution = { cwd: '/tmp/project', + llmConnectionId: 'connection-anthropic', llmConnectionSlug: 'anthropic', model: 'claude-sonnet-4-5-20250929', permissionMode: 'ask', @@ -206,6 +207,34 @@ describe('scheduled-task catalog', () => { } }); + it('requires an immutable Connection identity for new Agent tasks', () => { + const now = Date.UTC(2026, 0, 5, 8, 0, 0); + const result = normalizeCreateScheduledTaskInput( + { + title: 'Missing identity', + intentBody: 'run', + schedule: { kind: 'once', runAt: now + 60_000 }, + effect: { + kind: 'agent_run', + execution: { + cwd: '/tmp/project', + llmConnectionSlug: 'anthropic', + model: 'claude', + permissionMode: 'ask', + collaborationMode: 'agent', + orchestrationMode: 'default', + }, + }, + createdBy: { kind: 'user' }, + }, + now, + ); + assert.deepEqual(result, { + ok: false, + message: 'execution.llmConnectionId is required', + }); + }); + it('rejects future recurrence anchors outside the scheduling horizon', () => { const now = Date.UTC(2026, 0, 5, 8, 0, 0); for (const schedule of [ @@ -240,6 +269,7 @@ describe('decodePersistedScheduledTask', () => { kind: 'agent_run', execution: { cwd: '/repo', + llmConnectionId: 'connection-anthropic', llmConnectionSlug: 'anthropic', model: 'claude', permissionMode: 'ask', @@ -275,6 +305,19 @@ describe('decodePersistedScheduledTask', () => { assert.equal(decodePersistedScheduledTask(markPersisted(base)), base); }); + it('keeps legacy slug-only Agent tasks readable', () => { + const { llmConnectionId: _legacyId, ...legacyExecution } = + base.effect.kind === 'agent_run' ? base.effect.execution : {}; + const legacy = { + ...base, + effect: { kind: 'agent_run' as const, execution: legacyExecution }, + } as ScheduledTask; + const decoded = decodePersistedScheduledTask(markPersisted(legacy)); + assert.equal(decoded.effect.kind, 'agent_run'); + if (decoded.effect.kind !== 'agent_run') return; + assert.equal(decoded.effect.execution.llmConnectionId, undefined); + }); + it('leaves effects without an execution template alone', () => { const notify: ScheduledTask = { ...base, effect: { kind: 'notify', channel: 'local' } }; assert.equal(decodePersistedScheduledTask(markPersisted(notify)), notify); diff --git a/packages/core/src/agent-run.ts b/packages/core/src/agent-run.ts index ac0fcedb24..ffc7e44d4c 100644 --- a/packages/core/src/agent-run.ts +++ b/packages/core/src/agent-run.ts @@ -87,7 +87,12 @@ export type RootExecutionDescriptor = } | { kind: 'regenerate'; sourceTurnId: string } | { kind: 'context_compact' } - | { kind: 'scheduled_task'; scheduledTaskId: string } + | { + kind: 'scheduled_task'; + scheduledTaskId: string; + /** Includes the immutable Connection target for Agent ScheduledTasks. */ + executionFingerprint?: `sha256:${string}`; + } | { kind: 'legacy_automation'; automationId: string } | { kind: 'goal'; goalId: string } | { diff --git a/packages/core/src/scheduled-task.ts b/packages/core/src/scheduled-task.ts index e741812c34..d1eac46fb8 100644 --- a/packages/core/src/scheduled-task.ts +++ b/packages/core/src/scheduled-task.ts @@ -75,6 +75,8 @@ export type ScheduledTaskEffect = export interface ScheduledTaskExecutionTemplate { readonly cwd: string; readonly projectId?: string | null; + /** Immutable Connection entity identity. Omitted only on legacy slug-only rows. */ + readonly llmConnectionId?: string; readonly llmConnectionSlug: string; readonly model: string; readonly thinkingLevel?: ThinkingLevel; @@ -510,6 +512,9 @@ function normalizeExecution( ): ScheduledTaskNormalizeResult { if (!isObject(value)) return fail('agent_run requires execution template'); if (typeof value.cwd !== 'string' || !value.cwd.trim()) return fail('execution.cwd is required'); + if (typeof value.llmConnectionId !== 'string' || !value.llmConnectionId.trim()) { + return fail('execution.llmConnectionId is required'); + } if (typeof value.llmConnectionSlug !== 'string' || !value.llmConnectionSlug.trim()) { return fail('execution.llmConnectionSlug is required'); } @@ -541,6 +546,7 @@ function normalizeExecution( value: { cwd: value.cwd.trim(), ...(projectId === undefined ? {} : { projectId }), + llmConnectionId: value.llmConnectionId.trim(), llmConnectionSlug: value.llmConnectionSlug.trim(), model: value.model.trim(), ...(value.thinkingLevel === undefined ? {} : { thinkingLevel: value.thinkingLevel }), diff --git a/packages/runtime-host/src/__tests__/execution-host.test.ts b/packages/runtime-host/src/__tests__/execution-host.test.ts index 9e3705c332..cf523aaa67 100644 --- a/packages/runtime-host/src/__tests__/execution-host.test.ts +++ b/packages/runtime-host/src/__tests__/execution-host.test.ts @@ -35,6 +35,7 @@ import { createServer, type Server } from 'node:http'; import { connect, type Socket } from 'node:net'; import { tmpdir } from 'node:os'; import { dirname, join } from 'node:path'; +import { DatabaseSync } from 'node:sqlite'; import { test } from 'node:test'; import { TOOL_BOUNDARY_PROTOCOL_V1 } from '@maka/core/runtime-event'; import { canonicalToolArgsHash } from '@maka/core/tool-args-identity'; @@ -66,6 +67,7 @@ import { openInteractiveExecutionStoresForRead, openInteractiveExecutionStoresForWrite, } from '@maka/storage/execution-stores'; +import { OPERATIONAL_STATE_DATABASE_NAME } from '@maka/storage/operational-state-store'; import { openInteractiveRuntimePolicyStoresForWrite } from '@maka/storage/runtime-policy-stores'; import { resolveRootControlNamespace, @@ -162,6 +164,10 @@ test('production Host fails slug-only ScheduledTask Agent runs before binding ex timeout: 30_000, }, async () => { await withExecutionRoot(async (fixture) => { + const seededConnection = await fixture.seedConnectionEffect( + 'http://127.0.0.1:1', + 'test-secret', + ); const host = await fixture.startHost(); const desktop = await connectClient(fixture.root); try { @@ -175,8 +181,9 @@ test('production Host fails slug-only ScheduledTask Agent runs before binding ex kind: 'agent_run', execution: { cwd: fixture.root, - llmConnectionSlug: 'fake', - model: 'fake-model', + llmConnectionId: seededConnection.connectionId, + llmConnectionSlug: seededConnection.slug, + model: seededConnection.enabledModelIds[0]!, permissionMode: 'ask', collaborationMode: 'agent', orchestrationMode: 'default', @@ -187,6 +194,21 @@ test('production Host fails slug-only ScheduledTask Agent runs before binding ex assert.equal(created.kind, 'task'); if (created.kind !== 'task') return; + // Simulate a record written by a pre-#3927 build. Legacy rows remain + // readable, but must fail closed before a Session or AgentRun is bound. + const database = new DatabaseSync(join(fixture.root, OPERATIONAL_STATE_DATABASE_NAME)); + try { + database + .prepare( + `UPDATE workflow_scheduled_tasks + SET record_json = json_remove(record_json, '$.effect.execution.llmConnectionId') + WHERE task_id = ?`, + ) + .run(created.task.id); + } finally { + database.close(); + } + const fired = await desktop.request('scheduled-task.mutate', { kind: 'trigger_now', taskId: created.task.id, @@ -208,6 +230,105 @@ test('production Host fails slug-only ScheduledTask Agent runs before binding ex }); }); +test('two UDS Clients never rebind an Agent ScheduledTask after Connection slug reuse', { + timeout: 30_000, +}, async () => { + await withExecutionRoot(async (fixture) => { + const original = await fixture.seedConnectionEffect('http://127.0.0.1:1', 'test-secret'); + const model = original.enabledModelIds[0]!; + const host = await fixture.startHost(); + const creator = await connectClient(fixture.root); + const trigger = await connectClient(fixture.root); + try { + const created = await creator.request('scheduled-task.mutate', { + kind: 'create', + input: { + title: 'Connection slug reuse proof', + intentBody: 'The deleted account must never be replaced silently.', + schedule: { kind: 'once', runAt: Date.now() + 60_000 }, + effect: { + kind: 'agent_run', + execution: { + cwd: fixture.root, + llmConnectionId: original.connectionId, + llmConnectionSlug: original.slug, + model, + permissionMode: 'ask', + collaborationMode: 'agent', + orchestrationMode: 'default', + }, + }, + }, + }); + assert.equal(created.kind, 'task'); + if (created.kind !== 'task') return; + + const catalog = await trigger.request('connection.catalog.query', { kind: 'start' }); + assert.equal(catalog.kind, 'page'); + if (catalog.kind !== 'page') return; + const header = catalog.items.find( + (item) => item.kind === 'connection' && item.connectionId === original.connectionId, + ); + assert.equal(header?.kind, 'connection'); + if (header?.kind !== 'connection') return; + + const removed = await trigger.request('connection.catalog.remove', { + expected: { connectionId: original.connectionId, revision: header.revision }, + }); + assert.equal(removed.kind, 'committed'); + if (removed.kind !== 'committed') return; + const replacement = await trigger.request('connection.catalog.create', { + expectedCatalogRevision: removed.catalogRevision, + connection: { + slug: original.slug, + name: 'Replacement account', + providerType: original.providerType, + ...(original.baseUrl === undefined ? {} : { baseUrl: original.baseUrl }), + enabled: true, + enabledModelIds: [model], + }, + }); + assert.equal(replacement.kind, 'committed'); + if (replacement.kind !== 'committed') return; + assert.notEqual(replacement.connection.connectionId, original.connectionId); + + const fired = await trigger.request('scheduled-task.mutate', { + kind: 'trigger_now', + taskId: created.task.id, + }); + assert.equal(fired.kind, 'task'); + if (fired.kind !== 'task') return; + assert.equal(fired.task.runs[0]?.outcome, 'failed'); + assert.equal(fired.task.lastError, 'ScheduledTask model connection identity changed'); + const failedRun = fired.task.runs[0]; + assert.ok(failedRun?.sessionId); + assert.ok(failedRun?.runId); + const databaseAfterFire = new DatabaseSync( + join(fixture.root, OPERATIONAL_STATE_DATABASE_NAME), + ); + try { + assert.equal( + databaseAfterFire + .prepare('SELECT 1 AS present FROM session_metadata WHERE session_id = ?') + .get(failedRun.sessionId), + undefined, + ); + assert.equal( + databaseAfterFire + .prepare('SELECT 1 AS present FROM core_agent_runs WHERE run_id = ?') + .get(failedRun.runId), + undefined, + ); + } finally { + databaseAfterFire.close(); + } + } finally { + await Promise.allSettled([creator.close(), trigger.close()]); + await fixture.stopHost(host); + } + }); +}); + test('production Host settles dispatched Client Capabilities before publishing Ready', async () => { await withExecutionRoot(async (fixture) => { const prepared = await seedDispatchedClientCapability(fixture); diff --git a/packages/runtime-host/src/__tests__/scheduled-task-coordinator-recovery.test.ts b/packages/runtime-host/src/__tests__/scheduled-task-coordinator-recovery.test.ts index 073fa4c25e..be87701f85 100644 --- a/packages/runtime-host/src/__tests__/scheduled-task-coordinator-recovery.test.ts +++ b/packages/runtime-host/src/__tests__/scheduled-task-coordinator-recovery.test.ts @@ -25,7 +25,11 @@ import { test } from 'node:test'; import type { RootTurnAdmission } from '@maka/storage/execution-stores'; import { openInteractiveScheduledTaskStoreForWrite } from '@maka/storage/scheduled-task-store'; import { resolveStorageRoot, tryAcquireInteractiveRootOwner } from '@maka/storage/root-authority'; -import { HostScheduledTaskCoordinator } from '../server/scheduled-task-coordinator.js'; +import { SessionNotFoundError } from '@maka/storage/session-store'; +import { + HostScheduledTaskCoordinator, + scheduledTaskExecutionFingerprint, +} from '../server/scheduled-task-coordinator.js'; test('ScheduledTask recovery distinguishes a settled fire from a newer pending fire', async () => { const base = await mkdtemp(join(tmpdir(), 'maka-scheduled-task-recovery-')); @@ -57,6 +61,7 @@ test('ScheduledTask recovery distinguishes a settled fire from a newer pending f execution: { cwd: '/workspace', backend: 'ai-sdk', + llmConnectionId: 'connection-default', llmConnectionSlug: 'default', model: 'test-model', permissionMode: 'ask', @@ -83,9 +88,16 @@ test('ScheduledTask recovery distinguishes a settled fire from a newer pending f await store.bindFireExecution(newClaim.id, newExecution); await coordinator.prepareRecovery(); - await coordinator.assertRecoveryAdmission(admission(task.id, oldExecution), 'run_recorded'); + const fingerprint = + task.effect.kind === 'agent_run' + ? scheduledTaskExecutionFingerprint(task.effect.execution) + : undefined; await coordinator.assertRecoveryAdmission( - admission(task.id, newExecution), + admission(task.id, oldExecution, fingerprint), + 'run_recorded', + ); + await coordinator.assertRecoveryAdmission( + admission(task.id, newExecution, fingerprint), 'pending_fire_required', ); await assert.rejects( @@ -96,6 +108,14 @@ test('ScheduledTask recovery distinguishes a settled fire from a newer pending f ), /has no matching pending fire/, ); + await assert.rejects( + () => + coordinator.assertRecoveryAdmission( + admission(task.id, newExecution, `sha256:${'b'.repeat(64)}`), + 'pending_fire_required', + ), + /has no matching pending fire/, + ); await store.settleFire(newClaim.id, { at: 3_001, @@ -108,7 +128,11 @@ test('ScheduledTask recovery distinguishes a settled fire from a newer pending f const conflictingClaim = await store.claimNow(task.id, 4_000); await store.bindFireExecution(conflictingClaim.id, conflictingExecution); await assert.rejects( - () => coordinator.assertRecoveryAdmission(admission(task.id, oldExecution), 'run_recorded'), + () => + coordinator.assertRecoveryAdmission( + admission(task.id, oldExecution, fingerprint), + 'run_recorded', + ), /has no matching pending fire/, ); } finally { @@ -119,6 +143,246 @@ test('ScheduledTask recovery distinguishes a settled fire from a newer pending f } }); +test('ScheduledTask execution fails closed when the bound Connection identity is replaced', async () => { + const base = await mkdtemp(join(tmpdir(), 'maka-scheduled-task-identity-')); + const capability = await resolveStorageRoot({ path: join(base, 'root'), kind: 'interactive' }); + const owner = await tryAcquireInteractiveRootOwner(capability); + assert.ok(owner); + if (!owner) throw new Error('Unable to acquire the ScheduledTask identity test root'); + const store = await openInteractiveScheduledTaskStoreForWrite(owner.lease); + let createSessionCalls = 0; + let admitCalls = 0; + const coordinator = new HostScheduledTaskCoordinator({ + store, + sessions: { + readHeaderSnapshot: async () => { + throw new SessionNotFoundError('scheduled-task-session'); + }, + }, + runtime: { + sendMessage: async function* () { + // The replacement-identity path must fail before this stream is used. + }, + }, + root: { + admit: async () => { + admitCalls += 1; + throw new Error('stale ScheduledTask was admitted'); + }, + }, + runtimePolicy: { + runtimePolicy: { + getSnapshot: async () => ({ policy: { privacy: { incognitoActive: false } } }), + }, + connectionCatalog: null as never, + credentialVault: null as never, + operations: { + resolveExecutionConnection: async () => ({ kind: 'identity_mismatch' as const }), + }, + } as never, + nativeEffects: null as never, + createSession: async () => { + createSessionCalls += 1; + }, + changes: { publish: () => undefined }, + acquireResidency: () => ({ release: () => undefined }), + requestDrain: () => undefined, + }); + try { + await coordinator.prepareRecovery(); + const rejectedCreate = await coordinator.handlers['scheduled-task.mutate']( + { + kind: 'create', + input: { + title: 'Rejected replacement target', + intentBody: 'Must not persist an unresolvable Connection tuple.', + schedule: { kind: 'once', runAt: 2_000 }, + effect: { + kind: 'agent_run', + execution: { + cwd: '/workspace', + llmConnectionId: 'connection-a', + llmConnectionSlug: 'shared-slug', + model: 'model-a', + permissionMode: 'ask', + collaborationMode: 'agent', + orchestrationMode: 'default', + }, + }, + }, + }, + {} as never, + ); + assert.deepEqual(rejectedCreate, { + ok: false, + error: { + code: 'operation_conflict', + message: 'ScheduledTask model connection identity changed', + }, + }); + assert.equal((await store.list()).length, 0); + + const task = await store.create( + { + title: 'Replaced connection task', + intentBody: 'Must not use a same-slug replacement.', + schedule: { kind: 'once', runAt: 2_000 }, + effect: { + kind: 'agent_run', + execution: { + cwd: '/workspace', + llmConnectionId: 'connection-a', + llmConnectionSlug: 'shared-slug', + model: 'model-a', + permissionMode: 'ask', + collaborationMode: 'agent', + orchestrationMode: 'default', + }, + }, + createdBy: { kind: 'user' }, + }, + 1_000, + ); + const result = await coordinator.handlers['scheduled-task.mutate']( + { + kind: 'trigger_now', + taskId: task.id, + }, + {} as never, + ); + assert.equal(result.ok, true); + if (!result.ok || result.result.kind !== 'task') return; + assert.equal(result.result.task.runs[0]?.outcome, 'failed'); + assert.equal(result.result.task.lastError, 'ScheduledTask model connection identity changed'); + assert.equal(createSessionCalls, 0); + assert.equal(admitCalls, 0); + } finally { + await coordinator.close(); + store.close(); + await owner.close(); + await rm(base, { recursive: true, force: true }); + } +}); + +test('ScheduledTask with an exact Connection identity reaches Session and AgentRun admission', async () => { + const base = await mkdtemp(join(tmpdir(), 'maka-scheduled-task-success-')); + const capability = await resolveStorageRoot({ path: join(base, 'root'), kind: 'interactive' }); + const owner = await tryAcquireInteractiveRootOwner(capability); + assert.ok(owner); + if (!owner) throw new Error('Unable to acquire the ScheduledTask success test root'); + const store = await openInteractiveScheduledTaskStoreForWrite(owner.lease); + let createSessionCalls = 0; + let admitCalls = 0; + let expectedTaskId = ''; + const connection = { + connectionId: 'connection-a', + revision: 1, + slug: 'shared-slug', + name: 'Account A', + providerType: 'openai' as const, + enabled: true, + enabledModelIds: ['model-a'], + models: [], + }; + const coordinator = new HostScheduledTaskCoordinator({ + store, + sessions: { + readHeaderSnapshot: async () => { + throw new SessionNotFoundError('scheduled-task-session'); + }, + }, + runtime: { + sendMessage: async function* () { + // The admission authority owns execution startup in this unit test. + }, + }, + root: { + admit: async (input) => { + admitCalls += 1; + assert.equal(input.execution.kind, 'scheduled_task'); + if (input.execution.kind === 'scheduled_task') { + assert.equal(input.execution.scheduledTaskId, expectedTaskId); + assert.equal( + input.execution.executionFingerprint, + scheduledTaskExecutionFingerprint({ + cwd: '/workspace', + llmConnectionId: 'connection-a', + llmConnectionSlug: 'shared-slug', + model: 'model-a', + permissionMode: 'ask', + collaborationMode: 'agent', + orchestrationMode: 'default', + }), + ); + } + return {} as never; + }, + }, + runtimePolicy: { + runtimePolicy: { + getSnapshot: async () => ({ policy: { privacy: { incognitoActive: false } } }), + }, + connectionCatalog: null as never, + credentialVault: null as never, + operations: { + resolveExecutionConnection: async () => ({ kind: 'ready' as const, connection }), + }, + } as never, + nativeEffects: null as never, + createSession: async (input) => { + createSessionCalls += 1; + assert.deepEqual(input.modelTarget, { + kind: 'explicit', + connectionId: 'connection-a', + connectionSlug: 'shared-slug', + model: 'model-a', + }); + }, + changes: { publish: () => undefined }, + acquireResidency: () => ({ release: () => undefined }), + requestDrain: () => undefined, + }); + try { + const task = await store.create( + { + title: 'Exact connection task', + intentBody: 'Run with the frozen account.', + schedule: { kind: 'once', runAt: 2_000 }, + effect: { + kind: 'agent_run', + execution: { + cwd: '/workspace', + llmConnectionId: 'connection-a', + llmConnectionSlug: 'shared-slug', + model: 'model-a', + permissionMode: 'ask', + collaborationMode: 'agent', + orchestrationMode: 'default', + }, + }, + createdBy: { kind: 'user' }, + }, + 1_000, + ); + expectedTaskId = task.id; + await coordinator.prepareRecovery(); + const result = await coordinator.handlers['scheduled-task.mutate']( + { kind: 'trigger_now', taskId: task.id }, + {} as never, + ); + assert.equal(result.ok, true); + if (!result.ok || result.result.kind !== 'task') return; + assert.equal(result.result.task.runs[0]?.outcome, 'ok'); + assert.equal(createSessionCalls, 1); + assert.equal(admitCalls, 1); + } finally { + await coordinator.close(); + store.close(); + await owner.close(); + await rm(base, { recursive: true, force: true }); + } +}); + function execution(suffix: string) { return { sessionId: `session-${suffix}`, @@ -131,11 +395,16 @@ function execution(suffix: string) { function admission( scheduledTaskId: string, identity: ReturnType, + executionFingerprint?: `sha256:${string}`, ): RootTurnAdmission { return { schemaVersion: 1, ...identity, - execution: { kind: 'scheduled_task', scheduledTaskId }, + execution: { + kind: 'scheduled_task', + scheduledTaskId, + ...(executionFingerprint === undefined ? {} : { executionFingerprint }), + }, previousRootTurnId: null, normalizedInput: { text: 'Continue the scheduled work.' }, sourceMessages: [], diff --git a/packages/runtime-host/src/__tests__/scheduled-task-protocol.test.ts b/packages/runtime-host/src/__tests__/scheduled-task-protocol.test.ts index 3d9ea5e7d1..ea6452db95 100644 --- a/packages/runtime-host/src/__tests__/scheduled-task-protocol.test.ts +++ b/packages/runtime-host/src/__tests__/scheduled-task-protocol.test.ts @@ -97,6 +97,41 @@ describe('ScheduledTask protocol', () => { assertDropped(updated.kind === 'update' ? updated.patch.effect : undefined); }); + test('requires Connection identity on Agent task mutations but accepts legacy reads', () => { + const legacy = agentRunEffect('project-1'); + if (legacy.kind !== 'agent_run') return; + const { llmConnectionId: _legacyId, ...legacyExecution } = legacy.execution; + assert.throws( + () => + decodeScheduledTaskMutateInput({ + kind: 'create', + input: { + title: 'Legacy task', + intentBody: 'Run it', + schedule: { kind: 'once', runAt: 1 }, + effect: { kind: 'agent_run', execution: legacyExecution }, + }, + }), + /requires Connection id/u, + ); + assert.deepEqual( + decodeScheduledTaskQueryResult({ + kind: 'task', + task: { + ...scheduledTask('legacy-task'), + effect: { kind: 'agent_run', execution: legacyExecution }, + }, + }), + { + kind: 'task', + task: { + ...scheduledTask('legacy-task'), + effect: { kind: 'agent_run', execution: legacyExecution }, + }, + }, + ); + }); + test('accepts signal-only catalog changes', () => { const frame = { kind: 'scheduled-task.changed' as const, @@ -169,6 +204,7 @@ function agentRunEffect(projectId: string | null | undefined): ScheduledTaskEffe execution: { cwd: '/workspace', ...(projectId === undefined ? {} : { projectId }), + llmConnectionId: 'connection-openai', llmConnectionSlug: 'openai', model: 'gpt-5', permissionMode: 'ask', diff --git a/packages/runtime-host/src/protocol/index.ts b/packages/runtime-host/src/protocol/index.ts index deff39f1ce..cf441eb4e2 100644 --- a/packages/runtime-host/src/protocol/index.ts +++ b/packages/runtime-host/src/protocol/index.ts @@ -100,7 +100,10 @@ export const RUNTIME_HOST_REGISTRATION_SCHEMA_VERSION = 1 as const; export const RUNTIME_HOST_PROTOCOL_VERSION = 0 as const; // Increment when the same protocol version no longer guarantees safe Client-Host // interoperability. Mismatches are rejected before domain commands are admitted. -export const RUNTIME_HOST_COMPATIBILITY_EPOCH = 98 as const; +export const RUNTIME_HOST_COMPATIBILITY_EPOCH = 99 as const; +// 99: ScheduledTask Agent execution templates carry immutable Connection +// identity. Older peers cannot preserve the ID/slug/model binding and could +// silently route a deleted Connection to a same-slug replacement. // 98: Peer Mesh invitations carry signed reachability leases and member route // projections use the convergent recovery state machine. Older peers decode a // different strict wire shape. diff --git a/packages/runtime-host/src/protocol/scheduled-task.ts b/packages/runtime-host/src/protocol/scheduled-task.ts index 48b3d48746..de66544d2a 100644 --- a/packages/runtime-host/src/protocol/scheduled-task.ts +++ b/packages/runtime-host/src/protocol/scheduled-task.ts @@ -359,7 +359,7 @@ function decodeCreateInput(value: unknown): Omit { - if (input.kind === 'update') { - return this.#cancelWaitingNativeFireThen(input.taskId, () => - this.#store.update(input.taskId, input.patch, this.#now()), - ); - } - if (input.kind === 'pause') { - return this.#cancelWaitingNativeFireThen(input.taskId, () => - this.#store.pause(input.taskId, this.#now()), - ); - } - if (input.kind === 'resume') return this.#store.resume(input.taskId, this.#now()); - if (input.kind === 'snooze') { - return this.#cancelWaitingNativeFireThen(input.taskId, () => - this.#store.snooze(input.taskId, input.delayMs, this.#now()), - ); - } - return this.#store.clearRunHistory(input.taskId, this.#now()); - }); + const task = await this.#commitTask( + 'updated', + () => { + if (input.kind === 'update') { + return this.#cancelWaitingNativeFireThen(input.taskId, () => + this.#store.update(input.taskId, input.patch, this.#now()), + ); + } + if (input.kind === 'pause') { + return this.#cancelWaitingNativeFireThen(input.taskId, () => + this.#store.pause(input.taskId, this.#now()), + ); + } + if (input.kind === 'resume') return this.#store.resume(input.taskId, this.#now()); + if (input.kind === 'snooze') { + return this.#cancelWaitingNativeFireThen(input.taskId, () => + this.#store.snooze(input.taskId, input.delayMs, this.#now()), + ); + } + return this.#store.clearRunHistory(input.taskId, this.#now()); + }, + input.kind === 'update' + ? () => this.#validateAgentRunEffect(input.patch.effect) + : undefined, + ); return taskSuccess(task); } catch (error) { if (error instanceof ScheduledTaskStoreError) { @@ -453,7 +485,7 @@ export class HostScheduledTaskCoordinator implements ScheduledTaskToolAuthority } } - async #commitCreate(input: unknown): Promise { + async #commitCreate(input: CreateScheduledTaskInput): Promise { return this.#exclusive(async () => { const incognito = (await this.#runtimePolicy.runtimePolicy.getSnapshot()).policy.privacy .incognitoActive; @@ -469,6 +501,7 @@ export class HostScheduledTaskCoordinator implements ScheduledTaskToolAuthority 'ScheduledTask catalog limit reached', ); } + await this.#validateAgentRunEffect(input.effect); const task = await this.#store.create(input, this.#now()); this.#publish('created', task.id); await this.#refreshSchedule(); @@ -479,8 +512,10 @@ export class HostScheduledTaskCoordinator implements ScheduledTaskToolAuthority #commitTask( reason: ScheduledTaskChangedReason, mutate: () => Promise, + beforeMutate?: () => Promise, ): Promise { return this.#exclusive(async () => { + await beforeMutate?.(); const task = await mutate(); this.#publish(reason, task.id); await this.#refreshSchedule(); @@ -488,6 +523,15 @@ export class HostScheduledTaskCoordinator implements ScheduledTaskToolAuthority }); } + async #validateAgentRunEffect(effect: ScheduledTaskEffect | undefined): Promise { + if (effect?.kind !== 'agent_run') return; + try { + await this.#resolveAgentRunConnection(effect.execution); + } catch (error) { + throw new ScheduledTaskMutationError('operation_conflict', errorMessage(error)); + } + } + async #cancelWaitingNativeFireThen(taskId: string, operation: () => Promise): Promise { await this.#store.cancelWaitingNativeFire(taskId); return operation(); @@ -593,10 +637,10 @@ export class HostScheduledTaskCoordinator implements ScheduledTaskToolAuthority ); } - // Persisted agent-run templates currently identify their model connection - // by reusable slug only. Do not resolve that slug to a potentially - // different Connection entity. #3927 will make the exact ID durable. - if (task.effect.kind === 'agent_run') { + // Legacy persisted Agent-run templates may still identify their model + // connection by reusable slug only. Never resolve that slug to a + // potentially different Connection entity. + if (task.effect.kind === 'agent_run' && !task.effect.execution.llmConnectionId) { return this.#settleFailure(claim, SCHEDULED_AGENT_RUN_IDENTITY_REQUIRED); } @@ -639,20 +683,21 @@ export class HostScheduledTaskCoordinator implements ScheduledTaskToolAuthority await this.#readResumableSession(identity.sessionId); return; } + const execution = task.effect.execution; + const connection = await this.#resolveAgentRunConnection(execution); try { - await this.#sessions.readHeaderSnapshot(identity.sessionId); + const existing = await this.#sessions.readHeaderSnapshot(identity.sessionId); + if ( + existing.llmConnectionId !== execution.llmConnectionId || + existing.llmConnectionSlug !== execution.llmConnectionSlug || + existing.model !== execution.model + ) { + throw new Error('ScheduledTask Session model identity changed'); + } return; } catch (error) { if (!isSessionNotFoundError(error) && !isMissingRecord(error)) throw error; } - const execution = task.effect.execution; - const catalog = await this.#runtimePolicy.connectionCatalog.getSnapshot(); - const connection = catalog.connections.find( - (candidate) => candidate.slug === execution.llmConnectionSlug, - ); - if (!connection) { - throw new Error('ScheduledTask model connection does not exist'); - } await this.#createSession({ sessionId: identity.sessionId, workspace: @@ -674,11 +719,54 @@ export class HostScheduledTaskCoordinator implements ScheduledTaskToolAuthority }); } + async #resolveAgentRunConnection( + execution: ScheduledTaskExecutionTemplate, + ): Promise { + if (!execution.llmConnectionId) { + throw new Error(SCHEDULED_AGENT_RUN_IDENTITY_REQUIRED); + } + const resolved = await this.#runtimePolicy.operations.resolveExecutionConnection({ + kind: 'bound', + connectionId: execution.llmConnectionId, + connectionSlug: execution.llmConnectionSlug, + }); + if (resolved.kind !== 'ready') { + throw new Error( + resolved.kind === 'identity_mismatch' || resolved.kind === 'not_found' + ? 'ScheduledTask model connection identity changed' + : resolved.kind === 'disabled' + ? 'ScheduledTask model connection is disabled' + : resolved.kind === 'credential_not_configured' + ? 'ScheduledTask model connection is not ready' + : resolved.kind === 'provider_retired' + ? 'ScheduledTask model connection uses a retired provider' + : 'ScheduledTask model connection is unavailable', + ); + } + if (!authorizeConnectionModel(resolved.connection, execution.model)) { + throw new Error('ScheduledTask model is no longer enabled'); + } + return resolved.connection; + } + async #admitAgentRun(task: ScheduledTask, identity: ScheduledTaskFireExecution): Promise { + if (task.effect.kind === 'agent_run') { + // Re-read the bound Connection immediately before admission. The + // Session/Connection stores have independent write lanes, so this + // second check closes the delete-and-recreate-same-slug window between + // Session creation and AgentRun admission. + await this.#resolveAgentRunConnection(task.effect.execution); + } const content = { text: task.intent.body }; await this.#root.admit({ ...identity, - execution: { kind: 'scheduled_task', scheduledTaskId: task.id }, + execution: { + kind: 'scheduled_task', + scheduledTaskId: task.id, + ...(task.effect.kind === 'agent_run' + ? { executionFingerprint: scheduledTaskExecutionFingerprint(task.effect.execution) } + : {}), + }, content, start: ({ runId, userMessageId, onRunStarted }) => { if (runId !== identity.runId || userMessageId !== identity.userMessageId) { @@ -823,9 +911,13 @@ class ScheduledTaskMutationError extends Error { } function executionTemplateFromHeader(header: SessionHeader): ScheduledTaskExecutionTemplate { + if (!header.llmConnectionId) { + throw new Error(SCHEDULED_AGENT_RUN_IDENTITY_REQUIRED); + } return { cwd: header.cwd, ...(header.projectId === undefined ? {} : { projectId: header.projectId }), + llmConnectionId: header.llmConnectionId, llmConnectionSlug: header.llmConnectionSlug, model: header.model, ...(header.thinkingLevel === undefined ? {} : { thinkingLevel: header.thinkingLevel }), diff --git a/packages/runtime/src/scheduled-task-tools.ts b/packages/runtime/src/scheduled-task-tools.ts index 2ba8be19d0..7706adcb2d 100644 --- a/packages/runtime/src/scheduled-task-tools.ts +++ b/packages/runtime/src/scheduled-task-tools.ts @@ -178,6 +178,9 @@ export function buildAgentScheduledTaskCreatePayload(input: { if (input.effect === 'agent_run' && !input.execution) { return { error: 'agent_run requires a frozen execution template from the creator session' }; } + if (input.effect === 'agent_run' && !input.execution?.llmConnectionId) { + return { error: 'agent_run requires an immutable model connection identity' }; + } const schedule = input.schedule.kind === 'once' ? input.schedule diff --git a/packages/storage/src/__tests__/sqlite-workflow-store.test.ts b/packages/storage/src/__tests__/sqlite-workflow-store.test.ts index 4d512c0dd4..26a8084bc2 100644 --- a/packages/storage/src/__tests__/sqlite-workflow-store.test.ts +++ b/packages/storage/src/__tests__/sqlite-workflow-store.test.ts @@ -765,6 +765,7 @@ describe('SQLite workflow stores', () => { execution: { cwd: '/workspace', backend: 'ai-sdk', + llmConnectionId: 'connection-default', llmConnectionSlug: 'default', model: 'test-model', permissionMode: 'ask', @@ -776,6 +777,10 @@ describe('SQLite workflow stores', () => { }, now, ); + assert.equal( + task.effect.kind === 'agent_run' ? task.effect.execution.llmConnectionId : undefined, + 'connection-default', + ); const claim = await store.claimNow(task.id, now); await store.bindFireExecution(claim.id, { sessionId: 'session-1', @@ -816,6 +821,7 @@ describe('SQLite workflow stores', () => { kind: 'agent_run', execution: { cwd: '/workspace', + llmConnectionId: 'connection-default', llmConnectionSlug: 'default', model: 'test-model', permissionMode: 'execute', @@ -838,6 +844,7 @@ describe('SQLite workflow stores', () => { kind: 'agent_run', execution: { cwd: '/workspace', + llmConnectionId: 'connection-default', llmConnectionSlug: 'default', model: 'test-model', permissionMode: 'ask', diff --git a/packages/storage/src/agent-run-store.ts b/packages/storage/src/agent-run-store.ts index bb0bc592d4..645d26676a 100644 --- a/packages/storage/src/agent-run-store.ts +++ b/packages/storage/src/agent-run-store.ts @@ -2093,14 +2093,22 @@ function normalizeRootExecutionDescriptor(value: unknown): RootExecutionDescript return Object.freeze({ kind: 'context_compact' }); } if (value.kind === 'scheduled_task') { + const keys = ['kind', 'scheduledTaskId', 'executionFingerprint']; if ( - !hasExactKeys(value, ['kind', 'scheduledTaskId']) || + !Object.keys(value).every((key) => keys.includes(key)) || typeof value.scheduledTaskId !== 'string' || - !isSafeId(value.scheduledTaskId) + !isSafeId(value.scheduledTaskId) || + (value.executionFingerprint !== undefined && !isSha256Digest(value.executionFingerprint)) ) { throw new Error('Invalid root execution descriptor'); } - return Object.freeze({ kind: 'scheduled_task', scheduledTaskId: value.scheduledTaskId }); + return Object.freeze({ + kind: 'scheduled_task', + scheduledTaskId: value.scheduledTaskId, + ...(value.executionFingerprint !== undefined + ? { executionFingerprint: value.executionFingerprint } + : {}), + }); } if (value.kind === 'automation' || value.kind === 'legacy_automation') { if ( diff --git a/packages/ui/src/scheduled-task-form-dialog.tsx b/packages/ui/src/scheduled-task-form-dialog.tsx index dfaebc1acc..4d65f833cf 100644 --- a/packages/ui/src/scheduled-task-form-dialog.tsx +++ b/packages/ui/src/scheduled-task-form-dialog.tsx @@ -191,17 +191,25 @@ export function ScheduledTaskFormDialog(props: { schedule = { kind: 'calendar', recurrence, anchorAt: parsedRunAt }; } submitPendingRef.current = true; - const input = { + const baseInput = { title: title.trim(), intentBody: note.trim(), schedule, - effect, }; + // Preserve a pre-#3927 slug-only target without resubmitting it as a new + // effect; title, intent, and schedule remain editable. + const preservesLegacyEffect = + editingId !== null && + props.seed.lockedEffect?.kind === 'agent_run' && + !props.seed.lockedEffect.execution.llmConnectionId; setSubmitPending(true); try { const result = editingId - ? await props.onUpdate?.(editingId, input) - : await props.onCreate?.(input); + ? await props.onUpdate?.( + editingId, + preservesLegacyEffect ? baseInput : { ...baseInput, effect: effect! }, + ) + : await props.onCreate?.({ ...baseInput, effect: effect! }); if (result !== false && scheduledTaskMountedRef.current) { resetForm(); props.onOpenChange(false); From 2763d0581e6bce72597dc5bc205716d74a4fe774 Mon Sep 17 00:00:00 2001 From: testikun Date: Thu, 3 Sep 2026 09:54:51 +0800 Subject: [PATCH 2/2] fix(runtime-host): finish scheduled task identity review Generated-by: Codex --- ...cheduled-task-coordinator-recovery.test.ts | 29 ++++++--- .../src/server/scheduled-task-coordinator.ts | 65 +++++++------------ packages/storage/src/agent-run-store.ts | 7 +- .../ui/src/scheduled-task-form-dialog.tsx | 4 +- 4 files changed, 50 insertions(+), 55 deletions(-) diff --git a/packages/runtime-host/src/__tests__/scheduled-task-coordinator-recovery.test.ts b/packages/runtime-host/src/__tests__/scheduled-task-coordinator-recovery.test.ts index be87701f85..13b917e11d 100644 --- a/packages/runtime-host/src/__tests__/scheduled-task-coordinator-recovery.test.ts +++ b/packages/runtime-host/src/__tests__/scheduled-task-coordinator-recovery.test.ts @@ -190,13 +190,13 @@ test('ScheduledTask execution fails closed when the bound Connection identity is }); try { await coordinator.prepareRecovery(); - const rejectedCreate = await coordinator.handlers['scheduled-task.mutate']( + const created = await coordinator.handlers['scheduled-task.mutate']( { kind: 'create', input: { title: 'Rejected replacement target', intentBody: 'Must not persist an unresolvable Connection tuple.', - schedule: { kind: 'once', runAt: 2_000 }, + schedule: { kind: 'once', runAt: Date.now() + 60_000 }, effect: { kind: 'agent_run', execution: { @@ -213,14 +213,25 @@ test('ScheduledTask execution fails closed when the bound Connection identity is }, {} as never, ); - assert.deepEqual(rejectedCreate, { - ok: false, - error: { - code: 'operation_conflict', - message: 'ScheduledTask model connection identity changed', + assert.equal(created.ok, true); + if (!created.ok || created.result.kind !== 'task') return; + assert.equal(created.result.task.effect.kind, 'agent_run'); + assert.equal((await store.list()).length, 1); + + const updated = await coordinator.handlers['scheduled-task.mutate']( + { + kind: 'update', + taskId: created.result.task.id, + patch: { + title: 'Updated while temporarily unavailable', + effect: created.result.task.effect, + }, }, - }); - assert.equal((await store.list()).length, 0); + {} as never, + ); + assert.equal(updated.ok, true); + if (!updated.ok || updated.result.kind !== 'task') return; + assert.equal(updated.result.task.title, 'Updated while temporarily unavailable'); const task = await store.create( { diff --git a/packages/runtime-host/src/server/scheduled-task-coordinator.ts b/packages/runtime-host/src/server/scheduled-task-coordinator.ts index 8efe747bbe..2bbdbf7d51 100644 --- a/packages/runtime-host/src/server/scheduled-task-coordinator.ts +++ b/packages/runtime-host/src/server/scheduled-task-coordinator.ts @@ -17,7 +17,7 @@ * under the License. */ -import { createHash, randomUUID } from 'node:crypto'; +import { randomUUID } from 'node:crypto'; import { botDisplayLabel } from '@maka/core/bot-events'; import { isBotDeliveryProvider } from '@maka/core/bot-chat-settings'; import { messageContentsEqual } from '@maka/core/events'; @@ -26,7 +26,6 @@ import type { ConnectionCatalogEntry } from '@maka/core/runtime-policy'; import { type CreateScheduledTaskInput, type ScheduledTask, - type ScheduledTaskEffect, type ScheduledTaskExecutionTemplate, } from '@maka/core/scheduled-task'; import type { SessionHeader } from '@maka/core/session'; @@ -38,6 +37,7 @@ import { type ScheduledTaskToolAuthority, } from '@maka/runtime/scheduled-task-tools'; import { type MakaTool } from '@maka/runtime/tool-runtime'; +import { stableHash } from '@maka/runtime/request-shape'; import { type SessionManager } from '@maka/runtime/session-manager'; import { authenticateInteractiveScheduledTaskStoreWriter, @@ -114,13 +114,12 @@ export function scheduledTaskExecutionFingerprint( execution: ScheduledTaskExecutionTemplate, ): `sha256:${string}` | undefined { if (!execution.llmConnectionId) return undefined; - const identity = [ + return stableHash([ 'scheduled-task-agent-run.v1', execution.llmConnectionId, execution.llmConnectionSlug, execution.model, - ]; - return `sha256:${createHash('sha256').update(JSON.stringify(identity)).digest('hex')}`; + ]); } export interface HostScheduledTaskSessionRetirement { @@ -437,31 +436,25 @@ export class HostScheduledTaskCoordinator implements ScheduledTaskToolAuthority }), ); } - const task = await this.#commitTask( - 'updated', - () => { - if (input.kind === 'update') { - return this.#cancelWaitingNativeFireThen(input.taskId, () => - this.#store.update(input.taskId, input.patch, this.#now()), - ); - } - if (input.kind === 'pause') { - return this.#cancelWaitingNativeFireThen(input.taskId, () => - this.#store.pause(input.taskId, this.#now()), - ); - } - if (input.kind === 'resume') return this.#store.resume(input.taskId, this.#now()); - if (input.kind === 'snooze') { - return this.#cancelWaitingNativeFireThen(input.taskId, () => - this.#store.snooze(input.taskId, input.delayMs, this.#now()), - ); - } - return this.#store.clearRunHistory(input.taskId, this.#now()); - }, - input.kind === 'update' - ? () => this.#validateAgentRunEffect(input.patch.effect) - : undefined, - ); + const task = await this.#commitTask('updated', () => { + if (input.kind === 'update') { + return this.#cancelWaitingNativeFireThen(input.taskId, () => + this.#store.update(input.taskId, input.patch, this.#now()), + ); + } + if (input.kind === 'pause') { + return this.#cancelWaitingNativeFireThen(input.taskId, () => + this.#store.pause(input.taskId, this.#now()), + ); + } + if (input.kind === 'resume') return this.#store.resume(input.taskId, this.#now()); + if (input.kind === 'snooze') { + return this.#cancelWaitingNativeFireThen(input.taskId, () => + this.#store.snooze(input.taskId, input.delayMs, this.#now()), + ); + } + return this.#store.clearRunHistory(input.taskId, this.#now()); + }); return taskSuccess(task); } catch (error) { if (error instanceof ScheduledTaskStoreError) { @@ -501,7 +494,6 @@ export class HostScheduledTaskCoordinator implements ScheduledTaskToolAuthority 'ScheduledTask catalog limit reached', ); } - await this.#validateAgentRunEffect(input.effect); const task = await this.#store.create(input, this.#now()); this.#publish('created', task.id); await this.#refreshSchedule(); @@ -512,10 +504,8 @@ export class HostScheduledTaskCoordinator implements ScheduledTaskToolAuthority #commitTask( reason: ScheduledTaskChangedReason, mutate: () => Promise, - beforeMutate?: () => Promise, ): Promise { return this.#exclusive(async () => { - await beforeMutate?.(); const task = await mutate(); this.#publish(reason, task.id); await this.#refreshSchedule(); @@ -523,15 +513,6 @@ export class HostScheduledTaskCoordinator implements ScheduledTaskToolAuthority }); } - async #validateAgentRunEffect(effect: ScheduledTaskEffect | undefined): Promise { - if (effect?.kind !== 'agent_run') return; - try { - await this.#resolveAgentRunConnection(effect.execution); - } catch (error) { - throw new ScheduledTaskMutationError('operation_conflict', errorMessage(error)); - } - } - async #cancelWaitingNativeFireThen(taskId: string, operation: () => Promise): Promise { await this.#store.cancelWaitingNativeFire(taskId); return operation(); diff --git a/packages/storage/src/agent-run-store.ts b/packages/storage/src/agent-run-store.ts index 645d26676a..b10e782776 100644 --- a/packages/storage/src/agent-run-store.ts +++ b/packages/storage/src/agent-run-store.ts @@ -2093,9 +2093,12 @@ function normalizeRootExecutionDescriptor(value: unknown): RootExecutionDescript return Object.freeze({ kind: 'context_compact' }); } if (value.kind === 'scheduled_task') { - const keys = ['kind', 'scheduledTaskId', 'executionFingerprint']; if ( - !Object.keys(value).every((key) => keys.includes(key)) || + !hasExactKeys(value, [ + 'kind', + 'scheduledTaskId', + ...(Object.hasOwn(value, 'executionFingerprint') ? ['executionFingerprint'] : []), + ]) || typeof value.scheduledTaskId !== 'string' || !isSafeId(value.scheduledTaskId) || (value.executionFingerprint !== undefined && !isSha256Digest(value.executionFingerprint)) diff --git a/packages/ui/src/scheduled-task-form-dialog.tsx b/packages/ui/src/scheduled-task-form-dialog.tsx index 4d65f833cf..9f7bd624c6 100644 --- a/packages/ui/src/scheduled-task-form-dialog.tsx +++ b/packages/ui/src/scheduled-task-form-dialog.tsx @@ -207,9 +207,9 @@ export function ScheduledTaskFormDialog(props: { const result = editingId ? await props.onUpdate?.( editingId, - preservesLegacyEffect ? baseInput : { ...baseInput, effect: effect! }, + preservesLegacyEffect ? baseInput : { ...baseInput, effect }, ) - : await props.onCreate?.({ ...baseInput, effect: effect! }); + : await props.onCreate?.({ ...baseInput, effect }); if (result !== false && scheduledTaskMountedRef.current) { resetForm(); props.onOpenChange(false);