refactor(cloud-workers): centralize placement lifecycle fences (#121658)

* refactor(cloud-workers): centralize placement ownership fences Amp-Thread-ID: https://ampcode.com/threads/T-019feaaa-c7ed-769e-9f29-a3612bec72e7 * test: restore core test typechecking Amp-Thread-ID: https://ampcode.com/threads/T-019feaaa-c7ed-769e-9f29-a3612bec72e7 * test: repair current-main CI regressions Amp-Thread-ID: https://ampcode.com/threads/T-019feaaa-c7ed-769e-9f29-a3612bec72e7 * test: align shared main contracts Amp-Thread-ID: https://ampcode.com/threads/T-019feaaa-c7ed-769e-9f29-a3612bec72e7 * docs(cloud-workers): document placement fence Amp-Thread-ID: https://ampcode.com/threads/T-019feaaa-c7ed-769e-9f29-a3612bec72e7 * test(approvals): align account routing fixtures Amp-Thread-ID: https://ampcode.com/threads/T-019feaaa-c7ed-769e-9f29-a3612bec72e7 * fix(protocol): refresh Swift approval resolve models Amp-Thread-ID: https://ampcode.com/threads/T-019feaaa-c7ed-769e-9f29-a3612bec72e7 * test(approvals): use scoped event kind fixtures Amp-Thread-ID: https://ampcode.com/threads/T-019feaaa-c7ed-769e-9f29-a3612bec72e7 --------- Co-authored-by: Amp <amp@ampcode.com>
2026-08-27 21:07:01 -06:00 · 2026-08-10 15:15:24 -07:00
parent 1263450938
commit 5fb5eefa61
3 changed files with 110 additions and 119 deletions
@@ -15742,21 +15742,25 @@ public struct ApprovalResolveParams: Codable, Sendable {
    public let id: String
    public let kind: ApprovalKind
    public let decision: ApprovalDecision
+    public let reviewer: [String: AnyCodable]?

    public init(
        id: String,
        kind: ApprovalKind,
-        decision: ApprovalDecision)
+        decision: ApprovalDecision,
+        reviewer: [String: AnyCodable]? = nil)
    {
        self.id = id
        self.kind = kind
        self.decision = decision
+        self.reviewer = reviewer
    }

    private enum CodingKeys: String, CodingKey {
        case id
        case kind
        case decision
+        case reviewer
    }
 }

@@ -16143,18 +16147,22 @@ public struct ExecApprovalRequestParams: Codable, Sendable {
 public struct ExecApprovalResolveParams: Codable, Sendable {
    public let id: String
    public let decision: String
+    public let reviewer: [String: AnyCodable]?

    public init(
        id: String,
-        decision: String)
+        decision: String,
+        reviewer: [String: AnyCodable]? = nil)
    {
        self.id = id
        self.decision = decision
+        self.reviewer = reviewer
    }

    private enum CodingKeys: String, CodingKey {
        case id
        case decision
+        case reviewer
    }
 }

@@ -16513,18 +16521,22 @@ public struct PluginApprovalRequestParams: Codable, Sendable {
 public struct PluginApprovalResolveParams: Codable, Sendable {
    public let id: String
    public let decision: String
+    public let reviewer: [String: AnyCodable]?

    public init(
        id: String,
-        decision: String)
+        decision: String,
+        reviewer: [String: AnyCodable]? = nil)
    {
        self.id = id
        self.decision = decision
+        self.reviewer = reviewer
    }

    private enum CodingKeys: String, CodingKey {
        case id
        case decision
+        case reviewer
    }
 }

@@ -56,6 +56,58 @@ class WorkerDispatchTargetChangedError extends Error {
  readonly code = "invalid_state";
 }

