Promote stage to bkt3: plan supersession (#317) and Claude account revive queue (#322)

apps/server/src/auth/rpcForkScopes.ts

@@ -71,6 +71,7 @@ export const FORK_RPC_REQUIRED_SCOPES = {
   [WS_FORK_METHODS.subscribeClaudeAccounts]: AuthOrchestrationReadScope,
   [WS_FORK_METHODS.subscribeThreadClaudeAccount]: AuthOrchestrationReadScope,
   [WS_FORK_METHODS.claudeAccountsSetThreadMode]: AuthOrchestrationOperateScope,
+  [WS_FORK_METHODS.claudeAccountsLeaveQueue]: AuthOrchestrationOperateScope,
   // T3-CUSTOM(expbkt3): Claude account access per user. Listing is a read and
   // replacing an allow list is operate; both handlers also require an admin.
   [WS_FORK_METHODS.claudeAccountsAccessList]: AuthOrchestrationReadScope,

apps/server/src/claudeAccounts/ClaudeAutoswitchClient.test.ts

@@ -149,6 +149,37 @@ describe("ClaudeAutoswitchClient", () => {
     }),
   );
 
+  it.effect("decodes the space-share-v2 fields and keeps older answers without them", () =>
+    Effect.gen(function* () {
+      const answers = [
+        '{"status":"no_eligible_profile","chosen":null,"recovers_at":"2026-10-16T12:30:00Z","recoveries":[{"profile":"tushar","recovers_at":"2026-10-16T12:30:00Z"},{"profile":"agent","recovers_at":"2026-10-16T12:31:00Z"}],"snapshot":{"place_rule":"space-share-v2","profiles":[{"name":"sam","sessions":11,"fair_share":6,"load":12},{"name":"tushar","sessions":3,"fair_share":null,"load":3}]}}',
+        '{"status":"no_eligible_profile","chosen":null,"recovers_at":null}',
+      ];
+      const { layer } = harness(() => output({ stdout: answers.shift() ?? "" }));
+      const [current, older] = yield* Effect.gen(function* () {
+        const client = yield* ClaudeAutoswitchClient;
+        const first = yield* client.place({ pending: {}, avoid: [] });
+        const second = yield* client.place({ pending: {}, avoid: [] });
+        return [first, second] as const;
+      }).pipe(Effect.provide(layer));
+
+      assert.equal(current.kind, "ok");
+      if (current.kind === "ok") {
+        assert.deepEqual(current.value.recoveries, [
+          { profile: "tushar", recovers_at: "2026-10-16T12:30:00Z" },
+          { profile: "agent", recovers_at: "2026-10-16T12:31:00Z" },
+        ]);
+        const profiles = current.value.snapshot?.profiles ?? [];
+        assert.equal(current.value.snapshot?.place_rule, "space-share-v2");
+        assert.equal(profiles[0]?.fair_share, 6);
+        assert.equal(profiles[0]?.load, 12);
+        assert.equal(profiles[1]?.fair_share, null);
+      }
+      assert.equal(older.kind, "ok");
+      if (older.kind === "ok") assert.equal(older.value.recoveries, undefined);
+    }),
+  );
+
   it.effect(
     "passes the limit type and profile to --hard-limit and honours the configured path",
     () =>

apps/server/src/claudeAccounts/ClaudeAutoswitchClient.ts

@@ -68,6 +68,10 @@ export const SwitcherProfile = Schema.Struct({
   place_rank: Schema.optionalKey(NullableNumber),
   place_score: Schema.optionalKey(NullableNumber),
   guard_ok: Schema.optionalKey(Schema.NullOr(Schema.Boolean)),
+  // `space-share-v2`: the share of live sessions this account should carry
+  // (null when it is not placeable now) and its live plus pending sessions.
+  fair_share: Schema.optionalKey(NullableNumber),
+  load: Schema.optionalKey(NullableNumber),
 });
 export type SwitcherProfile = typeof SwitcherProfile.Type;
 
@@ -91,6 +95,20 @@ export const SwitcherPlaceResult = Schema.Struct({
   dir: Schema.optionalKey(NullableString),
   reason: Schema.optionalKey(NullableString),
   recovers_at: Schema.optionalKey(NullableString),
+  /**
+   * On `no_eligible_profile`, when each account is expected to have room,
+   * earliest first (`recovers_at` is the first). Absent from older switchers.
+   */
+  recoveries: Schema.optionalKey(
+    Schema.NullOr(
+      Schema.Array(
+        Schema.Struct({
+          profile: Schema.String,
+          recovers_at: Schema.String,
+        }),
+      ),
+    ),
+  ),
   snapshot: Schema.optionalKey(Schema.NullOr(SwitcherStatus)),
 });
 export type SwitcherPlaceResult = typeof SwitcherPlaceResult.Type;

apps/server/src/claudeAccounts/applyClaudeAccountProfile.ts

@@ -23,6 +23,9 @@ export class ClaudeAccountPolicyError extends Schema.TaggedError<ClaudeAccountPo
   {
     threadId: Schema.String,
     detail: Schema.String,
+    /** When no account had room: when the first one the owner may use is expected to. */
+    expectedAt: Schema.optionalKey(Schema.String),
+    expectedProfile: Schema.optionalKey(Schema.String),
   },
 ) {
   override get message(): string {

apps/server/src/mcp/toolkits/webUi/catalog.test.ts

@@ -54,7 +54,7 @@ const invocation = (
 
 it("generates one unique virtual tool and complete schemas for every web RPC", () => {
   // T3-CUSTOM(expbkt3): the catalog includes native V2 methods and every retained fork RPC.
-  expect(WEB_UI_VIRTUAL_TOOL_COUNT).toBe(236);
+  expect(WEB_UI_VIRTUAL_TOOL_COUNT).toBe(237);
   expect(WEB_UI_STREAM_TOOL_COUNT).toBe(31);
   expect(WEB_UI_VIRTUAL_TOOL_COUNT).toBe(WsRpcGroup.requests.size);
   expect(new Set(WEB_UI_VIRTUAL_TOOLS.map((tool) => tool.name)).size).toBe(

apps/server/src/mcp/toolkits/webUi/registration.test.ts

@@ -81,9 +81,9 @@ it.effect("registers four compact tools while listing the complete virtual surfa
       expect(listed.structuredContent).toMatchObject({
         ok: true,
         // T3-CUSTOM(expbkt3): registration exposes native V2 methods and all retained fork RPCs.
-        rpcCount: 236,
+        rpcCount: 237,
         streamCount: 31,
-        matchedCount: 236,
+        matchedCount: 237,
       });
 
       const schema = yield* withInvocation(

apps/server/src/orchestration-v2/Layers/OrchestrationEngine.ts

@@ -274,6 +274,23 @@ const make = Effect.gen(function* () {
         }
         case "thread.activity.append":
           return yield* dispatchNative(command, actorUserId);
+        // T3-CUSTOM(expbkt3): plan review records a decided plan through the legacy
+        // upsert; V2 keeps the decision as the plan artifact's status.
+        case "thread.proposed-plan.upsert":
+          if (command.proposedPlan.implementedAt === null) {
+            return yield* dispatchNative(command, actorUserId);
+          }
+          return yield* dispatchNative(
+            {
+              type: "thread.proposed-plan.retire",
+              commandId: command.commandId,
+              threadId: command.threadId,
+              planId: command.proposedPlan.id,
+              status:
+                command.proposedPlan.implementationThreadId === null ? "superseded" : "completed",
+            },
+            actorUserId,
+          );
         default:
           return yield* dispatchNative(command, actorUserId);
       }

apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts

@@ -334,6 +334,150 @@ it.effect("implements a proposed plan that the command projection leaves out", (
   }).pipe(Effect.provide(layerTest)),
 );
 
+const createPlanThread = (name: string) =>
+  Effect.gen(function* () {
+    const orchestrator = yield* Orchestrator.OrchestratorV2;
+    const threadId = ThreadId.make(`thread:${name}`);
+    yield* orchestrator.dispatch({
+      type: "thread.create",
+      commandId: CommandId.make(`create-${name}`),
+      threadId,
+      projectId: ProjectId.make(`project:${name}`),
+      title: "Plan",
+      modelSelection,
+      runtimeMode: "full-access",
+      interactionMode: "plan",
+      branch: null,
+      worktreePath: null,
+      createdBy: "user",
+      creationSource: "web",
+    });
+    return threadId;
+  });
+
+const applyProposedPlan = (
+  threadId: ThreadId,
+  name: string,
+  status: "active" | "completed" | "superseded" = "active",
+) =>
+  Effect.gen(function* () {
+    const projections = yield* ProjectionStore.ProjectionStoreV2;
+    const planId = PlanId.make(`plan:${name}`);
+    yield* projections.apply({
+      id: EventId.make(`plan:${name}:${status}`),
+      type: "plan.updated",
+      threadId,
+      occurredAt: yield* DateTime.now,
+      payload: {
+        id: planId,
+        threadId,
+        runId: null,
+        nodeId: NodeId.make(`node:${name}`),
+        kind: "proposed_plan",
+        status,
+        markdown: `# Plan ${name}`,
+      },
+    });
+    return planId;
+  });
+
+const continueThread = (
+  threadId: ThreadId,
+  name: string,
+  sourcePlanRef?: { readonly threadId: ThreadId; readonly planId: PlanId },
+  text = "What about the rollout?",
+) =>
+  Effect.gen(function* () {
+    const orchestrator = yield* Orchestrator.OrchestratorV2;
+    yield* orchestrator.dispatch({
+      type: "message.dispatch",
+      commandId: CommandId.make(`continue-${name}`),
+      threadId,
+      messageId: MessageId.make(`continue-${name}-input`),
+      text,
+      attachments: [],
+      ...(sourcePlanRef === undefined ? {} : { sourcePlanRef }),
+      dispatchMode: { type: "defer_start" },
+      createdBy: "user",
+      creationSource: "web",
+    });
+  });
+
+it.effect("supersedes an open plan when the thread continues without it", () =>
+  Effect.gen(function* () {
+    const projections = yield* ProjectionStore.ProjectionStoreV2;
+    const threadId = yield* createPlanThread("continue-plan");
+    const planId = yield* applyProposedPlan(threadId, "continue-plan");
+
+    yield* continueThread(threadId, "continue-plan");
+
+    assert.equal((yield* projections.getPlan(threadId, planId))?.status, "superseded");
+    assert.deepEqual(yield* projections.getActiveProposedPlans(threadId), []);
+  }).pipe(Effect.provide(layerTest)),
+);
+
+it.effect("leaves an open plan alone for a maintenance command", () =>
+  Effect.gen(function* () {
+    const projections = yield* ProjectionStore.ProjectionStoreV2;
+    const threadId = yield* createPlanThread("compact-plan");
+    const planId = yield* applyProposedPlan(threadId, "compact-plan");
+
+    yield* continueThread(threadId, "compact-plan", undefined, "/compact");
+
+    assert.equal((yield* projections.getPlan(threadId, planId))?.status, "active");
+  }).pipe(Effect.provide(layerTest)),
+);
+
+it.effect("completes the implemented plan and supersedes every other open plan", () =>
+  Effect.gen(function* () {
+    const projections = yield* ProjectionStore.ProjectionStoreV2;
+    const threadId = yield* createPlanThread("implement-newest");
+    const older = yield* applyProposedPlan(threadId, "implement-newest-r2");
+    const newest = yield* applyProposedPlan(threadId, "implement-newest-r3");
+
+    yield* continueThread(threadId, "implement-newest", { threadId, planId: newest });
+
+    assert.equal((yield* projections.getPlan(threadId, newest))?.status, "completed");
+    assert.equal((yield* projections.getPlan(threadId, older))?.status, "superseded");
+  }).pipe(Effect.provide(layerTest)),
+);
+
+it.effect("implements a superseded plan that the turn names", () =>
+  Effect.gen(function* () {
+    const projections = yield* ProjectionStore.ProjectionStoreV2;
+    const threadId = yield* createPlanThread("implement-superseded");
+    const planId = yield* applyProposedPlan(threadId, "implement-superseded", "superseded");
+
+    yield* continueThread(threadId, "implement-superseded", { threadId, planId });
+
+    assert.equal((yield* projections.getPlan(threadId, planId))?.status, "completed");
+  }).pipe(Effect.provide(layerTest)),
+);
+
+it.effect("retires an open plan whose review was decided", () =>
+  Effect.gen(function* () {
+    const orchestrator = yield* Orchestrator.OrchestratorV2;
+    const projections = yield* ProjectionStore.ProjectionStoreV2;
+    const threadId = yield* createPlanThread("retire-plan");
+    const open = yield* applyProposedPlan(threadId, "retire-plan-open");
+    const done = yield* applyProposedPlan(threadId, "retire-plan-done", "completed");
+
+    for (const planId of [open, done]) {
+      yield* orchestrator.dispatch({
+        type: "thread.proposed-plan.retire",
+        commandId: CommandId.make(`retire-${planId}`),
+        threadId,
+        planId,
+        status: "superseded",
+      });
+    }
+
+    assert.equal((yield* projections.getPlan(threadId, open))?.status, "superseded");
+    // A finished plan keeps its outcome.
+    assert.equal((yield* projections.getPlan(threadId, done))?.status, "completed");
+  }).pipe(Effect.provide(layerTest)),
+);
+
 // Stop's settle follow-up runs after the provider interrupt returns, possibly
 // long after the Stop (retries) or again (an effect replayed after a crash).
 // A later run's background work is not that Stop's to end.