Merge master b189f586d into Codex runner default candidate

Preserve master transport, lifecycle, and release verification controls alongside macOS qualification.

Co-Authored-By: Paperclip <noreply@paperclip.ing>

server/src/modules/agent-lifecycle/__tests__/diagnostics.test.ts

@@ -13,14 +13,58 @@ function fixture() {
   const store = { get: vi.fn(), claim: vi.fn().mockResolvedValue(agent), renew: vi.fn().mockResolvedValue(true),
     setRequiredPlugins: vi.fn().mockResolvedValue(true), completeHost: vi.fn().mockResolvedValue(true),
     completePlugin: vi.fn().mockResolvedValue(true), defer: vi.fn().mockResolvedValue(true),
-    change: vi.fn().mockResolvedValue({ ...agent, lifecycleState: "ready" }), pending: vi.fn().mockResolvedValue([agent.id]) } satisfies LifecycleStore;
+    change: vi.fn().mockResolvedValue({ ...agent, lifecycleState: "ready", lifecycleVersion: 4, lifecycleOperation: null }), pending: vi.fn().mockResolvedValue([agent.id]) } satisfies LifecycleStore;
   const driver = { requiredPluginIds: vi.fn().mockResolvedValue(["plugin"]), runHost: vi.fn().mockResolvedValue("complete"),
-    runPlugin: vi.fn().mockResolvedValue("complete") } satisfies LifecycleDriver;
+    runPlugin: vi.fn().mockResolvedValue("complete"), onReady: vi.fn().mockResolvedValue(undefined) } satisfies LifecycleDriver;
   const report = vi.fn();
   return { store, driver, report, worker: createLifecycleWorker(store, driver, report) };
 }
 afterEach(() => { vi.restoreAllMocks(); vi.useRealTimers(); });
 