+type WorkerPlacementSessionRuntime = Awaited<
+  ReturnType<typeof loadWorkerPlacementSessionRuntimeModule>
+>;
+type WorkerPlacementSessionTarget = ReturnType<
+  WorkerPlacementSessionRuntime["resolveGatewaySessionStoreTargetWithStore"]
+>;
+
+/** Keeps store identity, session incarnation, canonical ownership, and the live worktree
+ * in one cross-phase fence. Initial resolution throws normally; barrier revalidation
+ * supplies expectedTarget and yields an invalid_state retry when the target changed. */
+function resolveWorkerPlacementSessionTarget(params: {
+  sessionRuntime: WorkerPlacementSessionRuntime;
+  config: ReturnType<typeof getRuntimeConfig>;
+  sessionId: string;
+  sessionKey: string;
+  agentId: string;
+  expectedTarget?: WorkerPlacementSessionTarget;
+  errorMessage: string;
+}) {
+  const target = params.sessionRuntime.resolveGatewaySessionStoreTargetWithStore({
+    cfg: params.config,
+    key: params.sessionKey,
+    agentId: params.agentId,
+    clone: false,
+  });
+  const entry = params.sessionRuntime.resolveCanonicalSessionEntryFromStoreKeys(
+    target.store,
+    target.storeKeys,
+  );
+  const worktree = params.sessionRuntime.managedWorktrees.findLiveByOwner(
+    "session",
+    target.canonicalKey,
+  );
+  const expected = params.expectedTarget;
+  if (
+    (expected &&
+      (target.storePath !== expected.storePath ||
+        target.canonicalKey !== expected.canonicalKey ||
+        target.agentId !== expected.agentId)) ||
+    entry?.sessionId !== params.sessionId ||
+    !entry.worktree?.id ||
+    !worktree ||
+    worktree.id !== entry.worktree.id ||
+    worktree.ownerId !== target.canonicalKey
+  ) {
+    throw expected
+      ? new WorkerDispatchTargetChangedError(params.errorMessage)
+      : new Error(params.errorMessage);
+  }
+  return { config: params.config, target, entry, worktree };
+}
+
 /** Serializes reconciliation sweeps against in-flight dispatches so a sweep never
 * observes a placement mid-transition. Dispatches wait out any pending sweep. */
 export function coordinateWorkerPlacementDispatch(
@@ -200,28 +252,15 @@ export function createGatewayWorkerPlacementRuntime(params: GatewayWorkerPlaceme
    sessionKey: string;
    agentId: string;
  }): Promise<string> => {
-    const {
-      managedWorktrees,
-      resolveCanonicalSessionEntryFromStoreKeys,
-      resolveGatewaySessionStoreTargetWithStore,
-    } = await loadWorkerPlacementSessionRuntimeModule();
-    const target = resolveGatewaySessionStoreTargetWithStore({
-      cfg: getRuntimeConfig(),
-      key: sessionKey,
+    const sessionRuntime = await loadWorkerPlacementSessionRuntimeModule();
+    const { worktree } = resolveWorkerPlacementSessionTarget({
+      sessionRuntime,
+      config: getRuntimeConfig(),
+      sessionId,
+      sessionKey,
      agentId,
-      clone: false,
+      errorMessage: `Session ${sessionKey} dispatch requires a session-owned managed worktree`,
    });
-    const sessionEntry = resolveCanonicalSessionEntryFromStoreKeys(target.store, target.storeKeys);
-    const worktree = managedWorktrees.findLiveByOwner("session", target.canonicalKey);
-    if (
-      sessionEntry?.sessionId !== sessionId ||
-      !sessionEntry.worktree?.id ||
-      !worktree ||
-      worktree.id !== sessionEntry.worktree.id ||
-      worktree.ownerId !== target.canonicalKey
-    ) {
-      throw new Error(`Session ${sessionKey} dispatch requires a session-owned managed worktree`);
-    }
    return worktree.path;
  };
  const dispatchService = coordinateWorkerPlacementDispatch(
@@ -230,13 +269,12 @@ export function createGatewayWorkerPlacementRuntime(params: GatewayWorkerPlaceme
      environments: params.environments,
      ...workspaceConflictHandlers,
      runLocalBarrier: async ({ sessionId, sessionKey, agentId, startDispatch }) => {
+        const sessionRuntime = await loadWorkerPlacementSessionRuntimeModule();
        const {
          isWorkerPlacementSessionRuntimeSupported,
-          managedWorktrees,
-          resolveCanonicalSessionEntryFromStoreKeys,
          resolveGatewaySessionStoreTargetWithStore,
          resolveWorkerPlacementSessionRuntime,
-        } = await loadWorkerPlacementSessionRuntimeModule();
+        } = sessionRuntime;
        const target = resolveGatewaySessionStoreTargetWithStore({
          cfg: getRuntimeConfig(),
          key: sessionKey,
@@ -254,35 +292,20 @@ export function createGatewayWorkerPlacementRuntime(params: GatewayWorkerPlaceme
          scope: target.storePath,
          identities: lifecycleIdentities,
          prepare: async () => {
-            const currentConfig = getRuntimeConfig();
-            const currentTarget = resolveGatewaySessionStoreTargetWithStore({
-              cfg: currentConfig,
-              key: sessionKey,
+            const {
+              config: currentConfig,
+              target: currentTarget,
+              entry: currentEntry,
+              worktree,
+            } = resolveWorkerPlacementSessionTarget({
+              sessionRuntime,
+              config: getRuntimeConfig(),
+              sessionId,
+              sessionKey,
              agentId,
-              clone: false,
+              expectedTarget: target,
+              errorMessage: `Session ${sessionKey} changed before cloud worker dispatch. Retry.`,
            });
-            const currentEntry = resolveCanonicalSessionEntryFromStoreKeys(
-              currentTarget.store,
-              currentTarget.storeKeys,
-            );
-            const worktree = managedWorktrees.findLiveByOwner(
-              "session",
-              currentTarget.canonicalKey,
-            );
-            if (
-              currentTarget.storePath !== target.storePath ||
-              currentTarget.canonicalKey !== target.canonicalKey ||
-              currentTarget.agentId !== target.agentId ||
-              currentEntry?.sessionId !== sessionId ||
-              !currentEntry.worktree?.id ||
-              !worktree ||
-              worktree.id !== currentEntry.worktree.id ||
-              worktree.ownerId !== currentTarget.canonicalKey
-            ) {
-              throw new WorkerDispatchTargetChangedError(
-                `Session ${sessionKey} changed before cloud worker dispatch. Retry.`,
-              );
-            }
            if (currentEntry.archivedAt !== undefined) {
              throw new WorkerDispatchTargetChangedError(
                `Session ${sessionKey} was archived before cloud worker dispatch. Retry.`,
@@ -334,13 +357,12 @@ export function createGatewayWorkerPlacementRuntime(params: GatewayWorkerPlaceme
        return placement;
      },
      runActivationBarrier: async ({ sessionId, sessionKey, agentId, activate }) => {
+        const sessionRuntime = await loadWorkerPlacementSessionRuntimeModule();
        const {
          isWorkerPlacementSessionRuntimeSupported,
-          managedWorktrees,
-          resolveCanonicalSessionEntryFromStoreKeys,
          resolveGatewaySessionStoreTargetWithStore,
          resolveWorkerPlacementSessionRuntime,
-        } = await loadWorkerPlacementSessionRuntimeModule();
+        } = sessionRuntime;
        const target = resolveGatewaySessionStoreTargetWithStore({
          cfg: getRuntimeConfig(),
          key: sessionKey,
@@ -358,35 +380,19 @@ export function createGatewayWorkerPlacementRuntime(params: GatewayWorkerPlaceme
          scope: target.storePath,
          identities: lifecycleIdentities,
          run: async () => {
-            const currentConfig = getRuntimeConfig();
-            const currentTarget = resolveGatewaySessionStoreTargetWithStore({
-              cfg: currentConfig,
-              key: sessionKey,
+            const {
+              config: currentConfig,
+              target: currentTarget,
+              entry: currentEntry,
+            } = resolveWorkerPlacementSessionTarget({
+              sessionRuntime,
+              config: getRuntimeConfig(),
+              sessionId,
+              sessionKey,
              agentId,
-              clone: false,
+              expectedTarget: target,
+              errorMessage: `Session ${sessionKey} changed before cloud worker activation. Retry.`,
            });
-            const currentEntry = resolveCanonicalSessionEntryFromStoreKeys(
-              currentTarget.store,
-              currentTarget.storeKeys,
-            );
-            const worktree = managedWorktrees.findLiveByOwner(
-              "session",
-              currentTarget.canonicalKey,
-            );
-            if (
-              currentTarget.storePath !== target.storePath ||
-              currentTarget.canonicalKey !== target.canonicalKey ||
-              currentTarget.agentId !== target.agentId ||
-              currentEntry?.sessionId !== sessionId ||
-              !currentEntry.worktree?.id ||
-              !worktree ||
-              worktree.id !== currentEntry.worktree.id ||
-              worktree.ownerId !== currentTarget.canonicalKey
-            ) {
-              throw new WorkerDispatchTargetChangedError(
-                `Session ${sessionKey} changed before cloud worker activation. Retry.`,
-              );
-            }
            if (currentEntry.archivedAt !== undefined) {
              throw new WorkerDispatchTargetChangedError(
                `Session ${sessionKey} was archived before cloud worker activation. Retry.`,
@@ -412,11 +418,8 @@ export function createGatewayWorkerPlacementRuntime(params: GatewayWorkerPlaceme
        return activePlacement;
      },
      runReclaimBarrier: async ({ sessionId, sessionKey, agentId, reclaim }) => {
-        const {
-          managedWorktrees,
-          resolveCanonicalSessionEntryFromStoreKeys,
-          resolveGatewaySessionStoreTargetWithStore,
-        } = await loadWorkerPlacementSessionRuntimeModule();
+        const sessionRuntime = await loadWorkerPlacementSessionRuntimeModule();
+        const { resolveGatewaySessionStoreTargetWithStore } = sessionRuntime;
        const target = resolveGatewaySessionStoreTargetWithStore({
          cfg: getRuntimeConfig(),
          key: sessionKey,
@@ -435,34 +438,15 @@ export function createGatewayWorkerPlacementRuntime(params: GatewayWorkerPlaceme
          scope: target.storePath,
          identities: lifecycleIdentities,
          prepare: async () => {
-            const currentTarget = resolveGatewaySessionStoreTargetWithStore({
-              cfg: getRuntimeConfig(),
-              key: sessionKey,
+            const { worktree } = resolveWorkerPlacementSessionTarget({
+              sessionRuntime,
+              config: getRuntimeConfig(),
+              sessionId,
+              sessionKey,
              agentId,
-              clone: false,
+              expectedTarget: target,
+              errorMessage: `Session ${sessionKey} changed before cloud worker stop. Retry.`,
            });
-            const currentEntry = resolveCanonicalSessionEntryFromStoreKeys(
-              currentTarget.store,
-              currentTarget.storeKeys,
-            );
-            const worktree = managedWorktrees.findLiveByOwner(
-              "session",
-              currentTarget.canonicalKey,
-            );
-            if (
-              currentTarget.storePath !== target.storePath ||
-              currentTarget.canonicalKey !== target.canonicalKey ||
-              currentTarget.agentId !== target.agentId ||
-              currentEntry?.sessionId !== sessionId ||
-              !currentEntry.worktree?.id ||
-              !worktree ||
-              worktree.id !== currentEntry.worktree.id ||
-              worktree.ownerId !== currentTarget.canonicalKey
-            ) {
-              throw new WorkerDispatchTargetChangedError(
-                `Session ${sessionKey} changed before cloud worker stop. Retry.`,
-              );
-            }
            const placement = params.placements.get(sessionId);
            if (placement?.state !== "active" || placement.turnClaim) {
              throw new Error(
@@ -232,11 +232,6 @@ export function createPlacementFailureActions(deps: {
    environment: ReturnType<WorkerEnvironmentService["get"]>,
    claimedTurnError: Error,
  ): Promise<void> => {
-    if (placement.turnClaim) {
-      const draining = startDrain(placement);
-      await failDraining(draining, claimedTurnError, { forceClaimFence: true });
-      return;
-    }
    const draining = startDrain(placement);
    if (draining.turnClaim) {
      await failDraining(draining, claimedTurnError, { forceClaimFence: true });