fix(daemon): resume-choice review follow-ups — steer superpipe, dedicated park counter, RPC ordering

packages/daemon/src/lib/agent/handler-outcome-routing.ts

@@ -9,11 +9,11 @@ import {
 } from './message-delivery.ts';
 import {
   RESUME_CHOICE_DEAD_LETTER_REASON,
-  resumeChoiceParkDelayMs,
+  routeBlockedSteer,
   routeBlockedTurn,
 } from './message-delivery-pipeline.ts';
 
-export { RESUME_CHOICE_DEAD_LETTER_REASON, resumeChoiceParkDelayMs };
+export { RESUME_CHOICE_DEAD_LETTER_REASON };
 
 export type HandlerJobResult =
   | { outcome: 'completed'; skipped?: 'turn_terminated' }
@@ -107,15 +107,29 @@ export function routeFeedSteerOutcome(
   }
   if (result.outcome === 'park') {
     if (args.waitingForInput && args.resumeChoicePending === true) {
-      if (args.parkCount >= RESUME_CHOICE_PARK_BUDGET) {
-        return { deadLetter: RESUME_CHOICE_DEAD_LETTER_REASON };
+      const decision = routeBlockedSteer({
+        outcome: 'blocked',
+        parkCount: args.parkCount,
+        resumeChoiceResolved: false,
+        now: args.now,
+      });
+      if (decision.action === 'dead_letter') {
+        return { deadLetter: decision.reason };
+      }
+      if (decision.action === 'requeue_now') {
+        const retryAt = args.now;
+        return {
+          mutation: 'requeue',
+          retryAt,
+          settleSkipped: false,
+          result: { parked: 'sdk_resume_choice', retryAt },
+        };
       }
-      const retryAt = args.now + resumeChoiceParkDelayMs(args.parkCount);
       return {
         mutation: 'requeueParked',
-        retryAt,
+        retryAt: decision.retryAt,
         settleSkipped: false,
-        result: { parked: 'sdk_resume_choice', retryAt },
+        result: { parked: 'sdk_resume_choice', retryAt: decision.retryAt },
       };
     }
     if (args.waitingForInput) {

packages/daemon/src/lib/agent/message-delivery-pipeline.ts

@@ -258,3 +258,20 @@ export function routeBlockedTurn(input: BlockedTurnRouteInput): BlockedTurnRoute
     }
   );
 }
+
+const blockedSteerRouteRun = decisionRun('message-blocked-steer-route', [
+  applyResolvedChoiceGate,
+  applyResumeChoiceBudgetGate,
+  applyParkBackoffFinalGate,
+]);
+
+export function routeBlockedSteer(input: BlockedTurnRouteInput): BlockedTurnRouteDecision {
+  const ctx = blockedSteerRouteRun(input);
+  return (
+    ctx.decision ?? {
+      action: 'park',
+      retryAt: input.now + MESSAGE_DELIVERY_PARK_MS,
+      reason: 'sdk_resume_choice',
+    }
+  );
+}

packages/daemon/src/lib/job-handlers/message-delivery.handler.ts

@@ -81,6 +81,7 @@ export function createMessageDeliveryHandler(deps: MessageDeliveryHandlerDeps):
       const retryAt = Date.now() + MESSAGE_DELIVERY_PARK_MS;
       deps.jobQueue.requeueParked(job.id, retryAt, job.claimToken, {
         reason: 'acp_awaiting_acceptance',
+        counter: 'park',
       });
       return { parked: 'acp_awaiting_acceptance', retryAt };
     }
