diff --git a/api/events.md b/api/events.md index 0195db022..4f7741437 100644 --- a/api/events.md +++ b/api/events.md @@ -1,6 +1,6 @@ # WebSocket Events -最后更新：2026-09-02 +最后更新：2026-09-07 WebSocket 事件合同入口：协议边界、源码 owner、**重复投递/幂等语义**、验收命令。旧长版见 [../docs/history.md](../docs/history.md)。 @@ -30,14 +30,17 @@ Hub/Edge 实时面均为 **at-least-once**：重连、离线队列、outbox、 > **`seq_id`（Hub WS per-conn）vs `seq`（Edge per-bus）**：Hub `seq_id` 是 `PushToConn` 在单连接上单调递增的投递序号，重连从 1 计、跨连接不可比；Edge `EventEnvelope.seq` 是事件总线上的 stream 单调序号，与持久化 `agent_run_events.event_seq` / `messages.seq_id` 对齐。REST 增量同步接口（`GET .../messages/sync?after_seq=`、`GET .../events?after_seq=`）的 `after_seq` 一律指**持久化表的内部 seq**（`messages.seq_id` 或 `agent_run_events.event_seq`），**不是** WS 帧的 `seq_id`。客户端不得用 WS `seq_id` 作为 REST 游标。 -### Hub→Edge `delivery_id` 去重契约（#2101 G2） +### Hub→Edge `delivery_id` admission 契约（#2101 G2 / #2347） -Hub 向 Edge 投递任务有两条并行通道：WS `PushToConn(agent.dispatch)` 与 outbox redispatch HTTP POST `/v1/runs`。两者共享同一个 `delivery_id`（UUID，由 Hub dispatch/outbox 生成并附在 payload 顶层 `delivery_id` / `deliveryId`）。Edge **必须**在消费入口（POST `/v1/runs`）按 `delivery_id` 做进程内去重： +Hub 的 WS `agent.dispatch` 与 outbox HTTP POST `/v1/runs` 共享同一 `delivery_id`。Desktop 将事件中的 `delivery_id` / `deliveryId` 转交为 Edge 请求的 `deliveryId`；空值保留既有无去重路径。 -- **键**：`delivery_id` 字符串；空值视为遗留载荷，跳过 dedup 直接处理。 -- **存储**：进程内 LRU + TTL（参考实现 `edge-server/internal/deliverydedup`，默认 4096 条 / 5 分钟）。不持久化；崩溃后重复投递的最坏后果是幂等重放一次，可接受。 -- **语义**：TTL 窗口内同 `delivery_id` → 跳过 run 创建、返回成功（HTTP 202 + `{deduplicated:true}`），附带日志/指标；不同 `delivery_id` 正常处理。 -- **责任划分**：Hub 保证同一逻辑投递在所有通道使用相同 `delivery_id`；Edge 保证消费端幂等。任一侧失守都会产生重复 run。 +- **原子接收**：先保留 pending claim；仅在 run 接收成功后提交含原 `runId` 的回执，失败或放弃则释放 claim，允许同 ID 重试。 +- **容量与有效期**：进程内缓存默认共容纳 4096 个 pending claim / accepted receipt。成功回执从提交起保留 5 分钟，也可因 LRU 容量压力被淘汰；重放不续期。pending claim 不因 TTL/LRU 被移除，避免首个请求未完成时重复执行。 +- **绑定**：非空 `hubTaskId` 是业务绑定。同一 Hub task 的 HTTP 本地线程与 Desktop 会话线程可以不同，但回执指向同一个原 run；无 `hubTaskId` 的遗留请求按 `projectId` / `threadId` 绑定。缓存内同 delivery ID 的不同绑定返回 409 `delivery_conflict`。 +- **成功重放**：有效回执返回原 run 的正常 202 envelope，包括 `data.runId`、`data.deduplicated: true` 和 `data.deliveryId`；不新建 run、timeline 或 executor。原 run 已删除则返回 404 `not_found`，不静默重建。每次请求先通过 capability 校验；重放还按原 run 实际 project/thread scope 复验。 +- **临时拒绝**：同 ID 正在接收，或容量被 pending claim 占满时，返回 503 `delivery_busy` + `Retry-After`（秒），而不是成功回执。409 `active_run_exists` 表示线程被其他活动 run 占用，也不能当作本次投递成功。Desktop 对这两类拒绝不 ACK、不 FAIL，由现有 Hub outbox 负责重投。 +- **ACK 与业务状态**：每次成功接收或重放都幂等重发 task / relay ACK，以修复丢失的确认；已建立的 run 映射、输出和 running/terminal 状态不因重复投递而回退，业务接收通知只触发一次。 +- **进程边界**：回执不是持久化执行日志，重启后会丢失；跨重启身份核对与执行恢复不由此缓存保证。`queued` 状态本身不能证明子进程尚未启动，不能据此自动重启旧 run。 标签：**UPSERT by id**（稳定 id 合并，禁止第二行）；**idempotent on apply**（再应用不变）；**水位 / watermark**（只前进 `max`）；**ephemeral**（可丢可重，不写持久态）；**非幂等**（须自备去重或 REST）。 diff --git a/api/openapi.yaml b/api/openapi.yaml index 7ccc76272..f93e0cb22 100644 --- a/api/openapi.yaml +++ b/api/openapi.yaml @@ -717,7 +717,7 @@ paths: $ref: "#/components/schemas/StartRunRequest" responses: "202": - description: Run accepted. + description: Run accepted. On an accepted replay this returns the original run envelope (data.runId) with additive deduplicated/deliveryId; no new run/timeline/executor is created. content: application/json: schema: @@ -726,15 +726,30 @@ paths: # capability_token_invalid: the X-AgentHub-Capability-Token is missing, # fails validation, or does not bind this user/device/project/target. $ref: "#/components/responses/Error" - "503": - # not_configured: a Hub user identity reached the Edge but HubJWTSecret is - # empty/misconfigured. The dual-token policy fails closed rather than - # soft-skipping (#899), and the errcode table defines not_configured as 503 - # — an operator fault, not a client credential fault. Declared since #2245: - # before that this path hand-copied 403 at the call site, so a server - # misconfiguration was reported to the client as "your credentials are - # wrong, do not retry". + "409": + # delivery_conflict: delivery_id binds a different Hub task or legacy + # project/thread scope (#2347); active_run_exists: the thread already has + # an active run. Client must not treat it as the same work. + $ref: "#/components/responses/Error" + "404": + # not_found: the original run referenced by an accepted receipt was removed. $ref: "#/components/responses/Error" + "503": + description: > + not_configured means the Edge cannot validate a Hub identity because + its Hub credential policy is unconfigured; it fails closed. + delivery_busy is temporary contention for a pending delivery or + admission capacity and includes Retry-After. Neither is acceptance. + headers: + Retry-After: + description: Delay in seconds, present for delivery_busy responses. + schema: + type: integer + minimum: 1 + content: + application/json: + schema: + $ref: "#/components/schemas/ErrorResponse" /v1/runs/{runId}: get: @@ -8605,6 +8620,9 @@ components: threadId: type: string description: Thread to bind the run to. Defaults to the local thread in P0 compatibility mode. + deliveryId: + type: string + description: Hub delivery_id for dual-channel admission dedup. Empty string bypasses dedup (legacy payload). prompt: type: string description: User message or task description passed to the agent. @@ -8672,7 +8690,11 @@ components: description: Request a no-persistent-session run for runtimes that support it. hubTaskId: type: string - description: Optional Hub task id for Edge-owned Hub callbacks. Desktop bridge usually streams callbacks itself and should avoid duplicate callback paths. + description: > + Optional Hub task identity for Edge-owned callbacks and cross-transport + delivery admission. When set, receipts bind to this business task; + without it, legacy delivery scope binds projectId/threadId. Callback + ownership must still avoid duplicate Edge/Desktop output bridges. AgentInfo: type: object required: [id, name, status] @@ -9571,6 +9593,12 @@ components: finishedAt: type: string format: date-time + deduplicated: + type: boolean + description: True on an accepted replay of an already admitted delivery (same runId returned, no new run created). + deliveryId: + type: string + description: The delivery_id whose accepted receipt is being replayed; present on accepted replays. HubAuditEvent: type: object diff --git a/app/desktop/src/__e2e__/hub-delivery-admission.spec.ts b/app/desktop/src/__e2e__/hub-delivery-admission.spec.ts new file mode 100644 index 000000000..5a288ac7f --- /dev/null +++ b/app/desktop/src/__e2e__/hub-delivery-admission.spec.ts @@ -0,0 +1,237 @@ +import { expect, test, type Page, type WebSocketRoute } from '@playwright/test'; + +// Exercise the normal App -> DesktopHubTaskBridge -> useHubIntegration path. +// Only HTTP/WS boundaries are fixtures: no real token, Hub, Edge or model is used. +// Like oidc-login.spec.ts, CSP bypass lets the reserved .invalid Hub reach routing. +test.use({ bypassCSP: true, viewport: { width: 1440, height: 810 } }); + +const DEVICE = '00000000-0000-0000-0000-000000000123'; +const TASK = 'task-delivery-fixture'; +const RUN = 'run-delivery-fixture'; +const DELIVERY = 'delivery-fixture'; +const RELAY = 'relay-delivery-fixture'; +const TARGET = 'target-delivery-fixture'; +const THREAD = 'thread-delivery-fixture'; +const EMPTY_LIST = { items: [], page: { hasMore: false } }; + +async function readTaskState(page: Page) { + return page.evaluate(async () => { + // Read the actual store; do not seed business state or replace the hook. + const modulePath = '/src/stores/taskBridgeStore.ts'; + const { useTaskBridgeStore } = await import(/* @vite-ignore */ modulePath); + const state = useTaskBridgeStore.getState(); + return { + tasks: state.tasks.map((task: { taskId: string; status: string; runId?: string }) => ({ + taskId: task.taskId, status: task.status, runId: task.runId ?? null, + })), + runToTask: state.runToTask, + }; + }); +} + +async function installDispatchFixture(page: Page, baseURL: string, theme: 'light' | 'dark') { + const appOrigin = new URL(baseURL).origin; + const hubSockets = new Set(); + const edgeSockets = new Set(); + const calls = { + runs: [] as Record[], + acks: [] as Record[], + relayAcks: [] as Record[], + done: [] as Record[], + fails: [] as Record[], + registered: 0, + targetsRead: 0, + rejection: 503, + pageErrors: [] as string[], + unhandledWrites: [] as string[], + }; + + page.on('pageerror', (error) => calls.pageErrors.push(error.message)); + page.on('console', (message) => { + if (message.type() === 'error' && message.text().includes('[ErrorBoundary]')) { + calls.pageErrors.push(message.text()); + } + }); + + await page.addInitScript(({ device, color }) => { + sessionStorage.setItem('agenthub_hub_token', 'fixture-not-a-real-access-token'); + localStorage.setItem('agenthub_token_source', 'tokendance'); + localStorage.setItem('agenthub_device_id', device); + localStorage.setItem('agenthub_onboarding_seen', 'true'); + localStorage.setItem('agenthub-v4-theme', color); + // Select the non-demo application path. Evidence remains fixture-only: + // every Hub/Edge request is intercepted below, including WebSockets. + localStorage.setItem('agenthub.workbench.dataMode', 'observed'); + }, { device: DEVICE, color: theme }); + + await page.route('**/*', async (route) => { + const request = route.request(); + const url = new URL(request.url()); + if (url.origin === appOrigin && request.method() === 'GET' && !['fetch', 'xhr'].includes(request.resourceType())) { + await route.continue(); + return; + } + const json = (data: unknown, status = 200) => route.fulfill({ status, json: data }); + const list = () => json({ code: 'OK', data: EMPTY_LIST }); + if (url.origin === 'https://hub.test.invalid') { + if (url.pathname === '/client/auth/me') { + await json({ code: 'OK', data: { id: 'user-fixture', username: 'fixture', display_name: 'Fixture User' } }); + } else if (url.pathname === '/client/contacts' || url.pathname === '/client/sessions') { + await json({ code: 'OK', data: [] }); + } else if (url.pathname === '/edge/devices:register' || url.pathname === '/edge/devices/register') { + calls.registered++; + await json({ code: 'OK', data: { id: DEVICE } }); + } else if (url.pathname === '/web/execution-targets') { + calls.targetsRead++; + await json({ code: 'OK', data: { + items: [{ id: TARGET, name: 'Fixture Local Edge', device_id: DEVICE, target_type: 'local_edge', health_state: 'healthy', is_online: true }], + page: { hasMore: false }, + } }); + } else if (url.pathname === '/edge/agent-tasks/' + TASK + '/ack') { + calls.acks.push(request.postDataJSON()); + await json({ code: calls.acks.length === 1 ? 'ERROR' : 'OK' }, calls.acks.length === 1 ? 500 : 200); + } else if (url.pathname === '/web/relay/commands/' + RELAY + '/device-ack') { + calls.relayAcks.push(request.postDataJSON()); + await json({ code: calls.relayAcks.length === 1 ? 'ERROR' : 'OK' }, calls.relayAcks.length === 1 ? 500 : 200); + } else if (url.pathname === '/edge/agent-tasks/' + TASK + '/done') { + calls.done.push(request.postDataJSON()); + await json({ code: 'OK' }); + } else if (url.pathname === '/edge/agent-tasks/' + TASK + '/fail') { + calls.fails.push(request.postDataJSON()); + await json({ code: 'OK' }); + } else if (url.pathname === '/edge/agent-tasks/' + TASK + '/stream') { + await json({ code: 'OK' }); + } else if (request.method() === 'GET') { + await list(); + } else { + calls.unhandledWrites.push(request.method() + ' ' + url.pathname); + await route.abort(); + } + return; + } + if (url.origin === 'http://127.0.0.1:3210') { + if (url.pathname === '/v1/health') { + await json({ code: 'OK', data: { status: 'ok', version: 'fixture', edgeId: 'edge-fixture' } }); + } else if (url.pathname === '/v1/model-catalog') { + await json({ code: 'OK', data: { items: [], sources: [] } }); + } else if (url.pathname === '/v1/threads' && request.method() === 'POST') { + await json({ code: 'OK', data: { threadId: THREAD, projectId: 'proj_local' } }, 201); + } else if (url.pathname === '/v1/runs' && request.method() === 'POST') { + calls.runs.push(request.postDataJSON()); + if (calls.rejection) { + await route.fulfill({ status: calls.rejection, headers: { 'Retry-After': '1' }, json: { + error: { code: calls.rejection === 503 ? 'delivery_busy' : 'internal_error', message: 'fixture admission rejection', traceId: 'fixture-trace' }, + } }); + } else { + await json({ code: 'OK', data: { runId: RUN, projectId: 'proj_local', threadId: THREAD, status: 'queued', deduplicated: calls.runs.length > 2, deliveryId: DELIVERY } }, 202); + } + } else if (request.method() === 'GET') { + await list(); + } else { + calls.unhandledWrites.push(request.method() + ' ' + url.pathname); + await route.abort(); + } + return; + } + // Unknown APIs, font CDNs and any live host fail closed; no fallback network. + await route.abort(); + }); + + await page.routeWebSocket('**', (socket) => { + const url = new URL(socket.url()); + if (url.host === new URL(appOrigin).host) { + socket.connectToServer(); // Vite HMR only, on the task's own renderer server. + } else if (url.hostname === 'hub.test.invalid' && url.pathname === '/client/ws') { + hubSockets.add(socket); + socket.onClose(() => hubSockets.delete(socket)); + socket.send(JSON.stringify({ type: 'auth.ok', payload: null })); + } else if (url.origin === 'ws://127.0.0.1:3210' && url.pathname === '/v1/events') { + edgeSockets.add(socket); + socket.onClose(() => edgeSockets.delete(socket)); + } else { + socket.close(); + } + }); + + await page.goto('/'); + await expect(page.getByTestId('agenthub-workbench')).toBeVisible(); + await expect.poll(() => calls.registered).toBeGreaterThan(0); + await expect.poll(() => calls.targetsRead).toBeGreaterThan(0); + await expect.poll(() => hubSockets.size).toBeGreaterThan(0); + await expect.poll(() => edgeSockets.size).toBeGreaterThan(0); + // Wait for the target query's React effects before sending the first frame. + await page.evaluate(() => new Promise((resolve) => requestAnimationFrame(() => requestAnimationFrame(() => resolve())))); + + return { + calls, + async dispatch() { + const response = page.waitForResponse((result) => + result.url() === 'http://127.0.0.1:3210/v1/runs' && result.request().method() === 'POST', + ); + const frame = { type: 'agent.dispatch', payload: { + relay_command_id: RELAY, command_type: 'agent.dispatch', payload: JSON.stringify({ + task_id: TASK, delivery_id: DELIVERY, prompt: 'Fixture task', thread_id: THREAD, + agent_type: 'codex', target_id: TARGET, edge_device_id: DEVICE, + }), + } }; + for (const socket of hubSockets) socket.send(JSON.stringify(frame)); + await (await response).finished(); + // Negative assertions must wait until the error body and React effects + // have been consumed, not just until the mock records the request. + await page.evaluate(() => new Promise((resolve) => requestAnimationFrame(() => requestAnimationFrame(() => resolve())))); + }, + finish() { + for (const socket of edgeSockets) { + socket.send(JSON.stringify({ version: 'v1', id: 'fixture-finished', seq: 1, type: 'run.finished', ts: new Date().toISOString(), scope: { runId: RUN, threadId: THREAD }, payload: { runId: RUN } })); + } + }, + }; +} + +for (const theme of ['light', 'dark'] as const) { + test('actual Desktop bridge retries admission and repairs lost ACKs (' + theme + ')', async ({ page, baseURL }, testInfo) => { + if (!baseURL) throw new Error('Desktop E2E baseURL is required'); + const { calls, dispatch, finish } = await installDispatchFixture(page, baseURL, theme); + await dispatch(); + await expect.poll(() => calls.runs.length).toBe(1); + await expect.poll(() => readTaskState(page)).toEqual({ tasks: [{ taskId: TASK, status: 'queued', runId: null }], runToTask: {} }); + expect(calls.runs[0]).toMatchObject({ deliveryId: DELIVERY, hubTaskId: TASK, targetId: TARGET, edgeDeviceId: DEVICE }); + expect(calls.acks).toHaveLength(0); + expect(calls.relayAcks).toHaveLength(0); + expect(calls.fails).toHaveLength(0); + + calls.rejection = 0; + await dispatch(); + await expect.poll(() => calls.acks.length).toBe(1); + await expect.poll(() => calls.relayAcks.length).toBe(1); + await expect.poll(() => readTaskState(page)).toEqual({ tasks: [{ taskId: TASK, status: 'running', runId: RUN }], runToTask: { [RUN]: TASK } }); + + // Both first ACK requests failed. A successful replay must send them again. + await dispatch(); + await expect.poll(() => calls.acks.length).toBe(2); + await expect.poll(() => calls.relayAcks.length).toBe(2); + expect(calls.acks).toEqual([{ run_id: RUN }, { run_id: RUN }]); + expect(calls.runs).toHaveLength(3); + expect(calls.fails).toHaveLength(0); + + finish(); + await expect.poll(() => calls.done.length).toBe(1); + const finished = { tasks: [{ taskId: TASK, status: 'done', runId: RUN }], runToTask: { [RUN]: TASK } }; + await expect.poll(() => readTaskState(page)).toEqual(finished); + await dispatch(); + await expect.poll(() => calls.acks.length).toBe(3); + await expect.poll(() => calls.relayAcks.length).toBe(3); + await expect.poll(() => readTaskState(page)).toEqual(finished); + expect(calls.done).toHaveLength(1); + + calls.rejection = 500; + await dispatch(); + await expect.poll(() => calls.runs.length).toBe(5); + await expect.poll(() => readTaskState(page)).toEqual(finished); + expect(calls.fails).toHaveLength(0); + expect(calls.acks).toHaveLength(3); + expect(calls.pageErrors).toEqual([]); + expect(calls.unhandledWrites).toEqual([]); + await page.screenshot({ path: testInfo.outputPath('delivery-admission-' + theme + '.png') }); + }); +} diff --git a/app/desktop/src/__tests__/useHubIntegration.test.ts b/app/desktop/src/__tests__/useHubIntegration.test.ts index 4c8642515..564502044 100644 --- a/app/desktop/src/__tests__/useHubIntegration.test.ts +++ b/app/desktop/src/__tests__/useHubIntegration.test.ts @@ -213,6 +213,7 @@ describe('useHubIntegration', () => { request: vi.fn(), registerDevice: vi.fn().mockResolvedValue({ id: 'dev-1' }), ackTask: vi.fn().mockResolvedValue(undefined), + ackRelayCommand: vi.fn().mockResolvedValue(undefined), streamTask: vi.fn().mockResolvedValue(undefined), streamTaskEvent: vi.fn().mockResolvedValue(undefined), doneTask: vi.fn().mockResolvedValue(undefined), @@ -314,6 +315,25 @@ describe('useHubIntegration', () => { }); } + function mockRunCreateResponseWithStatus(body: Record, status: number) { + fetchMock.mockImplementation(async (input: unknown) => { + const url = String(input); + if (url.endsWith('/v1/threads')) { + return new Response(JSON.stringify({ threadId: 'thread-ok' }), { + status: 201, + headers: { 'Content-Type': 'application/json' }, + }); + } + if (url.endsWith('/v1/runs')) { + return new Response(JSON.stringify(body), { + status, + headers: { 'Content-Type': 'application/json' }, + }); + } + return new Response('{}', { status: 200, headers: { 'Content-Type': 'application/json' } }); + }); + } + // ── agent.dispatch → Edge run ────────────────────────── it('acks task and starts Edge run on agent.dispatch', async () => { @@ -600,6 +620,295 @@ describe('useHubIntegration', () => { expect(hubClient.ackTask).not.toHaveBeenCalled(); }); + it('forwards Hub delivery_id into the Edge run body', async () => { + renderHook(() => useHubIntegration({ hubWS, hubClient })); + + await act(async () => { + fireHubEvent( + HUB_EVENTS.AGENT_DISPATCH, + makeDispatchPayload({ delivery_id: 'delivery-snake' }), + ); + }); + + expect(fetchBodyFor('/v1/runs').deliveryId).toBe('delivery-snake'); + }); + + it('forwards Hub deliveryId (camelCase) into the Edge run body', async () => { + renderHook(() => useHubIntegration({ hubWS, hubClient })); + + await act(async () => { + fireHubEvent( + HUB_EVENTS.AGENT_DISPATCH, + makeDispatchPayload({ deliveryId: 'delivery-camel' }), + ); + }); + + expect(fetchBodyFor('/v1/runs').deliveryId).toBe('delivery-camel'); + }); + + it('omits deliveryId from the Edge run body when no Hub delivery id is present (legacy)', async () => { + renderHook(() => useHubIntegration({ hubWS, hubClient })); + + await act(async () => { + fireHubEvent(HUB_EVENTS.AGENT_DISPATCH, makeDispatchPayload()); + }); + + const body = fetchBodyFor('/v1/runs'); + expect('deliveryId' in body).toBe(false); + }); + + it('duplicate dispatch keeps one run mapping, does not overwrite progress, and idempotently re-acks', async () => { + mockRunSequence('run-1', 'run-1'); + const onDispatch = vi.fn(); + renderHook(() => useHubIntegration({ hubWS, hubClient, onDispatch })); + + await act(async () => { + fireHubEvent( + HUB_EVENTS.AGENT_DISPATCH, + makeDispatchPayload({ delivery_id: 'd1' }), + ); + }); + expect(hoisted.storeTasks).toHaveLength(1); + expect(hoisted.storeTasks[0]?.runId).toBe('run-1'); + expect(hoisted.storeTasks[0]?.status).toBe('running'); + expect(hubClient.ackTask).toHaveBeenCalledTimes(1); + expect(onDispatch).toHaveBeenCalledTimes(1); + + await act(async () => { + fireHubEvent( + HUB_EVENTS.AGENT_DISPATCH, + makeDispatchPayload({ delivery_id: 'd1' }), + ); + }); + + expect(hoisted.storeTasks).toHaveLength(1); + expect(hoisted.storeTasks[0]?.runId).toBe('run-1'); + expect(hoisted.storeTasks[0]?.status).toBe('running'); + expect(hoisted.storeRunToTask['run-1']).toBe('task-1'); + expect(hubClient.ackTask).toHaveBeenCalledTimes(2); + expect(hubClient.failTask).not.toHaveBeenCalled(); + expect(onDispatch).toHaveBeenCalledTimes(1); + }); + + it('does not downgrade a terminal task on a replayed dispatch, but re-acks idempotently', async () => { + mockRunSequence('run-1'); + renderHook(() => useHubIntegration({ hubWS, hubClient })); + + await act(async () => { + fireHubEvent( + HUB_EVENTS.AGENT_DISPATCH, + makeDispatchPayload({ delivery_id: 'd1' }), + ); + }); + act(() => { + fireEdgeEvent(makeEvent('run.finished', { runId: 'run-1' })); + }); + expect(hoisted.storeTasks[0]?.status).toBe('done'); + expect(hubClient.ackTask).toHaveBeenCalledTimes(1); + + mockRunSequence('run-1'); + await act(async () => { + fireHubEvent( + HUB_EVENTS.AGENT_DISPATCH, + makeDispatchPayload({ delivery_id: 'd1' }), + ); + }); + + expect(hoisted.storeTasks).toHaveLength(1); + expect(hoisted.storeTasks[0]?.status).toBe('done'); + expect(hoisted.storeTasks[0]?.runId).toBe('run-1'); + expect(hubClient.ackTask).toHaveBeenCalledTimes(2); + expect(hubClient.failTask).not.toHaveBeenCalled(); + }); + + it('keeps a delivery_busy rejection queued without acking or failing', async () => { + mockRunCreateResponseWithStatus({ error: { code: 'delivery_busy', message: 'busy', traceId: 'trace_001' } }, 503); + renderHook(() => useHubIntegration({ hubWS, hubClient })); + + await act(async () => { + fireHubEvent( + HUB_EVENTS.AGENT_DISPATCH, + makeDispatchPayload({ delivery_id: 'd1' }), + ); + }); + + expect(hoisted.storeTasks).toHaveLength(1); + expect(hoisted.storeTasks[0]?.status).toBe('queued'); + expect(hoisted.storeTasks[0]?.runId).toBeUndefined(); + expect(hubClient.ackTask).not.toHaveBeenCalled(); + expect(hubClient.failTask).not.toHaveBeenCalled(); + }); + + it('keeps an existing active run on active_run_exists without acking or failing', async () => { + mockRunSequence('run-1'); + renderHook(() => useHubIntegration({ hubWS, hubClient })); + + await act(async () => { + fireHubEvent( + HUB_EVENTS.AGENT_DISPATCH, + makeDispatchPayload({ delivery_id: 'd1' }), + ); + }); + expect(hoisted.storeTasks[0]?.status).toBe('running'); + + mockRunCreateResponseWithStatus({ error: { code: 'active_run_exists', message: 'active', traceId: 'trace_001' } }, 409); + await act(async () => { + fireHubEvent( + HUB_EVENTS.AGENT_DISPATCH, + makeDispatchPayload({ delivery_id: 'd1' }), + ); + }); + + expect(hoisted.storeTasks).toHaveLength(1); + expect(hoisted.storeTasks[0]?.status).toBe('running'); + expect(hoisted.storeTasks[0]?.runId).toBe('run-1'); + expect(hubClient.ackTask).toHaveBeenCalledTimes(1); + expect(hubClient.failTask).not.toHaveBeenCalled(); + }); + + it('does not ack a relay command on a delivery_busy rejection', async () => { + mockRunCreateResponseWithStatus({ error: { code: 'delivery_busy', message: 'busy', traceId: 'trace_001' } }, 503); + renderHook(() => + useHubIntegration({ + hubWS, + hubClient, + dispatchTarget: { targetId: 'target-current', deviceId: 'desktop-current' }, + }), + ); + + const relayFrame = { + relay_command_id: 'relay-1', + command_type: 'agent.dispatch', + payload: JSON.stringify( + makeDispatchPayload({ + target_id: 'target-current', + edge_device_id: 'desktop-current', + delivery_id: 'd1', + }), + ), + }; + await act(async () => { + fireHubEvent(HUB_EVENTS.AGENT_DISPATCH, relayFrame); + }); + + expect(hubClient.ackRelayCommand).not.toHaveBeenCalled(); + expect(hubClient.ackTask).not.toHaveBeenCalled(); + expect(hubClient.failTask).not.toHaveBeenCalled(); + }); + + it('still fails a permanent Edge admission rejection (e.g. 500)', async () => { + mockRunCreateResponseWithStatus({ error: { code: 'internal_error', message: 'boom', traceId: 'trace_001' } }, 500); + renderHook(() => useHubIntegration({ hubWS, hubClient })); + + await act(async () => { + fireHubEvent( + HUB_EVENTS.AGENT_DISPATCH, + makeDispatchPayload({ delivery_id: 'd1' }), + ); + }); + + expect(hoisted.storeTasks[0]?.status).toBe('failed'); + expect(hubClient.ackTask).not.toHaveBeenCalled(); + expect(hubClient.failTask).toHaveBeenCalled(); + }); + + it('re-acks a successful replay when the first ACK was lost in transit', async () => { + mockRunSequence('run-1', 'run-1'); + const ackTaskMock = hubClient.ackTask as ReturnType; + ackTaskMock.mockRejectedValueOnce(new Error('ACK transport lost')).mockResolvedValue(undefined); + + renderHook(() => useHubIntegration({ hubWS, hubClient })); + + await act(async () => { + fireHubEvent( + HUB_EVENTS.AGENT_DISPATCH, + makeDispatchPayload({ delivery_id: 'd1' }), + ); + }); + expect(hoisted.storeTasks).toHaveLength(1); + expect(hoisted.storeTasks[0]?.runId).toBe('run-1'); + expect(hoisted.storeTasks[0]?.status).toBe('running'); + expect(ackTaskMock).toHaveBeenCalledTimes(1); + + await act(async () => { + fireHubEvent( + HUB_EVENTS.AGENT_DISPATCH, + makeDispatchPayload({ delivery_id: 'd1' }), + ); + }); + + expect(ackTaskMock).toHaveBeenCalledTimes(2); + expect(hoisted.storeTasks[0]?.runId).toBe('run-1'); + expect(hoisted.storeTasks[0]?.status).toBe('running'); + expect(hubClient.failTask).not.toHaveBeenCalled(); + }); + + it('re-acks a relay command ACK on a successful replay when the first was lost', async () => { + mockRunSequence('run-1', 'run-1'); + const relayAckMock = hubClient.ackRelayCommand as ReturnType; + relayAckMock.mockRejectedValueOnce(new Error('relay ACK lost')).mockResolvedValue(undefined); + + renderHook(() => + useHubIntegration({ + hubWS, + hubClient, + dispatchTarget: { targetId: 'target-current', deviceId: 'desktop-current' }, + }), + ); + + const relayFrame = { + relay_command_id: 'relay-1', + command_type: 'agent.dispatch', + payload: JSON.stringify( + makeDispatchPayload({ + target_id: 'target-current', + edge_device_id: 'desktop-current', + delivery_id: 'd1', + }), + ), + }; + await act(async () => { + fireHubEvent(HUB_EVENTS.AGENT_DISPATCH, relayFrame); + }); + expect(relayAckMock).toHaveBeenCalledTimes(1); + expect(hubClient.ackTask).toHaveBeenCalledTimes(1); + + await act(async () => { + fireHubEvent(HUB_EVENTS.AGENT_DISPATCH, relayFrame); + }); + + expect(relayAckMock).toHaveBeenCalledTimes(2); + expect(hubClient.ackTask).toHaveBeenCalledTimes(2); + expect(hubClient.failTask).not.toHaveBeenCalled(); + }); + + it('does not downgrade an already-running task when a duplicate delivery fails permanently', async () => { + mockRunSequence('run-1'); + renderHook(() => useHubIntegration({ hubWS, hubClient })); + + await act(async () => { + fireHubEvent( + HUB_EVENTS.AGENT_DISPATCH, + makeDispatchPayload({ delivery_id: 'd1' }), + ); + }); + expect(hoisted.storeTasks[0]?.status).toBe('running'); + expect(hoisted.storeTasks[0]?.runId).toBe('run-1'); + + mockRunCreateResponseWithStatus({ error: { code: 'internal_error', message: 'boom', traceId: 'trace_001' } }, 500); + await act(async () => { + fireHubEvent( + HUB_EVENTS.AGENT_DISPATCH, + makeDispatchPayload({ delivery_id: 'd1' }), + ); + }); + + expect(hoisted.storeTasks).toHaveLength(1); + expect(hoisted.storeTasks[0]?.status).toBe('running'); + expect(hoisted.storeTasks[0]?.runId).toBe('run-1'); + expect(hubClient.failTask).not.toHaveBeenCalled(); + }); + // ── Edge events → Hub callbacks ────────────────────── it('streams text_delta to Hub', async () => { diff --git a/app/desktop/src/hooks/hubIntegrationHelpers.test.ts b/app/desktop/src/hooks/hubIntegrationHelpers.test.ts index 7d046fbda..0737dbf44 100644 --- a/app/desktop/src/hooks/hubIntegrationHelpers.test.ts +++ b/app/desktop/src/hooks/hubIntegrationHelpers.test.ts @@ -17,7 +17,9 @@ import { extractCreatedRunId, extractRunOutputBatch, getTeamRouteContext, + hasTaskProgressed, isTerminalBridgeTask, + isTransientAdmissionRejection, normalizeRouteDecision, normalizeRuntimeAgentId, parsePermissionDecisionControl, @@ -308,11 +310,67 @@ describe('hubIntegrationMappers', () => { expect(() => extractCreatedRunId({ code: 'ok', data: {} })).toThrow(/no id\/runId/); }); + it('buildEdgeRunBody forwards deliveryId from snake_case and camelCase aliases, omitting it for legacy', () => { + const base = { task_id: 'task-1' }; + const snake = buildEdgeRunBody({ ...base, delivery_id: 'd-1' }, 't', 'p', 'a', null); + expect(snake.deliveryId).toBe('d-1'); + + const camel = buildEdgeRunBody({ ...base, deliveryId: 'd-2' }, 't', 'p', 'a', null); + expect(camel.deliveryId).toBe('d-2'); + + const legacy = buildEdgeRunBody({ ...base }, 't', 'p', 'a', null); + expect(legacy.deliveryId).toBeUndefined(); + expect('deliveryId' in legacy).toBe(false); + }); + + it('extractCreatedRunId accepts a deduplicated accepted run envelope (normal replay)', () => { + expect( + extractCreatedRunId({ + code: 'ok', + data: { runId: 'run-replay-1', deduplicated: true, deliveryId: 'd-1' }, + }), + ).toBe('run-replay-1'); + expect( + extractCreatedRunId({ + code: 'ok', + data: { id: 'run-replay-2', deduplicated: true, deliveryId: 'd-2' }, + }), + ).toBe('run-replay-2'); + }); + + it('classifies transient admission rejections from the canonical Edge error envelope', () => { + const busyEnvelope = { error: { code: 'delivery_busy', message: 'busy', traceId: 'trace_001' } }; + const activeEnvelope = { error: { code: 'active_run_exists', message: 'active', traceId: 'trace_001' } }; + + expect(isTransientAdmissionRejection(503, busyEnvelope)).toBe(true); + expect(isTransientAdmissionRejection(409, activeEnvelope)).toBe(true); + // Raw JSON string form (what the hook passes from runResp.text()). + expect(isTransientAdmissionRejection(503, JSON.stringify(busyEnvelope))).toBe(true); + expect(isTransientAdmissionRejection(409, JSON.stringify(activeEnvelope))).toBe(true); + + // Non-transient / non-matching envelopes keep the existing failure path. + expect(isTransientAdmissionRejection(409, { error: { code: 'delivery_conflict', message: 'x', traceId: 't' } })).toBe(false); + expect(isTransientAdmissionRejection(500, busyEnvelope)).toBe(false); + expect(isTransientAdmissionRejection(503, { error: { code: 'internal', message: 'x', traceId: 't' } })).toBe(false); + expect(isTransientAdmissionRejection(503, 'not-json')).toBe(false); + expect(isTransientAdmissionRejection(200, {})).toBe(false); + // A flat top-level {code} is not the canonical envelope — must not match. + expect(isTransientAdmissionRejection(503, { code: 'delivery_busy' })).toBe(false); + }); + it('isTerminalBridgeTask detects done/failed only', () => { expect(isTerminalBridgeTask(makeTask({ status: 'done' }))).toBe(true); expect(isTerminalBridgeTask(makeTask({ status: 'failed' }))).toBe(true); expect(isTerminalBridgeTask(makeTask({ status: 'running' }))).toBe(false); }); + + it('hasTaskProgressed is true once a task has a runId or reached running/terminal', () => { + expect(hasTaskProgressed(undefined)).toBe(false); + expect(hasTaskProgressed(makeTask({ status: 'queued' }))).toBe(false); + expect(hasTaskProgressed(makeTask({ status: 'running', runId: 'r1' }))).toBe(true); + expect(hasTaskProgressed(makeTask({ status: 'done' }))).toBe(true); + expect(hasTaskProgressed(makeTask({ status: 'failed' }))).toBe(true); + }); }); describe('parseDispatchFrame', () => { diff --git a/app/desktop/src/hooks/hubIntegrationMappers.ts b/app/desktop/src/hooks/hubIntegrationMappers.ts index 3e772749f..ec3b4d8de 100644 --- a/app/desktop/src/hooks/hubIntegrationMappers.ts +++ b/app/desktop/src/hooks/hubIntegrationMappers.ts @@ -44,6 +44,19 @@ export function isTerminalBridgeTask(task: AgentTask): boolean { return task.status === 'done' || task.status === 'failed'; } +/** True once a task has been accepted by Edge (has a runId) or already reached + * a running/terminal status. Used to prevent a duplicate delivery from + * downgrading or failing an already-progressed task. */ +export function hasTaskProgressed(task: AgentTask | undefined): boolean { + return ( + task !== undefined && + (task.runId !== undefined || + task.status === 'running' || + task.status === 'done' || + task.status === 'failed') + ); +} + export function normalizeRouteDecision(value: unknown): CoordinatorRouteDecision | null { const record = parseRecord(value); const nested = parseRecord(record.decision); @@ -287,6 +300,7 @@ export function buildEdgeRunBody( parseStringRecord(modelParams.configOverrides), ephemeral: getFirstBoolean(modelParams.ephemeral, data.ephemeral), hubTaskId: getFirstString(data.task_id), + deliveryId: getFirstString(data.delivery_id, data.deliveryId), targetId: targetBinding?.expectedTargetId, edgeDeviceId: targetBinding?.expectedEdgeDeviceId, dispatchTargetEvidence: targetBinding @@ -324,6 +338,26 @@ export function extractCreatedRunId(value: unknown): string { return runId; } +/** + * Classify a definite transient Edge admission rejection. Only these two codes + * are safe to leave queued for the Hub outbox to retry: a busy delivery slot + * (503 delivery_busy) and a Hub thread already occupied by an active run + * (409 active_run_exists). Any other non-OK response keeps the existing + * failure handling rather than being treated as a retryable admission. + * + * Reads the canonical Edge error envelope ({ error: { code, message, traceId } }), + * which may arrive either as an already-parsed object or as a raw JSON string. + */ +export function isTransientAdmissionRejection(status: number, body: unknown): boolean { + const record = parseRecord(body); + const error = parseRecord(record.error); + const code = getFirstString(error.code); + return ( + (status === 503 && code === 'delivery_busy') || + (status === 409 && code === 'active_run_exists') + ); +} + export const FINAL_OUTPUT_MAX_CHARS = 32_000; /** diff --git a/app/desktop/src/hooks/useHubIntegration.ts b/app/desktop/src/hooks/useHubIntegration.ts index 8fbf621a4..0666a3e4b 100644 --- a/app/desktop/src/hooks/useHubIntegration.ts +++ b/app/desktop/src/hooks/useHubIntegration.ts @@ -33,7 +33,9 @@ import { extractRunOutputBatch, FINAL_OUTPUT_MAX_CHARS, getTeamRouteContext, + hasTaskProgressed, isTerminalBridgeTask, + isTransientAdmissionRejection, normalizeRuntimeAgentId, parsePermissionDecisionControl, permissionDecisionControlKey, @@ -339,19 +341,23 @@ export function useHubIntegration(options: HubIntegrationOptions): HubIntegratio console.info('[useHubIntegration] Relay command target mismatch, skipping:', relayCommandId, targetError); return; } - store.getState().addTask({ - taskId, - agentId: normalizeRuntimeAgentId( - getString(data, 'agent_type') || getString(data, 'agent_id'), - ), - prompt: getString(data, 'prompt') || getString(data, 'content'), - threadId: getString(data, 'thread_id') || getString(data, 'session_id') || 'hub-dispatch', - status: 'failed', - dispatchPayload, - error: targetError, - createdAt: new Date().toISOString(), - }); - void catchHubReport(`failTask:${taskId}`, hubClient.failTask(taskId, targetError)); + // A duplicate delivery with a target mismatch must not corrupt an already + // running/terminal task. Only a not-yet-accepted task is failed here. + if (!hasTaskProgressed(store.getState().tasks.find((t) => t.taskId === taskId))) { + store.getState().addTask({ + taskId, + agentId: normalizeRuntimeAgentId( + getString(data, 'agent_type') || getString(data, 'agent_id'), + ), + prompt: getString(data, 'prompt') || getString(data, 'content'), + threadId: getString(data, 'thread_id') || getString(data, 'session_id') || 'hub-dispatch', + status: 'failed', + dispatchPayload, + error: targetError, + createdAt: new Date().toISOString(), + }); + void catchHubReport(`failTask:${taskId}`, hubClient.failTask(taskId, targetError)); + } return; } @@ -395,19 +401,36 @@ export function useHubIntegration(options: HubIntegrationOptions): HubIntegratio ); if (!runResp.ok) { + // Classify definite transient admission rejections (delivery_busy / + // active_run_exists). Anything else keeps the existing permanent path. const errorText = await runResp.text().catch(() => 'Unknown error'); + if (isTransientAdmissionRejection(runResp.status, errorText)) { + // Keep the task queued/waiting; retry ownership stays with the Hub + // outbox. Do NOT ackTask / failTask / ackRelayCommand, and do not + // invent a second retry loop here. + return; + } throw new Error(`Edge POST /v1/runs returned ${runResp.status}: ${errorText}`); } const runId = extractCreatedRunId(await runResp.json()); - // Map taskId ↔ runId and mark running - store.getState().updateTask(taskId, { runId, status: 'running' }); + // Re-read before update to resist an async completion race: a duplicate + // dispatch must not downgrade a task that already reached running/done/ + // failed, nor clobber an existing runId/output mapping. + const existingTask = store.getState().tasks.find((t) => t.taskId === taskId); + const taskProgressed = hasTaskProgressed(existingTask); - // Acknowledge task to Hub (log failures — do not throw into dispatch handler) - void catchHubReport(`ackTask:${taskId}`, hubClient.ackTask(taskId, runId)); + // Business state/mapping is only written once (first accepted instance). + if (!taskProgressed) { + store.getState().updateTask(taskId, { runId, status: 'running' }); + } - // Ack relay command if this was a relay-dispatched task + // Every accepted delivery (first accept AND successful replay) is + // idempotently acknowledged. If the first ACK was lost in transit, Hub + // re-dispatches and we must re-ACK so the outbox/relay can converge; a + // double ACK here is expected recovery, not a race defect. + void catchHubReport(`ackTask:${taskId}`, hubClient.ackTask(taskId, runId)); if (relayCommandId && dispatchTarget?.deviceId) { void catchHubReport( `ackRelayCommand:${relayCommandId}`, @@ -415,18 +438,27 @@ export function useHubIntegration(options: HubIntegrationOptions): HubIntegratio ); } - // Notify consumer - const updatedTask = store.getState().tasks.find((t) => t.taskId === taskId); - if (updatedTask) { - onDispatch?.(updatedTask); + // onDispatch is a business "newly accepted" notification — a successful + // replay must not re-trigger it (transport ACK != business notification). + if (!taskProgressed) { + const updatedTask = store.getState().tasks.find((t) => t.taskId === taskId); + if (updatedTask) { + onDispatch?.(updatedTask); + } } } catch (err) { const errorMsg = err instanceof Error ? err.message : String(err); - store.getState().updateTask(taskId, { - status: 'failed', - error: errorMsg, - }); - void catchHubReport(`failTask:${taskId}`, hubClient.failTask(taskId, errorMsg)); + // A duplicate delivery's permanent failure must not corrupt an already + // running/terminal task. Only a not-yet-accepted delivery is failed here + // (permanent errors on new queued tasks still fail). + const currentTask = store.getState().tasks.find((t) => t.taskId === taskId); + if (!hasTaskProgressed(currentTask)) { + store.getState().updateTask(taskId, { + status: 'failed', + error: errorMsg, + }); + void catchHubReport(`failTask:${taskId}`, hubClient.failTask(taskId, errorMsg)); + } } }); diff --git a/edge-server/internal/api/delivery_admission_test.go b/edge-server/internal/api/delivery_admission_test.go new file mode 100644 index 000000000..e47d8bdce --- /dev/null +++ b/edge-server/internal/api/delivery_admission_test.go @@ -0,0 +1,581 @@ +package api + +import ( + "encoding/json" + "fmt" + "io" + "net/http" + "net/http/httptest" + "strings" + "sync" + "testing" + "time" + + "github.com/agenthub/edge-server/internal/lifecycle" + "github.com/agenthub/edge-server/internal/store" + "github.com/agenthub/pkg/jwtutil" +) + +// admissionExecutor is a controllable RunExecutor used to prove the +// delivery-admission contract without mocking PostRuns/runcontrol: +// - counts every Start call so tests can assert "second run started once" +// - can fail the next Start (owner release after a failed attempt) +// - can block the next Start until release (holds the admission window open) +type admissionExecutor struct { + mu sync.Mutex + starts []store.Run + err error + blockNext bool + failNext bool + entered chan string // buffered; signalled when a blocking Start is entered + release chan struct{} +} + +func (e *admissionExecutor) Start(run store.Run, ctx lifecycle.RunProcessContext) error { + e.mu.Lock() + e.starts = append(e.starts, run) + fail := e.failNext + block := e.blockNext + e.failNext = false + e.blockNext = false + e.mu.Unlock() + if fail { + return e.err + } + if block { + if e.entered != nil { + select { + case e.entered <- run.ID: + default: + } + } + if e.release != nil { + <-e.release + } + } + return nil +} + +func (e *admissionExecutor) StartCount() int { + e.mu.Lock() + defer e.mu.Unlock() + return len(e.starts) +} + +func (e *admissionExecutor) Cancel(runID string) lifecycle.CancelResult { + return lifecycle.CancelResult{Found: false} +} + +// newDeliveryTestServer builds a real httptest server with a controllable +// executor. config may override Handler fields (e.g. HubJWTSecret/EdgeDeviceID). +func newDeliveryTestServer(t *testing.T, exec lifecycle.RunExecutor, config func(*Handler)) (*httptest.Server, *Handler) { + t.Helper() + h := newTestHandler() + if h.Bus != nil { + t.Cleanup(func() { + if err := h.Bus.Close(); err != nil { + t.Errorf("close event bus: %v", err) + } + }) + } + if exec != nil { + h.Executor = exec + } + if config != nil { + config(h) + } + h.WorkspaceAllowlist = []string{t.TempDir()} + mux := http.NewServeMux() + h.RegisterRoutes(mux) + return httptest.NewServer(mux), h +} + +type postResult struct { + status int + body map[string]any + resp *http.Response +} + +// httpPostJSON performs POST /v1/runs without touching testing.T, so it is safe +// to call from a spawned goroutine (unlike helpers that call t.Fatalf). +func httpPostJSON(url, body string) (postResult, error) { + resp, err := http.Post(url+"/v1/runs", "application/json", strings.NewReader(body)) + if err != nil { + return postResult{}, err + } + defer resp.Body.Close() + raw, _ := io.ReadAll(resp.Body) + var m map[string]any + _ = json.Unmarshal(raw, &m) + return postResult{status: resp.StatusCode, body: m, resp: resp}, nil +} + +func postRunsRaw(t *testing.T, serverURL, body string) postResult { + t.Helper() + pr, err := httpPostJSON(serverURL, body) + if err != nil { + t.Fatalf("POST %s: %v", serverURL, err) + } + return pr +} + +func doReq(t *testing.T, req *http.Request) postResult { + t.Helper() + resp, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatalf("do req: %v", err) + } + defer resp.Body.Close() + raw, _ := io.ReadAll(resp.Body) + var m map[string]any + _ = json.Unmarshal(raw, &m) + return postResult{status: resp.StatusCode, body: m, resp: resp} +} + +func admissionRunBody(workDir, deliveryID, hubTaskID string, extra map[string]any) string { + m := map[string]any{"prompt": "admission-regression-run", "workDir": workDir} + if deliveryID != "" { + m["deliveryId"] = deliveryID + } + if hubTaskID != "" { + m["hubTaskId"] = hubTaskID + } + for k, v := range extra { + m[k] = v + } + b, _ := json.Marshal(m) + return string(b) +} + +func errCode(body map[string]any) string { + if e, ok := body["error"].(map[string]any); ok { + if c, ok := e["code"].(string); ok { + return c + } + } + return "" +} + +func signCapability(hmacSigningKey, userID, deviceID, projectID, purpose string, ttl time.Duration) string { + tok, err := jwtutil.IssueCapabilityToken([]byte(hmacSigningKey), userID, deviceID, projectID, purpose, ttl) + if err != nil { + panic(fmt.Sprintf("sign capability: %v", err)) + } + return tok +} + +// ── 1. Reject before admission, then retry the SAME delivery id must admit a +// real run (not a fake cached 202). ───────────────────────────────────────── + +func TestDeliveryAdmission_RejectBeforeAdmission_RetryGetsRealRun(t *testing.T) { + exec := &admissionExecutor{} + server, h := newDeliveryTestServer(t, exec, nil) + defer server.Close() + + // Occupy the default thread with an active run so admission is rejected. + h.ensureDefaults() + if _, err := h.Store.CreateRun("run-occupied", "proj_local", "thread_local"); err != nil { + t.Fatalf("create occupant run: %v", err) + } + body := admissionRunBody(h.WorkspaceAllowlist[0], "del-reject-retry", "", nil) + + resp1 := postRunsRaw(t, server.URL, body) + if resp1.status != http.StatusConflict { + t.Fatalf("first admission: expected 409 active_run_exists, got %d: %#v", resp1.status, resp1.body) + } + if errCode(resp1.body) != "active_run_exists" { + t.Fatalf("expected code active_run_exists, got %q", errCode(resp1.body)) + } + + // Free the occupant so the same delivery can now be admitted. + if _, ok := h.Store.SetRunStatus("run-occupied", "finished"); !ok { + t.Fatalf("could not free occupant run") + } + + resp2 := postRunsRaw(t, server.URL, body) + if resp2.status != http.StatusAccepted { + t.Fatalf("retry after freeing: expected 202, got %d: %#v", resp2.status, resp2.body) + } + data := unwrapSuccess(resp2.body) + runID, _ := data["runId"].(string) + if runID == "" { + t.Fatalf("retry must return a real runId; got a fake deduplicated 202: %#v", data) + } + if v, _ := data["deduplicated"].(bool); v { + t.Fatalf("retry after admission rejection must not be a deduplicated 202: %#v", data) + } + if exec.StartCount() != 1 { + t.Fatalf("expected exactly one executor start on admitted retry, got %d", exec.StartCount()) + } +} + +// ── 2. Success then replay returns the ORIGINAL run id, no second start. ──── + +func TestDeliveryAdmission_SuccessReplayReturnsOriginalRunID(t *testing.T) { + exec := &admissionExecutor{} + server, h := newDeliveryTestServer(t, exec, nil) + defer server.Close() + body := admissionRunBody(h.WorkspaceAllowlist[0], "del-same-run", "task-1", nil) + + resp1 := postRunsRaw(t, server.URL, body) + if resp1.status != http.StatusAccepted { + t.Fatalf("first: expected 202, got %d: %#v", resp1.status, resp1.body) + } + data1 := unwrapSuccess(resp1.body) + runID, _ := data1["runId"].(string) + if runID == "" { + t.Fatalf("first run missing runId: %#v", data1) + } + + resp2 := postRunsRaw(t, server.URL, body) + if resp2.status != http.StatusAccepted { + t.Fatalf("replay: expected 202, got %d: %#v", resp2.status, resp2.body) + } + data2 := unwrapSuccess(resp2.body) + if got, _ := data2["runId"].(string); got != runID { + t.Fatalf("replay must return original runId %q, got %#v", runID, data2) + } + if v, _ := data2["deduplicated"].(bool); !v { + t.Fatalf("replay must carry deduplicated=true: %#v", data2) + } + if got, _ := data2["deliveryId"].(string); got != "del-same-run" { + t.Fatalf("replay must carry deliveryId, got %#v", data2) + } + if exec.StartCount() != 1 { + t.Fatalf("replay must not start a second executor; starts=%d", exec.StartCount()) + } +} + +// ── 2c. Same deliveryId + same HubTaskId but a DIFFERENT thread (the two +// channels legitimately use different thread representations) must replay the +// original run and its real original scope, not conflict and not re-execute. ─ +func TestDeliveryAdmission_SameHubTaskIDDifferentThreadReplaysOriginalRun(t *testing.T) { + exec := &admissionExecutor{} + server, h := newDeliveryTestServer(t, exec, nil) + defer server.Close() + h.ensureDefaults() + if _, err := h.Store.CreateThread("thread-B", "proj_local", "Thread B", "direct", "", ""); err != nil { + t.Fatalf("create thread-B: %v", err) + } + wd := h.WorkspaceAllowlist[0] + + body1 := admissionRunBody(wd, "del-thread-diff", "task-1", nil) + resp1 := postRunsRaw(t, server.URL, body1) + if resp1.status != http.StatusAccepted { + t.Fatalf("first: expected 202, got %d: %#v", resp1.status, resp1.body) + } + data1 := unwrapSuccess(resp1.body) + runID, _ := data1["runId"].(string) + if runID == "" { + t.Fatalf("first run missing runId: %#v", data1) + } + if got, _ := data1["threadId"].(string); got != "thread_local" { + t.Fatalf("first run scope threadId = %q, want thread_local", got) + } + + // Same deliveryId + same HubTaskId, but the replay uses a different thread + // (channel representation). Must return the ORIGINAL run (thread_local + // scope) and not execute again. + body2 := admissionRunBody(wd, "del-thread-diff", "task-1", map[string]any{"threadId": "thread-B"}) + resp2 := postRunsRaw(t, server.URL, body2) + if resp2.status != http.StatusAccepted { + t.Fatalf("replay (different thread): expected 202, got %d: %#v", resp2.status, resp2.body) + } + data2 := unwrapSuccess(resp2.body) + if got, _ := data2["runId"].(string); got != runID { + t.Fatalf("replay must return original runId %q, got %#v", runID, data2) + } + if got, _ := data2["threadId"].(string); got != "thread_local" { + t.Fatalf("replay must assert the REAL original scope threadId=thread_local, got %q: %#v", got, data2) + } + if exec.StartCount() != 1 { + t.Fatalf("same HubTaskId replay must not start a second executor; starts=%d", exec.StartCount()) + } +} + +// ── 2b. Distinct delivery ids and legacy (no id) process independently. ───── + +func TestDeliveryAdmission_DistinctIDsAndLegacyProcessedIndependently(t *testing.T) { + exec := &admissionExecutor{} + server, h := newDeliveryTestServer(t, exec, nil) + defer server.Close() + h.ensureDefaults() + for _, th := range []string{"thread-B", "thread-C"} { + if _, err := h.Store.CreateThread(th, "proj_local", th, "direct", "", ""); err != nil { + t.Fatalf("create %s: %v", th, err) + } + } + wd := h.WorkspaceAllowlist[0] + + r1 := postRunsRaw(t, server.URL, admissionRunBody(wd, "del-A", "", nil)) + r2 := postRunsRaw(t, server.URL, admissionRunBody(wd, "del-B", "", map[string]any{"threadId": "thread-B"})) + r3 := postRunsRaw(t, server.URL, admissionRunBody(wd, "", "", map[string]any{"threadId": "thread-C"})) + + if r1.status != http.StatusAccepted || r2.status != http.StatusAccepted || r3.status != http.StatusAccepted { + t.Fatalf("independent runs: expected all 202, got %d/%d/%d", r1.status, r2.status, r3.status) + } + idA := unwrapSuccess(r1.body)["runId"] + idB := unwrapSuccess(r2.body)["runId"] + if idA == idB { + t.Fatalf("distinct delivery ids must produce distinct runs, both=%v", idA) + } + if v, _ := unwrapSuccess(r2.body)["deduplicated"].(bool); v { + t.Fatalf("distinct delivery id must NOT be deduplicated: %#v", unwrapSuccess(r2.body)) + } + if _, ok := unwrapSuccess(r3.body)["deduplicated"]; ok { + t.Fatalf("legacy (no delivery id) must not carry deduplicated: %#v", unwrapSuccess(r3.body)) + } + if exec.StartCount() != 3 { + t.Fatalf("expected 3 starts for 3 independent runs, got %d", exec.StartCount()) + } +} + +// ── 3. Concurrent same delivery id during admission returns 503 busy with +// Retry-After; replay after completion returns same run id. No time.Sleep. ── + +func TestDeliveryAdmission_ConcurrentSameIDReturnsBusy(t *testing.T) { + entered := make(chan string, 1) + release := make(chan struct{}) + + exec := &admissionExecutor{blockNext: true, entered: entered, release: release} + server, h := newDeliveryTestServer(t, exec, nil) + defer server.Close() + // Must be registered AFTER server.Close() so this runs first (LIFO) and + // unblocks the held admission request before the server is torn down. + defer func() { + select { + case <-release: + default: + close(release) + } + }() + + body := admissionRunBody(h.WorkspaceAllowlist[0], "del-busy", "", nil) + + // First request: admission is held open inside executor Start. + firstPost := make(chan postResult, 1) + go func() { + pr, err := httpPostJSON(server.URL, body) + if err != nil { + t.Errorf("first post: %v", err) + firstPost <- postResult{status: 0} + return + } + firstPost <- pr + }() + + select { + case rid := <-entered: + _ = rid + case <-time.After(5 * time.Second): + t.Fatal("first request did not enter admission in time") + } + + // Second same delivery id while the first admission is held -> busy, not 202. + resp2 := postRunsRaw(t, server.URL, body) + if resp2.status != http.StatusServiceUnavailable { + t.Fatalf("concurrent duplicate must be 503 delivery_busy, got %d: %#v", resp2.status, resp2.body) + } + if errCode(resp2.body) != "delivery_busy" { + t.Fatalf("expected code delivery_busy, got %q", errCode(resp2.body)) + } + if resp2.resp.Header.Get("Retry-After") == "" { + t.Fatalf("503 delivery_busy must carry a Retry-After header") + } + + // Release the first admission; it completes with a real run id. + close(release) + first := <-firstPost + if first.status != http.StatusAccepted { + t.Fatalf("first admission should complete 202, got %d: %#v", first.status, first.body) + } + runID, _ := unwrapSuccess(first.body)["runId"].(string) + if runID == "" { + t.Fatalf("first admitted run missing runId: %#v", unwrapSuccess(first.body)) + } + + // Replay after completion returns the same run id, no second start. + resp3 := postRunsRaw(t, server.URL, body) + if resp3.status != http.StatusAccepted { + t.Fatalf("replay after completion: expected 202, got %d: %#v", resp3.status, resp3.body) + } + if got, _ := unwrapSuccess(resp3.body)["runId"].(string); got != runID { + t.Fatalf("replay after completion must return same runId %q, got %#v", runID, unwrapSuccess(resp3.body)) + } + if exec.StartCount() != 1 { + t.Fatalf("expected one start across busy + replay, got %d", exec.StartCount()) + } +} + +// ── 3b. A failed owner releases the claim so a retry is admitted fresh. ───── + +func TestDeliveryAdmission_FailedOwnerReleasesClaim(t *testing.T) { + exec := &admissionExecutor{err: fmt.Errorf("start failed")} + exec.failNext = true + server, h := newDeliveryTestServer(t, exec, nil) + defer server.Close() + body := admissionRunBody(h.WorkspaceAllowlist[0], "del-fail-retry", "", nil) + + resp1 := postRunsRaw(t, server.URL, body) + if resp1.status == http.StatusAccepted { + t.Fatalf("failing first attempt must not be accepted, got 202: %#v", resp1.body) + } + + // Owner released the claim: the same delivery id can be admitted again. + resp2 := postRunsRaw(t, server.URL, body) + if resp2.status != http.StatusAccepted { + t.Fatalf("retry after failed owner must be accepted, got %d: %#v", resp2.status, resp2.body) + } + data := unwrapSuccess(resp2.body) + if runID, _ := data["runId"].(string); runID == "" { + t.Fatalf("retry must return a real runId after failure, got %#v", data) + } + if v, _ := data["deduplicated"].(bool); v { + t.Fatalf("retry after failed owner must not be a deduplicated 202: %#v", data) + } + if exec.StartCount() != 2 { + t.Fatalf("expected two starts (failed first + admitted retry), got %d", exec.StartCount()) + } +} + +// ── 4. A committed receipt cannot bypass capability validation. ───────────── + +func TestDeliveryAdmission_ReceiptCannotBypassCapability(t *testing.T) { + const hmacSigningKey = "0123456789abcdef0123456789abcdef01234567" + exec := &admissionExecutor{} + server, h := newDeliveryTestServer(t, exec, func(h *Handler) { + h.HubJWTSecret = hmacSigningKey + h.EdgeDeviceID = "test-device" + }) + defer server.Close() + body := admissionRunBody(h.WorkspaceAllowlist[0], "del-cap", "task-cap", nil) + + // Valid capability admits a real run. + req, _ := http.NewRequest(http.MethodPost, server.URL+"/v1/runs", strings.NewReader(body)) + req.Header.Set("Content-Type", "application/json") + req.Header.Set("X-AgentHub-Capability-Token", signCapability(hmacSigningKey, "user-1", "test-device", "proj_local", "run-start", time.Hour)) + resp1 := doReq(t, req) + if resp1.status != http.StatusAccepted { + t.Fatalf("valid capability must admit, got %d: %#v", resp1.status, resp1.body) + } + if runID, _ := unwrapSuccess(resp1.body)["runId"].(string); runID == "" { + t.Fatalf("valid-capability run missing runId: %#v", unwrapSuccess(resp1.body)) + } + + // Same delivery id but NO capability token must still be rejected, not a + // cached 202 that skipped the per-request capability gate. + req2, _ := http.NewRequest(http.MethodPost, server.URL+"/v1/runs", strings.NewReader(body)) + req2.Header.Set("Content-Type", "application/json") + resp2 := doReq(t, req2) + if resp2.status != http.StatusForbidden { + t.Fatalf("duplicate without capability must be 403, got %d: %#v", resp2.status, resp2.body) + } + if errCode(resp2.body) != "capability_token_invalid" { + t.Fatalf("expected capability_token_invalid, got %q", errCode(resp2.body)) + } +} + +// ── 4a2. A committed receipt on replay must re-validate capability against +// the ORIGINAL run scope: presenting a capability bound to a different scope +// cannot be used to read the original run. ────────────────────────────────── +func TestDeliveryAdmission_ReplayCannotUseDifferentScopeCapability(t *testing.T) { + const hmacSigningKey = "0123456789abcdef0123456789abcdef01234567" + exec := &admissionExecutor{} + server, h := newDeliveryTestServer(t, exec, func(h *Handler) { + h.HubJWTSecret = hmacSigningKey + h.EdgeDeviceID = "test-device" + }) + defer server.Close() + wd := h.WorkspaceAllowlist[0] + + // First: capability bound to proj_local, hubTaskId present. + body1 := admissionRunBody(wd, "del-cap-scope", "task-cap-scope", map[string]any{"projectId": "proj_local"}) + req1, _ := http.NewRequest(http.MethodPost, server.URL+"/v1/runs", strings.NewReader(body1)) + req1.Header.Set("Content-Type", "application/json") + req1.Header.Set("X-AgentHub-Capability-Token", signCapability(hmacSigningKey, "user-1", "test-device", "proj_local", "run-start", time.Hour)) + resp1 := doReq(t, req1) + if resp1.status != http.StatusAccepted { + t.Fatalf("first (proj_local capability): expected 202, got %d: %#v", resp1.status, resp1.body) + } + if runID, _ := unwrapSuccess(resp1.body)["runId"].(string); runID == "" { + t.Fatalf("first run missing runId: %#v", unwrapSuccess(resp1.body)) + } + + // Replay with same deliveryId + same HubTaskId, but a capability bound to a + // DIFFERENT scope (proj_other). The per-request capability matches the + // request body, but it must NOT be usable to read the original (proj_local) + // run → 403 capability_token_invalid. + body2 := admissionRunBody(wd, "del-cap-scope", "task-cap-scope", map[string]any{"projectId": "proj_other"}) + req2, _ := http.NewRequest(http.MethodPost, server.URL+"/v1/runs", strings.NewReader(body2)) + req2.Header.Set("Content-Type", "application/json") + req2.Header.Set("X-AgentHub-Capability-Token", signCapability(hmacSigningKey, "user-1", "test-device", "proj_other", "run-start", time.Hour)) + resp2 := doReq(t, req2) + if resp2.status != http.StatusForbidden { + t.Fatalf("replay with different-scope capability must be 403, got %d: %#v", resp2.status, resp2.body) + } + if errCode(resp2.body) != "capability_token_invalid" { + t.Fatalf("expected capability_token_invalid, got %q", errCode(resp2.body)) + } +} + +// ── 4b. Same delivery id with a different binding must conflict (409), not +// reuse another run. ──────────────────────────────────────────────────────── + +func TestDeliveryAdmission_SameDeliveryIDDifferentBindingConflicts(t *testing.T) { + exec := &admissionExecutor{} + server, h := newDeliveryTestServer(t, exec, nil) + defer server.Close() + wd := h.WorkspaceAllowlist[0] + + body1 := admissionRunBody(wd, "del-bind", "task-1", nil) + resp1 := postRunsRaw(t, server.URL, body1) + if resp1.status != http.StatusAccepted { + t.Fatalf("first: expected 202, got %d: %#v", resp1.status, resp1.body) + } + if runID, _ := unwrapSuccess(resp1.body)["runId"].(string); runID == "" { + t.Fatalf("first binding run missing runId: %#v", unwrapSuccess(resp1.body)) + } + + // Same delivery id, different hubTaskId binding → 409 delivery_conflict. + body2 := admissionRunBody(wd, "del-bind", "task-2", nil) + resp2 := postRunsRaw(t, server.URL, body2) + if resp2.status != http.StatusConflict { + t.Fatalf("same delivery id different binding must be 409 delivery_conflict, got %d: %#v", resp2.status, resp2.body) + } + if errCode(resp2.body) != "delivery_conflict" { + t.Fatalf("expected code delivery_conflict, got %q", errCode(resp2.body)) + } +} + +// ── 4c. Legacy (no HubTaskId): different project/thread on the same delivery +// id is a scope conflict → 409 delivery_conflict (no HubTaskId to bind on). ── +func TestDeliveryAdmission_LegacyDifferentThreadConflicts(t *testing.T) { + exec := &admissionExecutor{} + server, h := newDeliveryTestServer(t, exec, nil) + defer server.Close() + h.ensureDefaults() + if _, err := h.Store.CreateThread("thread-B", "proj_local", "Thread B", "direct", "", ""); err != nil { + t.Fatalf("create thread-B: %v", err) + } + wd := h.WorkspaceAllowlist[0] + + body1 := admissionRunBody(wd, "del-legacy-conflict", "", nil) // no hubTaskId, thread_local + resp1 := postRunsRaw(t, server.URL, body1) + if resp1.status != http.StatusAccepted { + t.Fatalf("first legacy: expected 202, got %d: %#v", resp1.status, resp1.body) + } + if runID, _ := unwrapSuccess(resp1.body)["runId"].(string); runID == "" { + t.Fatalf("first legacy run missing runId: %#v", unwrapSuccess(resp1.body)) + } + + // Same delivery id, no hubTaskId, but a different thread → legacy scope + // conflict (there is no HubTaskId primary binding to absorb the difference). + body2 := admissionRunBody(wd, "del-legacy-conflict", "", map[string]any{"threadId": "thread-B"}) + resp2 := postRunsRaw(t, server.URL, body2) + if resp2.status != http.StatusConflict { + t.Fatalf("legacy different thread must be 409 delivery_conflict, got %d: %#v", resp2.status, resp2.body) + } + if errCode(resp2.body) != "delivery_conflict" { + t.Fatalf("expected code delivery_conflict, got %q", errCode(resp2.body)) + } +} diff --git a/edge-server/internal/api/handlers_run_delivery.go b/edge-server/internal/api/handlers_run_delivery.go new file mode 100644 index 000000000..40470f7af --- /dev/null +++ b/edge-server/internal/api/handlers_run_delivery.go @@ -0,0 +1,59 @@ +package api + +import ( + "log/slog" + "net/http" + + "github.com/agenthub/edge-server/internal/deliverydedup" + "github.com/agenthub/edge-server/internal/errcode" + "github.com/agenthub/edge-server/internal/store" +) + +// beginRunDelivery runs only after request capability validation. A successful +// replay is a receipt for the original run, never an assertion that a process +// is still running or that a rejected request was accepted. +func (h *Handler) beginRunDelivery(w http.ResponseWriter, r *http.Request, req runRequest, repository store.Repository) (*deliverydedup.Claim, bool) { + if h.DeliveryDedup == nil || req.DeliveryID == "" { + return nil, false + } + admission := h.DeliveryDedup.Begin(req.DeliveryID, deliverydedup.Scope{ + HubTaskID: req.HubTaskID, + ProjectID: req.ProjectID, + ThreadID: req.ThreadID, + }) + switch admission.State { + case deliverydedup.Claimed: + return admission.Claim, false + case deliverydedup.Busy: + w.Header().Set("Retry-After", "1") + errcode.Write(w, errcode.ErrDeliveryBusy) + case deliverydedup.Conflict: + errcode.Write(w, errcode.ErrDeliveryConflict) + case deliverydedup.Accepted: + run, ok := repository.GetRun(admission.RunID) + if !ok { + // Retention/deletion must not turn an old receipt into new execution. + errcode.Write(w, errcode.ErrNotFound.WithMessage("accepted run is no longer available")) + return nil, true + } + // HTTP and Desktop can use different local thread representations for + // the same Hub task. Authorize the actual stored scope too: a capability + // for the incoming representation must not expose a different resource. + replayRequest := req + replayRequest.ProjectID = run.ProjectID + replayRequest.ThreadID = run.ThreadID + if err := h.validateCapabilityRequest(r, &replayRequest); err != nil { + errcode.Write(w, err) + return nil, true + } + slog.Info("run.create.dedup", "deliveryId", req.DeliveryID, "hubTaskId", req.HubTaskID, "runId", run.ID, "result", "accepted_replay") + data := runToResponse(run) + data["deduplicated"] = true + data["deliveryId"] = req.DeliveryID + writeSuccess(w, http.StatusAccepted, acceptedResponse(data)) + default: + w.Header().Set("Retry-After", "1") + errcode.Write(w, errcode.ErrDeliveryBusy) + } + return nil, true +} diff --git a/edge-server/internal/api/handlers_runs.go b/edge-server/internal/api/handlers_runs.go index 74fe9ca81..9ecb13973 100644 --- a/edge-server/internal/api/handlers_runs.go +++ b/edge-server/internal/api/handlers_runs.go @@ -304,11 +304,6 @@ func (h *Handler) PostRuns(w http.ResponseWriter, r *http.Request) { return } - // #2101 G2: short-circuit duplicate deliveries before any run side-effects. - if h.handleDeliveryDedup(w, req.DeliveryID, req.HubTaskID, req.ThreadID) { - return - } - // Merge profile defaults: profile fields fill in blanks, request fields win. h.applyProfileDefaults(&req) @@ -339,6 +334,13 @@ func (h *Handler) PostRuns(w http.ResponseWriter, r *http.Request) { } repository := ensureStore(h) + claim, handled := h.beginRunDelivery(w, r, req, repository) + if handled { + return + } + if claim != nil { + defer claim.Release() + } // Auto-detect continue: when the thread has prior assistant messages, // set ContinueLast = true so adapters can resume the conversation. @@ -420,6 +422,10 @@ func (h *Handler) PostRuns(w http.ResponseWriter, r *http.Request) { errcode.Write(w, errcode.ErrInternal.WithMessagef("%v", err)) return } + if claim != nil && !claim.Commit(run.ID) { + errcode.Write(w, errcode.ErrInternal.WithMessage("run admission receipt could not be committed")) + return + } writeSuccess(w, http.StatusAccepted, acceptedResponse(runToResponse(run))) } @@ -493,32 +499,3 @@ func (h *Handler) GetMetrics(w http.ResponseWriter, r *http.Request) { } h.Metrics.Handler().ServeHTTP(w, r) } - -// --------------------------------------------------------------------------- -// GET /v1/events (WebSocket) -// --------------------------------------------------------------------------- - -// handleDeliveryDedup short-circuits POST /v1/runs when the request carries a -// delivery_id that was already seen within the TTL window (#2101 G2). Returns -// true when the response has been written and the caller must return; false -// means the request should continue normal processing. Empty delivery_id or a -// nil dedup cache bypass dedup so legacy payloads keep working. -func (h *Handler) handleDeliveryDedup(w http.ResponseWriter, deliveryID, hubTaskID, threadID string) bool { - if h.DeliveryDedup == nil || deliveryID == "" { - return false - } - if h.DeliveryDedup.Seen(deliveryID) { - slog.Info("run.create.dedup", - "deliveryId", deliveryID, - "hubTaskId", hubTaskID, - "threadId", threadID, - "result", "duplicate_skipped") - writeSuccess(w, http.StatusAccepted, map[string]any{ - "deduplicated": true, - "deliveryId": deliveryID, - }) - return true - } - h.DeliveryDedup.Record(deliveryID) - return false -} diff --git a/edge-server/internal/deliverydedup/deliverydedup.go b/edge-server/internal/deliverydedup/deliverydedup.go index 8e24432ef..50d810640 100644 --- a/edge-server/internal/deliverydedup/deliverydedup.go +++ b/edge-server/internal/deliverydedup/deliverydedup.go @@ -1,60 +1,72 @@ -// Package deliverydedup provides an in-process, TTL-bounded LRU cache of -// recently seen Hub→Edge delivery_id values. It closes the G2 gap identified -// in #2101: Hub's outbox redispatch loop and WS PushToConn path can both -// deliver the same task to Edge; without a dedup contract at the Edge entry -// point, the run could be created twice. -// -// Design choices (documented here so callers don't re-litigate them): -// - In-process only. A crash clears the cache; post-crash duplicate -// deliveries replay once. This is acceptable because runs are idempotent -// at the adapter layer (same prompt + hubTaskId resumes rather than -// duplicates work) and adding persistence would couple dedup to a store -// lifecycle that outlives the problem (#2101 G2 scope). -// - LRU eviction with a hard capacity cap. Prevents unbounded growth when -// Hub floods Edge during recovery. Capacity and TTL are exported constants -// so tuning stays in one place and tests can assert the boundary. -// - Empty delivery_id is never recorded. Legacy payloads without the field -// must continue to flow through; treating "" as a key would falsely dedup -// unrelated legacy dispatches. +// Package deliverydedup tracks bounded, in-process Hub-to-Edge admission +// receipts. Seeing a delivery is not accepting it: an owner must commit a run +// ID, and a rejected or abandoned request releases its claim for retry. +// This cache is not a process-recovery log. A restart loses receipts; persisted +// Hub task/run identity and execution recovery remain separate concerns. package deliverydedup import ( + "strings" "sync" "time" ) const ( - // DefaultCapacity bounds the number of distinct delivery_ids held in - // memory. 4096 covers sustained burst from a full outbox retry sweep - // (~hundreds of tasks) plus steady-state traffic without approaching - // heap pressure on the 4C8G dev-class machines Edge typically runs on. DefaultCapacity = 4096 + DefaultTTL = 5 * time.Minute +) + +// Scope binds a delivery to its business task, not to a transport-specific +// thread representation. Legacy deliveries without a Hub task bind directly +// to project/thread. This is idempotency metadata, never an authorization grant. +type Scope struct { + HubTaskID string + ProjectID string + ThreadID string +} + +func (s Scope) matches(other Scope) bool { + if s.HubTaskID != "" || other.HubTaskID != "" { + return s.HubTaskID != "" && s.HubTaskID == other.HubTaskID + } + return s.ProjectID == other.ProjectID && s.ThreadID == other.ThreadID +} - // DefaultTTL is how long a delivery_id remains "recently seen". 5 minutes - // exceeds the Hub outbox retry base interval (exponential backoff from - // DeliveryRetryBaseInterval) so a single redispatch cycle cannot slip - // past the window, while still expiring stale entries before they pin - // memory across long idle periods. - DefaultTTL = 5 * time.Minute +type State uint8 + +const ( + Busy State = iota + Claimed + Accepted + Conflict ) -// entry pairs an absolute expiry with its position in the LRU list. +// Admission reports an atomic claim, a committed receipt, or a safe rejection. +// Empty delivery IDs bypass the cache: Claimed with a nil Claim. +type Admission struct { + State State + RunID string + Claim *Claim +} + type entry struct { + scope Scope + runID string expiry time.Time + owner *Claim } -// Deduper is a concurrency-safe, TTL-bounded LRU of recently seen delivery IDs. -// Zero-value is NOT usable; construct with New. +// Deduper bounds pending claims and accepted receipts together. Pending claims +// are never evicted or expired while their request may still accept work. type Deduper struct { mu sync.Mutex cap int ttl time.Duration - clock func() time.Time // injectable for tests + clock func() time.Time items map[string]entry - order []string // front = oldest, back = newest (LRU) + order []string // accepted receipts only, oldest use first } -// New constructs a Deduper with explicit capacity and TTL. Both must be > 0. func New(capacity int, ttl time.Duration) *Deduper { if capacity <= 0 { panic("deliverydedup: capacity must be > 0") @@ -62,16 +74,9 @@ func New(capacity int, ttl time.Duration) *Deduper { if ttl <= 0 { panic("deliverydedup: ttl must be > 0") } - return &Deduper{ - cap: capacity, - ttl: ttl, - clock: time.Now, - items: make(map[string]entry, capacity), - order: make([]string, 0, capacity), - } + return &Deduper{cap: capacity, ttl: ttl, clock: time.Now, items: make(map[string]entry, capacity)} } -// WithClock overrides the time source (for tests). Returns d for chaining. func (d *Deduper) WithClock(clock func() time.Time) *Deduper { d.mu.Lock() defer d.mu.Unlock() @@ -79,98 +84,106 @@ func (d *Deduper) WithClock(clock func() time.Time) *Deduper { return d } -// Seen reports whether deliveryID was already recorded within the TTL window. -// Empty string always returns false (legacy payloads pass through, see pkg doc). -func (d *Deduper) Seen(deliveryID string) bool { +// Begin reserves a delivery before side effects. Callers must defer Release +// and call Commit only after run admission succeeds. Busy is retryable, never +// a successful duplicate response. Replays retain their original expiry. +func (d *Deduper) Begin(deliveryID string, scope Scope) Admission { if deliveryID == "" { - return false + return Admission{State: Claimed} } d.mu.Lock() defer d.mu.Unlock() - e, ok := d.items[deliveryID] - if !ok { - return false + d.purgeExpiredLocked(d.clock()) + if e, ok := d.items[deliveryID]; ok { + if !e.scope.matches(scope) { + return Admission{State: Conflict} + } + if e.owner != nil { + return Admission{State: Busy} + } + d.promoteLocked(deliveryID) + return Admission{State: Accepted, RunID: e.runID} } - if d.clock().After(e.expiry) { - d.removeLocked(deliveryID) - return false + for len(d.items) >= d.cap { + if len(d.order) == 0 { + return Admission{State: Busy} + } + oldest := d.order[0] + d.order = d.order[1:] + delete(d.items, oldest) } - return true + claim := &Claim{cache: d, id: deliveryID} + d.items[deliveryID] = entry{scope: scope, owner: claim} + return Admission{State: Claimed, Claim: claim} } -// Record marks deliveryID as seen for the configured TTL. Empty string is a -// no-op. If the ID already exists and is not expired, its TTL is refreshed -// and it is promoted to MRU. Returns true if the ID was newly inserted (i.e. -// not previously seen within TTL); false means it was a duplicate or refresh. -func (d *Deduper) Record(deliveryID string) bool { - if deliveryID == "" { +// Claim is an ownership token, so a stale release cannot delete a newer claim +// for the same ID. Its methods are safe to call more than once or concurrently. +type Claim struct { + cache *Deduper + id string +} + +func (c *Claim) Commit(runID string) bool { + if c == nil || strings.TrimSpace(runID) == "" { return false } - now := d.clock() + d := c.cache d.mu.Lock() defer d.mu.Unlock() - if e, ok := d.items[deliveryID]; ok && !now.After(e.expiry) { - // Refresh TTL + promote to MRU. - d.promoteLocked(deliveryID) - d.items[deliveryID] = entry{expiry: now.Add(d.ttl)} + e, ok := d.items[c.id] + if !ok || e.owner != c { return false } - // Evict expired first, then LRU if still at capacity. - d.purgeExpiredLocked(now) - for len(d.order) >= d.cap { - d.evictOldestLocked() - } - d.items[deliveryID] = entry{expiry: now.Add(d.ttl)} - d.order = append(d.order, deliveryID) + e.owner = nil + e.runID = runID + e.expiry = d.clock().Add(d.ttl) + d.items[c.id] = e + d.order = append(d.order, c.id) return true } -// Len returns the current number of tracked IDs (including any that may have -// expired but not yet been purged). Exported for test assertions. -func (d *Deduper) Len() int { +func (c *Claim) Release() { + if c == nil { + return + } + d := c.cache d.mu.Lock() defer d.mu.Unlock() - return len(d.order) + if e, ok := d.items[c.id]; ok && e.owner == c { + delete(d.items, c.id) + } } -func (d *Deduper) removeLocked(id string) { - delete(d.items, id) - for i, v := range d.order { - if v == id { - d.order = append(d.order[:i], d.order[i+1:]...) - return - } - } +func (d *Deduper) Len() int { + d.mu.Lock() + defer d.mu.Unlock() + return len(d.items) } func (d *Deduper) promoteLocked(id string) { for i, v := range d.order { if v == id { - d.order = append(d.order[:i], d.order[i+1:]...) - d.order = append(d.order, id) + copy(d.order[i:], d.order[i+1:]) + d.order[len(d.order)-1] = id return } } } -func (d *Deduper) evictOldestLocked() { - if len(d.order) == 0 { - return - } - oldest := d.order[0] - d.order = d.order[1:] - delete(d.items, oldest) -} - func (d *Deduper) purgeExpiredLocked(now time.Time) { - // order is oldest-first; stop at first non-expired. - for len(d.order) > 0 { - id := d.order[0] + // LRU use order differs from expiry order because replay does not renew TTL. + kept := d.order[:0] + for _, id := range d.order { e, ok := d.items[id] - if !ok || !now.After(e.expiry) { - return + if !ok { + continue } - d.order = d.order[1:] - delete(d.items, id) + if now.After(e.expiry) { + delete(d.items, id) + continue + } + kept = append(kept, id) } + d.order = kept } diff --git a/edge-server/internal/deliverydedup/deliverydedup_test.go b/edge-server/internal/deliverydedup/deliverydedup_test.go index f9c09529e..9f60f3b46 100644 --- a/edge-server/internal/deliverydedup/deliverydedup_test.go +++ b/edge-server/internal/deliverydedup/deliverydedup_test.go @@ -6,150 +6,238 @@ import ( "time" ) -// fakeClock is a controllable time source for deterministic TTL tests. -type fakeClock struct { - mu sync.Mutex - now time.Time +func TestAdmissionCommitsOneReceipt(t *testing.T) { + d := New(2, time.Minute) + scope := Scope{HubTaskID: "task-a", ProjectID: "project", ThreadID: "thread"} + first := d.Begin("delivery", scope) + if first.State != Claimed || first.Claim == nil { + t.Fatalf("first admission = %#v", first) + } + if got := d.Begin("delivery", scope); got.State != Busy || got.RunID != "" { + t.Fatalf("uncommitted duplicate = %#v", got) + } + if !first.Claim.Commit("run-a") { + t.Fatal("commit failed") + } + first.Claim.Release() + got := d.Begin("delivery", scope) + if got.State != Accepted || got.RunID != "run-a" || got.Claim != nil { + t.Fatalf("committed replay = %#v", got) + } + if d.Len() != 1 { + t.Fatalf("tracked admissions = %d", d.Len()) + } } -func newFakeClock(t time.Time) *fakeClock { return &fakeClock{now: t} } -func (c *fakeClock) Now() time.Time { c.mu.Lock(); defer c.mu.Unlock(); return c.now } -func (c *fakeClock) Advance(d time.Duration) { c.mu.Lock(); defer c.mu.Unlock(); c.now = c.now.Add(d) } - -func TestRecord_NewReturnsTrue_DuplicateReturnsFalse(t *testing.T) { - d := New(8, time.Minute) - if !d.Record("a") { - t.Fatal("first Record(a) should return true") +func TestReleasedClaimCannotCommitOrReleaseItsReplacement(t *testing.T) { + d := New(1, time.Minute) + first := d.Begin("delivery", Scope{}) + first.Claim.Release() + retry := d.Begin("delivery", Scope{}) + if retry.State != Claimed || retry.Claim == nil { + t.Fatalf("retry = %#v", retry) + } + if first.Claim.Commit("stale-run") { + t.Fatal("stale owner committed a replacement claim") + } + first.Claim.Release() + if got := d.Begin("delivery", Scope{}); got.State != Busy { + t.Fatalf("stale release removed replacement: %#v", got) } - if d.Record("a") { - t.Fatal("second Record(a) should return false (duplicate)") + if !retry.Claim.Commit("retry-run") { + t.Fatal("retry commit failed") } - if !d.Record("b") { - t.Fatal("Record(b) should return true") + if got := d.Begin("delivery", Scope{}); got.RunID != "retry-run" { + t.Fatalf("wrong receipt: %#v", got) } } -func TestSeen_EmptyStringAlwaysFalse(t *testing.T) { - d := New(8, time.Minute) - d.Record("") - if d.Seen("") { - t.Fatal("Seen(\"\") must be false even after Record(\"\")") +func TestInvalidReceiptDoesNotBecomeAccepted(t *testing.T) { + d := New(1, time.Minute) + attempt := d.Begin("delivery", Scope{}) + if attempt.Claim.Commit(" ") { + t.Fatal("empty run id was committed") } - if d.Len() != 0 { - t.Fatalf("Len after Record(\"\") = %d, want 0", d.Len()) + if got := d.Begin("delivery", Scope{}); got.State != Busy { + t.Fatalf("invalid receipt was accepted: %#v", got) + } + attempt.Claim.Release() + if retry := d.Begin("delivery", Scope{}); retry.State != Claimed { + t.Fatalf("failed receipt could not retry: %#v", retry) + } else { + retry.Claim.Release() } } -func TestSeen_WithinTTL_True_AfterExpiry_False(t *testing.T) { - c := newFakeClock(time.Unix(0, 0)) - d := New(8, 10*time.Second).WithClock(c.Now) - d.Record("x") - if !d.Seen("x") { - t.Fatal("Seen(x) within TTL should be true") +func TestAdmissionBindsBusinessIdentity(t *testing.T) { + cases := []struct { + name string + first, replay Scope + want State + }{ + {"hub channel thread aliases", Scope{HubTaskID: "task", ProjectID: "local", ThreadID: "local-thread"}, Scope{HubTaskID: "task", ProjectID: "local", ThreadID: "conversation-thread"}, Accepted}, + {"other Hub task", Scope{HubTaskID: "task-a"}, Scope{HubTaskID: "task-b"}, Conflict}, + {"Hub identity cannot become legacy", Scope{HubTaskID: "task"}, Scope{}, Conflict}, + {"same legacy scope", Scope{ProjectID: "project", ThreadID: "thread"}, Scope{ProjectID: "project", ThreadID: "thread"}, Accepted}, + {"other legacy thread", Scope{ProjectID: "project", ThreadID: "a"}, Scope{ProjectID: "project", ThreadID: "b"}, Conflict}, + {"other legacy project", Scope{ProjectID: "a", ThreadID: "thread"}, Scope{ProjectID: "b", ThreadID: "thread"}, Conflict}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + d := New(2, time.Minute) + first := d.Begin("delivery", tc.first) + if !first.Claim.Commit("original-run") { + t.Fatal("commit failed") + } + got := d.Begin("delivery", tc.replay) + if got.State != tc.want { + t.Fatalf("replay = %#v, want state %v", got, tc.want) + } + if tc.want == Accepted && got.RunID != "original-run" { + t.Fatalf("changed original run: %#v", got) + } + if tc.want == Conflict && got.RunID != "" { + t.Fatalf("conflict exposed another run: %#v", got) + } + }) + } +} + +func TestConcurrentAdmissionHasOneOwner(t *testing.T) { + d := New(64, time.Minute) + start := make(chan struct{}) + results := make(chan Admission, 32) + var wg sync.WaitGroup + for range 32 { + wg.Go(func() { <-start; results <- d.Begin("delivery", Scope{HubTaskID: "task"}) }) + } + close(start) + wg.Wait() + close(results) + var owner *Claim + busy := 0 + for result := range results { + switch result.State { + case Claimed: + if owner != nil { + t.Fatal("two concurrent owners") + } + owner = result.Claim + case Busy: + busy++ + default: + t.Fatalf("uncommitted request returned %#v", result) + } } - c.Advance(9 * time.Second) - if !d.Seen("x") { - t.Fatal("Seen(x) just before expiry should still be true") + if owner == nil || busy != 31 { + t.Fatalf("owner=%v busy=%d", owner != nil, busy) } - c.Advance(2 * time.Second) // now 11s > 10s TTL - if d.Seen("x") { - t.Fatal("Seen(x) after TTL should be false") + if !owner.Commit("run") { + t.Fatal("owner commit failed") + } + if got := d.Begin("delivery", Scope{HubTaskID: "task"}); got.State != Accepted || got.RunID != "run" { + t.Fatalf("replay = %#v", got) } } -func TestCapacity_EvictsOldest_NotRecent(t *testing.T) { - c := newFakeClock(time.Unix(0, 0)) - d := New(3, time.Minute).WithClock(c.Now) - d.Record("a") - c.Advance(time.Millisecond) - d.Record("b") - c.Advance(time.Millisecond) - d.Record("c") - if d.Len() != 3 { - t.Fatalf("Len = %d, want 3", d.Len()) - } - // Adding "d" must evict "a" (oldest), not "c" (most recent). - d.Record("d") - if d.Seen("a") { - t.Fatal("expected a to be evicted when capacity exceeded") - } - if !d.Seen("b") || !d.Seen("c") || !d.Seen("d") { - t.Fatalf("recent ids must survive eviction: b=%v c=%v d=%v", - d.Seen("b"), d.Seen("c"), d.Seen("d")) - } - if d.Len() != 3 { - t.Fatalf("Len after eviction = %d, want 3", d.Len()) +func TestPendingClaimsAreBoundedAndNotExpiredOrEvicted(t *testing.T) { + now := time.Unix(1000, 0) + d := New(2, time.Minute).WithClock(func() time.Time { return now }) + first := d.Begin("pending-a", Scope{}) + second := d.Begin("pending-b", Scope{}) + now = now.Add(10 * time.Minute) + if got := d.Begin("pending-a", Scope{}); got.State != Busy { + t.Fatalf("expired a live owner: %#v", got) + } + if got := d.Begin("other", Scope{}); got.State != Busy { + t.Fatalf("evicted a live owner: %#v", got) + } + if d.Len() != 2 { + t.Fatalf("capacity exceeded: %d", d.Len()) + } + second.Claim.Release() + other := d.Begin("other", Scope{}) + if other.State != Claimed { + t.Fatalf("release did not free capacity: %#v", other) + } + if !first.Claim.Commit("first-run") || !other.Claim.Commit("other-run") { + t.Fatal("live owner lost after pressure") } } -func TestCapacity_PromotionPreventsEvictionOfRefreshedID(t *testing.T) { - c := newFakeClock(time.Unix(0, 0)) - d := New(3, time.Minute).WithClock(c.Now) - d.Record("a") - c.Advance(time.Millisecond) - d.Record("b") - c.Advance(time.Millisecond) - d.Record("c") - // Refresh "a" so it becomes MRU; next insert should now evict "b". - c.Advance(time.Millisecond) - d.Record("a") - c.Advance(time.Millisecond) - d.Record("d") - if d.Seen("b") { - t.Fatal("expected b to be evicted after a was promoted") - } - if !d.Seen("a") || !d.Seen("c") || !d.Seen("d") { - t.Fatalf("a/c/d should survive: a=%v c=%v d=%v", - d.Seen("a"), d.Seen("c"), d.Seen("d")) +func TestAcceptedReceiptsUseLRUEviction(t *testing.T) { + d := New(2, time.Minute) + for _, id := range []string{"a", "b"} { + if !d.Begin(id, Scope{}).Claim.Commit("run-" + id) { + t.Fatal("commit failed") + } + } + if got := d.Begin("a", Scope{}); got.State != Accepted { + t.Fatal("a missing") + } + if !d.Begin("c", Scope{}).Claim.Commit("run-c") { + t.Fatal("c commit failed") + } + if got := d.Begin("a", Scope{}); got.State != Accepted || got.RunID != "run-a" { + t.Fatalf("recent receipt evicted: %#v", got) + } + if got := d.Begin("b", Scope{}); got.State != Claimed { + t.Fatalf("oldest receipt was not evicted: %#v", got) + } else { + got.Claim.Release() } } -func TestExpiredEntriesArePurgedOnRecord(t *testing.T) { - c := newFakeClock(time.Unix(0, 0)) - d := New(4, 5*time.Second).WithClock(c.Now) - d.Record("old1") - c.Advance(time.Millisecond) - d.Record("old2") - c.Advance(6 * time.Second) // both expired - d.Record("fresh") - if d.Seen("old1") || d.Seen("old2") { - t.Fatal("expired entries must not appear Seen after purge") - } - if !d.Seen("fresh") { - t.Fatal("fresh entry must be present") - } - // Len reflects only live entries after purge-on-record. - if d.Len() != 1 { - t.Fatalf("Len = %d, want 1 after expired purge", d.Len()) +func TestReplayDoesNotRenewReceiptTTL(t *testing.T) { + now := time.Unix(1000, 0) + d := New(3, time.Minute).WithClock(func() time.Time { return now }) + if !d.Begin("a", Scope{}).Claim.Commit("run-a") { + t.Fatal("commit a") + } + now = now.Add(30 * time.Second) + if !d.Begin("b", Scope{}).Claim.Commit("run-b") { + t.Fatal("commit b") + } + if got := d.Begin("a", Scope{}); got.State != Accepted { + t.Fatal("a expired early") + } + now = now.Add(31 * time.Second) + if got := d.Begin("a", Scope{}); got.State != Claimed { + t.Fatalf("replay renewed TTL or hid an expired MRU entry: %#v", got) + } else { + got.Claim.Release() + } + if got := d.Begin("b", Scope{}); got.State != Accepted || got.RunID != "run-b" { + t.Fatalf("unexpired receipt lost: %#v", got) } } -func TestConcurrency_NoRaceUnderLoad(t *testing.T) { - d := New(256, time.Minute) - var wg sync.WaitGroup - for i := 0; i < 16; i++ { - wg.Add(1) - go func(base int) { - defer wg.Done() - for j := 0; j < 200; j++ { - id := "id-" + string(rune('A'+base%26)) + "-" + string(rune('0'+j%10)) - d.Record(id) - _ = d.Seen(id) - } - }(i) +func TestEmptyDeliveryDoesNotConsumeCapacity(t *testing.T) { + d := New(1, time.Minute) + for range 3 { + got := d.Begin("", Scope{}) + if got.State != Claimed || got.Claim != nil || got.RunID != "" { + t.Fatalf("legacy admission = %#v", got) + } } - wg.Wait() - // No panic / race detector trip = pass. Len sanity check. - if d.Len() > 256 { - t.Fatalf("Len exceeded capacity: %d", d.Len()) + if d.Len() != 0 { + t.Fatalf("legacy deliveries were cached: %d", d.Len()) } } -func TestNew_PanicsOnInvalidArgs(t *testing.T) { - defer func() { - if r := recover(); r == nil { - t.Fatal("expected panic on zero capacity") - } - }() - New(0, time.Minute) +func TestNewRejectsInvalidBounds(t *testing.T) { + for _, tc := range []struct { + name string + capacity int + ttl time.Duration + }{{"zero capacity", 0, time.Minute}, {"negative capacity", -1, time.Minute}, {"zero TTL", 1, 0}, {"negative TTL", 1, -time.Second}} { + t.Run(tc.name, func(t *testing.T) { + defer func() { + if recover() == nil { + t.Fatal("invalid cache bounds were accepted") + } + }() + New(tc.capacity, tc.ttl) + }) + } } diff --git a/edge-server/internal/errcode/codes.go b/edge-server/internal/errcode/codes.go index 94d3626d6..7fa5665b5 100644 --- a/edge-server/internal/errcode/codes.go +++ b/edge-server/internal/errcode/codes.go @@ -56,7 +56,9 @@ var ( ErrTooManyConcurrentRuns = New("too_many_concurrent_runs", "too many concurrent runs", http.StatusTooManyRequests) // Run - ErrActiveRunExists = New("active_run_exists", "thread already has an active run", http.StatusConflict) + ErrActiveRunExists = New("active_run_exists", "thread already has an active run", http.StatusConflict) + ErrDeliveryBusy = New("delivery_busy", "delivery admission is busy; retry later", http.StatusServiceUnavailable) + ErrDeliveryConflict = New("delivery_conflict", "delivery id belongs to another task or legacy scope", http.StatusConflict) // Agent discovery ErrInvalidAgentID = New("invalid_agent_id", "unknown agent adapter", http.StatusBadRequest)