+describe("lifecycle readiness handoff", () => {
+  it("notifies the scheduler only after host, plugins, and readiness have committed", async () => {
+    const { store, driver, worker } = fixture();
+    let commit!: (ready: LifecycleAgent) => void;
+    store.change.mockReturnValue(new Promise(resolve => { commit = resolve; }));
+    driver.onReady.mockImplementation(async ready => {
+      expect(store.completeHost).toHaveBeenCalledOnce();
+      expect(store.completePlugin).toHaveBeenCalledOnce();
+      expect(store.change).toHaveBeenCalledOnce();
+      expect(ready).toMatchObject({ lifecycleState: "ready", lifecycleVersion: 4, lifecycleOperation: null });
+    });
+    const work = worker.process(agent.id);
+    await vi.waitFor(() => expect(store.change).toHaveBeenCalledOnce());
+    expect(driver.onReady).not.toHaveBeenCalled();
+    commit({ ...agent, lifecycleState: "ready", lifecycleVersion: 4, lifecycleOperation: null });
+    await work;
+    expect(driver.onReady).toHaveBeenCalledOnce();
+    await worker.stop();
+  });
+
+  it.each(["host", "plugin", "transition", "hold"])("does not dispatch when %s prevents readiness", async stage => {
+    const { store, driver, worker } = fixture();
+    if (stage === "host") driver.runHost.mockResolvedValue("pending");
+    if (stage === "plugin") driver.runPlugin.mockResolvedValue("pending");
+    if (stage === "transition") store.change.mockResolvedValue(null);
+    if (stage === "hold") store.change.mockResolvedValue({ ...agent, lifecycleState: "paused" });
+    await worker.process(agent.id);
+    expect(driver.onReady).not.toHaveBeenCalled();
+    await worker.stop();
+  });
+
+  it("preserves committed readiness after a dispatch failure for the queue sweep to recover", async () => {
+    const { store, driver, report, worker } = fixture();
+    const error = new Error("queue unavailable");
+    driver.onReady.mockRejectedValue(error);
+    await worker.process(agent.id);
+    expect(report).toHaveBeenCalledWith(expect.objectContaining({ stage: "ready", phase: "ready", version: 4 }), error);
+    expect(store.defer).not.toHaveBeenCalled();
+    expect(driver.runHost).toHaveBeenCalledOnce();
+    expect(store.change).toHaveBeenCalledOnce();
+    await worker.stop();
+  });
+});
+
 describe("lifecycle failure diagnostics", () => {
   it.each(["host", "plugin", "claim", "renew", "scan", "defer"] as const)("reports %s failures with the available context", async stage => {
     const { store, driver, report, worker } = fixture();

server/src/modules/agent-lifecycle/__tests__/lifecycle.test.ts

@@ -486,6 +486,51 @@ const support = await getEmbeddedPostgresTestSupport();
     await expect(driver.runPlugin(snapshot, pluginId)).rejects.toThrow("Invalid lifecycle result");
   });
 
+  it("drives required resource work through setup, pause, resume, and confirmed cleanup", async () => {
+    const pluginId = randomUUID();
+    await db.insert(plugins).values({ id: pluginId, pluginKey: pluginId, packageName: "resource-lifecycle-test", version: "1.0.0", status: "ready",
+      manifestJson: { agentLifecycle: true, capabilities: ["agents.lifecycle.manage"] } as never });
+    let resourcesReady = false;
+    let resourcesRemoved = false;
+    const requests: import("@paperclipai/shared").AgentLifecycleRequest[] = [];
+    const call = async (id: string, method: string, input: import("@paperclipai/shared").AgentLifecycleRequest) => {
+      expect(id).toBe(pluginId);
+      expect(method).toBe("agentLifecycle");
+      expect(input.companyId).toBe(companyId);
+      expect((await current(input.agentId)).lifecycleOperation?.hostComplete).toBe(true);
+      requests.push(input);
+      const pending = (input.phase === "preparing" && !resourcesReady) || (input.phase === "cleaning_up" && !resourcesRemoved);
+      return { operationId: input.operationId, version: input.version, status: pending ? "pending" : "complete" };
+    };
+    const driver = createLifecycleDriver(db, { call } as never);
+    // Provider/harness execution is stubbed; plugin dispatch and durable transitions are real.
+    const work = configureAgentLifecycle(db, { ...driver, requiredPluginIds: async () => [pluginId], runHost: async () => "complete" });
+    workers.push(work);
+    const lifecycle = createAgentLifecycle(db);
+    const agent = await hire();
+    await work.process(agent.id);
+    expect(await current(agent.id)).toMatchObject({ lifecycleState: "preparing", status: "paused", lifecycleError: null });
+    const preparation = requests[0];
+    resourcesReady = true;
+    await lifecycle.retry(agent.id); await work.process(agent.id);
+    expect(requests[1]).toEqual(preparation);
+    expect((await current(agent.id)).lifecycleState).toBe("ready");
+    await lifecycle.pauseAgent(agent.id); await work.process(agent.id);
+    expect((await current(agent.id)).lifecycleState).toBe("paused");
+    await lifecycle.resumeAgent(agent.id); await work.process(agent.id);
+    expect((await current(agent.id)).lifecycleState).toBe("ready");
+    await lifecycle.terminateAgent(agent.id); await work.process(agent.id);
+    expect((await current(agent.id)).lifecycleState).toBe("cleaning_up");
+    await expect(lifecycle.purgeAgent(agent.id)).rejects.toThrow("Complete termination");
+    resourcesRemoved = true;
+    await lifecycle.retry(agent.id); await work.process(agent.id);
+    expect((await current(agent.id)).lifecycleState).toBe("terminated");
+    expect(requests.map(request => request.phase)).toEqual([
+      "preparing", "preparing", "verifying", "pausing", "resuming", "terminating", "cleaning_up", "cleaning_up",
+    ]);
+    expect(requests.every(request => request.agentId === agent.id)).toBe(true);
+  });
+
   it("keeps host completion and required plugin results across a worker restart", async () => {
     const pluginIds = [randomUUID(), randomUUID()];
     const runHost = vi.fn(async () => "complete" as const);

server/src/modules/agent-lifecycle/application/ports.ts

@@ -27,11 +27,13 @@ export interface LifecycleDriver {
   requiredPluginIds(agent: LifecycleAgent): Promise<string[]>;
   runHost(agent: LifecycleAgent): Promise<"complete" | "pending">;
   runPlugin(agent: LifecycleAgent, pluginId: string): Promise<"complete" | "pending">;
+  /** Notify the scheduler only after readiness has committed. */
+  onReady?(agent: LifecycleAgent): Promise<void>;
 }
 
 export interface LifecycleFailure {
   stage: "claim" | "renew" | "select_plugins" | "save_plugins" | "host" | "plugin"
-    | "complete_host" | "complete_plugin" | "transition" | "defer" | "scan" | "policy_scan";
+    | "complete_host" | "complete_plugin" | "transition" | "defer" | "scan" | "policy_scan" | "ready";
   agentId?: string;
   companyId?: string;
   phase?: AgentLifecycleState;

server/src/modules/agent-lifecycle/application/worker.ts

@@ -56,6 +56,14 @@ export function createLifecycleWorker(store: LifecycleStore, driver: LifecycleDr
           if (!await runStep("complete_plugin", () => store.completePlugin(agent, owner, pluginId), pluginId)) continue phase;
         }
         const changed = await runStep("transition", () => store.change(id, "complete", { version: agent.lifecycleVersion, owner }));
+        if (changed?.lifecycleState === "ready" && !stopped && canRun()) {
+          // Accepted messages may already be waiting for verification. Do not
+          // leave them until the periodic queue sweep. Readiness is committed;
+          // a dispatch failure must not replay setup or defer the old phase.
+          await attempt({ ...context, stage: "ready", phase: changed.lifecycleState,
+            version: changed.lifecycleVersion, operationId: changed.lifecycleOperation?.id },
+          async () => { await driver.onReady?.(changed); }).catch(() => {});
+        }
         if (changed && !isLifecycleWorkPending(changed.lifecycleState)) return;
       } catch {
         if (!await runStep("defer", () => store.defer(agent, owner, "The lifecycle step failed. Retry the operation.", new Date()))) continue;

server/src/modules/run-dispatch/adapters/postgres.ts

@@ -6,9 +6,7 @@ import type { Db } from "@paperclipai/db";
 import {
   agentWakeupRequests,
   agents,
-  approvals,
-  issueApprovals,
-  issueThreadInteractions,
+  connectionIntentDeliveries,
   heartbeatRuns,
   issueRecoveryActions,
   issueComments,
@@ -22,6 +20,8 @@ import { evaluateAgentInvokabilityFromDb } from "../../../services/agent-invokab
 import { budgetService, budgetServiceInTransaction } from "../../../services/budgets.js";
 import { isCompletedOnboardingHandoffWake } from "../../../services/chat-completion-delivery.js";
 import { isHeartbeatWakeOnDemandEnabled } from "../../../services/heartbeat-policy.js";
+import { DELIVERY_QUEUES, notifyDeliveryWork } from "../../../services/delivery-work-notifications.js";
+import { connectionContinuationPendingResponse } from "../../../services/satisfied-connection-intents.js";
 import { collectDispositionRepairSourceState } from "../../../services/recovery/disposition-repair.js";
 import { legacyDispositionEpisode, legacyDispositionFingerprint } from "../../../services/recovery/legacy-continuation.js";
 import { appendHeartbeatRunEvent } from "../../../services/heartbeat-run-events.js";
@@ -72,6 +72,7 @@ import { RunDispatchApplicationError } from "../application/types.js";
 type HeartbeatRun = typeof heartbeatRuns.$inferSelect;
 type LoadGateFactsInput = {
   conversationContinuation: boolean;
+  responsibleUserId: string | null;
   runId: string;
   companyId: string;
   agentId: string;
@@ -84,6 +85,8 @@ type LoadGateFactsResult =
   | { agentFound: true; facts: ScheduledRetryFacts }
   | { agentFound: false; issueId: string | null };
 type LoadStalenessFactsInput = {
+  responsibleUserId: string | null;
+  conversationContinuation: boolean;
   runId: string;
   companyId: string;
   agentId: string;
@@ -337,6 +340,7 @@ export function createPostgresRunDispatchAdapter(
       .select({
         id: issues.id,
         companyId: issues.companyId,
+        projectId: issues.projectId,
         status: issues.status,
         assigneeAgentId: issues.assigneeAgentId,
         assigneeUserId: issues.assigneeUserId,
@@ -364,19 +368,8 @@ export function createPostgresRunDispatchAdapter(
     facts.issueExecutionRunId = issue.executionRunId;
     facts.issueCheckoutRunId = issue.checkoutRunId;
     if (input.conversationContinuation) {
-      const [interactions, linkedApprovals] = await Promise.all([
-        dbOrTx.select({ id: issueThreadInteractions.id }).from(issueThreadInteractions).where(and(
-          eq(issueThreadInteractions.companyId, input.companyId),
-          eq(issueThreadInteractions.issueId, issueId), eq(issueThreadInteractions.status, "pending"),
-        )).limit(1),
-        dbOrTx.select({ id: approvals.id }).from(issueApprovals).innerJoin(approvals, and(
-          eq(approvals.id, issueApprovals.approvalId), eq(approvals.companyId, issueApprovals.companyId),
-        )).where(and(
-          eq(issueApprovals.companyId, input.companyId), eq(issueApprovals.issueId, issueId),
-          inArray(approvals.status, ["pending", "revision_requested"]),
-        )).limit(1),
-      ]);
-      facts.pendingResponse = interactions.length > 0 ? "interaction" : linkedApprovals.length > 0 ? "approval" : null;
+      facts.pendingResponse = await connectionContinuationPendingResponse(dbOrTx as unknown as Db,
+        issue, input.responsibleUserId, input.runId);
     }
     facts.reviewParticipant = await readNativeReviewParticipantFacts(dbOrTx, {
       companyId: input.companyId, issueId, agentId: input.agentId,
@@ -464,6 +457,7 @@ export function createPostgresRunDispatchAdapter(
         companyId: run.companyId,
         agentId: run.agentId,
         conversationContinuation: run.runtimeMode === "legacy" && hasConversationContinuationPolicy(run.resultJson),
+        responsibleUserId: run.responsibleUserId,
         contextSnapshot: parseObject(run.contextSnapshot),
         scheduledRetryReason: run.scheduledRetryReason,
         retryReasonOverride: input.retryReasonOverride,
@@ -519,6 +513,9 @@ export function createPostgresRunDispatchAdapter(
     const issueQuery = dbOrTx
       .select({
         id: issues.id,
+        companyId: issues.companyId,
+        projectId: issues.projectId,
+        assigneeUserId: issues.assigneeUserId,
         status: issues.status,
         assigneeAgentId: issues.assigneeAgentId,
         executionRunId: issues.executionRunId,
@@ -600,6 +597,9 @@ export function createPostgresRunDispatchAdapter(
         unresolvedBlockerIssueIds: readiness.unresolvedBlockerIssueIds,
         unresolvedBlockerCount: readiness.unresolvedBlockerCount,
       } : null,
+      pendingResponse: issue && (input.conversationContinuation || context.connectionIntentResolution === "existing_connection")
+        ? await connectionContinuationPendingResponse(dbOrTx as unknown as Db, issue, input.responsibleUserId, input.runId)
+        : null,
       issueFound: issue !== null,
       issueStatus: issue?.status ?? null,
       issueAssigneeAgentId: issue?.assigneeAgentId ?? null,
@@ -770,6 +770,7 @@ export function createPostgresRunDispatchAdapter(
           companyId: run.companyId,
           agentId: run.agentId,
           conversationContinuation: run.runtimeMode === "legacy" && hasConversationContinuationPolicy(run.resultJson),
+          responsibleUserId: run.responsibleUserId,
           contextSnapshot: parseObject(run.contextSnapshot),
           scheduledRetryReason: run.scheduledRetryReason,
           retryReasonOverride: run.scheduledRetryReason,
@@ -881,6 +882,7 @@ export function createPostgresRunDispatchAdapter(
     decision: Extract<ReturnType<typeof decideQueuedRunStaleness>, { stale: true }>,
     expectedStatus: "queued" | "running",
     now: Date,
+    providerDispatchNotStarted = false,
   ): Promise<CancelStaleQueuedRunOutcome> {
       const [row] = await tx
         .update(heartbeatRuns)
@@ -892,6 +894,9 @@ export function createPostgresRunDispatchAdapter(
           resultJson: {
             ...parseObject(run.resultJson),
             stopReason: decision.errorCode,
+            ...(providerDispatchNotStarted && decision.errorCode === "issue_waiting_for_response"
+              ? { preDispatchResponseWait: { version: 1, providerWorkStarted: false } }
+              : {}),
             ...(decision.errorCode === "execution_reconciliation_required"
               ? { executionWait: decision.details }
               : {}),
@@ -927,6 +932,20 @@ export function createPostgresRunDispatchAdapter(
           );
       }
 
+      const context = parseObject(row.contextSnapshot);
+      if (decision.errorCode === "issue_waiting_for_response" && context.connectionIntentResolution === "existing_connection"
+        && context.source === "connection_intent.resolved" && context.interactionKind === "connection_intent"
+        && typeof context.interactionId === "string") {
+        // The partial wake-key uniqueness fence permits another delivery after
+        // this skipped receipt. Rearm atomically so restored access or a later
+        // answer can resume the untouched request after a restart.
+        await tx.update(connectionIntentDeliveries).set({ deliveredAt: null, nextAttemptAt: now }).where(and(
+          eq(connectionIntentDeliveries.companyId, row.companyId),
+          eq(connectionIntentDeliveries.interactionId, context.interactionId),
+        ));
+        await notifyDeliveryWork(tx, DELIVERY_QUEUES.connection);
+      }
+
       await tx
         .update(issues)
         .set({
@@ -976,6 +995,8 @@ export function createPostgresRunDispatchAdapter(
         runId: run.id,
         companyId: run.companyId,
         agentId: run.agentId,
+        responsibleUserId: run.responsibleUserId,
+        conversationContinuation: run.runtimeMode === "legacy" && hasConversationContinuationPolicy(run.resultJson),
         issueId,
         contextSnapshot,
         scheduledRetryReason: run.scheduledRetryReason,
@@ -1129,6 +1150,7 @@ export function createPostgresRunDispatchAdapter(
           decision,
           input.expectedStatus,
           input.now,
+          true, // This locked gate has not handed off to the provider callback.
         );
         return { dispatched: false as const, cancellation };
       }

server/src/modules/run-dispatch/domain/policy.ts

@@ -108,6 +108,7 @@ export type ScheduledRetryFacts = {
 };
 
 export type QueuedRunStalenessErrorCode =
+  | "issue_waiting_for_response"
   | "execution_reconciliation_required"
   | "issue_dependencies_blocked"
   | "issue_not_found"
@@ -129,6 +130,7 @@ export type StalenessDecision =
     };
 
 export type QueuedRunFacts = {
+  pendingResponse?: "interaction" | "approval" | null;
   /** Rechecked for automatic native replacements immediately before dispatch. */
   dependenciesBlocked?: DependencyBlockFacts | null;
   runId: string;
@@ -546,6 +548,12 @@ export function decideQueuedRunStaleness(
     }
   }
 
+  if (facts.pendingResponse) {
+    return { stale: true, errorCode: "issue_waiting_for_response",
+      reason: "Cancelled because the continuation is still waiting for an authorized response",
+      details: { issueId: facts.issueId, pendingResponse: facts.pendingResponse } };
+  }
+
   if (facts.continuationParkApplies && facts.continuationParksExecutor) {
     return {
       stale: true,

server/src/modules/run-dispatch/domain/wake-context.test.ts

@@ -9,6 +9,13 @@ import {
 } from "./wake-context.js";
 
 describe("wake context", () => {
+  it("recognizes a retired available connection without accepting ordinary expiry", () => {
+    const context = { interactionId: "connection", interactionStatus: "expired", interactionKind: "connection_intent",
+      connectionIntentResolution: "existing_connection", mutation: "interaction", wakeReason: "issue_commented" };
+    expect(isResolvedInteractionContinuationWakeContext(context)).toBe(true);
+    expect(isResolvedInteractionContinuationWakeContext({ ...context, interactionKind: "request_confirmation" })).toBe(false);
+    expect(isResolvedInteractionContinuationWakeContext({ ...context, connectionIntentResolution: undefined })).toBe(false);
+  });
   it("requires the subscription-specific non-assignee receipt for a subscription wait", () => {
     expect(isNonAssigneeWorkspaceBusyRetry("ai_connection_busy", {
       aiConnectionBusyDeferredWhileAssignee: false,

server/src/modules/run-dispatch/domain/wake-context.ts

@@ -95,7 +95,10 @@ export function isResolvedInteractionContinuationWakeContext(contextSnapshot: un
   const interactionId = readNonEmptyString(context.interactionId);
   const interactionStatus = readNonEmptyString(context.interactionStatus);
   if (!interactionId || !interactionStatus) return false;
-  if (!RESOLVED_INTERACTION_CONTINUATION_STATUSES.has(interactionStatus)) return false;
+  const retiredConnection = interactionStatus === "expired"
+    && context.interactionKind === "connection_intent"
+    && context.connectionIntentResolution === "existing_connection";
+  if (!retiredConnection && !RESOLVED_INTERACTION_CONTINUATION_STATUSES.has(interactionStatus)) return false;
 
   const mutation = readNonEmptyString(context.mutation);
   const wakeReason = readNonEmptyString(context.wakeReason);