From b89213f4cbc6adcd88cf477bf7b32a443fd2c125 Mon Sep 17 00:00:00 2001 From: Matt Houghton <15250993+mattdevdba@users.noreply.github.com> Date: Tue, 27 Oct 2026 22:11:56 +0000 Subject: [PATCH 1/4] fix(v2-orchestrator): initialize workspaces through a durable callback init-ws ran as one synchronous step, so cloning a large repository was bounded by the orchestrator Lambda's 900-second timeout and the durable retry re-ran init-ws against a half-written checkout. New runs now dispatch an init-ws-start background job and suspend on a durable callback with a heartbeat, as runStage does. The callback has a 2-hour timeout and a 15-minute heartbeat timeout, and failures still surface as init_ws_failed. The mode is memoized with the run claim, so runs that checkpointed before this change replay the original init-ws step. A side-effect-free probe detects environment revisions built before init-ws-start and keeps the synchronous step for them. The intent API exposes orchestratorStartedAt so the intent page no longer flags a claimed run that is still initializing as stalled after 2 minutes. Fixes #637. --- docs/concepts/execution.md | 1 + frontend/src/pages/IntentView.test.tsx | 49 +++ frontend/src/pages/IntentView.tsx | 19 +- frontend/src/services/intents.ts | 3 + lambda/agentcore/commands/init-ws-start.js | 175 ++++++++++ lambda/agentcore/http-server.js | 16 + lambda/agentcore/test/init-ws-start.test.js | 283 ++++++++++++++++ lambda/intents/index.js | 4 + lambda/shared/agent-command-registry.js | 1 + lambda/v2-orchestrator/index.js | 154 +++++++-- .../test/orchestrator-replay.test.js | 162 ++++++++- .../v2-orchestrator/test/orchestrator.test.js | 319 +++++++++++++++--- .../test/section-integration.test.js | 19 +- .../test/skeleton-switch.test.js | 12 + 14 files changed, 1114 insertions(+), 103 deletions(-) create mode 100644 lambda/agentcore/commands/init-ws-start.js create mode 100644 lambda/agentcore/test/init-ws-start.test.js diff --git a/docs/concepts/execution.md b/docs/concepts/execution.md index 91af53121..e109fbe15 100644 --- a/docs/concepts/execution.md +++ b/docs/concepts/execution.md @@ -47,6 +47,7 @@ Agents run on **Amazon Bedrock AgentCore** — a serverless runtime that gives e - Each stage spawns a **headless agent CLI** (Kiro, Claude Code, OpenCode, or Codex — selected per project) inside the session. - The CLI's only interface to the platform is a stdio **MCP server**. Business reads and writes go to the graph as typed, provenance-stamped artifacts; questions, outputs, and metrics go to the process store and stream live to the browser. - Stages run as background jobs with durable callbacks, so a stage can run for hours while the orchestrator is suspended at zero compute. +- Workspace initialization (clone, intent-branch publish, graph anchor) runs the same way: the orchestrator dispatches `init-ws-start` and waits on a durable callback, so a large clone is not bounded by the orchestrator Lambda's 900-second timeout. The job heartbeats while it runs. The run fails with `init_ws_failed` when the job reports a failure, stops heartbeating for 15 minutes, or has not finished within 2 hours. The intent stays `CREATED` while it runs. Environment revisions built before `init-ws-start` existed, and runs already in flight at deployment, keep the synchronous `init-ws` call. - Session storage persists across park/resume, so a stage that asked a question continues its conversation with full context when you answer — even days later. The CLI authenticates to its model with a bearer token (Bedrock API key) or a Kiro API key configured in [Platform Admin → Agents](../using-the-platform/platform-settings.md#agents). The runtime's IAM role deliberately has no model-invocation permissions — token auth is the only path. diff --git a/frontend/src/pages/IntentView.test.tsx b/frontend/src/pages/IntentView.test.tsx index e2dca5032..3b6c03ad5 100644 --- a/frontend/src/pages/IntentView.test.tsx +++ b/frontend/src/pages/IntentView.test.tsx @@ -347,6 +347,55 @@ describe('IntentView', () => { expect(rewind).not.toHaveBeenCalled(); }); + it('does not call a claimed run stalled while its workspace is still initializing', async () => { + // An orchestrator claimed the run 10 minutes ago and is still cloning a + // large repository: CREATED is expected, and a restart would race the clone. + const tenMinutesAgo = new Date(Date.now() - 10 * 60_000).toISOString(); + get.mockResolvedValue( + baseDetail({ + status: 'CREATED', + updatedAt: tenMinutesAgo, + orchestratorStartedAt: tenMinutesAgo, + }), + ); + + renderAt(); + + expect((await screen.findAllByText('My intent')).length).toBeGreaterThan(0); + expect(screen.queryByText('Run stalled — never started')).toBeNull(); + expect(screen.queryByRole('button', { name: 'Restart workflow' })).toBeNull(); + }); + + it('calls an unclaimed CREATED hand-off stalled after two minutes', async () => { + get.mockResolvedValue( + baseDetail({ + status: 'CREATED', + updatedAt: new Date(Date.now() - 3 * 60_000).toISOString(), + orchestratorStartedAt: null, + }), + ); + + renderAt(); + + expect(await screen.findByText('Run stalled — never started')).toBeTruthy(); + expect(screen.getByRole('button', { name: 'Restart workflow' })).toBeTruthy(); + }); + + it('calls a claimed run stalled once the init callback window has passed', async () => { + const threeHoursAgo = new Date(Date.now() - 3 * 60 * 60_000).toISOString(); + get.mockResolvedValue( + baseDetail({ + status: 'CREATED', + updatedAt: threeHoursAgo, + orchestratorStartedAt: threeHoursAgo, + }), + ); + + renderAt(); + + expect(await screen.findByText('Run stalled — never started')).toBeTruthy(); + }); + it('renders one QuestionEditor for the active gate (exclusive expansion)', async () => { get.mockResolvedValue({ ...baseDetail({ status: 'WAITING', pendingHumanTaskId: 'h1' }), diff --git a/frontend/src/pages/IntentView.tsx b/frontend/src/pages/IntentView.tsx index 4ff6cb4f1..b0bdf0676 100644 --- a/frontend/src/pages/IntentView.tsx +++ b/frontend/src/pages/IntentView.tsx @@ -77,6 +77,13 @@ const CREDENTIAL_FAILURE_CODES = new Set([ 'credential_quota_exhausted', ]); const NON_EXPORTABLE_STATUSES = new Set(['DRAFT', 'CREATED']); +// A CREATED run no orchestrator has claimed should leave CREATED within +// seconds of the hand-off. +const UNCLAIMED_STALL_AFTER_MS = 2 * 60_000; +// A claimed run in CREATED is initializing its workspace. The orchestrator's +// init callback times out after 2 h (v2-orchestrator INIT_WS_CALLBACK_TIMEOUT) +// and fails the run; the slack covers that failure write. +const CLAIMED_INIT_STALL_AFTER_MS = 2 * 60 * 60_000 + 10 * 60_000; const EXPORT_HARNESSES: Array<{ value: NativeExportHarness; label: string }> = [ { value: 'claude', label: 'Claude' }, { value: 'codex', label: 'Codex' }, @@ -391,13 +398,19 @@ export default function IntentView() { const fanoutActive = isFanoutActive(detail); const reviewGate = reviewGateId ? gates.find((g) => g.humanTaskId === reviewGateId) : null; // Stalled detection: a CREATED run whose hand-off never reached a live - // orchestrator strands here (init-ws should flip it to RUNNING within - // seconds). After >2 min untouched, offer a restart instead of spinning. + // orchestrator strands here. After >2 min untouched, offer a restart instead + // of spinning. Once an orchestrator has claimed the run, CREATED means the + // workspace is still initializing — a large clone legitimately takes many + // minutes, and the orchestrator fails the run itself if the init job dies — + // so only flag it after the init callback window has clearly passed. const lastTouch = intent.updatedAt ?? intent.createdAt; + const stallAfterMs = intent.orchestratorStartedAt + ? CLAIMED_INIT_STALL_AFTER_MS + : UNCLAIMED_STALL_AFTER_MS; const isStalled = intent.status === 'CREATED' && !!lastTouch && - Date.now() - new Date(lastTouch).getTime() > 120_000; + Date.now() - new Date(lastTouch).getTime() > stallAfterMs; const isCredentialFailure = isFailed && Boolean(intent.failure?.code && CREDENTIAL_FAILURE_CODES.has(intent.failure.code)); const credentialSettingsPath = isCredentialFailure diff --git a/frontend/src/services/intents.ts b/frontend/src/services/intents.ts index bf14e8fb9..1d98c0ecd 100644 --- a/frontend/src/services/intents.ts +++ b/frontend/src/services/intents.ts @@ -167,6 +167,9 @@ export interface Intent { createdAt: string | null; updatedAt: string | null; completedAt: string | null; + // When an orchestrator run claimed this execution; null until claimed and + // after a relaunch resets it. Absent from older API responses. + orchestratorStartedAt?: string | null; } export interface IntentAttachment { diff --git a/lambda/agentcore/commands/init-ws-start.js b/lambda/agentcore/commands/init-ws-start.js new file mode 100644 index 000000000..2bf5ec2a9 --- /dev/null +++ b/lambda/agentcore/commands/init-ws-start.js @@ -0,0 +1,175 @@ +import { runCredentialJob } from '../credential-session.js'; +// init-ws-start — the async workspace initialization (the run-stage-start +// pattern applied to init-ws). +// +// The synchronous init-ws path holds the AgentCore request — and the +// orchestrator's durable step — open for the whole clone. A large repository +// (field incident: a 1.8 GB / 100k-commit GitLab repo) outlives the +// orchestrator Lambda's 900 s function timeout; the durable engine then +// retries init-ws against the half-written checkout. This command accepts the +// same payload as init-ws PLUS a `callbackId`, runs init-ws as a BACKGROUND +// job in the container, and returns in milliseconds. The orchestrator suspends +// on the durable callback at zero compute; the job completes it when init-ws +// returns. +// +// Reliability contract (mirrors run-stage-start): +// - the accept response only says "job started"; the init verdict always +// travels through the callback, in init-ws's own {ok, reason, detail} +// shape — the orchestrator's existing decode applies unchanged; +// - EVERY termination of the job sends a callback: success, init-ws's +// ok:false failures, and uncaught throws (init_ws_job_crashed); +// - the first heartbeat is sent BEFORE the clone starts, so a callback the +// container cannot reach fails the dispatch synchronously; +// - heartbeats fire while the clone runs so the orchestrator's +// heartbeatTimeout separates "large repository" from "dead container"; +// - one init job per execution: a duplicate start for the SAME callback is +// an idempotent accept (durable dispatch retry), a different callback is +// refused (init_job_already_running) — two clones must never race into +// the same workspace directory; +// - the busy tracker is held for the job's lifetime so /ping reports +// HealthyBusy and AgentCore keeps the session (and its mount) alive. + +import { Logger } from '@aws-lambda-powertools/logger'; + +const logger = new Logger({ + persistentKeys: { component: 'agentcore', module: 'init-ws-start' }, +}); + +const jobKey = (p) => `${p.executionId}:init-ws`; + +// init-ws returns {ok:false, reason, detail} on every handled failure and +// {ok:true, ...} on success. Anything else is a contract break — surface it as +// a failure rather than letting the orchestrator mark the workspace ready. +export const normalizeInitResult = (result) => { + if (result && typeof result === 'object' && typeof result.ok === 'boolean') return result; + return { ok: false, reason: 'init_ws_no_result', detail: String(result) }; +}; + +export const createInitWsStart = ({ + initWs, + sendCallbackSuccess, + sendCallbackHeartbeat, + busy = null, + heartbeatIntervalMs = 60_000, + activeJobs = new Map(), + log = logger, +}) => { + const start = async (payload) => { + const { callbackId, executionId } = payload ?? {}; + if (!callbackId) return { ok: false, reason: 'missing_callback_id' }; + if (!executionId) return { ok: false, reason: 'missing_execution_id' }; + + const key = jobKey(payload); + const inFlight = activeJobs.get(key); + if (inFlight) { + if (inFlight.callbackId === callbackId) { + return { ok: true, accepted: true, alreadyRunning: true, callbackId }; + } + return { ok: false, reason: 'init_job_already_running', detail: key }; + } + + let initialHeartbeat; + try { + initialHeartbeat = await sendCallbackHeartbeat(callbackId); + } catch (err) { + initialHeartbeat = { delivered: false, error: err?.message ?? String(err) }; + } + if (!initialHeartbeat?.delivered) { + log.error('init-ws callback heartbeat could not be established', { + executionId, + error: initialHeartbeat?.error ?? null, + }); + return { + ok: false, + reason: 'init_ws_callback_heartbeat_failed', + detail: initialHeartbeat?.error ?? null, + }; + } + + const jobState = { startedAt: Date.now(), callbackId, heartbeatTimer: null }; + activeJobs.set(key, jobState); + busy?.enter(); + log.info('init-ws job accepted', { executionId }); + + // The job payload is init-ws's own argument bag; the callback id is + // transport, not an init-ws input. + const initPayload = { ...payload }; + delete initPayload.callbackId; + delete initPayload.command; + + const job = runCredentialJob(async () => { + try { + const scheduleHeartbeat = () => { + jobState.heartbeatTimer = setTimeout(async () => { + let sent; + try { + sent = await sendCallbackHeartbeat(callbackId); + } catch (err) { + sent = { delivered: false, error: err?.message ?? String(err) }; + } + if (!sent?.delivered) { + log.error('init-ws callback heartbeat failed', { + executionId, + error: sent?.error ?? null, + }); + } + if (activeJobs.get(key) === jobState) scheduleHeartbeat(); + }, heartbeatIntervalMs); + }; + scheduleHeartbeat(); + + let result; + try { + result = normalizeInitResult(await initWs(initPayload)); + } catch (err) { + result = { + ok: false, + reason: 'init_ws_job_crashed', + detail: err?.message ?? String(err), + }; + } + // The synchronous path logs failures in dispatchInvocation; a + // background job bypasses it, so keep the same diagnosable record. + if (result.ok === false) { + log.warn('command returned failure', { + command: 'init-ws-start', + executionId, + reason: result.reason, + detail: result.detail, + }); + } else { + log.info('init-ws job completed', { + executionId, + durationMs: Date.now() - jobState.startedAt, + }); + } + + const sent = await sendCallbackSuccess(callbackId, result); + if (!sent?.delivered) { + // The orchestrator's callback timeout / heartbeatTimeout is the + // backstop; log loudly so an operator can complete it manually. + log.error('init-ws callback could not be delivered', { + executionId, + error: sent?.error ?? null, + }); + } + return result; + } finally { + if (jobState.heartbeatTimer) clearTimeout(jobState.heartbeatTimer); + activeJobs.delete(key); + busy?.leave(); + } + }); + job.catch((err) => + log.error('init-ws job promise rejected unexpectedly', { + executionId, + error: err?.message ?? String(err), + }), + ); + + return { ok: true, accepted: true, callbackId, jobKey: key }; + }; + + start.activeJobs = activeJobs; + return start; +}; diff --git a/lambda/agentcore/http-server.js b/lambda/agentcore/http-server.js index b73e63241..41e339168 100644 --- a/lambda/agentcore/http-server.js +++ b/lambda/agentcore/http-server.js @@ -17,6 +17,10 @@ import { // // Invocation payloads: // { "command": "init-ws", ...initWs args } +// { "command": "init-ws-start", ...initWs args, callbackId } +// → accepts in ms, runs init-ws as a background job (clone + branch +// publish can outlive the orchestrator Lambda), completes the +// orchestrator's durable callback with init-ws's {ok, reason, detail}. // { "command": "run-stage", ...runStage args } // { "command": "run-stage-start", ...runStage args, stageCallbackId } // → accepts in ms, runs the stage as a background job, completes the @@ -235,6 +239,7 @@ const main = async () => { const { initWs } = await import('./commands/init-ws.js'); const { runStage } = await import('./commands/run-stage.js'); const { createRunStageStart } = await import('./commands/run-stage-start.js'); + const { createInitWsStart } = await import('./commands/init-ws-start.js'); const { createDiscussionAssistStart } = await import('./commands/discussion-assist-start.js'); const { createComposePlanStart } = await import('./commands/compose-plan-start.js'); const { createQuorumEditPlanStart } = await import('./commands/quorum-edit-plan-start.js'); @@ -399,6 +404,17 @@ const main = async () => { busy, activeJobs: stageJobs, })(p); + // Async workspace initialization: init-ws as a background job so a large + // clone is not bounded by the orchestrator Lambda's function timeout. + const initJobs = new Map(); + handlers.initWsStart = (p) => + createInitWsStart({ + initWs: (q) => handlers.initWs(q), + sendCallbackSuccess: sendStageCallbackSuccess, + sendCallbackHeartbeat: sendStageCallbackHeartbeat, + busy, + activeJobs: initJobs, + })(p); const discussionJobs = new Map(); handlers.discussionAssistStart = (p, context) => createDiscussionAssistStart({ diff --git a/lambda/agentcore/test/init-ws-start.test.js b/lambda/agentcore/test/init-ws-start.test.js new file mode 100644 index 000000000..3623f6c2a --- /dev/null +++ b/lambda/agentcore/test/init-ws-start.test.js @@ -0,0 +1,283 @@ +import { describe, it, expect, vi } from 'vitest'; +import { createInitWsStart, normalizeInitResult } from '../commands/init-ws-start.js'; +import { createBusyTracker, dispatchInvocation } from '../http-server.js'; + +// Async workspace initialization: init-ws as a background job so a large clone +// is not bounded by the orchestrator Lambda. The command must accept fast, +// ALWAYS complete the durable callback with init-ws's {ok, reason, detail}, +// heartbeat while the clone runs, hold the busy tracker for the job's +// lifetime, and never let two clones race into one workspace. + +const flush = () => new Promise((r) => setTimeout(r, 0)); + +const makeGate = () => { + let resolve, reject; + const promise = new Promise((res, rej) => { + resolve = res; + reject = rej; + }); + return { promise, resolve, reject }; +}; + +const basePayload = { + command: 'init-ws-start', + projectId: 'p1', + intentId: 'i1', + executionId: 'e1', + repos: ['acme/api'], + branch: 'aidlc/i1', + orchestratorRunId: 'run-1', + callbackId: 'cb-init', +}; + +const quietLog = { info: () => {}, warn: vi.fn(), error: vi.fn() }; + +const makeDeps = (over = {}) => { + const sent = []; + const beats = []; + return { + sent, + beats, + deps: { + initWs: vi.fn(async () => ({ ok: true, intentId: 'i1', executionId: 'e1', repos: [] })), + sendCallbackSuccess: vi.fn(async (callbackId, result) => { + sent.push({ callbackId, result }); + return { delivered: true }; + }), + sendCallbackHeartbeat: vi.fn(async (callbackId) => { + beats.push(callbackId); + return { delivered: true }; + }), + log: { ...quietLog, warn: vi.fn(), error: vi.fn() }, + ...over, + }, + }; +}; + +describe('normalizeInitResult', () => { + it("passes through init-ws's own success and failure shapes", () => { + expect(normalizeInitResult({ ok: true, repos: [] })).toEqual({ ok: true, repos: [] }); + expect(normalizeInitResult({ ok: false, reason: 'checkout_failed' })).toEqual({ + ok: false, + reason: 'checkout_failed', + }); + }); + + it('turns a result without an ok flag into a failure, never a ready workspace', () => { + expect(normalizeInitResult(undefined)).toMatchObject({ + ok: false, + reason: 'init_ws_no_result', + }); + expect(normalizeInitResult({ repos: [] })).toMatchObject({ ok: false }); + }); +}); + +describe('createInitWsStart', () => { + it('refuses the capability probe (no callbackId) without starting a job', async () => { + const { deps } = makeDeps(); + const start = createInitWsStart(deps); + const res = await start({ command: 'init-ws-start', executionId: 'e1' }); + expect(res).toEqual({ ok: false, reason: 'missing_callback_id' }); + expect(deps.initWs).not.toHaveBeenCalled(); + expect(deps.sendCallbackHeartbeat).not.toHaveBeenCalled(); + }); + + it('rejects a payload without executionId', async () => { + const { deps } = makeDeps(); + const res = await createInitWsStart(deps)({ callbackId: 'cb-1' }); + expect(res).toEqual({ ok: false, reason: 'missing_execution_id' }); + expect(deps.initWs).not.toHaveBeenCalled(); + }); + + it('accepts while the clone is still running, then completes the callback with the init verdict', async () => { + const gate = makeGate(); + const { deps, sent } = makeDeps({ initWs: vi.fn(() => gate.promise) }); + const start = createInitWsStart(deps); + + const res = await start(basePayload); + expect(res).toMatchObject({ ok: true, accepted: true, callbackId: 'cb-init' }); + expect(sent).toHaveLength(0); + expect(start.activeJobs.size).toBe(1); + + gate.resolve({ ok: true, intentId: 'i1', executionId: 'e1', repos: ['acme/api'] }); + await flush(); + + expect(sent).toEqual([ + { + callbackId: 'cb-init', + result: { ok: true, intentId: 'i1', executionId: 'e1', repos: ['acme/api'] }, + }, + ]); + expect(start.activeJobs.size).toBe(0); + }); + + it('hands init-ws its own arguments, without the transport fields', async () => { + const { deps } = makeDeps(); + await createInitWsStart(deps)(basePayload); + await flush(); + const arg = deps.initWs.mock.calls[0][0]; + expect(arg).toMatchObject({ executionId: 'e1', branch: 'aidlc/i1', repos: ['acme/api'] }); + expect(arg).not.toHaveProperty('callbackId'); + expect(arg).not.toHaveProperty('command'); + }); + + it('delivers an init-ws failure through the callback and logs its detail', async () => { + const failure = { + ok: false, + reason: 'intent_branch_push_failed', + detail: 'acme/api (base master: push_failed)', + }; + const { deps, sent } = makeDeps({ initWs: vi.fn(async () => failure) }); + await createInitWsStart(deps)(basePayload); + await flush(); + expect(sent).toEqual([{ callbackId: 'cb-init', result: failure }]); + // The synchronous path logs failures in dispatchInvocation; the background + // job must keep that diagnosable record. + expect(deps.log.warn).toHaveBeenCalledWith( + 'command returned failure', + expect.objectContaining({ + reason: 'intent_branch_push_failed', + detail: 'acme/api (base master: push_failed)', + }), + ); + }); + + it('completes the callback even when init-ws throws', async () => { + const { deps, sent } = makeDeps({ + initWs: vi.fn(async () => { + throw new Error('ENOSPC: no space left on device'); + }), + }); + const start = createInitWsStart(deps); + await start(basePayload); + await flush(); + expect(sent).toEqual([ + { + callbackId: 'cb-init', + result: { + ok: false, + reason: 'init_ws_job_crashed', + detail: 'ENOSPC: no space left on device', + }, + }, + ]); + expect(start.activeJobs.size).toBe(0); + }); + + it('establishes the heartbeat before cloning and fails the dispatch when it cannot', async () => { + const { deps } = makeDeps({ + sendCallbackHeartbeat: vi.fn(async () => ({ delivered: false, error: 'AccessDenied' })), + }); + const start = createInitWsStart(deps); + const res = await start(basePayload); + expect(res).toEqual({ + ok: false, + reason: 'init_ws_callback_heartbeat_failed', + detail: 'AccessDenied', + }); + expect(deps.initWs).not.toHaveBeenCalled(); + expect(start.activeJobs.size).toBe(0); + }); + + it('keeps heartbeating while a long clone runs', async () => { + vi.useFakeTimers(); + try { + const gate = makeGate(); + const { deps, beats } = makeDeps({ initWs: vi.fn(() => gate.promise) }); + const start = createInitWsStart({ ...deps, heartbeatIntervalMs: 1_000 }); + await start(basePayload); + expect(beats).toEqual(['cb-init']); // the synchronous first beat + await vi.advanceTimersByTimeAsync(3_500); + expect(beats.length).toBe(4); + gate.resolve({ ok: true }); + await vi.advanceTimersByTimeAsync(0); + await vi.advanceTimersByTimeAsync(5_000); + // Heartbeats stop once the job completes. + expect(beats.length).toBe(4); + } finally { + vi.useRealTimers(); + } + }); + + it('acknowledges a duplicate dispatch of the SAME init attempt without a second clone', async () => { + const gate = makeGate(); + const { deps } = makeDeps({ initWs: vi.fn(() => gate.promise) }); + const start = createInitWsStart(deps); + await start(basePayload); + const again = await start(basePayload); + expect(again).toMatchObject({ ok: true, accepted: true, alreadyRunning: true }); + expect(deps.initWs).toHaveBeenCalledTimes(1); + gate.resolve({ ok: true }); + await flush(); + }); + + it('refuses a DIFFERENT init attempt while a clone is still writing the workspace', async () => { + const gate = makeGate(); + const { deps } = makeDeps({ initWs: vi.fn(() => gate.promise) }); + const start = createInitWsStart(deps); + await start(basePayload); + const relaunch = await start({ ...basePayload, callbackId: 'cb-relaunch' }); + expect(relaunch).toEqual({ + ok: false, + reason: 'init_job_already_running', + detail: 'e1:init-ws', + }); + expect(deps.initWs).toHaveBeenCalledTimes(1); + gate.resolve({ ok: true }); + await flush(); + // Once the first job finishes, a new attempt is accepted again. + const later = await start({ ...basePayload, callbackId: 'cb-relaunch' }); + expect(later).toMatchObject({ ok: true, accepted: true }); + await flush(); + }); + + it('holds the busy tracker for the whole job so the session stays HealthyBusy', async () => { + const gate = makeGate(); + const busy = createBusyTracker(); + const { deps } = makeDeps({ initWs: vi.fn(() => gate.promise) }); + await createInitWsStart({ ...deps, busy })(basePayload); + expect(busy.status).toBe('HealthyBusy'); + gate.resolve({ ok: true }); + await flush(); + expect(busy.status).toBe('Healthy'); + }); +}); + +describe('init-ws-start through dispatchInvocation', () => { + it('is routed by the command registry and returns the accept on HTTP 200', async () => { + const { deps } = makeDeps(); + const start = createInitWsStart(deps); + const res = await dispatchInvocation({ + payload: basePayload, + handlers: { initWsStart: start }, + busy: createBusyTracker(), + now: () => 'T', + }); + expect(res.statusCode).toBe(200); + expect(res.body).toMatchObject({ ok: true, accepted: true, command: 'init-ws-start' }); + await flush(); + }); + + it('answers the probe on HTTP 200, so the orchestrator can tell it from an older image', async () => { + const { deps } = makeDeps(); + const res = await dispatchInvocation({ + payload: { command: 'init-ws-start', executionId: 'e1' }, + handlers: { initWsStart: createInitWsStart(deps) }, + busy: createBusyTracker(), + now: () => 'T', + }); + expect(res).toEqual({ + statusCode: 200, + body: { ok: false, reason: 'missing_callback_id', command: 'init-ws-start', at: 'T' }, + }); + }); + + it('an image without the handler answers 400 unknown command (the older-image signal)', async () => { + const res = await dispatchInvocation({ + payload: { command: 'init-ws-start', executionId: 'e1' }, + handlers: {}, + busy: createBusyTracker(), + }); + expect(res).toEqual({ statusCode: 400, body: { error: 'unknown command "init-ws-start"' } }); + }); +}); diff --git a/lambda/intents/index.js b/lambda/intents/index.js index 50594eb1d..d59999b04 100644 --- a/lambda/intents/index.js +++ b/lambda/intents/index.js @@ -1421,6 +1421,10 @@ const mapIntent = (meta) => ({ createdAt: meta.startedAt ?? null, updatedAt: meta.updatedAt ?? null, completedAt: meta.completedAt ?? null, + // Set when an orchestrator run claims the execution (cleared on relaunch). + // Lets the client tell a long workspace initialization from a hand-off that + // never reached an orchestrator. + orchestratorStartedAt: meta.orchestratorStartedAt ?? null, }); const NATIVE_EXPORT_HARNESSES = new Set(['claude', 'codex', 'kiro', 'kiro-ide', 'opencode']); diff --git a/lambda/shared/agent-command-registry.js b/lambda/shared/agent-command-registry.js index 38fd19edd..c8a573bf1 100644 --- a/lambda/shared/agent-command-registry.js +++ b/lambda/shared/agent-command-registry.js @@ -15,6 +15,7 @@ const command = (handler, agentAuth = false) => Object.freeze({ handler, agentAu export const COMMANDS = Object.freeze({ 'init-ws': command('initWs'), + 'init-ws-start': command('initWsStart'), 'run-stage': command('runStage', AGENT_AUTH_MODES.EXECUTION), 'run-stage-start': command('runStageStart', AGENT_AUTH_MODES.EXECUTION), 'promote-units': command('promoteUnits'), diff --git a/lambda/v2-orchestrator/index.js b/lambda/v2-orchestrator/index.js index df4122511..4e0e7e685 100644 --- a/lambda/v2-orchestrator/index.js +++ b/lambda/v2-orchestrator/index.js @@ -573,6 +573,10 @@ const handler = async (event, ctx, deps = defaultDeps()) => { return { runId: updated?.orchestratorRunId ?? token, orchestratorExpiresAt: updated?.orchestratorExpiresAt ?? expiresAt, + // Replay gate for the workspace initialization shape (see + // initWorkspace). Memoized with the claim, so a run that checkpointed + // before this field existed replays the legacy `init-ws` step. + initWsMode: INIT_WS_MODE_CALLBACK, }; }); runId = claim.runId; @@ -624,31 +628,30 @@ const handler = async (event, ctx, deps = defaultDeps()) => { 'v2.workspace.initializing', `Initializing workspace (${(meta.repos ?? []).length} repo(s))…`, ); - const initResult = await ctx.step('init-ws', () => - invokeIntentRuntime( - { - command: 'init-ws', - projectId, - intentId, - executionId, - repos: meta.repos ?? [], - branch: meta.branch, - baseBranch: meta.baseBranch, - baseBranches: meta.baseBranches, - gitProvider, - repoProviders: meta.repoProviders ?? null, - ...(gitAuthor ? { gitAuthor } : {}), - title: meta.title, - prompt: meta.prompt, - workflowId, - workflowVersion, - scope, - startedBy: meta.startedBy, - attachments: meta.attachments ?? [], - }, - sessionId, - ), - ); + const initResult = await initWorkspace(ctx, invokeIntentRuntime, { + mode: claim.initWsMode ?? null, + sessionId, + orchestratorRunId: runId, + initPayload: { + projectId, + intentId, + executionId, + repos: meta.repos ?? [], + branch: meta.branch, + baseBranch: meta.baseBranch, + baseBranches: meta.baseBranches, + gitProvider, + repoProviders: meta.repoProviders ?? null, + ...(gitAuthor ? { gitAuthor } : {}), + title: meta.title, + prompt: meta.prompt, + workflowId, + workflowVersion, + scope, + startedBy: meta.startedBy, + attachments: meta.attachments ?? [], + }, + }); // The runtime returns {ok:false, reason} on checkout / vertex / seed failure; // it does NOT throw. Detect it here or the run would march on into stages // against an un-initialized workspace. @@ -2618,6 +2621,107 @@ const handler = async (event, ctx, deps = defaultDeps()) => { const STAGE_CALLBACK_TIMEOUT = { hours: 8 }; const STAGE_CALLBACK_HEARTBEAT_TIMEOUT = { minutes: 15 }; +// Workspace initialization (clone, intent-branch publish, Intent anchor, state +// seed). The synchronous `init-ws` step holds the whole clone inside one +// orchestrator Lambda invocation, so a large repository outlives the 900 s +// function timeout and the durable retry lands on a half-written checkout. +// Runs whose claim carries INIT_WS_MODE_CALLBACK instead dispatch +// `init-ws-start` and suspend on a durable callback, mirroring runStage. +// +// Replay compatibility: the mode rides the memoized `mint-run-id` result, so a +// run that checkpointed under the old code replays exactly `init-ws`; only new +// runs use the probe/callback operation names. +// +// Older environment images: a published environment revision pins the +// AgentCore container it was built from, and one built before init-ws-start +// answers `unknown command`. The side-effect-free probe (no callbackId, so a +// current container refuses with missing_callback_id) detects that BEFORE a +// callback exists, and such runs keep the synchronous step. +// +// Every outcome is returned in init-ws's own {ok, reason, detail} shape so the +// caller's single decode path is unchanged. +export const INIT_WS_MODE_CALLBACK = 'callback'; +// Bounds a clone that is still heartbeating but never finishes; generous +// because a repository's size is not known up front. +const INIT_WS_CALLBACK_TIMEOUT = { hours: 2 }; +const INIT_WS_CALLBACK_HEARTBEAT_TIMEOUT = { minutes: 15 }; + +// The container answers an unregistered command with HTTP 400, which AgentCore +// raises as RuntimeClientError ("Received error (400) from runtime") rather +// than returning the body. The probe sends a command and valid JSON, so a 400 +// from the runtime can only mean the command is unknown. A body-shaped +// `unknown command` error is accepted too (direct/local transports). +const isUnknownCommandResponse = (res) => + typeof res?.error === 'string' && res.error.startsWith('unknown command'); +const isUnknownCommandError = (err) => + err?.name === 'RuntimeClientError' && /\(400\)/.test(err?.message ?? ''); + +const initWorkspace = async ( + ctx, + invokeRuntime, + { mode, sessionId, orchestratorRunId, initPayload }, +) => { + const runSynchronously = () => + ctx.step('init-ws', () => invokeRuntime({ command: 'init-ws', ...initPayload }, sessionId)); + if (mode !== INIT_WS_MODE_CALLBACK) return runSynchronously(); + + const support = await ctx.step('init-ws-probe', async () => { + try { + const res = await invokeRuntime( + { command: 'init-ws-start', executionId: initPayload.executionId }, + sessionId, + ); + return { callback: !isUnknownCommandResponse(res) }; + } catch (err) { + if (isUnknownCommandError(err)) return { callback: false }; + // Transport failures (cold-start timeout, throttling) keep the durable + // step's retry behavior, exactly as the synchronous init-ws step does. + throw err; + } + }); + if (!support?.callback) return runSynchronously(); + + const [initDone, callbackId] = await ctx.createCallback('init-ws-cb', { + timeout: INIT_WS_CALLBACK_TIMEOUT, + heartbeatTimeout: INIT_WS_CALLBACK_HEARTBEAT_TIMEOUT, + }); + const dispatch = await ctx.step('init-ws-dispatch', () => + invokeRuntime( + { command: 'init-ws-start', ...initPayload, orchestratorRunId, callbackId }, + sessionId, + ), + ); + // The accept only says "job started"; a refusal (a concurrent init job, + // heartbeat channel unavailable) fails initialization here. + if (!dispatch || dispatch.ok === false || dispatch.error) { + return { + ok: false, + reason: dispatch?.reason ?? 'init_ws_dispatch_failed', + detail: dispatch?.detail ?? dispatch?.error ?? null, + }; + } + try { + // Callback serdes is pass-through: the container's body arrives as the raw + // JSON string (see runStage). + const raw = await initDone; + const result = typeof raw === 'string' ? JSON.parse(raw) : raw; + if (!result || typeof result !== 'object') { + return { ok: false, reason: 'init_ws_bad_callback_result' }; + } + return result; + } catch (err) { + if (err instanceof SyntaxError) { + return { ok: false, reason: 'init_ws_bad_callback_result', detail: err.message }; + } + // Callback timeout / heartbeat expiry (dead container) / explicit failure. + return { + ok: false, + reason: 'init_ws_callback_failed', + detail: err?.message ?? String(err), + }; + } +}; + const runStage = async ( ctx, invokeRuntime, diff --git a/lambda/v2-orchestrator/test/orchestrator-replay.test.js b/lambda/v2-orchestrator/test/orchestrator-replay.test.js index daf42f0c1..96e4dfeb8 100644 --- a/lambda/v2-orchestrator/test/orchestrator-replay.test.js +++ b/lambda/v2-orchestrator/test/orchestrator-replay.test.js @@ -38,6 +38,27 @@ beforeAll(async () => { await LocalDurableTestRunner.setupTestEnvironment(); }); +// The runner the current test drives. The fake container needs it to complete +// the workspace-init callback out-of-band, as the real init-ws-start job does. +let currentRunner = null; +const trackedRunner = (options) => { + currentRunner = new LocalDurableTestRunner(options); + return currentRunner; +}; + +// The container side of init-ws-start: the side-effect-free capability probe +// (no callbackId) is refused with missing_callback_id; a dispatch is accepted +// and its background job completes the `init-ws-cb` callback with `verdict`. +const acceptInitWs = (payload, verdict = { ok: true }) => { + if (!payload.callbackId) return { ok: false, reason: 'missing_callback_id' }; + const runner = currentRunner; + (async () => { + const op = await runner.getOperation('init-ws-cb').waitForData(WaitingOperationStatus.STARTED); + await op.sendCallbackSuccess(JSON.stringify(verdict)); + })(); + return { ok: true, accepted: true, callbackId: payload.callbackId }; +}; + afterAll(async () => { await LocalDurableTestRunner.teardownTestEnvironment(); }); @@ -74,7 +95,7 @@ it('resumes deployed engine-gate checkpoints with the recovery implementation', const suspension = new Promise((resolve) => { suspended = resolve; }); - const runner = new LocalDurableTestRunner({ + const runner = trackedRunner({ handlerFunction: async (...input) => { const result = await (upgraded ? newHandler : oldHandler)(...input); if (result.Status === 'PENDING') { @@ -133,7 +154,7 @@ it('completes an engine gate on a racing saved answer without a callback deliver { name: 'halt-s1-r1', prompt: 'Retry?' }, ), ); - const runner = new LocalDurableTestRunner({ handlerFunction: handler }); + const runner = trackedRunner({ handlerFunction: handler }); const execution = await runner.run({ payload: {} }); expect(execution.getResult()).toMatchObject({ gate: { status: 'answered', answer: { decision: 'retry' } }, @@ -200,7 +221,7 @@ it('replays an unpark whose META write committed but whose step result was lost' { name: 'replay', prompt: 'Continue?' }, ), ); - const runner = new LocalDurableTestRunner({ handlerFunction: handler }); + const runner = trackedRunner({ handlerFunction: handler }); expect((await runner.run({ payload: {} })).getResult()).toMatchObject({ gate: { status: 'answered' }, }); @@ -276,7 +297,7 @@ const makeWorld = ({ stages = [{ stageId: 'a' }, { stageId: 'b' }] } = {}) => { loadPlan: async () => ({ valid: true, plan: { stages } }), invokeRuntime: async (payload) => { world.invokes.push(payload); - if (payload.command === 'init-ws') return { ok: true }; + if (payload.command === 'init-ws-start') return acceptInitWs(payload); // ACCEPT only — the verdict arrives via the stage callback, later. return { ok: true, accepted: true, stageId: payload.stageId }; }, @@ -469,7 +490,7 @@ describe('orchestrator on the real durable runner (replay semantics)', () => { }); return __durableHandler(event, wrapped, world.deps); }); - const runner = new LocalDurableTestRunner({ handlerFunction: handler }); + const runner = trackedRunner({ handlerFunction: handler }); const completion = runner.run({ payload: { action: 'start', intentId: 'i1', executionId: 'i1' }, }); @@ -544,7 +565,7 @@ describe('orchestrator on the real durable runner (replay semantics)', () => { return __durableHandler(event, wrapped, world.deps); }); const current = withDurableExecution((event, ctx) => __durableHandler(event, ctx, world.deps)); - const runner = new LocalDurableTestRunner({ + const runner = trackedRunner({ handlerFunction: async (...input) => { const result = await (upgraded ? current : legacy)(...input); if (result.Status === 'PENDING') suspended(); @@ -574,10 +595,115 @@ describe('orchestrator on the real durable runner (replay semantics)', () => { expect(world.statusWrites).not.toContain('FAILED'); }); + it('replays a run that checkpointed the synchronous init-ws step under the previous code', async () => { + // In-flight executions at deploy time hold a memoized claim WITHOUT + // initWsMode and a checkpointed `init-ws` step. Record that history, then + // finish the run on the unmodified handler: the durable SDK must see the + // same operations in the same order (no NonDeterministicExecutionError), + // and init-ws must not run again. + const world = makeWorld({ stages: [{ stageId: 'a', stageInstanceId: 'si-a' }] }); + let upgraded = false; + let suspended; + const suspension = new Promise((resolve) => { + suspended = resolve; + }); + const legacy = withDurableExecution((event, ctx) => { + const wrapped = new Proxy(ctx, { + get: (target, key) => + key === 'step' + ? (name, fn, opts) => + name === 'mint-run-id' + ? target.step( + name, + async () => { + const { initWsMode: _absent, ...claim } = await fn(); + return claim; + }, + opts, + ) + : target.step(name, fn, opts) + : typeof target[key] === 'function' + ? target[key].bind(target) + : target[key], + }); + return __durableHandler(event, wrapped, world.deps); + }); + const current = withDurableExecution((event, ctx) => __durableHandler(event, ctx, world.deps)); + const runner = trackedRunner({ + handlerFunction: async (...input) => { + const result = await (upgraded ? current : legacy)(...input); + if (result.Status === 'PENDING') suspended(); + return result; + }, + }); + const completion = runner.run({ + payload: { action: 'start', intentId: 'i1', executionId: 'i1' }, + }); + await suspension; + upgraded = true; + await completeStage(runner, 'stage-cb-a', { ok: true, state: 'SUCCEEDED' }); + const execution = await completion; + expect(execution.getResult()).toEqual({ ok: true, intentId: 'i1', stages: 1 }); + const names = execution.getOperations().map((op) => op.getOperationData()?.Name); + expect(names).toContain('init-ws'); + expect(names).not.toContain('init-ws-probe'); + expect(names).not.toContain('init-ws-cb'); + expect(world.invokes.map((p) => p.command)).toEqual([ + 'init-ws', + 'create-workflow-checkpoint', + 'run-stage-start', + 'create-workflow-checkpoint', + ]); + }); + + it('records the probe and the init callback as durable operations for a new run', async () => { + const world = makeWorld({ stages: [{ stageId: 'a', stageInstanceId: 'si-a' }] }); + const handler = withDurableExecution((event, ctx) => __durableHandler(event, ctx, world.deps)); + const runner = trackedRunner({ handlerFunction: handler }); + const completion = runner.run({ + payload: { action: 'start', intentId: 'i1', executionId: 'i1' }, + }); + await completeStage(runner, 'stage-cb-a', { ok: true, state: 'SUCCEEDED' }); + const execution = await completion; + expect(execution.getResult()).toEqual({ ok: true, intentId: 'i1', stages: 1 }); + const names = execution.getOperations().map((op) => op.getOperationData()?.Name); + const order = ['init-ws-start', 'init-ws-probe', 'init-ws-cb', 'init-ws-dispatch'].map((n) => + names.indexOf(n), + ); + expect(order.every((i) => i >= 0)).toBe(true); + expect(order).toEqual(order.toSorted((a, b) => a - b)); + expect(names).not.toContain('init-ws'); + // The dispatch carried the REAL durable callback id the job completed. + const dispatch = world.invokes.find((p) => p.command === 'init-ws-start' && p.callbackId); + const initOp = runner.getOperation('init-ws-cb'); + expect(dispatch.callbackId).toBe(initOp.getCallbackDetails().callbackId); + expect(dispatch.orchestratorRunId).toMatch(/^run-/); + }); + + it('fails the run with the init-ws reason delivered through the init callback', async () => { + const world = makeWorld({ stages: [{ stageId: 'a', stageInstanceId: 'si-a' }] }); + world.deps.invokeRuntime = async (payload) => { + world.invokes.push(payload); + return acceptInitWs(payload, { + ok: false, + reason: 'checkout_failed', + detail: 'clone failed for owner/repo', + }); + }; + const handler = withDurableExecution((event, ctx) => __durableHandler(event, ctx, world.deps)); + trackedRunner({ handlerFunction: handler }); + const execution = await currentRunner.run({ + payload: { action: 'start', intentId: 'i1', executionId: 'i1' }, + }); + expect(execution.getResult()).toMatchObject({ ok: false, reason: 'init_ws_failed' }); + expect(world.statusWrites.filter((s) => s === 'FAILED')).toHaveLength(1); + expect(world.invokes.map((p) => p.command)).toEqual(['init-ws-start', 'init-ws-start']); + }); + it('does not fail a replacement attempt when an old worker returns retired', async () => { const world = makeWorld({ stages: [{ stageId: 'a', stageInstanceId: 'si-a' }] }); const handler = withDurableExecution((event, ctx) => __durableHandler(event, ctx, world.deps)); - const runner = new LocalDurableTestRunner({ handlerFunction: handler }); + const runner = trackedRunner({ handlerFunction: handler }); const completion = runner.run({ payload: { action: 'start', intentId: 'i1', executionId: 'i1' }, }); @@ -590,7 +716,7 @@ describe('orchestrator on the real durable runner (replay semantics)', () => { it('runs init-ws → stage a (park → human answer → resume) → stage b to SUCCEEDED with exactly-once side effects', async () => { const world = makeWorld(); const handler = withDurableExecution((event, ctx) => __durableHandler(event, ctx, world.deps)); - const runner = new LocalDurableTestRunner({ handlerFunction: handler }); + const runner = trackedRunner({ handlerFunction: handler }); const executionPromise = runner.run({ payload: { action: 'start', intentId: 'i1', executionId: 'i1' }, @@ -631,7 +757,8 @@ describe('orchestrator on the real durable runner (replay semantics)', () => { // Exactly-once side effects across every replay: expect(world.invokes.map((p) => p.command)).toEqual([ - 'init-ws', + 'init-ws-start', // capability probe + 'init-ws-start', // init dispatch (verdict via the init-ws-cb callback) 'create-workflow-checkpoint', // initial boundary 'run-stage-start', // a (fresh) 'run-stage-start', // a (resume h1) @@ -660,7 +787,7 @@ describe('orchestrator on the real durable runner (replay semantics)', () => { it('a stage verdict of FAILED delivered through the callback fails the run after replay', async () => { const world = makeWorld({ stages: [{ stageId: 'a', stageInstanceId: 'si-a' }] }); const handler = withDurableExecution((event, ctx) => __durableHandler(event, ctx, world.deps)); - const runner = new LocalDurableTestRunner({ handlerFunction: handler }); + const runner = trackedRunner({ handlerFunction: handler }); const executionPromise = runner.run({ payload: { action: 'start', intentId: 'i1', executionId: 'i1' }, @@ -693,7 +820,7 @@ describe('orchestrator on the real durable runner (replay semantics)', () => { it('a cancel sentinel on the gate callback retires the run without terminal writes', async () => { const world = makeWorld({ stages: [{ stageId: 'a' }] }); const handler = withDurableExecution((event, ctx) => __durableHandler(event, ctx, world.deps)); - const runner = new LocalDurableTestRunner({ handlerFunction: handler }); + const runner = trackedRunner({ handlerFunction: handler }); const executionPromise = runner.run({ payload: { action: 'start', intentId: 'i1', executionId: 'i1' }, @@ -802,7 +929,7 @@ describe('WP5 sections on the real durable runner', () => { const handler = withDurableExecution((event, ctx) => __durableHandler(event, ctx, world.deps), ); - const runner = new LocalDurableTestRunner({ handlerFunction: handler }); + const runner = trackedRunner({ handlerFunction: handler }); const executionPromise = runner.run({ payload: { action: 'start', intentId: 'i1', executionId: 'i1' }, @@ -851,7 +978,8 @@ describe('WP5 sections on the real durable runner', () => { // Exactly-once dispatches: each lane got ONE init-lane, its stage legs, // and ONE merge-lane, in skeleton-then-remaining order. expect(world.invokes.map((p) => `${p.command}:${p.stageId ?? p.unitSlug ?? '-'}`)).toEqual([ - 'init-ws:-', + 'init-ws-start:-', // capability probe + 'init-ws-start:-', // init dispatch 'create-workflow-checkpoint:-', 'run-stage-start:gen', 'derive-artifacts:-', @@ -927,7 +1055,7 @@ describe('WP5 sections on the real durable runner', () => { const handler = withDurableExecution((event, ctx) => __durableHandler(event, ctx, world.deps), ); - const runner = new LocalDurableTestRunner({ handlerFunction: handler }); + const runner = trackedRunner({ handlerFunction: handler }); const executionPromise = runner.run({ payload: { action: 'start', intentId: 'i1', executionId: 'i1' }, @@ -992,7 +1120,7 @@ describe('WP5 sections on the real durable runner', () => { const handler = withDurableExecution((event, ctx) => __durableHandler(event, ctx, world.deps), ); - const runner = new LocalDurableTestRunner({ handlerFunction: handler }); + const runner = trackedRunner({ handlerFunction: handler }); const executionPromise = runner.run({ payload: { action: 'start', intentId: 'i1', executionId: 'i1' }, @@ -1061,7 +1189,7 @@ describe('orchestrator gate bind CAS classification (answer-before-bind race)', const runTo = async (world) => { const handler = withDurableExecution((event, ctx) => __durableHandler(event, ctx, world.deps)); - const runner = new LocalDurableTestRunner({ handlerFunction: handler }); + const runner = trackedRunner({ handlerFunction: handler }); const executionPromise = runner.run({ payload: { action: 'start', intentId: 'i1', executionId: 'i1' }, }); @@ -1215,7 +1343,7 @@ describe('an answered-early gate never awaits its dangling callback', () => { }; world.deps.store.getHumanTask = async () => answeredGate; const handler = withDurableExecution((event, ctx) => __durableHandler(event, ctx, world.deps)); - const runner = new LocalDurableTestRunner({ handlerFunction: handler }); + const runner = trackedRunner({ handlerFunction: handler }); const executionPromise = runner.run({ payload: { action: 'start', intentId: 'i1', executionId: 'i1' }, }); diff --git a/lambda/v2-orchestrator/test/orchestrator.test.js b/lambda/v2-orchestrator/test/orchestrator.test.js index 6e9191ac6..8dedb718f 100644 --- a/lambda/v2-orchestrator/test/orchestrator.test.js +++ b/lambda/v2-orchestrator/test/orchestrator.test.js @@ -26,7 +26,9 @@ const makeCtx = (over = {}) => { logger: { info() {}, debug() {}, error() {} }, step: async (_name, fn) => fn(), createCallback: async (name) => { - if (String(name).startsWith('stage-cb-')) { + // The workspace-init callback is completed by the fake init-ws-start job + // through the same registry as stage callbacks. + if (isJobCallback(name)) { let resolve; const promise = new Promise((r) => { resolve = r; @@ -57,6 +59,26 @@ const makeCtx = (over = {}) => { // for the nth runtime call (init-ws result or the stage result the container // would deliver through the callback). For run-stage-start the verdict is // routed through the ctx's stage callback — exactly the production path. +// A current container answers the side-effect-free init-ws-start capability +// probe (no callbackId) with missing_callback_id. +const isInitProbe = (payload) => payload.command === 'init-ws-start' && !payload.callbackId; +const INIT_PROBE_ACK = { ok: false, reason: 'missing_callback_id' }; +// Callbacks the harness completes itself (stage jobs + the workspace-init job); +// ctx overrides keep these on the base registry. +const isJobCallback = (name) => String(name).startsWith('stage-cb-') || name === 'init-ws-cb'; + +// Wrap an inline runtime fake so workspace initialization completes through +// the production probe → callback → dispatch path with `verdict`; `fn` sees +// every other command. +const withInitWs = (ctx, fn, verdict = { ok: true }) => + vi.fn(async (payload, ...rest) => { + if (payload.command !== 'init-ws-start') return fn(payload, ...rest); + if (!payload.callbackId) return INIT_PROBE_ACK; + invokes.push(payload); + ctx.stageCallbackResolvers.get(payload.callbackId)(JSON.stringify(verdict)); + return { ok: true, accepted: true, callbackId: payload.callbackId }; + }); + const makeRuntime = (ctx, script) => { let n = 0; return vi.fn(async (payload, sessionId) => { @@ -64,10 +86,17 @@ const makeRuntime = (ctx, script) => { checkpointInvokes.push(payload); return { ok: true, checkpointId: `cp-${checkpointInvokes.length}` }; } + if (isInitProbe(payload)) return INIT_PROBE_ACK; invokes.push(payload); sessions.push(sessionId); n += 1; const verdict = await script(payload, n); + if (payload.command === 'init-ws-start') { + const resolve = ctx.stageCallbackResolvers.get(payload.callbackId); + if (!resolve) throw new Error(`no init callback registered: ${payload.callbackId}`); + resolve(JSON.stringify(verdict)); + return { ok: true, accepted: true, callbackId: payload.callbackId }; + } if (payload.command === 'run-stage-start') { const resolve = ctx.stageCallbackResolvers.get(payload.stageCallbackId); if (!resolve) throw new Error(`no stage callback registered: ${payload.stageCallbackId}`); @@ -82,7 +111,7 @@ const makeRuntime = (ctx, script) => { // Default script: init-ws ok, every stage SUCCEEDED. const okScript = (payload) => - payload.command === 'init-ws' ? { ok: true } : { ok: true, state: 'SUCCEEDED' }; + payload.command === 'init-ws-start' ? { ok: true } : { ok: true, state: 'SUCCEEDED' }; const MANAGED_RUNTIME_TARGET = { agentRuntimeArn: 'arn:aws:bedrock-agentcore:eu-west-1:123:runtime/managed', @@ -170,7 +199,7 @@ describe('orchestrator durable handler', () => { ); expect(res).toEqual({ ok: true, intentId: 'i1', stages: 2 }); expect(invokes.map((p) => p.command)).toEqual([ - 'init-ws', + 'init-ws-start', 'run-stage-start', 'run-stage-start', ]); @@ -179,17 +208,22 @@ describe('orchestrator durable handler', () => { const checkpointRunIds = [...new Set(checkpointInvokes.map((p) => p.orchestratorRunId))]; expect(checkpointRunIds).toHaveLength(1); expect(checkpointRunIds[0]).toMatch(/^run-/); - // init-ws carries the provider so the runtime picks the right clone scheme. - const initWs = invokes.find((p) => p.command === 'init-ws'); + // init-ws carries the provider so the runtime picks the right clone scheme, + // plus the init callback and the run's ownership token. + const initWs = invokes.find((p) => p.command === 'init-ws-start'); expect(initWs).toMatchObject({ projectId: 'p1', executionId: 'i1', gitProvider: 'github', repos: ['owner/repo'], + callbackId: 'cb-init-ws-cb', + orchestratorRunId: checkpointRunIds[0], }); expect(initWs).not.toHaveProperty('gitToken'); + // Probe + init dispatch + 2 stage dispatches + 3 checkpoints, all on the + // pinned runtime target. expect(deps.invokeRuntime.mock.calls.map(([, , target]) => target)).toEqual( - Array.from({ length: 6 }, () => MANAGED_RUNTIME_TARGET), + Array.from({ length: 7 }, () => MANAGED_RUNTIME_TARGET), ); const statuses = deps.store.updateExecution.mock.calls.map((c) => c[0].status); expect(statuses).toContain('RUNNING'); @@ -492,7 +526,7 @@ describe('orchestrator durable handler', () => { }), ); expect(invokes.map((p) => p.command)).toEqual([ - 'init-ws', + 'init-ws-start', 'run-stage-start', 'derive-artifacts', ]); @@ -716,7 +750,7 @@ describe('orchestrator durable handler', () => { const raceCtx = { ...base, createCallback: async (name) => { - if (String(name).startsWith('stage-cb-')) return base.createCallback(name); + if (isJobCallback(name)) return base.createCallback(name); return [ new Promise((resolve) => setTimeout(() => resolve({ answer: null }), 5)), `cb-${name}`, @@ -765,7 +799,7 @@ describe('orchestrator durable handler', () => { }, }, createCallback: async (name) => { - if (String(name).startsWith('stage-cb-')) return base.createCallback(name); + if (isJobCallback(name)) return base.createCallback(name); // Gate callback: NEVER resolves — awaiting it would hang the test. return [new Promise(() => {}), `cb-${name}`]; }, @@ -909,9 +943,8 @@ describe('orchestrator durable handler', () => { it('fails the stage when the container REFUSES the dispatch (accept-time failure)', async () => { deps.loadPlan.mockResolvedValue({ valid: true, plan: { stages: [{ stageId: 'a' }] } }); - deps.invokeRuntime = vi.fn(async (payload) => { + deps.invokeRuntime = withInitWs(ctx, async (payload) => { invokes.push(payload); - if (payload.command === 'init-ws') return { ok: true }; // e.g. duplicate job / old container without run-stage-start. return { ok: false, reason: 'job_already_running' }; }); @@ -927,11 +960,16 @@ describe('orchestrator durable handler', () => { }); it('fails (with reason + event) when init-ws returns ok:false instead of marching on', async () => { - deps.invokeRuntime = vi.fn(async (payload) => { - invokes.push(payload); - // init-ws reports a checkout failure (the runtime returns, does not throw). - return { ok: false, reason: 'checkout_failed', detail: 'auth' }; - }); + // init-ws reports a checkout failure through the init callback (the + // background job returns its verdict, it does not throw). + deps.invokeRuntime = withInitWs( + ctx, + async (payload) => { + invokes.push(payload); + return { ok: true, state: 'SUCCEEDED' }; + }, + { ok: false, reason: 'checkout_failed', detail: 'auth' }, + ); const res = await __durableHandler( { action: 'start', intentId: 'i1', executionId: 'i1' }, ctx, @@ -940,7 +978,7 @@ describe('orchestrator durable handler', () => { expect(res.ok).toBe(false); expect(res.reason).toBe('init_ws_failed'); // Never advanced past init-ws into the stage loop. - expect(invokes.map((p) => p.command)).toEqual(['init-ws']); + expect(invokes.map((p) => p.command)).toEqual(['init-ws-start']); // Status FAILED with a human-readable reason was persisted... const failCall = deps.store.updateExecution.mock.calls.find((c) => c[0].status === 'FAILED'); expect(failCall).toBeTruthy(); @@ -952,11 +990,8 @@ describe('orchestrator durable handler', () => { it('records FAILED + reason when a durable step throws (no silent death)', async () => { // A run-stage-start transport error throws out of the dispatch step. - let n = 0; - deps.invokeRuntime = vi.fn(async (payload) => { + deps.invokeRuntime = withInitWs(ctx, async (payload) => { invokes.push(payload); - n += 1; - if (n === 1) return { ok: true }; // init-ws throw new Error('AgentCore 500'); }); deps.loadPlan.mockResolvedValue({ valid: true, plan: { stages: [{ stageId: 'a' }] } }); @@ -977,17 +1012,18 @@ describe('orchestrator durable handler', () => { valid: true, plan: { stages: [{ stageId: 'a', stageInstanceId: 'si-a' }] }, }); - const dead = makeCtx({ + const base = makeCtx(); + const dead = { + ...base, createCallback: async (name) => { if (String(name).startsWith('stage-cb-')) { return [Promise.reject(new Error('callback timed out')), `cb-${name}`]; } - return [Promise.resolve({ answer: null }), `cb-${name}`]; + return base.createCallback(name); }, - }); - deps.invokeRuntime = vi.fn(async (payload) => { + }; + deps.invokeRuntime = withInitWs(dead, async (payload) => { invokes.push(payload); - if (payload.command === 'init-ws') return { ok: true }; return { ok: true, accepted: true }; }); const res = await __durableHandler( @@ -1007,6 +1043,171 @@ describe('orchestrator durable handler', () => { }); }); + describe('asynchronous workspace initialization', () => { + const stepNames = () => ctx.step.mock.calls.map(([name]) => name); + const initSteps = () => + stepNames().filter((name) => ['init-ws', 'init-ws-probe', 'init-ws-dispatch'].includes(name)); + beforeEach(() => { + ctx.step = vi.fn(async (_name, fn) => fn()); + ctx.createCallback = vi.fn(ctx.createCallback); + }); + + it('dispatches init-ws-start and waits on the init callback instead of holding the clone in a step', async () => { + const res = await __durableHandler( + { action: 'start', intentId: 'i1', executionId: 'i1' }, + ctx, + deps, + ); + expect(res.ok).toBe(true); + expect(initSteps()).toEqual(['init-ws-probe', 'init-ws-dispatch']); + expect(ctx.createCallback).toHaveBeenCalledWith('init-ws-cb', { + timeout: { hours: 2 }, + heartbeatTimeout: { minutes: 15 }, + }); + // The probe is side-effect free: it carries no init inputs and no callback. + const probe = deps.invokeRuntime.mock.calls.find(([p]) => isInitProbe(p))[0]; + expect(probe).toEqual({ command: 'init-ws-start', executionId: 'i1' }); + }); + + it.each([ + [ + 'AgentCore reports the 400', + async () => { + throw Object.assign(new Error('Received error (400) from runtime.'), { + name: 'RuntimeClientError', + }); + }, + ], + ['the body names the unknown command', async () => ({ error: 'unknown command "x"' })], + ])( + 'keeps the synchronous init-ws step on an older environment image (%s)', + async (_label, probeAnswer) => { + const inner = deps.invokeRuntime; + deps.invokeRuntime = vi.fn(async (payload, ...rest) => { + if (payload.command === 'init-ws-start') return probeAnswer(); + if (payload.command === 'init-ws') { + invokes.push(payload); + return { ok: true }; + } + return inner(payload, ...rest); + }); + const res = await __durableHandler( + { action: 'start', intentId: 'i1', executionId: 'i1' }, + ctx, + deps, + ); + expect(res).toEqual({ ok: true, intentId: 'i1', stages: 2 }); + expect(initSteps()).toEqual(['init-ws-probe', 'init-ws']); + expect(ctx.createCallback).not.toHaveBeenCalledWith('init-ws-cb', expect.anything()); + expect(invokes.map((p) => p.command)).toEqual([ + 'init-ws', + 'run-stage-start', + 'run-stage-start', + ]); + }, + ); + + it('does not mistake a probe transport failure for an older image', async () => { + deps.invokeRuntime = vi.fn(async () => { + throw Object.assign(new Error('Received error (503) from runtime.'), { + name: 'RuntimeClientError', + }); + }); + const res = await __durableHandler( + { action: 'start', intentId: 'i1', executionId: 'i1' }, + ctx, + deps, + ); + expect(res).toMatchObject({ ok: false, reason: 'orchestrator_error' }); + expect(stepNames()).not.toContain('init-ws'); + }); + + it('fails initialization when the container refuses the init job', async () => { + deps.invokeRuntime = vi.fn(async (payload) => { + if (isInitProbe(payload)) return INIT_PROBE_ACK; + invokes.push(payload); + return { ok: false, reason: 'init_job_already_running', detail: 'i1:init-ws' }; + }); + const res = await __durableHandler( + { action: 'start', intentId: 'i1', executionId: 'i1' }, + ctx, + deps, + ); + expect(res).toMatchObject({ ok: false, reason: 'init_ws_failed' }); + const failCall = deps.store.updateExecution.mock.calls.find((c) => c[0].status === 'FAILED'); + expect(failCall[0].failureReason).toBe('init_ws_failed: init_job_already_running'); + expect(stageStarts()).toHaveLength(0); + }); + + it('fails initialization when the init job dies (callback heartbeat expiry)', async () => { + const base = ctx.createCallback; + ctx.createCallback = vi.fn(async (name, options) => + name === 'init-ws-cb' + ? [Promise.reject(new Error('Callback heartbeat timed out')), `cb-${name}`] + : base(name, options), + ); + deps.invokeRuntime = vi.fn(async (payload) => { + if (isInitProbe(payload)) return INIT_PROBE_ACK; + invokes.push(payload); + return { ok: true, accepted: true }; + }); + const res = await __durableHandler( + { action: 'start', intentId: 'i1', executionId: 'i1' }, + ctx, + deps, + ); + expect(res).toMatchObject({ ok: false, reason: 'init_ws_failed' }); + const failCall = deps.store.updateExecution.mock.calls.find((c) => c[0].status === 'FAILED'); + expect(failCall[0].failureReason).toBe('init_ws_failed: init_ws_callback_failed'); + expect(stageStarts()).toHaveLength(0); + }); + + it('fails initialization on a malformed init callback body', async () => { + deps.invokeRuntime = vi.fn(async (payload) => { + if (isInitProbe(payload)) return INIT_PROBE_ACK; + invokes.push(payload); + ctx.stageCallbackResolvers.get(payload.callbackId)('not json'); + return { ok: true, accepted: true }; + }); + const res = await __durableHandler( + { action: 'start', intentId: 'i1', executionId: 'i1' }, + ctx, + deps, + ); + expect(res).toMatchObject({ ok: false, reason: 'init_ws_failed' }); + const failCall = deps.store.updateExecution.mock.calls.find((c) => c[0].status === 'FAILED'); + expect(failCall[0].failureReason).toBe('init_ws_failed: init_ws_bad_callback_result'); + }); + + it('replays the synchronous init-ws step for a run claimed before the mode existed', async () => { + // A run checkpointed by the previous code memoized a claim without + // initWsMode. Its replay must keep the `init-ws` operation at that + // position, or the durable SDK rejects the history as non-deterministic. + ctx.step = vi.fn(async (name, fn) => { + const value = await fn(); + if (name !== 'mint-run-id') return value; + const { initWsMode: _dropped, ...legacyClaim } = value; + return legacyClaim; + }); + const inner = deps.invokeRuntime; + deps.invokeRuntime = vi.fn(async (payload, ...rest) => { + if (payload.command === 'init-ws') { + invokes.push(payload); + return { ok: true }; + } + return inner(payload, ...rest); + }); + const res = await __durableHandler( + { action: 'start', intentId: 'i1', executionId: 'i1' }, + ctx, + deps, + ); + expect(res.ok).toBe(true); + expect(initSteps()).toEqual(['init-ws']); + expect(deps.invokeRuntime.mock.calls.map(([p]) => p.command)).not.toContain('init-ws-start'); + }); + }); + it('emits workspace lifecycle events so init-ws is visible in the feed', async () => { await __durableHandler({ action: 'start', intentId: 'i1', executionId: 'i1' }, ctx, deps); const evTypes = deps.store.appendEvent.mock.calls.map((c) => c[0].type); @@ -1035,10 +1236,14 @@ describe('orchestrator durable handler', () => { }); it('broadcasts FAILED live so the UI flips without a manual refresh', async () => { - deps.invokeRuntime = vi.fn(async (payload) => { - invokes.push(payload); - return { ok: false, reason: 'checkout_failed' }; // init-ws fails - }); + deps.invokeRuntime = withInitWs( + ctx, + async (payload) => { + invokes.push(payload); + return { ok: true, accepted: true }; + }, + { ok: false, reason: 'checkout_failed' }, // init-ws fails + ); await __durableHandler({ action: 'start', intentId: 'i1', executionId: 'i1' }, ctx, deps); const failed = deps.broadcast.mock.calls.find( (c) => c[1].action === 'agent.execution' && c[1].status === 'FAILED', @@ -1121,7 +1326,7 @@ describe('orchestrator durable handler', () => { // createCallback + wait are durable primitives, legitimately called outside // a step (they ARE the checkpoint), so they don't assert. createCallback: async (name) => { - if (String(name).startsWith('stage-cb-')) { + if (isJobCallback(name)) { let resolve; const promise = new Promise((r) => { resolve = r; @@ -1163,7 +1368,11 @@ describe('orchestrator durable handler', () => { }); deps.invokeRuntime = vi.fn(async (payload) => { assertInStep(`invokeRuntime:${payload.command}`); - if (payload.command === 'init-ws') return { ok: true }; + if (isInitProbe(payload)) return INIT_PROBE_ACK; + if (payload.command === 'init-ws-start') { + stageCallbacks.get(payload.callbackId)(JSON.stringify({ ok: true })); + return { ok: true, accepted: true }; + } // Deliver the verdict via the stage callback like the harness does. const resolve = stageCallbacks.get(payload.stageCallbackId); resolve({ ok: true, state: 'SUCCEEDED' }); @@ -1208,7 +1417,7 @@ describe('rewind relaunch (startAtStageId)', () => { expect(res).toEqual({ ok: true, intentId: 'i1', stages: 2 }); // init-ws still runs (idempotent workspace heal); the loop starts at 'b'. expect(invokes.map((p) => p.command)).toEqual([ - 'init-ws', + 'init-ws-start', 'run-stage-start', 'run-stage-start', ]); @@ -1369,7 +1578,7 @@ describe('unit DAG promotion after the producing stage succeeds', () => { it('dispatches promote-units exactly once, after the producing stage, before the next stage', async () => { deps.loadPlan.mockResolvedValue(PLAN_WITH_DAG_PRODUCER); deps.invokeRuntime = makeRuntime(ctx, (payload) => { - if (payload.command === 'init-ws') return { ok: true }; + if (payload.command === 'init-ws-start') return { ok: true }; if (payload.command === 'promote-units') return { ok: true, unitCount: 3, batchCount: 2, walkingSkeleton: 'auth' }; return { ok: true, state: 'SUCCEEDED' }; @@ -1382,7 +1591,7 @@ describe('unit DAG promotion after the producing stage succeeds', () => { expect(res.ok).toBe(true); const commands = invokes.map((p) => p.command); expect(commands).toEqual([ - 'init-ws', + 'init-ws-start', 'run-stage-start', // units-generation 'derive-artifacts', // projection before downstream consumers run 'promote-units', // right after the producer succeeded @@ -1424,7 +1633,7 @@ describe('unit DAG promotion after the producing stage succeeds', () => { deps.store.getExecution = vi.fn(async () => ({ ...META, deriveEnrichment: 'llm' })); deps.loadPlan.mockResolvedValue(PLAN_WITH_DAG_PRODUCER); deps.invokeRuntime = makeRuntime(ctx, (payload) => { - if (payload.command === 'init-ws') return { ok: true }; + if (payload.command === 'init-ws-start') return { ok: true }; if (payload.command === 'promote-units') return { ok: true, unitCount: 1, batchCount: 1 }; return { ok: true, state: 'SUCCEEDED' }; }); @@ -1437,7 +1646,7 @@ describe('unit DAG promotion after the producing stage succeeds', () => { it('fails the run (units_promotion_failed) when promotion is refused', async () => { deps.loadPlan.mockResolvedValue(PLAN_WITH_DAG_PRODUCER); deps.invokeRuntime = makeRuntime(ctx, (payload) => { - if (payload.command === 'init-ws') return { ok: true }; + if (payload.command === 'init-ws-start') return { ok: true }; if (payload.command === 'promote-units') return { ok: false, reason: 'dag_malformed', detail: 'duplicate: auth' }; return { ok: true, state: 'SUCCEEDED' }; @@ -1464,7 +1673,7 @@ describe('unit DAG promotion after the producing stage succeeds', () => { plan: { stages: [PLAN_WITH_DAG_PRODUCER.plan.stages[0]] }, }); deps.invokeRuntime = makeRuntime(ctx, (payload, n) => { - if (payload.command === 'init-ws') return { ok: true }; + if (payload.command === 'init-ws-start') return { ok: true }; if (payload.command === 'promote-units') return { ok: true, unitCount: 1, batchCount: 1, walkingSkeleton: 'auth' }; // First stage leg parks (approval question), resume leg succeeds. @@ -1480,7 +1689,7 @@ describe('unit DAG promotion after the producing stage succeeds', () => { const commands = invokes.map((p) => p.command); // fresh leg → resume leg → THEN projection and promotion. expect(commands).toEqual([ - 'init-ws', + 'init-ws-start', 'run-stage-start', 'run-stage-start', 'derive-artifacts', @@ -1536,7 +1745,7 @@ const UNIT_PLAN = (over = {}) => ({ }); const sectionScript = (payload) => { - if (payload.command === 'init-ws') return { ok: true }; + if (payload.command === 'init-ws-start') return { ok: true }; if (payload.command === 'promote-units') return { ok: true, unitCount: 2, batchCount: 2, walkingSkeleton: 'auth' }; return { ok: true, state: 'SUCCEEDED' }; @@ -1574,7 +1783,7 @@ describe('WP5 — parallel sections: lanes, skeleton, ladder, halt-and-ask', () // (init → fd → cg → merge) completes before billing's lane opens; bt runs // after fan-in. expect(invokes.map((p) => `${p.command}:${p.stageId ?? p.unitSlug ?? '-'}`)).toEqual([ - 'init-ws:-', + 'init-ws-start:-', 'run-stage-start:units-gen', 'derive-artifacts:-', 'promote-units:-', @@ -1724,7 +1933,7 @@ describe('WP5 — parallel sections: lanes, skeleton, ladder, halt-and-ask', () it('a lane failure halts-and-asks; an unanswered/abort decision fails the run with work preserved', async () => { deps.invokeRuntime = makeRuntime(ctx, (payload) => { - if (payload.command === 'init-ws') return { ok: true }; + if (payload.command === 'init-ws-start') return { ok: true }; if (payload.command === 'promote-units') return { ok: true, unitCount: 2, batchCount: 2, walkingSkeleton: 'auth' }; if (payload.stageId === 'cg' && payload.unitSlug === 'auth') @@ -1751,7 +1960,7 @@ describe('WP5 — parallel sections: lanes, skeleton, ladder, halt-and-ask', () it('treats a lane stage failed for a missing checkpoint like any lane failure', async () => { deps.invokeRuntime = makeRuntime(ctx, (payload) => { - if (payload.command === 'init-ws') return { ok: true }; + if (payload.command === 'init-ws-start') return { ok: true }; if (payload.command === 'promote-units') return { ok: true, unitCount: 2, batchCount: 2, walkingSkeleton: 'auth' }; if (payload.stageId === 'cg' && payload.unitSlug === 'auth') @@ -1794,7 +2003,7 @@ describe('WP5 — parallel sections: lanes, skeleton, ladder, halt-and-ask', () deps.store.getHumanTask = vi.fn(async () => ({ status: 'answered' })); let parked = false; deps.invokeRuntime = makeRuntime(ctx, (payload) => { - if (payload.command === 'init-ws') return { ok: true }; + if (payload.command === 'init-ws-start') return { ok: true }; if (payload.command === 'promote-units') return { ok: true, unitCount: 2, batchCount: 2 }; if (payload.stageId === 'fd' && payload.unitSlug === 'auth' && !parked) { parked = true; @@ -1836,7 +2045,7 @@ describe('WP5 — parallel sections: lanes, skeleton, ladder, halt-and-ask', () deps.store.getHumanTask = vi.fn(async () => ({ status: 'answered' })); let parked = false; deps.invokeRuntime = makeRuntime(ctx, (payload) => { - if (payload.command === 'init-ws') return { ok: true }; + if (payload.command === 'init-ws-start') return { ok: true }; if (payload.command === 'promote-units') return { ok: true, unitCount: 2, batchCount: 2 }; if (payload.stageId === 'fd' && payload.unitSlug === 'auth' && !parked) { parked = true; @@ -2260,7 +2469,7 @@ describe('PR per unit delivery', () => { }, }); deps.invokeRuntime = makeRuntime(ctx, (payload) => { - if (payload.command === 'init-ws') return { ok: true }; + if (payload.command === 'init-ws-start') return { ok: true }; if (payload.reviewFeedback) { return { ok: true, @@ -2697,7 +2906,7 @@ describe('WP5 — engine-gate decisions', () => { let cgBillingAttempts = 0; deps.store.getHumanTask = gateReads({ 'eg-halt': { answer: { decision: 'retry' } } }); deps.invokeRuntime = makeRuntime(ctx, (payload) => { - if (payload.command === 'init-ws') return { ok: true }; + if (payload.command === 'init-ws-start') return { ok: true }; if (payload.command === 'promote-units') return { ok: true, unitCount: 2, batchCount: 2 }; if (payload.stageId === 'cg' && payload.unitSlug === 'billing') { cgBillingAttempts += 1; @@ -2762,7 +2971,7 @@ describe('WP5 — engine-gate decisions', () => { deps.stopSession = vi.fn(async () => stopResult); let cgBillingAttempts = 0; deps.invokeRuntime = makeRuntime(ctx, (payload) => { - if (payload.command === 'init-ws') return { ok: true }; + if (payload.command === 'init-ws-start') return { ok: true }; if (payload.command === 'promote-units') return { ok: true, unitCount: 2, batchCount: 2 }; if (payload.stageId === 'cg' && payload.unitSlug === 'billing') { cgBillingAttempts += 1; @@ -2792,7 +3001,7 @@ describe('WP5 — engine-gate decisions', () => { it('halt-and-ask SKIP preserves merged lanes and lets the run continue without the failed unit', async () => { deps.store.getHumanTask = gateReads({ 'eg-halt': { answer: { decision: 'skip' } } }); deps.invokeRuntime = makeRuntime(ctx, (payload) => { - if (payload.command === 'init-ws') return { ok: true }; + if (payload.command === 'init-ws-start') return { ok: true }; if (payload.command === 'promote-units') return { ok: true, unitCount: 2, batchCount: 2 }; if (payload.stageId === 'cg' && payload.unitSlug === 'billing') return { ok: false, state: 'FAILED', reason: 'sensor_blocked' }; @@ -2837,7 +3046,7 @@ describe('WP5 — engine-gate decisions', () => { 'eg-halt': { answer: { decision: 'skip' } }, }); deps.invokeRuntime = makeRuntime(ctx, (payload) => { - if (payload.command === 'init-ws') return { ok: true }; + if (payload.command === 'init-ws-start') return { ok: true }; if (payload.command === 'promote-units') return { ok: true, unitCount: 3, batchCount: 2 }; if (payload.stageId === 'cg' && payload.unitSlug === 'b') return { ok: false, state: 'FAILED', reason: 'sensor_blocked' }; @@ -2878,7 +3087,7 @@ describe('WP5 — engine-gate decisions', () => { }); let assetWaitedForCharts = false; deps.invokeRuntime = makeRuntime(ctx, async (payload) => { - if (payload.command === 'init-ws') return { ok: true }; + if (payload.command === 'init-ws-start') return { ok: true }; if (payload.command === 'promote-units') { return { ok: true, unitCount: 3, batchCount: 1 }; } @@ -3054,7 +3263,7 @@ describe('WP5 — engine-gate decisions', () => { 'eg-halt-s1-w1-r0': { answer: { decision: 'garbage' } }, }); deps.invokeRuntime = makeRuntime(ctx, (payload) => { - if (payload.command === 'init-ws') return { ok: true }; + if (payload.command === 'init-ws-start') return { ok: true }; if (payload.command === 'promote-units') return { ok: true, unitCount: 2, batchCount: 2 }; if (payload.stageId === 'cg' && payload.unitSlug === 'billing') return { ok: false, state: 'FAILED', reason: 'sensor_blocked' }; @@ -3116,7 +3325,7 @@ describe('WP6 — merge conflict → resolution stage → retry → halt on repe it('one conflicted lane: resolve-conflict runs in the LANE session, the merge retry lands, lane MERGED', async () => { let authMerges = 0; deps.invokeRuntime = makeRuntime(ctx, (payload) => { - if (payload.command === 'init-ws') return { ok: true }; + if (payload.command === 'init-ws-start') return { ok: true }; if (payload.command === 'promote-units') return { ok: true, unitCount: 2, batchCount: 2 }; if (payload.command === 'merge-lane' && payload.unitSlug === 'auth') { authMerges += 1; @@ -3176,7 +3385,7 @@ describe('WP6 — merge conflict → resolution stage → retry → halt on repe : { status: 'answered' }, ); deps.invokeRuntime = makeRuntime(ctx, (payload) => { - if (payload.command === 'init-ws') return { ok: true }; + if (payload.command === 'init-ws-start') return { ok: true }; if (payload.command === 'promote-units') return { ok: true, unitCount: 2, batchCount: 2 }; if (payload.command === 'merge-lane' && payload.unitSlug === 'auth') return { ok: false, reason: 'merge_conflict', conflicts: ['o/r:shared.txt'] }; @@ -3651,7 +3860,7 @@ describe('credential-free runtime payloads', () => { const result = await start(); expect(result.ok).toBe(true); const gitPayloads = invokes.filter((payload) => - ['init-ws', 'run-stage-start'].includes(payload.command), + ['init-ws-start', 'run-stage-start'].includes(payload.command), ); expect(gitPayloads.length).toBeGreaterThan(0); for (const payload of gitPayloads) expect(payload.gitAuthor).toEqual(author); diff --git a/lambda/v2-orchestrator/test/section-integration.test.js b/lambda/v2-orchestrator/test/section-integration.test.js index 9833e49b5..a37e59911 100644 --- a/lambda/v2-orchestrator/test/section-integration.test.js +++ b/lambda/v2-orchestrator/test/section-integration.test.js @@ -210,9 +210,22 @@ const makeWorld = ({ remote, unitPlan, fileFor, beforeStage = null, conflictAgen world.invokes.push(payload); world.sessions.push(sessionId); const ws = wsFor(sessionId); - if (payload.command === 'init-ws') { - await realCheckoutRepo(remote)({ branch: payload.branch, targetDir: ws }); - return { ok: true }; + if (payload.command === 'init-ws-start') { + // Capability probe (no callbackId); then the accepted init job runs the + // REAL checkout in the background and completes the init callback. + if (!payload.callbackId) return { ok: false, reason: 'missing_callback_id' }; + (async () => { + try { + await realCheckoutRepo(remote)({ branch: payload.branch, targetDir: ws }); + const op = await world.runner + .getOperation('init-ws-cb') + .waitForData(WaitingOperationStatus.STARTED); + await op.sendCallbackSuccess(JSON.stringify({ ok: true })); + } catch (e) { + console.error('[fixture] init-ws job failed:', e?.message); + } + })(); + return { ok: true, accepted: true, callbackId: payload.callbackId }; } if (payload.command === 'promote-units') { return { ok: true, unitCount: unitPlan.units.length, batchCount: unitPlan.batches.length }; diff --git a/lambda/v2-orchestrator/test/skeleton-switch.test.js b/lambda/v2-orchestrator/test/skeleton-switch.test.js index 5878f5ec9..09c0cd5ee 100644 --- a/lambda/v2-orchestrator/test/skeleton-switch.test.js +++ b/lambda/v2-orchestrator/test/skeleton-switch.test.js @@ -125,6 +125,18 @@ const makeWorld = (skeleton) => { const invokeRuntime = async (payload) => { world.invokes.push(payload); + if (payload.command === 'init-ws-start') { + // Capability probe (no callbackId), then the accepted init job, which + // completes the init callback out-of-band. + if (!payload.callbackId) return { ok: false, reason: 'missing_callback_id' }; + (async () => { + const op = await world.runner + .getOperation('init-ws-cb') + .waitForData(WaitingOperationStatus.STARTED); + await op.sendCallbackSuccess(JSON.stringify({ ok: true })); + })(); + return { ok: true, accepted: true, callbackId: payload.callbackId }; + } if (payload.command === 'promote-units') { return { ok: true, unitCount: 2, batchCount: 1 }; } From c106ef08f51b5ce6c860c753e9ffe7825dda2e53 Mon Sep 17 00:00:00 2001 From: Matt Houghton <15250993+mattdevdba@users.noreply.github.com> Date: Thu, 29 Oct 2026 19:28:25 +0000 Subject: [PATCH 2/4] fix(agentcore): reserve the init-ws job before the first heartbeat Two init-ws-start dispatches arriving while the first heartbeat was pending both saw no active job and started a clone into the same workspace. The second also overwrote the tracked entry, which the first job's cleanup could then remove while the second was still running. The job is now reserved before the heartbeat is awaited. A duplicate of the same callback waits for and shares the first dispatch's acceptance, a different callback is refused, and cleanup only releases an entry the job still owns. --- lambda/agentcore/commands/init-ws-start.js | 38 ++++++++--- lambda/agentcore/test/init-ws-start.test.js | 75 +++++++++++++++++++++ 2 files changed, 103 insertions(+), 10 deletions(-) diff --git a/lambda/agentcore/commands/init-ws-start.js b/lambda/agentcore/commands/init-ws-start.js index 2bf5ec2a9..099efd00e 100644 --- a/lambda/agentcore/commands/init-ws-start.js +++ b/lambda/agentcore/commands/init-ws-start.js @@ -22,10 +22,12 @@ import { runCredentialJob } from '../credential-session.js'; // container cannot reach fails the dispatch synchronously; // - heartbeats fire while the clone runs so the orchestrator's // heartbeatTimeout separates "large repository" from "dead container"; -// - one init job per execution: a duplicate start for the SAME callback is -// an idempotent accept (durable dispatch retry), a different callback is -// refused (init_job_already_running) — two clones must never race into -// the same workspace directory; +// - one init job per execution: the workspace is reserved before the first +// heartbeat is awaited; a duplicate start for the SAME callback shares the +// first dispatch's acceptance (durable dispatch retry), a different +// callback is refused (init_job_already_running) — two clones must never +// race into the same workspace directory, and a job only releases the +// reservation it owns; // - the busy tracker is held for the job's lifetime so /ping reports // HealthyBusy and AgentCore keeps the session (and its mount) alive. @@ -62,12 +64,29 @@ export const createInitWsStart = ({ const key = jobKey(payload); const inFlight = activeJobs.get(key); if (inFlight) { - if (inFlight.callbackId === callbackId) { - return { ok: true, accepted: true, alreadyRunning: true, callbackId }; + if (inFlight.callbackId !== callbackId) { + return { ok: false, reason: 'init_job_already_running', detail: key }; } - return { ok: false, reason: 'init_job_already_running', detail: key }; + // A duplicate of the same attempt shares the first dispatch's outcome, + // including one whose initial heartbeat is still pending. + const first = await inFlight.acceptance; + return first.ok ? { ok: true, accepted: true, alreadyRunning: true, callbackId } : first; } + // Reserve the workspace before the first await, so an overlapping + // dispatch sees this attempt instead of starting a second clone. + const jobState = { startedAt: Date.now(), callbackId, heartbeatTimer: null, acceptance: null }; + activeJobs.set(key, jobState); + jobState.acceptance = accept(payload, key, jobState); + return jobState.acceptance; + }; + + const release = (key, jobState) => { + if (activeJobs.get(key) === jobState) activeJobs.delete(key); + }; + + const accept = async (payload, key, jobState) => { + const { callbackId, executionId } = payload; let initialHeartbeat; try { initialHeartbeat = await sendCallbackHeartbeat(callbackId); @@ -75,6 +94,7 @@ export const createInitWsStart = ({ initialHeartbeat = { delivered: false, error: err?.message ?? String(err) }; } if (!initialHeartbeat?.delivered) { + release(key, jobState); log.error('init-ws callback heartbeat could not be established', { executionId, error: initialHeartbeat?.error ?? null, @@ -86,8 +106,6 @@ export const createInitWsStart = ({ }; } - const jobState = { startedAt: Date.now(), callbackId, heartbeatTimer: null }; - activeJobs.set(key, jobState); busy?.enter(); log.info('init-ws job accepted', { executionId }); @@ -156,7 +174,7 @@ export const createInitWsStart = ({ return result; } finally { if (jobState.heartbeatTimer) clearTimeout(jobState.heartbeatTimer); - activeJobs.delete(key); + release(key, jobState); busy?.leave(); } }); diff --git a/lambda/agentcore/test/init-ws-start.test.js b/lambda/agentcore/test/init-ws-start.test.js index 3623f6c2a..08b7de6e4 100644 --- a/lambda/agentcore/test/init-ws-start.test.js +++ b/lambda/agentcore/test/init-ws-start.test.js @@ -231,6 +231,81 @@ describe('createInitWsStart', () => { await flush(); }); + describe('dispatches that overlap the first heartbeat', () => { + const makeOverlap = () => { + const beat = makeGate(); + const clone = makeGate(); + const { deps } = makeDeps({ + initWs: vi.fn(() => clone.promise), + sendCallbackHeartbeat: vi.fn(() => beat.promise), + }); + return { beat, clone, deps, start: createInitWsStart(deps) }; + }; + + it('runs one clone for two dispatches of the SAME attempt, both accepted', async () => { + const { beat, clone, deps, start } = makeOverlap(); + const first = start(basePayload); + const second = start(basePayload); + beat.resolve({ delivered: true }); + await expect(first).resolves.toMatchObject({ ok: true, accepted: true }); + await expect(second).resolves.toMatchObject({ + ok: true, + accepted: true, + alreadyRunning: true, + }); + expect(deps.sendCallbackHeartbeat).toHaveBeenCalledTimes(1); + expect(deps.initWs).toHaveBeenCalledTimes(1); + clone.resolve({ ok: true }); + await flush(); + expect(start.activeJobs.size).toBe(0); + }); + + it('refuses a DIFFERENT attempt while the first heartbeat is pending', async () => { + const { beat, clone, deps, start } = makeOverlap(); + const first = start(basePayload); + const relaunch = await start({ ...basePayload, callbackId: 'cb-relaunch' }); + expect(relaunch).toEqual({ + ok: false, + reason: 'init_job_already_running', + detail: 'e1:init-ws', + }); + beat.resolve({ delivered: true }); + await expect(first).resolves.toMatchObject({ ok: true, accepted: true }); + expect(deps.initWs).toHaveBeenCalledTimes(1); + clone.resolve({ ok: true }); + await flush(); + }); + + it('shares a failed first heartbeat with the duplicate and frees the workspace', async () => { + const { beat, deps, start } = makeOverlap(); + const first = start(basePayload); + const second = start(basePayload); + beat.resolve({ delivered: false, error: 'AccessDenied' }); + const failure = { + ok: false, + reason: 'init_ws_callback_heartbeat_failed', + detail: 'AccessDenied', + }; + await expect(first).resolves.toEqual(failure); + await expect(second).resolves.toEqual(failure); + expect(deps.initWs).not.toHaveBeenCalled(); + expect(start.activeJobs.size).toBe(0); + }); + }); + + it("a finished job's cleanup never removes another job's reservation", async () => { + const clone = makeGate(); + const { deps } = makeDeps({ initWs: vi.fn(() => clone.promise) }); + const start = createInitWsStart(deps); + await start(basePayload); + // Simulate a later attempt owning the entry when the first job ends. + const other = { callbackId: 'cb-other' }; + start.activeJobs.set('e1:init-ws', other); + clone.resolve({ ok: true }); + await flush(); + expect(start.activeJobs.get('e1:init-ws')).toBe(other); + }); + it('holds the busy tracker for the whole job so the session stays HealthyBusy', async () => { const gate = makeGate(); const busy = createBusyTracker(); From 534b1fe29d94da368dc55d44674597ffbd2ff725 Mon Sep 17 00:00:00 2001 From: Matt Houghton <15250993+mattdevdba@users.noreply.github.com> Date: Fri, 30 Oct 2026 19:39:07 +0000 Subject: [PATCH 3/4] fix(agentcore): answer a dispatch retry after the init-ws job finished The init-ws-dispatch step retries every error. If the first response was lost and the job finished before the retry arrived, the reservation was already released, so the retry started a second clone and heartbeated a callback that was already completed. init-ws-start now remembers a finished job's callback id for an hour and answers a same-callback dispatch with alreadyCompleted. A different callback still starts a new job. The init-ws-start log records now carry the callbackId. Also give the final-completion and frozen-checkpoint replay tests added on main the init-ws-start container behaviour they now need. --- lambda/agentcore/commands/init-ws-start.js | 33 +++++++++++- lambda/agentcore/test/init-ws-start.test.js | 51 +++++++++++++++++++ .../test/final-completion.test.js | 5 ++ .../test/orchestrator-replay.test.js | 4 +- 4 files changed, 90 insertions(+), 3 deletions(-) diff --git a/lambda/agentcore/commands/init-ws-start.js b/lambda/agentcore/commands/init-ws-start.js index 099efd00e..fdd96fa01 100644 --- a/lambda/agentcore/commands/init-ws-start.js +++ b/lambda/agentcore/commands/init-ws-start.js @@ -28,6 +28,9 @@ import { runCredentialJob } from '../credential-session.js'; // callback is refused (init_job_already_running) — two clones must never // race into the same workspace directory, and a job only releases the // reservation it owns; +// - a finished job's callback id is remembered for a bounded time, so a +// dispatch retry that arrives after the job completed is answered +// alreadyCompleted instead of cloning again; // - the busy tracker is held for the job's lifetime so /ping reports // HealthyBusy and AgentCore keeps the session (and its mount) alive. @@ -39,6 +42,11 @@ const logger = new Logger({ const jobKey = (p) => `${p.executionId}:init-ws`; +// How long a finished job's callback id is remembered, so a durable dispatch +// retry (lost response) is answered instead of starting a second clone. +// Covers the step's retry backoff with a wide margin. +const COMPLETED_RETENTION_MS = 60 * 60_000; + // init-ws returns {ok:false, reason, detail} on every handled failure and // {ok:true, ...} on success. Anything else is a contract break — surface it as // a failure rather than letting the orchestrator mark the workspace ready. @@ -54,14 +62,30 @@ export const createInitWsStart = ({ busy = null, heartbeatIntervalMs = 60_000, activeJobs = new Map(), + completedJobs = new Map(), + completedRetentionMs = COMPLETED_RETENTION_MS, + now = Date.now, log = logger, }) => { + const pruneCompleted = () => { + for (const [key, done] of completedJobs) { + if (now() - done.completedAt > completedRetentionMs) completedJobs.delete(key); + } + }; + const start = async (payload) => { const { callbackId, executionId } = payload ?? {}; if (!callbackId) return { ok: false, reason: 'missing_callback_id' }; if (!executionId) return { ok: false, reason: 'missing_execution_id' }; const key = jobKey(payload); + pruneCompleted(); + // A dispatch retry that arrives after its job finished (the first + // response was lost) must not clone again or heartbeat a completed + // callback: the verdict has already been sent. + if (completedJobs.get(key)?.callbackId === callbackId) { + return { ok: true, accepted: true, alreadyCompleted: true, callbackId }; + } const inFlight = activeJobs.get(key); if (inFlight) { if (inFlight.callbackId !== callbackId) { @@ -97,6 +121,7 @@ export const createInitWsStart = ({ release(key, jobState); log.error('init-ws callback heartbeat could not be established', { executionId, + callbackId, error: initialHeartbeat?.error ?? null, }); return { @@ -107,7 +132,7 @@ export const createInitWsStart = ({ } busy?.enter(); - log.info('init-ws job accepted', { executionId }); + log.info('init-ws job accepted', { executionId, callbackId }); // The job payload is init-ws's own argument bag; the callback id is // transport, not an init-ws input. @@ -128,6 +153,7 @@ export const createInitWsStart = ({ if (!sent?.delivered) { log.error('init-ws callback heartbeat failed', { executionId, + callbackId, error: sent?.error ?? null, }); } @@ -152,12 +178,14 @@ export const createInitWsStart = ({ log.warn('command returned failure', { command: 'init-ws-start', executionId, + callbackId, reason: result.reason, detail: result.detail, }); } else { log.info('init-ws job completed', { executionId, + callbackId, durationMs: Date.now() - jobState.startedAt, }); } @@ -168,12 +196,14 @@ export const createInitWsStart = ({ // backstop; log loudly so an operator can complete it manually. log.error('init-ws callback could not be delivered', { executionId, + callbackId, error: sent?.error ?? null, }); } return result; } finally { if (jobState.heartbeatTimer) clearTimeout(jobState.heartbeatTimer); + completedJobs.set(key, { callbackId, completedAt: now() }); release(key, jobState); busy?.leave(); } @@ -181,6 +211,7 @@ export const createInitWsStart = ({ job.catch((err) => log.error('init-ws job promise rejected unexpectedly', { executionId, + callbackId, error: err?.message ?? String(err), }), ); diff --git a/lambda/agentcore/test/init-ws-start.test.js b/lambda/agentcore/test/init-ws-start.test.js index 08b7de6e4..3e2c224b0 100644 --- a/lambda/agentcore/test/init-ws-start.test.js +++ b/lambda/agentcore/test/init-ws-start.test.js @@ -306,6 +306,57 @@ describe('createInitWsStart', () => { expect(start.activeJobs.get('e1:init-ws')).toBe(other); }); + describe('a dispatch retry after the job finished', () => { + it('answers alreadyCompleted without a second clone or a heartbeat to the completed callback', async () => { + const { deps, sent, beats } = makeDeps(); + const start = createInitWsStart(deps); + await start(basePayload); + await flush(); + expect(sent).toHaveLength(1); + + const retry = await start(basePayload); + + expect(retry).toEqual({ + ok: true, + accepted: true, + alreadyCompleted: true, + callbackId: 'cb-init', + }); + expect(deps.initWs).toHaveBeenCalledOnce(); + expect(beats).toEqual(['cb-init']); + expect(sent).toHaveLength(1); + }); + + it('starts a new job for a DIFFERENT callback once the previous one finished', async () => { + const { deps } = makeDeps(); + const start = createInitWsStart(deps); + await start(basePayload); + await flush(); + + const next = await start({ ...basePayload, callbackId: 'cb-next' }); + await flush(); + + expect(next).toMatchObject({ ok: true, accepted: true, callbackId: 'cb-next' }); + expect(next).not.toHaveProperty('alreadyCompleted'); + expect(deps.initWs).toHaveBeenCalledTimes(2); + }); + + it('forgets a finished callback after the retention window', async () => { + let clock = 1_000; + const { deps } = makeDeps(); + const start = createInitWsStart({ ...deps, completedRetentionMs: 100, now: () => clock }); + await start(basePayload); + await flush(); + clock += 101; + + const late = await start(basePayload); + await flush(); + + expect(late).not.toHaveProperty('alreadyCompleted'); + expect(deps.initWs).toHaveBeenCalledTimes(2); + }); + }); + it('holds the busy tracker for the whole job so the session stays HealthyBusy', async () => { const gate = makeGate(); const busy = createBusyTracker(); diff --git a/lambda/v2-orchestrator/test/final-completion.test.js b/lambda/v2-orchestrator/test/final-completion.test.js index d054406d1..835c1f791 100644 --- a/lambda/v2-orchestrator/test/final-completion.test.js +++ b/lambda/v2-orchestrator/test/final-completion.test.js @@ -56,6 +56,11 @@ const setup = () => { name.startsWith('section-completion-v1-') || name.startsWith('load-completion-v1-') ? { state: 'complete', unresolved: [] } : fn(), + // The accepted init-ws-start job completes the workspace-init callback. + createCallback: async (name) => { + if (name !== 'init-ws-cb') throw new Error(`unexpected callback ${name}`); + return [Promise.resolve(JSON.stringify({ ok: true })), `cb-${name}`]; + }, logger: { info() {}, error() {} }, }; return { diff --git a/lambda/v2-orchestrator/test/orchestrator-replay.test.js b/lambda/v2-orchestrator/test/orchestrator-replay.test.js index 96e4dfeb8..f3bb42122 100644 --- a/lambda/v2-orchestrator/test/orchestrator-replay.test.js +++ b/lambda/v2-orchestrator/test/orchestrator-replay.test.js @@ -364,7 +364,7 @@ describe('orchestrator on the real durable runner (replay semantics)', () => { }); return __durableHandler(event, wrapped, world.deps); }); - const runner = new LocalDurableTestRunner({ handlerFunction: handler }); + const runner = trackedRunner({ handlerFunction: handler }); const completion = runner.run({ payload: { action: 'start', intentId: 'i1', executionId: 'i1' }, }); @@ -421,7 +421,7 @@ describe('orchestrator on the real durable runner (replay semantics)', () => { }); return __durableHandler(event, wrapped, world.deps); }); - const runner = new LocalDurableTestRunner({ + const runner = trackedRunner({ handlerFunction: async (...input) => { const result = await handler(...input); if (result.Status === 'PENDING') { From f794a6ef0e271fc9c5b3af5c7133c1ef85c4c79c Mon Sep 17 00:00:00 2001 From: Matt Houghton <15250993+mattdevdba@users.noreply.github.com> Date: Mon, 2 Nov 2026 16:36:53 +0000 Subject: [PATCH 4/4] fix(agentcore): share the finished init-ws jobs across requests http-server.js builds createInitWsStart per request, so its default completedJobs map was new on every call and a dispatch retry after the job finished started a second job. Keep the map for the container's lifetime beside initJobs. --- lambda/agentcore/http-server.js | 6 +++++- lambda/agentcore/test/init-ws-start.test.js | 16 ++++++++++++++++ 2 files changed, 21 insertions(+), 1 deletion(-) diff --git a/lambda/agentcore/http-server.js b/lambda/agentcore/http-server.js index 41e339168..39b63b4c9 100644 --- a/lambda/agentcore/http-server.js +++ b/lambda/agentcore/http-server.js @@ -405,8 +405,11 @@ const main = async () => { activeJobs: stageJobs, })(p); // Async workspace initialization: init-ws as a background job so a large - // clone is not bounded by the orchestrator Lambda's function timeout. + // clone is not bounded by the orchestrator Lambda's function timeout. The + // job maps live for the container, not the request, so a dispatch retry + // sees the running or finished job. const initJobs = new Map(); + const initCompletedJobs = new Map(); handlers.initWsStart = (p) => createInitWsStart({ initWs: (q) => handlers.initWs(q), @@ -414,6 +417,7 @@ const main = async () => { sendCallbackHeartbeat: sendStageCallbackHeartbeat, busy, activeJobs: initJobs, + completedJobs: initCompletedJobs, })(p); const discussionJobs = new Map(); handlers.discussionAssistStart = (p, context) => diff --git a/lambda/agentcore/test/init-ws-start.test.js b/lambda/agentcore/test/init-ws-start.test.js index 3e2c224b0..ff6fab8d2 100644 --- a/lambda/agentcore/test/init-ws-start.test.js +++ b/lambda/agentcore/test/init-ws-start.test.js @@ -327,6 +327,22 @@ describe('createInitWsStart', () => { expect(sent).toHaveLength(1); }); + it('answers alreadyCompleted when each request builds its own handler over the shared maps', async () => { + // http-server.js builds createInitWsStart per request, so the retry + // only sees the finished job through the container-lifetime maps. + const { deps, sent, beats } = makeDeps(); + const shared = { activeJobs: new Map(), completedJobs: new Map() }; + await createInitWsStart({ ...deps, ...shared })(basePayload); + await flush(); + + const retry = await createInitWsStart({ ...deps, ...shared })(basePayload); + + expect(retry).toMatchObject({ ok: true, accepted: true, alreadyCompleted: true }); + expect(deps.initWs).toHaveBeenCalledOnce(); + expect(beats).toEqual(['cb-init']); + expect(sent).toHaveLength(1); + }); + it('starts a new job for a DIFFERENT callback once the previous one finished', async () => { const { deps } = makeDeps(); const start = createInitWsStart(deps);