@@ -122,7 +123,8 @@ export function createMessageDeliveryHandler(deps: MessageDeliveryHandlerDeps):
       );
       const result = await turn;
       const route = routeDriveTurnOutcome(result, {
-        parkCount: result.outcome === 'blocked' ? deps.jobQueue.getParkCount(job.id) : 0,
+        parkCount:
+          result.outcome === 'blocked' ? deps.jobQueue.getParkCount(job.id, 'resumeChoice') : 0,
         now: Date.now(),
         resumeChoiceResolved:
           result.outcome === 'blocked'
@@ -150,11 +152,13 @@ export function createMessageDeliveryHandler(deps: MessageDeliveryHandlerDeps):
       if (route.mutation === 'requeue' && route.retryAt !== undefined) {
         deps.jobQueue.requeue(job.id, route.retryAt, job.claimToken);
       } else if (route.mutation === 'requeueParked' && route.retryAt !== undefined) {
+        const parkedReason = (route.result as { parked?: string }).parked;
         deps.jobQueue.requeueParked(job.id, route.retryAt, job.claimToken, {
-          reason: (route.result as { parked?: string }).parked,
+          reason: parkedReason,
+          counter: parkedReason === 'sdk_resume_choice' ? 'resumeChoice' : 'park',
         });
         if (
-          (route.result as { parked?: string }).parked === 'sdk_resume_choice' &&
+          parkedReason === 'sdk_resume_choice' &&
           deps.isResumeChoiceResolved?.(payload.sessionId) === true
         ) {
           deps.jobQueue.rescheduleSessionDeliveries(job.queue, payload.sessionId, Date.now(), {
@@ -184,11 +188,14 @@ export function createMessageDeliveryHandler(deps: MessageDeliveryHandlerDeps):
     const needsParkBudget = result.outcome === 'park' || result.outcome === 'awaiting_acceptance';
     const waitingForInput =
       result.outcome === 'park' ? (session.isWaitingForInput?.() ?? false) : false;
+    const resumeChoicePending =
+      waitingForInput && deps.isResumeChoiceResolved?.(payload.sessionId) === false;
     const route = routeFeedSteerOutcome(result, {
-      parkCount: needsParkBudget ? deps.jobQueue.getParkCount(job.id) : 0,
+      parkCount: needsParkBudget
+        ? deps.jobQueue.getParkCount(job.id, resumeChoicePending ? 'resumeChoice' : 'park')
+        : 0,
       waitingForInput,
-      resumeChoicePending:
-        waitingForInput && deps.isResumeChoiceResolved?.(payload.sessionId) === false,
+      resumeChoicePending,
       now: Date.now(),
     });
     if ('deadLetter' in route) {
@@ -200,9 +207,22 @@ export function createMessageDeliveryHandler(deps: MessageDeliveryHandlerDeps):
     if (route.mutation === 'requeue' && route.retryAt !== undefined) {
       deps.jobQueue.requeue(job.id, route.retryAt, job.claimToken);
     } else if (route.mutation === 'requeueParked' && route.retryAt !== undefined) {
+      const parkedReason = (route.result as { parked?: string }).parked;
       deps.jobQueue.requeueParked(job.id, route.retryAt, job.claimToken, {
-        reason: (route.result as { parked?: string }).parked,
+        reason: parkedReason,
+        counter: parkedReason === 'sdk_resume_choice' ? 'resumeChoice' : 'park',
       });
+      if (
+        parkedReason === 'sdk_resume_choice' &&
+        deps.isResumeChoiceResolved?.(payload.sessionId) === true
+      ) {
+        deps.jobQueue.rescheduleSessionDeliveries(job.queue, payload.sessionId, Date.now(), {
+          parkReason: 'sdk_resume_choice',
+        });
+        log.info(
+          `message_delivery: resume choice resolved while parking steer ${payload.messageUuid}; waking immediately`
+        );
+      }
     } else if (route.mutation === 'requeueAs') {
       try {
         deps.jobQueue.requeueAs(

packages/daemon/src/lib/rpc-handlers/session-handlers.ts

@@ -1331,13 +1331,6 @@ export function setupSessionHandlers(
 
     const db = sessionManager.getDatabase();
 
-    if (choice === 'start_fresh') {
-      db.updateSession(targetSessionId, { sdkSessionId: undefined, sdkOriginPath: undefined });
-      const session = agentSession.getSessionData();
-      session.sdkSessionId = undefined;
-      session.sdkOriginPath = undefined;
-    }
-
     const existingAction = db
       .getSDKMessageRepo()
       .getHyperNeoActionMessageByUuid(targetSessionId, messageUuid);
@@ -1348,6 +1341,13 @@ export function setupSessionHandlers(
       return { retried: true, alreadyResolved: true, messageId: messageUuid };
     }
 
+    if (choice === 'start_fresh') {
+      db.updateSession(targetSessionId, { sdkSessionId: undefined, sdkOriginPath: undefined });
+      const session = agentSession.getSessionData();
+      session.sdkSessionId = undefined;
+      session.sdkOriginPath = undefined;
+    }
+
     const resolvedMessage: HyperNeoActionMessage = {
       type: 'hyperneo_action',
       uuid: messageUuid,

packages/daemon/src/storage/repositories/job-queue-repository.ts

@@ -232,9 +232,10 @@ export class JobQueueRepository {
     return this.getJob(jobId);
   }
 
-  getParkCount(jobId: string): number {
+  getParkCount(jobId: string, counter: 'park' | 'resumeChoice' = 'park'): number {
+    const field = counter === 'resumeChoice' ? '$.__resumeChoiceParks' : '$.__parkCount';
     const row = this.db
-      .prepare(`SELECT json_extract(payload, '$.__parkCount') AS c FROM job_queue WHERE id = ?`)
+      .prepare(`SELECT json_extract(payload, '${field}') AS c FROM job_queue WHERE id = ?`)
       .get(jobId) as { c: number | null } | undefined;
     return Number(row?.c ?? 0) || 0;
   }
@@ -243,25 +244,22 @@ export class JobQueueRepository {
     jobId: string,
     runAt: number,
     claimToken?: string | null,
-    opts?: { reason?: string }
+    opts?: { reason?: string; counter?: 'park' | 'resumeChoice' }
   ): Job | null {
-    const payloadSet =
-      opts?.reason !== undefined
-        ? `payload = json_set(payload, '$.__parkCount',
-               COALESCE(json_extract(payload, '$.__parkCount'), 0) + 1, '$.__parkReason', ?)`
-        : `payload = json_set(payload, '$.__parkCount',
-               COALESCE(json_extract(payload, '$.__parkCount'), 0) + 1, '$.__parkReason', NULL)`;
+    const counterField =
+      opts?.counter === 'resumeChoice' ? '$.__resumeChoiceParks' : '$.__parkCount';
+    const reasonSet =
+      opts?.reason !== undefined ? `, '$.__parkReason', ?` : `, '$.__parkReason', NULL`;
+    const reasonParams = opts?.reason !== undefined ? [opts.reason] : [];
     const stmt = this.db.prepare(
       `UPDATE job_queue
          SET status = 'pending', run_at = ?, started_at = NULL, heartbeat_at = NULL,
-             ${payloadSet}
+             payload = json_set(payload, '${counterField}',
+               COALESCE(json_extract(payload, '${counterField}'), 0) + 1${reasonSet})
        WHERE id = ? AND status = 'processing'
          AND (? IS NULL OR json_extract(payload, '$.__claimToken') = ?)`
     );
-    const params =
-      opts?.reason !== undefined
-        ? [runAt, opts.reason, jobId, claimToken ?? null, claimToken ?? null]
-        : [runAt, jobId, claimToken ?? null, claimToken ?? null];
+    const params = [runAt, ...reasonParams, jobId, claimToken ?? null, claimToken ?? null];
     const res = withBusyRetry(() => stmt.run(...params));
     if (res.changes === 0) return null;
     return this.getJob(jobId);
@@ -275,7 +273,7 @@ export class JobQueueRepository {
     opts?: { resetParkCount?: boolean }
   ): Job | null {
     const payloadSet = opts?.resetParkCount
-      ? `payload = json_set(payload, '$.role', ?, '$.__parkCount', 0),`
+      ? `payload = json_set(payload, '$.role', ?, '$.__parkCount', 0, '$.__resumeChoiceParks', 0),`
       : `payload = json_set(payload, '$.role', ?),`;
     const stmt = this.db.prepare(
       `UPDATE job_queue
@@ -306,7 +304,7 @@ export class JobQueueRepository {
             SET run_at = CASE WHEN run_at > ? THEN ? ELSE run_at END,
                 started_at = NULL,
                 heartbeat_at = NULL,
-                payload = json_set(payload, '$.__parkCount', 0)
+                payload = json_set(payload, '$.__parkCount', 0, '$.__resumeChoiceParks', 0)
           WHERE queue = ? AND status = 'pending'
             AND json_extract(payload, '$.sessionId') = ?
             AND json_extract(payload, '$.__parkReason') = ?`
@@ -780,7 +778,8 @@ export class JobQueueRepository {
           .prepare(
             `UPDATE job_queue
                 SET retry_count = retry_count + 1, status = 'pending', error = ?,
-                    run_at = ?, started_at = NULL, heartbeat_at = NULL
+                    run_at = ?, started_at = NULL, heartbeat_at = NULL,
+                    payload = json_set(payload, '$.__parkReason', NULL)
               WHERE id = ?
                 AND (? IS NULL OR (status = 'processing'
                   AND json_extract(payload, '$.__claimToken') = ?))`

packages/daemon/tests/unit/4-space-storage/storage/job-queue-repository.test.ts

@@ -87,6 +87,56 @@ describe('JobQueueRepository', () => {
       expect(changed).toBe(0);
     });
 
+    it('resume-choice parks track a dedicated counter isolated from generic parks', () => {
+      const future = Date.now() + 600_000;
+      repository.enqueue({
+        queue: 'message_delivery',
+        payload: { sessionId: 'sess-a', messageUuid: 'u1' },
+        runAt: Date.now() - 1000,
+      });
+      let [job] = repository.dequeue('message_delivery', 1);
+      for (let i = 0; i < 3; i++) {
+        repository.requeueParked(job.id, future, job.claimToken, {
+          reason: 'turn_blocked',
+          counter: 'park',
+        });
+        db.prepare(`UPDATE job_queue SET run_at = 0 WHERE id = ?`).run(job.id);
+        [job] = repository.dequeue('message_delivery', 1);
+      }
+      expect(repository.getParkCount(job.id)).toBe(3);
+      expect(repository.getParkCount(job.id, 'resumeChoice')).toBe(0);
+      repository.requeueParked(job.id, future, job.claimToken, {
+        reason: 'sdk_resume_choice',
+        counter: 'resumeChoice',
+      });
+      expect(repository.getParkCount(job.id)).toBe(3);
+      expect(repository.getParkCount(job.id, 'resumeChoice')).toBe(1);
+    });
+
+    it('fail() on the retry path clears the park reason stamp', () => {
+      const future = Date.now() + 600_000;
+      repository.enqueue({
+        queue: 'message_delivery',
+        payload: { sessionId: 'sess-a', messageUuid: 'u1' },
+        runAt: Date.now() - 1000,
+      });
+      const [job] = repository.dequeue('message_delivery', 1);
+      repository.requeueParked(job.id, future, job.claimToken, {
+        reason: 'sdk_resume_choice',
+        counter: 'resumeChoice',
+      });
+      db.prepare(`UPDATE job_queue SET run_at = 0 WHERE id = ?`).run(job.id);
+      const [reclaimed] = repository.dequeue('message_delivery', 1);
+      repository.fail(reclaimed.id, 'transient boom', reclaimed.claimToken);
+      const changed = repository.rescheduleSessionDeliveries(
+        'message_delivery',
+        'sess-a',
+        Date.now(),
+        { parkReason: 'sdk_resume_choice' }
+      );
+      expect(changed).toBe(0);
+    });
+
     it('a reason-filtered wake resets already-due parked jobs without delaying them', () => {
       const past = Date.now() - 600_000;
       repository.enqueue({

packages/daemon/tests/unit/4-space-storage/storage/message-delivery-v2.test.ts

@@ -822,7 +822,7 @@ describe('message-delivery v2 — handler (conformance)', () => {
     expect(result.retryAt).toBeGreaterThanOrEqual(before + MESSAGE_DELIVERY_PARK_MS);
     expect(repo.getJob(job.id)?.status).toBe('pending');
     expect(repo.getJob(job.id)?.runAt).toBe(result.retryAt);
-    expect(repo.getParkCount(job.id)).toBe(1);
+    expect(repo.getParkCount(job.id, 'resumeChoice')).toBe(1);
   });
 
   it('promoting a heavily parked steer to a turn resets the resume-choice park budget', async () => {
@@ -850,7 +850,7 @@ describe('message-delivery v2 — handler (conformance)', () => {
     const result = await handler(promoted);
     expect(result).toMatchObject({ parked: 'sdk_resume_choice' });
     expect(result.retryAt as number).toBeGreaterThan(Date.now());
-    expect(repo.getParkCount(steer.id)).toBe(1);
+    expect(repo.getParkCount(steer.id, 'resumeChoice')).toBe(1);
     expect(repo.getJob(steer.id)?.status).toBe('pending');
   });
 
@@ -866,7 +866,7 @@ describe('message-delivery v2 — handler (conformance)', () => {
     let current = first;
     for (let i = 0; i < RESUME_CHOICE_PARK_BUDGET; i++) {
       await handler(current);
-      expect(repo.getParkCount(first.id)).toBe(i + 1);
+      expect(repo.getParkCount(first.id, 'resumeChoice')).toBe(i + 1);
       if (i < RESUME_CHOICE_PARK_BUDGET - 1) {
         db.prepare(`UPDATE job_queue SET run_at = 0 WHERE id = ?`).run(current.id);
         current = repo.dequeue(MESSAGE_DELIVERY, 1)[0]!;
@@ -875,7 +875,7 @@ describe('message-delivery v2 — handler (conformance)', () => {
     db.prepare(`UPDATE job_queue SET run_at = 0 WHERE id = ?`).run(current.id);
     const [last] = repo.dequeue(MESSAGE_DELIVERY, 1);
     await expect(handler(last!)).rejects.toThrow('past its budget');
-    expect(repo.getParkCount(first.id)).toBe(RESUME_CHOICE_PARK_BUDGET);
+    expect(repo.getParkCount(first.id, 'resumeChoice')).toBe(RESUME_CHOICE_PARK_BUDGET);
     expect(repo.getJob(first.id)?.status).toBe('processing');
   });