fix(dispatch): reconcile durable Hub task admission outcomes

api/events.md

@@ -40,7 +40,11 @@ Hub 的 WS `agent.dispatch` 与 outbox HTTP POST `/v1/runs` 共享同一 `delive
 - **成功重放**：有效回执返回原 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。
+- **Hub task 接收证据**：非空 `hubTaskId` 在启动执行器前，与 `admissionState: pending` 一起写入 run；File/SQLite 在返回前同步保存。执行器返回后只允许转为 `accepted` 或带 `admissionErrorCode` 的 `rejected`，不改写执行状态。没有 Hub task 的本地/MCP 请求保留原路径。
+- **冷重放**：进程缓存丢失或 delivery ID 改变时，按 Hub task 查最新 attempt，并复验原 run scope。`accepted` 返回原 run，不重新执行；上次最终证据保存失败时只重试保存。已接收 run 后续 `failed` 不等于接收拒绝。
+- **拒绝与未决**：429 `too_many_concurrent_runs` 是执行器持有执行权之前的容量拒绝；503 `admission_persist_failed` 是证据保存失败（执行器可能已经接收），两者都不应 ACK/FAIL，由 Hub 重投向 Edge 核对。只有明确的容量拒绝或调用 Start 之前的保存失败，才允许创建新 attempt；普通 `executor_start_failed` 保持拒绝，不伪装成功。
+- **结果不明**：同一 Hub task 的当前接收者仍在处理时返回 503 `delivery_busy`。恢复后的 `pending`、未知 admission state、无 `startedAt` 的旧 run 返回 409 `admission_uncertain`；Desktop 保持待核对错误并通过现有通知提示用户，同一原因不重复提示，不 ACK/FAIL、不自动启动。旧 run 只有明确 `startedAt` 才能按原身份重放。`GET /v1/runs/{runId}` 暴露已记录的 `admissionState` / `admissionErrorCode` 供核对。
+- **恢复边界**：缓存不是恢复日志；持久化接收证据证明的是是否接收，不保证进程仍在运行，也不提供自动进程恢复。`queued`/`failed` 本身不能证明没有外部副作用。未决 admission 不参与终态自动清理；原 run 因显式删除或正常 retention 消失后，不宣称永久保留 Hub task 的幂等身份。
 
 标签：**UPSERT by id**（稳定 id 合并，禁止第二行）；**idempotent on apply**（再应用不变）；**水位 / watermark**（只前进 `max`）；**ephemeral**（可丢可重，不写持久态）；**非幂等**（须自备去重或 REST）。
 

api/openapi.yaml

@@ -729,8 +729,21 @@ paths:
         "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.
+          # an active run. admission_uncertain requires reconciliation of a retained
+          # run without proof of admission; clients must not ACK/FAIL or restart.
           $ref: "#/components/responses/Error"
+        "429":
+          description: Executor capacity rejected admission before execution; retry the same Hub task later.
+          headers:
+            Retry-After:
+              description: Delay in seconds before retrying admission.
+              schema:
+                type: integer
+                minimum: 1
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/ErrorResponse"
         "404":
           # not_found: the original run referenced by an accepted receipt was removed.
           $ref: "#/components/responses/Error"
@@ -739,10 +752,12 @@ paths:
             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.
+            admission capacity and includes Retry-After. admission_persist_failed
+            means admission evidence could not be saved; execution may already
+            have been accepted. Retry to reconcile, not to force another start.
           headers:
             Retry-After:
-              description: Delay in seconds, present for delivery_busy responses.
+              description: Delay in seconds, present for delivery_busy and admission_persist_failed responses.
               schema:
                 type: integer
                 minimum: 1
@@ -9593,6 +9608,13 @@ components:
         finishedAt:
           type: string
           format: date-time
+        admissionState:
+          type: string
+          enum: [pending, accepted, rejected]
+          description: Hub task admission evidence, independent of execution status. Absent on legacy or non-Hub runs.
+        admissionErrorCode:
+          type: string
+          description: Public-safe rejection code recorded with rejected admission; not an execution failure reason.
         deduplicated:
           type: boolean
           description: True on an accepted replay of an already admitted delivery (same runId returned, no new run created).

app/desktop/src/__e2e__/hub-delivery-admission.spec.ts

@@ -29,6 +29,15 @@ async function readTaskState(page: Page) {
   });
 }
 
+async function readTaskError(page: Page) {
+  return page.evaluate(async () => {
+    const modulePath = '/src/stores/taskBridgeStore.ts';
+    const { useTaskBridgeStore } = await import(/* @vite-ignore */ modulePath);
+    const state = useTaskBridgeStore.getState();
+    return state.tasks[0]?.error ?? null;
+  });
+}
+
 async function installDispatchFixture(page: Page, baseURL: string, theme: 'light' | 'dark') {
   const appOrigin = new URL(baseURL).origin;
   const hubSockets = new Set<WebSocketRoute>();
@@ -41,7 +50,7 @@ async function installDispatchFixture(page: Page, baseURL: string, theme: 'light
     fails: [] as Record<string, unknown>[],
     registered: 0,
     targetsRead: 0,
-    rejection: 503,
+    rejection: { status: 503, code: 'delivery_busy' } as { status: number; code: string } | null,
     pageErrors: [] as string[],
     unhandledWrites: [] as string[],
   };
@@ -119,8 +128,8 @@ async function installDispatchFixture(page: Page, baseURL: string, theme: 'light
       } 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' },
+          await route.fulfill({ status: calls.rejection.status, headers: calls.rejection.status === 503 ? { 'Retry-After': '1' } : {}, json: {
+            error: { code: calls.rejection.code, message: 'fixture admission rejection: ' + calls.rejection.code, 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);
@@ -192,26 +201,62 @@ 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);
+    const queued = { tasks: [{ taskId: TASK, status: 'queued', runId: null }], runToTask: {} };
+
     await dispatch();
     await expect.poll(() => calls.runs.length).toBe(1);
-    await expect.poll(() => readTaskState(page)).toEqual({ tasks: [{ taskId: TASK, status: 'queued', runId: null }], runToTask: {} });
+    await expect.poll(() => readTaskState(page)).toEqual(queued);
     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;
+    calls.rejection = { status: 429, code: 'too_many_concurrent_runs' };
+    await dispatch();
+    await expect.poll(() => calls.runs.length).toBe(2);
+    await expect.poll(() => readTaskState(page)).toEqual(queued);
+    expect(calls.acks).toHaveLength(0);
+    expect(calls.relayAcks).toHaveLength(0);
+    expect(calls.fails).toHaveLength(0);
+
+    calls.rejection = { status: 503, code: 'admission_persist_failed' };
     await dispatch();
+    await expect.poll(() => calls.runs.length).toBe(3);
+    await expect.poll(() => readTaskState(page)).toEqual(queued);
+    expect(calls.acks).toHaveLength(0);
+    expect(calls.relayAcks).toHaveLength(0);
+    expect(calls.fails).toHaveLength(0);
+
+    // admission_uncertain is human-review territory: queued, no ACK/FAIL/relay-ACK.
+    calls.rejection = { status: 409, code: 'admission_uncertain' };
+    await dispatch();
+    await expect.poll(() => calls.runs.length).toBe(4);
+    await expect.poll(() => readTaskState(page)).toEqual(queued);
+    await expect.poll(() => readTaskError(page)).toContain('fixture admission rejection: admission_uncertain');
+    const reviewNotice = page.getByText(/Edge admission result is uncertain.*fixture admission rejection: admission_uncertain/);
+    await expect(reviewNotice).toBeVisible();
+    await expect(reviewNotice).toHaveCount(1);
+    await page.screenshot({ path: testInfo.outputPath('admission-uncertain-' + theme + '.png') });
+    expect(calls.acks).toHaveLength(0);
+    expect(calls.relayAcks).toHaveLength(0);
+    expect(calls.fails).toHaveLength(0);
+
+    // The test drives the next dispatch after manual review; this is not an
+    // automatic client-side retry and the hook itself does not start one.
+    calls.rejection = null;
+    await dispatch();
+    await expect.poll(() => calls.runs.length).toBe(5);
     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 } });
+    await expect.poll(() => readTaskError(page)).toBeNull();
 
     // 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.runs).toHaveLength(6);
     expect(calls.fails).toHaveLength(0);
 
     finish();
@@ -224,9 +269,9 @@ for (const theme of ['light', 'dark'] as const) {
     await expect.poll(() => readTaskState(page)).toEqual(finished);
     expect(calls.done).toHaveLength(1);
 
-    calls.rejection = 500;
+    calls.rejection = { status: 500, code: 'internal_error' };
     await dispatch();
-    await expect.poll(() => calls.runs.length).toBe(5);
+    await expect.poll(() => calls.runs.length).toBe(8);
     await expect.poll(() => readTaskState(page)).toEqual(finished);
     expect(calls.fails).toHaveLength(0);
     expect(calls.acks).toHaveLength(3);

app/desktop/src/__tests__/useHubIntegration.test.ts

@@ -118,6 +118,7 @@ import { createServer, type IncomingMessage } from 'node:http';
 import type { HubWSHandle } from '@shared/hub/hubWS';
 import type { HubClient } from '@/api/hubClient';
 import { HUB_EVENTS } from '@shared/hubEvents';
+import { useToastStore } from '@shared/ui/toast';
 import { useHubIntegration } from '@/hooks/useHubIntegration';
 
 // ── Helpers ─────────────────────────────────────────────
@@ -739,6 +740,116 @@ describe('useHubIntegration', () => {
     expect(hubClient.failTask).not.toHaveBeenCalled();
   });
 
+  it('keeps a too_many_concurrent_runs rejection queued without acking or failing', async () => {
+    mockRunCreateResponseWithStatus({ error: { code: 'too_many_concurrent_runs', message: 'too many', traceId: 'trace_001' } }, 429);
+    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 admission_persist_failed rejection queued without acking or failing', async () => {
+    mockRunCreateResponseWithStatus({ error: { code: 'admission_persist_failed', message: 'persist failed', 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 admission_uncertain rejection queued for manual review without acking or failing', async () => {
+    const notice = vi.spyOn(useToastStore.getState(), 'showToast').mockReturnValue('notice-fixture');
+    try {
+      mockRunCreateResponseWithStatus({ error: { code: 'admission_uncertain', message: 'manual review', traceId: 'trace_001' } }, 409);
+      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(hoisted.storeTasks).toHaveLength(1);
+      expect(hoisted.storeTasks[0]?.status).toBe('queued');
+      expect(hoisted.storeTasks[0]?.runId).toBeUndefined();
+      expect(hoisted.storeTasks[0]?.error).toContain('Edge admission result is uncertain');
+      expect(hoisted.storeTasks[0]?.error).toContain('manual review');
+      expect(hubClient.ackTask).not.toHaveBeenCalled();
+      expect(hubClient.failTask).not.toHaveBeenCalled();
+      expect(hubClient.ackRelayCommand).not.toHaveBeenCalled();
+
+      mockRunCreateResponseWithStatus({ error: { code: 'admission_uncertain', message: 'manual review', traceId: 'trace_002' } }, 409);
+      await act(async () => { fireHubEvent(HUB_EVENTS.AGENT_DISPATCH, relayFrame); });
+      expect(notice).toHaveBeenCalledTimes(1);
+      expect(notice).toHaveBeenCalledWith('warning', expect.stringContaining('manual review'), { duration: 10_000 });
+    } finally {
+      notice.mockRestore();
+    }
+  });
+
+  it('clears a queued admission_uncertain error when the same delivery is accepted', async () => {
+    mockRunCreateResponseWithStatus({ error: { code: 'admission_uncertain', message: 'manual review', traceId: 'trace_001' } }, 409);
+    renderHook(() => useHubIntegration({ hubWS, hubClient }));
+
+    await act(async () => {
+      fireHubEvent(
+        HUB_EVENTS.AGENT_DISPATCH,
+        makeDispatchPayload({ delivery_id: 'd1' }),
+      );
+    });
+    expect(hoisted.storeTasks[0]?.status).toBe('queued');
+    expect(hoisted.storeTasks[0]?.error).toContain('manual review');
+
+    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('running');
+    expect(hoisted.storeTasks[0]?.runId).toBe('run-1');
+    expect(hoisted.storeTasks[0]?.error).toBeUndefined();
+    expect(hubClient.ackTask).toHaveBeenCalledTimes(1);
+    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 }));
@@ -766,6 +877,62 @@ describe('useHubIntegration', () => {
     expect(hubClient.failTask).not.toHaveBeenCalled();
   });
 
+  it('does not downgrade an already-running task on admission_uncertain', 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: 'admission_uncertain', message: 'manual review', 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(hoisted.storeTasks[0]?.error).toBeUndefined();
+    expect(hubClient.ackTask).toHaveBeenCalledTimes(1);
+    expect(hubClient.failTask).not.toHaveBeenCalled();
+  });
+
+  it('does not clear an existing error on a progressed successful replay', async () => {
+    mockRunSequence('run-1');
+    renderHook(() => useHubIntegration({ hubWS, hubClient }));
+
+    await act(async () => {
+      fireHubEvent(
+        HUB_EVENTS.AGENT_DISPATCH,
+        makeDispatchPayload({ delivery_id: 'd1' }),
+      );
+    });
+    hoisted.getStoreState().updateTask('task-1', { error: 'existing running error' });
+
+    mockRunSequence('run-1');
+    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');
+    expect(hoisted.storeTasks[0]?.error).toBe('existing running error');
+    expect(hubClient.ackTask).toHaveBeenCalledTimes(2);
+    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(() =>