diff --git a/apps/server/src/auth/rpcForkScopes.ts b/apps/server/src/auth/rpcForkScopes.ts index 791e0b576b058..2519a525eabb4 100644 --- a/apps/server/src/auth/rpcForkScopes.ts +++ b/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, diff --git a/apps/server/src/claudeAccounts/ClaudeAccountsService.test.ts b/apps/server/src/claudeAccounts/ClaudeAccountsService.test.ts index 4017e80b1ebd7..6a59576d32be2 100644 --- a/apps/server/src/claudeAccounts/ClaudeAccountsService.test.ts +++ b/apps/server/src/claudeAccounts/ClaudeAccountsService.test.ts @@ -18,7 +18,9 @@ import { ThreadId, UserId, } from "@t3tools/contracts"; +import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; +import * as Fiber from "effect/Fiber"; import * as FileSystem from "effect/FileSystem"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; @@ -155,6 +157,8 @@ interface HarnessOptions { readonly placeFailed?: boolean; /** `--place` dies with a defect, as a bug in the client would. */ readonly placeDefect?: boolean; + /** Runs inside every `--place` call before it answers, e.g. to hold it in flight. */ + readonly beforePlace?: () => Effect.Effect; /** Replaces the thread-account repository, e.g. with one whose reads fail. */ readonly threadRepository?: Layer.Layer; } @@ -228,15 +232,21 @@ const makeHarness = (options: HarnessOptions = {}) => place: (input) => Effect.sync(() => { placeCalls.push({ pending: { ...input.pending }, avoid: input.avoid }); - if (options.placeUnavailable) { - return { kind: "unavailable", detail: "old switcher" } as const; - } - if (options.placeFailed) { - return { kind: "failed", detail: "timed out after 10s" } as const; - } - if (options.placeDefect) throw new Error("place exploded"); - return { kind: "ok", value: place(input) } as const; - }), + }).pipe( + Effect.andThen(options.beforePlace?.() ?? Effect.void), + Effect.andThen( + Effect.sync(() => { + if (options.placeUnavailable) { + return { kind: "unavailable", detail: "old switcher" } as const; + } + if (options.placeFailed) { + return { kind: "failed", detail: "timed out after 10s" } as const; + } + if (options.placeDefect) throw new Error("place exploded"); + return { kind: "ok", value: place(input) } as const; + }), + ), + ), hardLimit: (input) => Effect.sync(() => { hardLimitCalls.push(input); @@ -1371,9 +1381,14 @@ describe("ClaudeAccountsService hard limits", () => { "tushar resets at 00:02 UTC — this thread continues on it at 00:03 UTC.", ); assert.equal(waiting.resolvedProfile, "tushar"); + assert.equal(waiting.wait?.expectedProfile, "tushar"); assert.equal(harness.placeCalls.length, placementsBefore); + // The wait is a queue entry: the status poll after the reset continues it. + yield* service.refreshStatus(); + assert.equal(harness.dispatched.length, 0); yield* TestClock.adjust("3 minutes"); + yield* service.refreshStatus(); yield* takeUntil(seen, (view) => view.notice === undefined); assert.equal(harness.dispatched.length, 1); const { command, actorUserId } = harness.dispatched[0]!; @@ -1384,9 +1399,12 @@ describe("ClaudeAccountsService hard limits", () => { const resumed = yield* service.resolveForSession(threadId); assert.equal(resumed?.profile, "tushar"); assert.equal(resumed?.reason, "sticky"); + assert.equal(harness.placeCalls.length, placementsBefore); // Claude still refuses with the old reset: no second wait, the thread moves. + const started = yield* watchThreadInto(service, threadId); yield* PubSub.publish(harness.providerEvents, providerEvent("turn.started", threadId)); + yield* Queue.take(started.seen); yield* service.handleHardLimit({ threadId, rateLimitType: "five_hour", @@ -1441,20 +1459,29 @@ describe("ClaudeAccountsService hard limits", () => { }), ); - scenario("drops the wait when the thread's account mode changes", {}, (harness, service) => - Effect.gen(function* () { - yield* service.refreshStatus(); - const threadId = ThreadId.make("thread-reset-wait-mode"); - yield* service.resolveForSession(threadId); - yield* service.handleHardLimit({ - threadId, - rateLimitType: "five_hour", - resetAt: resetInTwoMinutes, - }); - yield* service.setThreadMode({ threadId, mode: { kind: "profile", profile: "agent" } }); - yield* TestClock.adjust("10 minutes"); - assert.equal(harness.dispatched.length, 0); - }), + scenario( + "drops the wait when the thread's account mode changes, continuing on the chosen account now", + {}, + (harness, service) => + Effect.gen(function* () { + yield* service.refreshStatus(); + const threadId = ThreadId.make("thread-reset-wait-mode"); + yield* service.resolveForSession(threadId); + yield* service.handleHardLimit({ + threadId, + rateLimitType: "five_hour", + resetAt: resetInTwoMinutes, + }); + const view = yield* service.setThreadMode({ + threadId, + mode: { kind: "profile", profile: "agent" }, + }); + assert.equal(view.wait, undefined); + assert.equal(harness.dispatched.length, 1); + yield* TestClock.adjust("10 minutes"); + yield* service.refreshStatus(); + assert.equal(harness.dispatched.length, 1); + }), ); scenario("moves instead when the reset is more than five minutes away", {}, (harness, service) => @@ -2037,3 +2064,668 @@ describe("ClaudeAccountsService access per user: owners, probing, hard limits, r }), ); }); + +/** + * A switcher whose `--place` the test opens and closes: no room until + * `state.open` names accounts, then the open account with the fewest pending + * placements. With no room it lists when agent, then barsha, recover. + * `state.hold` keeps every `--place` in flight until it is released. + */ +function gatedPlace() { + const state: { + open: ReadonlyArray; + hold?: { readonly entered: Deferred.Deferred; readonly release: Deferred.Deferred }; + } = { open: [] }; + const beforePlace = (): Effect.Effect => { + const hold = state.hold; + return hold === undefined + ? Effect.void + : Effect.andThen(Deferred.succeed(hold.entered, undefined), Deferred.await(hold.release)); + }; + const place = (input: PlaceInput): SwitcherPlaceResult => { + let chosen: string | undefined; + for (const name of state.open) { + if (input.avoid.includes(name)) continue; + if (chosen === undefined || (input.pending[name] ?? 0) < (input.pending[chosen] ?? 0)) { + chosen = name; + } + } + if (chosen === undefined) { + return { + status: "no_eligible_profile", + chosen: null, + dir: null, + reason: "every profile is over its limit", + recovers_at: "1970-01-01T00:30:00Z", + recoveries: [ + { profile: "agent", recovers_at: "1970-01-01T00:30:00Z" }, + { profile: "barsha", recovers_at: "1970-01-01T00:45:00Z" }, + ], + }; + } + return { status: "placed", chosen, dir: null, reason: "fewest sessions" }; + }; + return { state, place, beforePlace }; +} + +/** Domain events are handled in order on one fiber: a deleted, watched thread marks them done. */ +const drainDomainEvents = (harness: Harness, service: ClaudeAccountsService["Service"]) => + Effect.gen(function* () { + const drainId = ThreadId.make(`thread-drain-${(eventCounter += 1)}`); + const drain = yield* watchThreadInto(service, drainId); + yield* PubSub.publish(harness.domainEvents, { + type: "thread.deleted", + payload: { threadId: drainId, deletedAt: "1970-01-01T00:00:00.000Z" }, + } as unknown as OrchestrationEvent); + yield* Queue.take(drain.seen); + }); + +const sessionSet = ( + threadId: ThreadId, + status: "ready" | "running", + activeTurnId: string | null, +): OrchestrationEvent => + ({ + type: "thread.session-set", + payload: { threadId, session: orchestrationSession(threadId, status, activeTurnId) }, + }) as unknown as OrchestrationEvent; + +/** Marks a turn as running on the thread, as the settled projection would. */ +const markTurnRunning = ( + harness: Harness, + service: ClaudeAccountsService["Service"], + threadId: ThreadId, +) => + Effect.gen(function* () { + yield* PubSub.publish(harness.domainEvents, sessionSet(threadId, "running", "turn-1")); + yield* drainDomainEvents(harness, service); + }); + +/** A turn runs and settles on the thread, as the projection reports a turn boundary. */ +const endTurn = (harness: Harness, threadId: ThreadId) => + Effect.gen(function* () { + yield* PubSub.publish(harness.domainEvents, sessionSet(threadId, "running", "turn-1")); + yield* PubSub.publish(harness.domainEvents, sessionSet(threadId, "ready", null)); + }); + +const fiveHourAt = (threadId: ThreadId, resetAt: number) => ({ + threadId, + rateLimitType: "five_hour", + resetAt, +}); + +describe("ClaudeAccountsService revive queue", () => { + const farReset = 1_790_000_000; + + { + const gate = gatedPlace(); + scenario( + "handles a new turn's rejection after a failed move, even with the same reset", + { place: gate.place }, + (harness, service) => + Effect.gen(function* () { + yield* service.refreshStatus(); + const threadId = ThreadId.make("thread-queue-same-reset"); + gate.state.open = ["tushar"]; + assert.equal((yield* service.resolveForSession(threadId))?.profile, "tushar"); + gate.state.open = []; + + assert.isTrue(yield* service.handleHardLimit(fiveHourAt(threadId, farReset))); + assert.deepEqual(harness.stoppedThreads, [threadId]); + assert.equal(harness.dispatched.length, 0); + assert.equal((yield* service.getThread(threadId)).wait?.position, 1); + // Copies that arrive while it waits are consumed quietly. + yield* service.handleHardLimit(fiveHourAt(threadId, farReset)); + assert.equal(harness.hardLimitCalls.length, 1); + + // The person writes: the message takes the thread out of the queue. + const { seen } = yield* watchThreadInto(service, threadId); + yield* PubSub.publish(harness.domainEvents, { + type: "thread.turn-start-requested", + payload: { threadId }, + } as unknown as OrchestrationEvent); + yield* takeUntil(seen, (view) => view.wait === undefined); + gate.state.open = ["agent"]; + yield* service.refreshStatus(); + assert.equal(harness.dispatched.length, 0); + + // That turn is refused with the same reset: it is a new rejection, and agent has room. + assert.isTrue(yield* service.handleHardLimit(fiveHourAt(threadId, farReset))); + assert.equal(harness.hardLimitCalls.length, 2); + assert.equal((yield* service.getThread(threadId)).resolvedProfile, "agent"); + assert.equal(harness.dispatched.length, 1); + }), + ); + } + + { + const gate = gatedPlace(); + scenario( + "parks Auto threads no account can take and revives them when one has room", + { place: gate.place }, + (harness, service) => + Effect.gen(function* () { + yield* service.refreshStatus(); + const active = ThreadId.make("thread-queue-active"); + const idle = ThreadId.make("thread-queue-idle"); + gate.state.open = ["tushar"]; + yield* service.resolveForSession(active); + yield* service.resolveForSession(idle); + gate.state.open = []; + yield* markTurnRunning(harness, service, active); + + yield* service.handleHardLimit(fiveHourAt(active, farReset)); + yield* service.handleHardLimit(fiveHourAt(idle, farReset)); + assert.includeMembers(harness.stoppedThreads, [active, idle]); + assert.equal(harness.dispatched.length, 0); + + const activeView = yield* service.getThread(active); + assert.deepEqual(activeView.wait && { ...activeView.wait, queuedAt: "" }, { + position: 1, + queuedAt: "", + expectedAt: "1970-01-01T00:30:00.000Z", + expectedProfile: "agent", + wasActive: true, + }); + assert.equal( + activeView.notice, + "All Claude accounts are at their limit. This thread is number 1 in the queue and continues on agent at about 00:30 UTC.", + ); + const idleView = yield* service.getThread(idle); + assert.equal(idleView.wait?.position, 2); + assert.equal( + idleView.notice, + "All Claude accounts are at their limit. This thread is number 2 in the queue and can continue on agent at about 00:30 UTC when you send a message.", + ); + + // Still no room: both keep waiting. + yield* service.refreshStatus(); + assert.equal(harness.dispatched.length, 0); + assert.equal((yield* service.getThread(active)).wait?.position, 1); + + gate.state.open = ["agent"]; + yield* service.refreshStatus(); + const revivedActive = yield* service.getThread(active); + const revivedIdle = yield* service.getThread(idle); + assert.equal(revivedActive.wait, undefined); + assert.equal(revivedActive.notice, undefined); + assert.equal(revivedActive.resolvedProfile, "agent"); + assert.equal(revivedIdle.wait, undefined); + assert.equal(revivedIdle.resolvedProfile, "agent"); + // Only the thread that was mid-turn is continued. + assert.equal(harness.dispatched.length, 1); + const { command, actorUserId } = harness.dispatched[0]!; + assert.equal(actorUserId, owner); + assert.equal(command.type, "thread.turn.start"); + if (command.type === "thread.turn.start") { + assert.equal(command.threadId, active); + assert.equal(command.message.text, "Continue where you left off."); + } + }), + ); + } + + { + const gate = gatedPlace(); + scenario( + "revives without a continue when the shell already shows the next turn", + { place: gate.place }, + (harness, service) => + Effect.gen(function* () { + yield* service.refreshStatus(); + const threadId = ThreadId.make("thread-queue-raced"); + gate.state.open = ["tushar"]; + yield* service.resolveForSession(threadId); + gate.state.open = []; + yield* markTurnRunning(harness, service, threadId); + yield* service.handleHardLimit(fiveHourAt(threadId, farReset)); + assert.equal((yield* service.getThread(threadId)).wait?.wasActive, true); + + // The person's message was accepted before the revive pass ran. + yield* Ref.set(harness.shellSession, orchestrationSession(threadId, "running", "turn-2")); + gate.state.open = ["agent"]; + yield* service.refreshStatus(); + const view = yield* service.getThread(threadId); + assert.equal(view.wait, undefined); + assert.equal(view.resolvedProfile, "agent"); + assert.equal(harness.dispatched.length, 0); + }), + ); + } + + { + const gate = gatedPlace(); + scenario( + "spreads a wave of waiting threads over the accounts that free up", + { + place: gate.place, + status: (dir) => + switcherStatus([ + switcherProfile(dir, "tushar"), + switcherProfile(dir, "agent", { rank: 2, elected: false }), + switcherProfile(dir, "barsha", { rank: 3, elected: false }), + switcherProfile(dir, "carol", { rank: 4, elected: false }), + ]), + }, + (harness, service) => + Effect.gen(function* () { + yield* service.refreshStatus(); + const threads = [1, 2, 3].map((index) => ThreadId.make(`thread-wave-${index}`)); + gate.state.open = ["tushar"]; + for (const threadId of threads) yield* service.resolveForSession(threadId); + gate.state.open = []; + for (const threadId of threads) { + yield* markTurnRunning(harness, service, threadId); + yield* service.handleHardLimit(fiveHourAt(threadId, farReset)); + } + for (const [index, threadId] of threads.entries()) { + assert.equal((yield* service.getThread(threadId)).wait?.position, index + 1); + } + + gate.state.open = ["agent", "barsha", "carol"]; + yield* service.refreshStatus(); + const placed = new Set(); + for (const threadId of threads) { + const view = yield* service.getThread(threadId); + assert.equal(view.wait, undefined); + placed.add(view.resolvedProfile ?? ""); + } + assert.deepEqual([...placed].toSorted(), ["agent", "barsha", "carol"]); + assert.equal(harness.dispatched.length, 3); + }), + ); + } + + { + const gate = gatedPlace(); + scenario( + "takes a thread out of the queue without moving it", + { place: gate.place }, + (harness, service) => + Effect.gen(function* () { + yield* service.refreshStatus(); + const threadId = ThreadId.make("thread-queue-leave"); + gate.state.open = ["tushar"]; + yield* service.resolveForSession(threadId); + gate.state.open = []; + yield* markTurnRunning(harness, service, threadId); + yield* service.handleHardLimit(fiveHourAt(threadId, farReset)); + assert.isDefined((yield* service.getThread(threadId)).wait); + + const left = yield* service.leaveQueue(threadId); + assert.equal(left.wait, undefined); + assert.equal(left.notice, undefined); + assert.equal(left.resolvedProfile, "tushar"); + // Leaving twice is harmless. + yield* service.leaveQueue(threadId); + + gate.state.open = ["agent"]; + yield* service.refreshStatus(); + assert.equal(harness.dispatched.length, 0); + assert.equal((yield* service.getThread(threadId)).resolvedProfile, "tushar"); + }), + ); + } + + { + const gate = gatedPlace(); + scenario( + "fails a start with the account expected to have room first, and does not queue it", + { place: gate.place }, + (_harness, service) => + Effect.gen(function* () { + yield* service.refreshStatus(); + const threadId = ThreadId.make("thread-queue-spawn"); + const result = yield* service.resolveForSession(threadId).pipe(Effect.result); + assert.equal(result._tag, "Failure"); + if (result._tag === "Failure") { + assert.equal( + result.failure.detail, + "All Claude accounts are at their limit — agent is expected to have room at 00:30 UTC.", + ); + } + assert.equal((yield* service.getThread(threadId)).wait, undefined); + }), + ); + } + + { + const gate = gatedPlace(); + scenario( + "does not revive a thread that left the queue while its placement was in flight", + { place: gate.place, beforePlace: gate.beforePlace }, + (harness, service) => + Effect.gen(function* () { + yield* service.refreshStatus(); + const threadId = ThreadId.make("thread-queue-race"); + gate.state.open = ["tushar"]; + yield* service.resolveForSession(threadId); + gate.state.open = []; + yield* markTurnRunning(harness, service, threadId); + yield* service.handleHardLimit(fiveHourAt(threadId, farReset)); + assert.isDefined((yield* service.getThread(threadId)).wait); + + const entered = yield* Deferred.make(); + const release = yield* Deferred.make(); + gate.state.open = ["agent"]; + gate.state.hold = { entered, release }; + const pass = yield* service.refreshStatus().pipe(Effect.forkScoped); + // The switcher is answering for the thread; the person leaves the queue now. + yield* Deferred.await(entered); + const left = yield* service.leaveQueue(threadId); + assert.equal(left.wait, undefined); + yield* Deferred.succeed(release, undefined); + yield* Fiber.join(pass); + + assert.equal(harness.dispatched.length, 0); + const view = yield* service.getThread(threadId); + assert.equal(view.wait, undefined); + assert.equal(view.resolvedProfile, "tushar"); + }), + ); + } + + { + const gate = gatedPlace(); + scenario( + "takes an archived thread out of the queue so it is not continued", + { place: gate.place }, + (harness, service) => + Effect.gen(function* () { + yield* service.refreshStatus(); + const threadId = ThreadId.make("thread-queue-archived"); + gate.state.open = ["tushar"]; + yield* service.resolveForSession(threadId); + gate.state.open = []; + yield* markTurnRunning(harness, service, threadId); + yield* service.handleHardLimit(fiveHourAt(threadId, farReset)); + const { seen } = yield* watchThreadInto(service, threadId); + + yield* PubSub.publish(harness.domainEvents, { + type: "thread.archived", + payload: { + threadId, + archivedAt: "1970-01-01T00:00:00.000Z", + updatedAt: "1970-01-01T00:00:00.000Z", + }, + } as unknown as OrchestrationEvent); + yield* takeUntil(seen, (view) => view.wait === undefined); + gate.state.open = ["agent"]; + yield* service.refreshStatus(); + assert.equal(harness.dispatched.length, 0); + // The account row stays for an unarchive. + assert.equal((yield* service.getThread(threadId)).resolvedProfile, "tushar"); + }), + ); + } + + scenario( + "handles a model-scoped rejection again once the thread is back on that account", + {}, + (harness, service) => + Effect.gen(function* () { + yield* service.refreshStatus(); + const threadId = ThreadId.make("thread-scoped-return"); + assert.equal((yield* service.resolveForSession(threadId))?.profile, "tushar"); + const opus = { threadId, rateLimitType: "seven_day_opus", resetAt: farReset }; + + yield* service.handleHardLimit(opus); + assert.equal((yield* service.getThread(threadId)).resolvedProfile, "agent"); + // A burst of copies of that one rejection moves the thread once. + yield* service.handleHardLimit(opus); + yield* service.handleHardLimit(opus); + assert.equal(harness.hardLimitCalls.length, 1); + assert.equal(harness.dispatched.length, 1); + + // The first turn on agent; later the thread is placed back on tushar, + // which a model-scoped limit left in rotation. + const { seen } = yield* watchThreadInto(service, threadId); + yield* PubSub.publish(harness.providerEvents, providerEvent("turn.started", threadId)); + yield* Queue.take(seen); + assert.equal( + (yield* service.reassign(threadId, { avoid: ["agent", "barsha"] }))?.profile, + "tushar", + ); + + // Claude refuses Opus on tushar again, same reset: a real rejection, handled. + assert.isTrue(yield* service.handleHardLimit(opus)); + assert.deepEqual(harness.hardLimitCalls, [ + { type: "seven_day_opus", profile: "tushar" }, + { type: "seven_day_opus", profile: "tushar" }, + ]); + assert.notEqual((yield* service.getThread(threadId)).resolvedProfile, "tushar"); + assert.equal(harness.dispatched.length, 2); + }), + ); + + it.effect("keeps the queue across a restart and continues the thread there", () => + Effect.gen(function* () { + const repository = yield* Layer.build( + Layer.mergeAll(ThreadClaudeAccountRepo.layer, MigrationsLive).pipe( + Layer.provide(SqlitePersistenceMemory), + ), + ); + const gate = gatedPlace(); + const harness = yield* makeHarness({ + place: gate.place, + threadRepository: Layer.succeedContext(repository), + }); + const threadId = ThreadId.make("thread-queue-restart"); + + yield* Effect.gen(function* () { + const service = yield* ClaudeAccountsService; + yield* service.refreshStatus(); + gate.state.open = ["tushar"]; + yield* service.resolveForSession(threadId); + gate.state.open = []; + yield* markTurnRunning(harness, service, threadId); + yield* service.handleHardLimit(fiveHourAt(threadId, farReset)); + assert.equal((yield* service.getThread(threadId)).wait?.position, 1); + }).pipe(Effect.provide(harness.layer)); + + // A new server over the same table. + yield* Effect.gen(function* () { + const service = yield* ClaudeAccountsService; + const view = yield* service.getThread(threadId); + assert.equal(view.wait?.position, 1); + assert.equal(view.wait?.wasActive, true); + assert.equal(harness.dispatched.length, 0); + + gate.state.open = ["agent"]; + yield* service.refreshStatus(); + const revived = yield* service.getThread(threadId); + assert.equal(revived.wait, undefined); + assert.equal(revived.resolvedProfile, "agent"); + assert.equal(harness.dispatched.length, 1); + const { command } = harness.dispatched[0]!; + assert.equal(command.type === "thread.turn.start" ? command.threadId : null, threadId); + }).pipe(Effect.provide(harness.layer)); + }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)), + ); +}); + +describe("ClaudeAccountsService notices", () => { + scenario("drops a notice tied to a reset once that time passes", {}, (harness, service) => + Effect.gen(function* () { + yield* service.refreshStatus(); + const threadId = ThreadId.make("thread-notice-expiry"); + yield* service.setThreadMode({ threadId, mode: { kind: "profile", profile: "agent" } }); + yield* service.resolveForSession(threadId); + // Ten minutes after the test clock's epoch start. + yield* service.handleHardLimit(fiveHourAt(threadId, 600)); + assert.equal( + (yield* service.getThread(threadId)).notice, + "agent is at its usage limit until 00:10 UTC — switch to Auto or another account.", + ); + const { seen } = yield* watchThreadInto(service, threadId); + + yield* TestClock.adjust("11 minutes"); + yield* service.refreshStatus(); + yield* takeUntil(seen, (view) => view.notice === undefined); + assert.equal((yield* service.getThread(threadId)).notice, undefined); + assert.deepEqual(harness.stoppedThreads, []); + }), + ); +}); + +describe("ClaudeAccountsService rebalancing at turn boundaries", () => { + const sharedStatus = (dir: DirFor) => ({ + ...switcherStatus([ + switcherProfile(dir, "tushar", { sessions: 5, fair_share: 2, load: 5 }), + switcherProfile(dir, "agent", { rank: 2, elected: false, fair_share: 2, load: 0 }), + switcherProfile(dir, "barsha", { rank: 3, elected: false, fair_share: 2, load: 0 }), + ]), + place_rule: "space-share-v2", + }); + + { + const gate = gatedPlace(); + scenario( + "moves an idle Auto thread off an over-loaded account once, not again within the hour", + { place: gate.place, status: sharedStatus }, + (harness, service) => + Effect.gen(function* () { + yield* service.refreshStatus(); + const threadId = ThreadId.make("thread-rebalance"); + gate.state.open = ["tushar"]; + yield* service.resolveForSession(threadId); + gate.state.open = ["agent"]; + yield* Ref.set(harness.sessions, [liveSession(threadId, "ready")]); + const { seen } = yield* watchThreadInto(service, threadId); + + yield* endTurn(harness, threadId); + yield* takeUntil(seen, (view) => view.resolvedProfile === "agent"); + yield* drainDomainEvents(harness, service); + assert.deepEqual(harness.stoppedThreads, [threadId]); + assert.include(harness.placeCalls.at(-1)!.avoid, "tushar"); + // Nothing asked the thread to continue: its next message spawns on agent. + assert.equal(harness.dispatched.length, 0); + assert.equal((yield* service.resolveForSession(threadId))?.reason, "sticky"); + + // agent is over its share now too, but the thread moved less than an hour ago. + yield* Ref.update(harness.status, (current) => ({ + ...current, + profiles: current.profiles.map((profile) => + profile.name === "agent" ? { ...profile, sessions: 6 } : profile, + ), + })); + yield* service.refreshStatus(); + gate.state.open = ["barsha"]; + const placements = harness.placeCalls.length; + yield* endTurn(harness, threadId); + yield* drainDomainEvents(harness, service); + assert.equal(harness.placeCalls.length, placements); + assert.equal((yield* service.getThread(threadId)).resolvedProfile, "agent"); + + yield* TestClock.adjust("61 minutes"); + yield* endTurn(harness, threadId); + yield* takeUntil(seen, (view) => view.resolvedProfile === "barsha"); + yield* drainDomainEvents(harness, service); + assert.deepEqual(harness.stoppedThreads, [threadId, threadId]); + }), + ); + } + + { + const gate = gatedPlace(); + scenario( + "never moves a pinned thread, and leaves a thread with a turn running alone", + { place: gate.place, status: sharedStatus }, + (harness, service) => + Effect.gen(function* () { + yield* service.refreshStatus(); + const pinned = ThreadId.make("thread-rebalance-pinned"); + const busy = ThreadId.make("thread-rebalance-busy"); + yield* service.setThreadMode({ + threadId: pinned, + mode: { kind: "profile", profile: "tushar" }, + }); + yield* service.resolveForSession(pinned); + gate.state.open = ["tushar"]; + yield* service.resolveForSession(busy); + gate.state.open = ["agent"]; + yield* Ref.set(harness.sessions, [ + liveSession(pinned, "ready"), + liveSession(busy, "running"), + ]); + const placements = harness.placeCalls.length; + + yield* endTurn(harness, pinned); + yield* endTurn(harness, busy); + yield* drainDomainEvents(harness, service); + assert.equal(harness.placeCalls.length, placements); + assert.deepEqual(harness.stoppedThreads, []); + assert.equal((yield* service.getThread(pinned)).resolvedProfile, "tushar"); + assert.equal((yield* service.getThread(busy)).resolvedProfile, "tushar"); + }), + ); + } + + { + const gate = gatedPlace(); + scenario( + "does not ask the switcher again within five minutes after a probe found nowhere better", + { place: gate.place, status: sharedStatus }, + (harness, service) => + Effect.gen(function* () { + yield* service.refreshStatus(); + const threadId = ThreadId.make("thread-rebalance-backoff"); + gate.state.open = ["tushar"]; + yield* service.resolveForSession(threadId); + gate.state.open = []; + yield* Ref.set(harness.sessions, [liveSession(threadId, "ready")]); + const before = harness.placeCalls.length; + + yield* endTurn(harness, threadId); + yield* drainDomainEvents(harness, service); + assert.equal(harness.placeCalls.length, before + 1); + assert.deepEqual(harness.stoppedThreads, []); + + yield* endTurn(harness, threadId); + yield* drainDomainEvents(harness, service); + assert.equal(harness.placeCalls.length, before + 1); + + yield* TestClock.adjust("6 minutes"); + yield* endTurn(harness, threadId); + yield* drainDomainEvents(harness, service); + assert.equal(harness.placeCalls.length, before + 2); + assert.equal((yield* service.getThread(threadId)).resolvedProfile, "tushar"); + }), + ); + } + + { + const gate = gatedPlace(); + scenario( + "moves one thread per status read off an over-loaded account, whose count is stale until then", + { place: gate.place, status: sharedStatus }, + (harness, service) => + Effect.gen(function* () { + yield* service.refreshStatus(); + const first = ThreadId.make("thread-rebalance-first"); + const second = ThreadId.make("thread-rebalance-second"); + gate.state.open = ["tushar"]; + yield* service.resolveForSession(first); + yield* service.resolveForSession(second); + gate.state.open = ["agent", "barsha"]; + yield* Ref.set(harness.sessions, [ + liveSession(first, "ready"), + liveSession(second, "ready"), + ]); + + yield* endTurn(harness, first); + yield* endTurn(harness, second); + yield* drainDomainEvents(harness, service); + assert.deepEqual(harness.stoppedThreads, [first]); + assert.notEqual((yield* service.getThread(first)).resolvedProfile, "tushar"); + assert.equal((yield* service.getThread(second)).resolvedProfile, "tushar"); + + // The next status read counts the move; tushar is still over its share. + yield* service.refreshStatus(); + yield* endTurn(harness, second); + yield* drainDomainEvents(harness, service); + assert.deepEqual(harness.stoppedThreads, [first, second]); + }), + ); + } +}); diff --git a/apps/server/src/claudeAccounts/ClaudeAccountsService.ts b/apps/server/src/claudeAccounts/ClaudeAccountsService.ts index 84be8780fdebc..78c11b3fad3aa 100644 --- a/apps/server/src/claudeAccounts/ClaudeAccountsService.ts +++ b/apps/server/src/claudeAccounts/ClaudeAccountsService.ts @@ -19,6 +19,14 @@ * affected Auto thread. Pinned threads get a notice instead. An Auto * thread whose window resets within five minutes keeps its account: the * session stops and the thread continues there once the window resets; + * - the **revive queue**: an Auto thread that no account can take stops and + * waits in `thread_claude_account` (so a restart keeps it). After every + * status poll the waiting threads are placed again in queue order, and one + * that was mid-turn gets the continue turn; + * - **rebalancing**: when an idle Auto thread's turn ends on an account that + * carries two or more sessions above its fair share (or can no longer take + * new ones), it moves to where the switcher would place it now, at most once + * an hour per thread; * - **access per user**: an account with no users assigned is open to * everyone; once an admin assigns users, only they may use it. Placement * avoids the accounts the thread's owner may not use, a pin to one fails, @@ -51,7 +59,6 @@ import * as Crypto from "effect/Crypto"; import * as DateTime from "effect/DateTime"; import * as Duration from "effect/Duration"; import * as Effect from "effect/Effect"; -import * as FiberMap from "effect/FiberMap"; import * as FileSystem from "effect/FileSystem"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; @@ -97,6 +104,12 @@ const PENDING_PLACEMENT_WINDOW_MS = 90_000; /** How long an exhausted mark lasts when Claude did not say when the window resets. */ const DEFAULT_EXHAUSTED_MS = 60 * 60 * 1000; const MAX_HANDLED_CONDITIONS = 256; +/** A thread is moved off an over-loaded account at most this often. */ +export const REBALANCE_COOLDOWN_MS = 60 * 60 * 1000; +/** Sessions above the account's fair share before a finished turn moves its thread. */ +const REBALANCE_EXCESS = 2; +/** After a rebalance probe finds nowhere better, the thread is not probed again for this long. */ +export const REBALANCE_PROBE_BACKOFF_MS = 5 * 60 * 1000; /** A limit that resets within this long is waited out on the same account instead of moving. */ export const RESET_WAIT_THRESHOLD_MS = 5 * 60 * 1000; /** Margin after the reset so the status poll sees the account free before the thread continues. */ @@ -121,7 +134,11 @@ const sameThreadAccount = (left: ThreadClaudeAccount, right: ThreadClaudeAccount left.resolvedProfile === right.resolvedProfile && left.resolvedAt === right.resolvedAt && left.notice === right.notice && - left.pendingRestart === right.pendingRestart; + left.pendingRestart === right.pendingRestart && + left.wait?.position === right.wait?.position && + left.wait?.expectedAt === right.wait?.expectedAt && + left.wait?.expectedProfile === right.wait?.expectedProfile && + left.wait?.wasActive === right.wait?.wasActive; /** Published after each status read: watched threads re-read their observed account. */ const LIVE_SESSIONS_CHANGE = "\u0000live-sessions"; @@ -177,6 +194,10 @@ export interface ClaudeAccountsServiceShape { /** Re-reads `--status --json` now; the poller calls this on its own schedule. */ readonly refreshStatus: () => Effect.Effect; readonly markExhausted: (profile: string, resetsAt: string | undefined) => Effect.Effect; + /** Takes the thread out of the revive queue without moving it. */ + readonly leaveQueue: ( + threadId: ThreadId, + ) => Effect.Effect; /** Places the thread again, avoiding the given accounts, and persists the result. */ readonly reassign: ( threadId: ThreadId, @@ -202,6 +223,27 @@ interface ExhaustedMark { readonly resetsAt: string | undefined; } +/** A composer notice; one tied to a time disappears at that time. */ +interface Notice { + readonly text: string; + readonly untilMs?: number; +} + +/** A thread in the revive queue: the stored row plus what only memory knows. */ +interface WaitEntry { + readonly queuedAt: string; + readonly wasActive: boolean; + readonly preferProfile: string | null; + readonly notBefore: string | null; + /** The account a model-scoped limit refused; the revive avoids it. Memory only. */ + readonly avoid?: string; + /** Whether the owner may use only some accounts, for the notice's wording. */ + readonly restricted: boolean; + /** When the first account the owner may use is expected to have room. */ + readonly expectedAt?: string; + readonly expectedProfile?: string; +} + const nowIso = Effect.map(DateTime.now, DateTime.formatIso); function clampPercent(value: number): number { @@ -314,7 +356,7 @@ export const make = Effect.gen(function* () { const fiveHourHistory: FiveHourHistory = new Map(); // Per-thread state the row does not carry. const threadProfiles = new Map(); - const notices = new Map(); + const notices = new Map(); const pendingRestart = new Set(); /** Threads with a turn in flight, as seen on the provider event bus. */ const activeTurns = new Set(); @@ -326,12 +368,23 @@ export const make = Effect.gen(function* () { const movedAwaitingTurn = new Set(); const handledConditions = new Set(); /** - * Auto threads waiting out a window that resets soon, each with the timer - * that continues it. A thread leaves the set just before its continue turn - * is sent, so that turn's own start does not cancel anything. + * The revive queue, mirrored from `thread_claude_account` and read back at + * start. A thread leaves it just before its continue turn is sent, so that + * turn's own start does not cancel anything. */ - const waitingForReset = new Set(); - const resetWaits = yield* FiberMap.make(); + const waits = new Map(); + /** When each thread was last moved by a rebalance. */ + const lastRebalancedAt = new Map(); + /** When a rebalance probe last found nowhere better for each thread. */ + const lastFailedProbeAt = new Map(); + /** + * Accounts a rebalance moved a thread off since the last status read. Their + * session counts are stale until the next read, so each gives up at most + * one thread per read. + */ + const rebalancedFrom = new Set(); + /** Threads whose settled projection last showed a running turn. */ + const projectedRunning = new Set(); /** * Live Claude session id -> the account it runs on, from each account's * `sessions/.json` (the CLI writes one at start and never removes it). @@ -344,6 +397,22 @@ export const make = Effect.gen(function* () { const announce = (key: string) => PubSub.publish(changes, key).pipe(Effect.ignore); const announceSnapshot = announce(SNAPSHOT_CHANGE); const placementLock = yield* Semaphore.make(1); + /** One revive pass at a time; a pass started meanwhile runs after it. */ + const reviveLock = yield* Semaphore.make(1); + + const storedWaits = yield* repository + .listWaiting() + .pipe( + Effect.catchCause((cause) => + Effect.as( + Effect.logWarning("claude.account.queue-load-failed", { cause }), + [] as ReadonlyArray, + ), + ), + ); + for (const row of storedWaits) { + if (row.wait !== null) waits.set(row.threadId, { ...row.wait, restricted: false }); + } const readSettings: Effect.Effect = settingsService.getSettings.pipe( @@ -714,6 +783,11 @@ export const make = Effect.gen(function* () { lastStatus = status; lastStatusAtMs = yield* Clock.currentTimeMillis; unavailableReason = undefined; + // One rebalance per source account per status read. A moved session + // whose restart was deferred is still counted in this read, so a second + // thread can leave the account one read early; the slack of 2 over the + // fair share absorbs that. + rebalancedFrom.clear(); const generatedMs = typeof status.generated_ms === "number" ? status.generated_ms : lastStatusAtMs; for (const profile of status.profiles) { @@ -745,6 +819,17 @@ export const make = Effect.gen(function* () { ); }); + /** Drops notices whose time has passed and tells their threads. */ + const expireNotices = Effect.gen(function* () { + const now = yield* Clock.currentTimeMillis; + for (const [threadId, notice] of notices) { + if (notice.untilMs !== undefined && notice.untilMs <= now) { + notices.delete(threadId); + yield* announce(threadId); + } + } + }); + const refreshStatus: ClaudeAccountsServiceShape["refreshStatus"] = () => Effect.gen(function* () { const input = yield* rankInput; @@ -764,6 +849,9 @@ export const make = Effect.gen(function* () { liveSessionProfiles = yield* indexLiveSessions(outcome.value); yield* announce(LIVE_SESSIONS_CHANGE); } + yield* expireNotices; + // Accounts may have room now: continue the threads waiting for one. + yield* revivePass; }); // Poll only while the setting is on; a disabled feature costs nothing but @@ -795,8 +883,81 @@ export const make = Effect.gen(function* () { // ---------------------------------------------------------------- threads - const threadView = (row: ThreadClaudeAccountRow): ThreadClaudeAccount => { - const notice = notices.get(row.threadId); + const setNotice = (threadId: ThreadId, text: string, until?: string) => { + const untilMs = until === undefined ? Number.NaN : Date.parse(until); + notices.set(threadId, Number.isNaN(untilMs) ? { text } : { text, untilMs }); + }; + + const isDue = (entry: WaitEntry, nowMs: number) => + entry.notBefore !== null && Date.parse(entry.notBefore) <= nowMs; + + /** The revive order: waits whose reset has passed, then interrupted turns, then the oldest. */ + const queueOrder = (nowMs: number): ReadonlyArray => + [...waits].toSorted(([, left], [, right]) => { + const byDue = Number(isDue(right, nowMs)) - Number(isDue(left, nowMs)); + if (byDue !== 0) return byDue; + const byActive = Number(right.wasActive) - Number(left.wasActive); + if (byActive !== 0) return byActive; + return left.queuedAt.localeCompare(right.queuedAt); + }); + + /** The queue state a waiting thread shows, with the notice derived from it. */ + const waitView = (threadId: ThreadId, entry: WaitEntry, nowMs: number) => { + const position = queueOrder(nowMs).findIndex(([id]) => id === threadId) + 1; + const notBeforeMs = entry.notBefore === null ? Number.NaN : Date.parse(entry.notBefore); + // A short reset wait still ahead: the thread continues on its own account. + if (entry.preferProfile !== null && entry.notBefore !== null && notBeforeMs > nowMs) { + return { + value: { + position, + queuedAt: entry.queuedAt, + expectedAt: entry.notBefore, + expectedProfile: entry.preferProfile, + wasActive: entry.wasActive, + }, + notice: `${entry.preferProfile} resets at ${formatResetClock( + isoFromUnixSeconds((notBeforeMs - RESET_WAIT_BUFFER_MS) / 1000), + )} — this thread continues on it at ${formatResetClock(entry.notBefore)}.`, + }; + } + // A reset wait whose time has passed says nothing about when room comes: + // only the last placement's answer does. + const expectedAt = entry.expectedAt; + const expectedProfile = entry.expectedProfile; + const limitLine = entry.restricted + ? "All Claude accounts you can use are at their limit." + : "All Claude accounts are at their limit."; + const target = expectedProfile === undefined ? "" : ` on ${expectedProfile}`; + const when = + expectedAt === undefined + ? " once an account has room" + : ` at about ${formatResetClock(expectedAt)}`; + const next = entry.wasActive + ? `continues${target}${when}` + : `can continue${target}${when} when you send a message`; + return { + value: { + position, + queuedAt: entry.queuedAt, + ...(expectedAt !== undefined ? { expectedAt } : {}), + ...(expectedProfile !== undefined ? { expectedProfile } : {}), + wasActive: entry.wasActive, + }, + notice: `${limitLine} This thread is number ${position} in the queue and ${next}.`, + }; + }; + + const threadView = (row: ThreadClaudeAccountRow, nowMs: number): ThreadClaudeAccount => { + const entry = waits.get(row.threadId); + const wait = entry === undefined ? undefined : waitView(row.threadId, entry, nowMs); + const stored = notices.get(row.threadId); + // A notice tied to a time never outlives it. + const live = + stored !== undefined && (stored.untilMs === undefined || stored.untilMs > nowMs) + ? stored.text + : undefined; + if (stored !== undefined && live === undefined) notices.delete(row.threadId); + const notice = wait?.notice ?? live; return { threadId: row.threadId, mode: row.mode, @@ -804,6 +965,7 @@ export const make = Effect.gen(function* () { ...(row.resolvedAt ? { resolvedAt: row.resolvedAt } : {}), ...(notice ? { notice } : {}), ...(pendingRestart.has(row.threadId) ? { pendingRestart: true } : {}), + ...(wait ? { wait: wait.value } : {}), }; }; @@ -821,13 +983,14 @@ export const make = Effect.gen(function* () { const getThread: ClaudeAccountsServiceShape["getThread"] = (threadId) => Effect.gen(function* () { const row = yield* repository.get(threadId).pipe(Effect.mapError(internal("getThread"))); - if (row.resolvedProfile || row.mode.kind !== "auto") return threadView(row); + const nowMs = yield* Clock.currentTimeMillis; + if (row.resolvedProfile || row.mode.kind !== "auto") return threadView(row, nowMs); // A session that started before placement (on the shared link) has no // resolved account yet: show the one its live process really uses. const observed = yield* observedProfile(threadId); - if (observed === undefined) return threadView(row); + if (observed === undefined) return threadView(row, nowMs); threadProfiles.set(threadId, observed); - return threadView({ ...row, resolvedProfile: observed }); + return threadView({ ...row, resolvedProfile: observed }, nowMs); }); const persistResolved = (threadId: ThreadId, profile: string) => @@ -901,10 +1064,24 @@ export const make = Effect.gen(function* () { }); }); + interface PlaceOptions { + /** Why the placement runs, for the log. */ + readonly purpose: "spawn" | "reassign" | "revive" | "rebalance"; + /** + * Accept only a real `placed` answer: no `fallback` account above its trip + * line and no degraded active link. A revive or a rebalance onto either + * would only fail again, so it waits for a real answer instead. + */ + readonly requirePlaced?: boolean; + /** Counts the answer as a pending placement (the default); a probe passes false. */ + readonly commit?: boolean; + } + /** Asks the switcher for an account. Must run under `placementLock`. */ const placeWith = ( threadId: ThreadId, avoid: ReadonlyArray, + options: PlaceOptions, ): Effect.Effect => Effect.gen(function* () { const pending = yield* pendingCounts; @@ -931,7 +1108,14 @@ export const make = Effect.gen(function* () { threadId, kind: outcome.kind, detail: outcome.detail, + purpose: options.purpose, }); + if (options.requirePlaced) { + return yield* new ClaudeAccountPolicyError({ + threadId, + detail: "Claude account placement is unavailable right now — try again shortly.", + }); + } if (outcome.kind === "unavailable") { const fallback = yield* currentProfileFromActiveLink; if (fallback !== undefined && !avoidSet.has(fallback.profile)) return fallback; @@ -950,9 +1134,35 @@ export const make = Effect.gen(function* () { chosen: result.chosen, }); } - if (result.status === "no_eligible_profile" || chosen === null) { + if ( + result.status === "no_eligible_profile" || + chosen === null || + (options.requirePlaced === true && result.status !== "placed") + ) { + yield* Effect.logInfo("claude.account.place-none", { + threadId, + status: result.status, + recoversAt: result.recovers_at ?? null, + recoveries: result.recoveries ?? [], + reason: result.reason ?? null, + purpose: options.purpose, + }); if (mayUseNone(restriction)) return yield* noAccountPolicyError(threadId); - const recoversAt = isoOrUndefined(result.recovers_at) ?? firstRecovery(disallowed); + // The first account this owner may use that is expected to have room. + // A switcher that lists recoveries speaks for every account; an older + // one only gives the first reset, which may be an account off limits. + let first: { readonly profile: string; readonly at: string } | undefined; + for (const recovery of result.recoveries ?? []) { + const at = isoOrUndefined(recovery.recovers_at); + if (at !== undefined && !disallowed.has(recovery.profile)) { + first = { profile: recovery.profile, at }; + break; + } + } + const recoversAt = + first?.at ?? + (result.recoveries ? undefined : isoOrUndefined(result.recovers_at)) ?? + firstRecovery(disallowed); // Nothing the owner may use is waiting on a reset, and whatever else // is left is off limits: that is "no access", not "at its limit". The // access list decides this; the host list, when known, only narrows it. @@ -971,13 +1181,19 @@ export const make = Effect.gen(function* () { return yield* new ClaudeAccountPolicyError({ threadId, detail: - recoversAt === undefined - ? `${limitLine}.` - : `${limitLine} — first resets at ${formatResetClock(recoversAt)}.`, + first !== undefined + ? `${limitLine} — ${first.profile} is expected to have room at ${formatResetClock(first.at)}.` + : recoversAt === undefined + ? `${limitLine}.` + : `${limitLine} — first resets at ${formatResetClock(recoversAt)}.`, + ...(recoversAt !== undefined ? { expectedAt: recoversAt } : {}), + ...(first !== undefined ? { expectedProfile: first.profile } : {}), }); } - const atMs = yield* Clock.currentTimeMillis; - pendingPlacements.push({ profile: chosen, atMs }); + if (options.commit !== false) { + const atMs = yield* Clock.currentTimeMillis; + pendingPlacements.push({ profile: chosen, atMs }); + } return { profile: chosen, dir: result.dir ? result.dir : profileDir(chosen), @@ -986,10 +1202,14 @@ export const make = Effect.gen(function* () { } as const; }); - const placeAndPersist = (threadId: ThreadId, avoid: ReadonlyArray) => + const placeAndPersist = ( + threadId: ThreadId, + avoid: ReadonlyArray, + purpose: PlaceOptions["purpose"], + ) => Semaphore.withPermit(placementLock)( Effect.gen(function* () { - const resolution = yield* placeWith(threadId, avoid); + const resolution = yield* placeWith(threadId, avoid, { purpose }); if (resolution !== undefined) { yield* persistResolved(threadId, resolution.profile).pipe( Effect.catchCause((cause) => @@ -1059,8 +1279,10 @@ export const make = Effect.gen(function* () { // Sticky: an account the thread already ran on is kept until a hard // limit, a sign-out, an exhausted mark or its removal takes it out. - // Nearing a limit never moves an existing thread; only new placements - // avoid it. Losing access to the account re-places it too. + // Nearing a limit never moves a running thread here; only new + // placements avoid it, and the turn-boundary rebalance may move an idle + // thread off an over-loaded account (it writes the row this reads). + // Losing access to the account re-places it too. if ( row.resolvedProfile !== null && !disallowed.has(row.resolvedProfile) && @@ -1074,7 +1296,11 @@ export const make = Effect.gen(function* () { reason: "sticky", } as const; } - return yield* placeAndPersist(threadId, row.resolvedProfile ? [row.resolvedProfile] : []); + return yield* placeAndPersist( + threadId, + row.resolvedProfile ? [row.resolvedProfile] : [], + "spawn", + ); }); /** @@ -1097,7 +1323,7 @@ export const make = Effect.gen(function* () { ); const reassign: ClaudeAccountsServiceShape["reassign"] = (threadId, input) => - placeAndPersist(threadId, input.avoid); + placeAndPersist(threadId, input.avoid, "reassign"); const markExhausted: ClaudeAccountsServiceShape["markExhausted"] = (profile, resetsAt) => Effect.gen(function* () { @@ -1156,7 +1382,7 @@ export const make = Effect.gen(function* () { }); /** Restarts the thread's provider session so its next turn spawns on the new account. */ - const restartForNewAccount = (threadId: ThreadId) => + const restartForNewAccount = (threadId: ThreadId, why = "mode-changed") => Effect.gen(function* () { const sessions = yield* providers.listSessions(); const live = sessions.find( @@ -1168,7 +1394,7 @@ export const make = Effect.gen(function* () { yield* Effect.logInfo("claude.account.restart-deferred", { threadId }); return; } - yield* stopSessionQuietly(threadId, "mode-changed"); + yield* stopSessionQuietly(threadId, why); }); /** @@ -1188,15 +1414,100 @@ export const make = Effect.gen(function* () { yield* announce(threadId); }); - /** Drops a pending wait-for-reset; `true` when there was one. */ - const cancelResetWait = (threadId: ThreadId) => + // ------------------------------------------------------------ revive queue + + /** Forgets the rejections already handled for a thread, so the next one is judged afresh. */ + const forgetConditions = (threadId: ThreadId) => { + const prefix = `${threadId}\u0000`; + for (const key of handledConditions) { + if (key.startsWith(prefix)) handledConditions.delete(key); + } + }; + + /** Tells every waiting thread to re-read its place in the queue. */ + const announceQueue = Effect.suspend(() => + Effect.forEach([...waits.keys()], announce, { discard: true }), + ); + + /** + * Puts the thread in the revive queue (keeping its place when it is already + * there). `false` when the row could not be written: the caller falls back. + */ + const park = ( + threadId: ThreadId, + input: Omit & { + readonly preferProfile?: string; + readonly notBefore?: string; + }, + ) => + Effect.gen(function* () { + const queuedAt = waits.get(threadId)?.queuedAt ?? (yield* nowIso); + const entry: WaitEntry = { + ...input, + queuedAt, + preferProfile: input.preferProfile ?? null, + notBefore: input.notBefore ?? null, + }; + yield* repository.setWait({ + threadId, + wait: { + queuedAt, + wasActive: entry.wasActive, + preferProfile: entry.preferProfile, + notBefore: entry.notBefore, + }, + }); + waits.set(threadId, entry); + notices.delete(threadId); + yield* Effect.logInfo("claude.account.queued", { + threadId, + wasActive: entry.wasActive, + preferProfile: entry.preferProfile, + notBefore: entry.notBefore, + expectedAt: entry.expectedAt, + expectedProfile: entry.expectedProfile, + }); + yield* announceQueue; + return true; + }).pipe( + Effect.catchCause((cause) => + Effect.as( + Effect.logWarning("claude.account.queue-write-failed", { threadId, cause }), + false, + ), + ), + ); + + /** Takes the thread out of the revive queue; `true` when it was waiting. */ + const dequeue = (threadId: ThreadId, why: string) => Effect.gen(function* () { - if (!waitingForReset.delete(threadId)) return false; - yield* FiberMap.remove(resetWaits, threadId); - yield* Effect.logInfo("claude.account.reset-wait-cancelled", { threadId }); + if (!waits.delete(threadId)) return false; + forgetConditions(threadId); + yield* Effect.logInfo("claude.account.dequeued", { threadId, why }); + yield* announceQueue; + yield* repository.clearWait(threadId); return true; }); + /** `dequeue` for event handlers: a failed write is logged, never raised. */ + const dequeueQuietly = (threadId: ThreadId, why: string) => + dequeue(threadId, why).pipe( + Effect.catchCause((cause) => + Effect.as( + Effect.logWarning("claude.account.queue-clear-failed", { threadId, cause }), + true, + ), + ), + ); + + const leaveQueue: ClaudeAccountsServiceShape["leaveQueue"] = (threadId) => + Effect.gen(function* () { + if (yield* dequeue(threadId, "left-queue").pipe(Effect.mapError(internal("leaveQueue")))) { + yield* announce(threadId); + } + return yield* getThread(threadId); + }); + const setThreadMode: ClaudeAccountsServiceShape["setThreadMode"] = ({ threadId, mode, @@ -1238,7 +1549,9 @@ export const make = Effect.gen(function* () { yield* repository .setMode({ threadId, mode, updatedAt }) .pipe(Effect.mapError(internal("setThreadMode"))); - yield* cancelResetWait(threadId); + const waiting = waits.get(threadId); + yield* dequeue(threadId, "mode-changed").pipe(Effect.mapError(internal("setThreadMode"))); + forgetConditions(threadId); notices.delete(threadId); movedAwaitingTurn.delete(threadId); yield* pruneExhausted; @@ -1250,7 +1563,7 @@ export const make = Effect.gen(function* () { (!mayUse(access, owner, row.resolvedProfile) || !(yield* usableAndPresent(row.resolvedProfile))); if (mode.kind === "profile" && knownProfile(mode.profile)?.auth?.state === "logged_out") { - notices.set( + setNotice( threadId, `Claude account ${mode.profile} is logged out — sign it in on the Claude Accounts page or switch this thread to Auto.`, ); @@ -1260,6 +1573,9 @@ export const make = Effect.gen(function* () { } else { pendingRestart.delete(threadId); } + // Choosing an account for a thread that stopped mid-turn to wait is + // "continue now on that account": its session already stopped. + if (waiting?.wasActive) yield* dispatchContinueTurn(threadId, CONTINUE_TEXT); yield* announce(threadId); return yield* getThread(threadId); }); @@ -1273,6 +1589,21 @@ export const make = Effect.gen(function* () { */ const CONTINUE_TEXT = "Continue where you left off."; + /** Whether the thread's shell shows a turn accepted or running. */ + const shellTurnInFlight = (threadId: ThreadId) => + query.getThreadShellById(threadId).pipe( + Effect.map((shell) => { + const session = Option.isSome(shell) ? shell.value.session : null; + return ( + session !== null && + (session.activeTurnId !== null || + session.status === "running" || + session.status === "starting") + ); + }), + Effect.orElseSucceed(() => false), + ); + const dispatchContinueTurn = (threadId: ThreadId, text: string) => Effect.gen(function* () { const shell = yield* query.getThreadShellById(threadId); @@ -1318,6 +1649,8 @@ export const make = Effect.gen(function* () { const handleHardLimit: ClaudeAccountsServiceShape["handleHardLimit"] = (condition) => Effect.gen(function* () { + // Read first: the turn this rejection ends may finish while this runs. + const wasActive = activeTurns.has(condition.threadId); const settings = yield* readSettings; if (!settings.enabled) return false; const { threadId, rateLimitType } = condition; @@ -1352,10 +1685,18 @@ export const make = Effect.gen(function* () { yield* Effect.logInfo("claude.account.hard-limit-stale", { threadId, rateLimitType }); return true; } + // A waiting thread's session is stopped; its next turn takes it out of + // the queue first. A rejection now is a late copy from before the wait. + if (waits.has(threadId)) { + yield* Effect.logInfo("claude.account.hard-limit-queued", { threadId, rateLimitType }); + return true; + } // Repeated copies of one rejection against one account are consumed // once. Accounts often share a reset instant (5-hour windows are // hour-aligned), so the account is part of the key. Without a reset time // nothing identifies the rejection, so every copy is judged on its own. + // The key stays only while the thread runs where this left it: a thread + // that waits forgets it, so a new turn's rejection is handled again. const conditionKey = `${threadId}\u0000${profile}\u0000${rateLimitType}\u0000${resetsAt}`; if (resetsAt !== undefined && !rememberCondition(conditionKey)) { return true; @@ -1373,6 +1714,7 @@ export const make = Effect.gen(function* () { rateLimitType, resetsAt, accountWide, + wasActive, election: election.kind === "ok" ? election.value.status : election.kind, }); // Only an account-wide window takes the account out of rotation; a @@ -1380,11 +1722,12 @@ export const make = Effect.gen(function* () { if (accountWide) yield* markExhausted(profile, resetsAt); if (row.mode.kind === "profile") { - notices.set( + setNotice( threadId, accountWide ? `${profile} is at its usage limit until ${formatResetClock(resetsAt)} — switch to Auto or another account.` : `${profile} has no ${scope} allowance left until ${formatResetClock(resetsAt)} — switch the model or account.`, + resetsAt, ); yield* announce(threadId); return true; @@ -1392,64 +1735,300 @@ export const make = Effect.gen(function* () { // A window that resets within the threshold is cheaper to wait out than // to leave: the thread keeps its account and its cache, and continues - // there once the window resets. The timer lives in memory, so a restart - // in between leaves the notice and a stopped thread. + // there once the window resets. The wait is a queue entry, so it + // survives a restart; it always continues, as part of the stopped turn. const nowMs = yield* Clock.currentTimeMillis; const resetMs = resetsAt === undefined ? Number.NaN : Date.parse(resetsAt); // A reset already in the past while Claude still refuses is not worth // another wait: the thread moves instead. if (resetMs > nowMs && resetMs - nowMs <= RESET_WAIT_THRESHOLD_MS) { - const continueAtMs = resetMs + RESET_WAIT_BUFFER_MS; - const continueAt = isoFromUnixSeconds(continueAtMs / 1000); - notices.set( - threadId, - `${profile} resets at ${formatResetClock(resetsAt)} — this thread continues on it at ${formatResetClock(continueAt)}.`, - ); - waitingForReset.add(threadId); - yield* stopSessionQuietly(threadId, "hard-limit-reset-wait"); - yield* FiberMap.run( - resetWaits, - threadId, - Effect.gen(function* () { - yield* Effect.sleep(Duration.millis(continueAtMs - nowMs)); - if (!waitingForReset.delete(threadId)) return; - // The continue asks the same account again, so a fresh rejection - // carrying the same reset is real, not a copy of this one. - handledConditions.delete(conditionKey); - notices.delete(threadId); - yield* pruneExhausted; - yield* Effect.logInfo("claude.account.reset-wait-continue", { threadId, profile }); - yield* dispatchContinueTurn(threadId, CONTINUE_TEXT); - yield* announce(threadId); - }), - ); - yield* Effect.logInfo("claude.account.reset-wait", { threadId, profile, continueAt }); - yield* announce(threadId); - return true; + const continueAt = isoFromUnixSeconds((resetMs + RESET_WAIT_BUFFER_MS) / 1000); + const parked = yield* park(threadId, { + wasActive: true, + restricted: false, + preferProfile: profile, + ...(continueAt !== undefined ? { notBefore: continueAt } : {}), + }); + if (parked) { + handledConditions.delete(conditionKey); + yield* stopSessionQuietly(threadId, "hard-limit-reset-wait"); + yield* Effect.logInfo("claude.account.reset-wait", { threadId, profile, continueAt }); + return true; + } } const moved = yield* reassign(threadId, { avoid: [profile] }).pipe(Effect.result); - if (moved._tag === "Failure") { - notices.set(threadId, moved.failure.detail); + if (moved._tag === "Success" && moved.success !== undefined) { + // Copies of this rejection are caught as stale until the first turn + // starts on the new account; past that, the key would only swallow a + // real rejection if the thread came back to this account. + handledConditions.delete(conditionKey); + notices.delete(threadId); + movedAwaitingTurn.add(threadId); + yield* stopSessionQuietly(threadId, "hard-limit"); + yield* dispatchContinueTurn(threadId, CONTINUE_TEXT); yield* announce(threadId); return true; } - if (moved.success === undefined) { - notices.set( + + // No account can take the thread now. It stops and waits in the queue; + // the revive pass after the next status poll continues it once an + // account has room. + const failure = moved._tag === "Failure" ? moved.failure : undefined; + const restriction = yield* disallowedFor(threadId).pipe( + Effect.orElseSucceed(() => UNRESTRICTED), + ); + const parked = yield* park(threadId, { + wasActive, + restricted: isRestricted(restriction), + // A model-scoped limit leaves the account in rotation for others. + ...(accountWide ? {} : { avoid: profile }), + ...(failure?.expectedAt !== undefined ? { expectedAt: failure.expectedAt } : {}), + ...(failure?.expectedProfile !== undefined + ? { expectedProfile: failure.expectedProfile } + : {}), + }); + if (!parked) { + setNotice( threadId, - `${profile} is at its usage limit until ${formatResetClock(resetsAt)} and no other account could be chosen.`, + failure?.detail ?? + `${profile} is at its usage limit until ${formatResetClock(resetsAt)} and no other account could be chosen.`, + failure?.expectedAt ?? resetsAt, ); yield* announce(threadId); return true; } - notices.delete(threadId); - movedAwaitingTurn.add(threadId); - yield* stopSessionQuietly(threadId, "hard-limit"); - yield* dispatchContinueTurn(threadId, CONTINUE_TEXT); - yield* announce(threadId); + // While it waits, copies are caught by the queue check above. + handledConditions.delete(conditionKey); + yield* stopSessionQuietly(threadId, "hard-limit-queued"); return true; }); + /** + * Places one waiting thread if an account has room, under the placement + * lock. Answers whether it left the queue and should get the continue turn, + * which the caller sends outside the lock. + */ + const reviveOne = (threadId: ThreadId, nowMs: number, anyRoom: boolean) => + Semaphore.withPermit(placementLock)( + Effect.gen(function* () { + const entry = waits.get(threadId); + if (entry === undefined) return undefined; + // Still waiting out a reset on its own account. + if (entry.notBefore !== null && Date.parse(entry.notBefore) > nowMs) return undefined; + const row = yield* repository.get(threadId); + if (row.mode.kind !== "auto") { + yield* dequeue(threadId, "pinned"); + return { continueTurn: false } as const; + } + const restriction = yield* disallowedFor(threadId); + const prefer = entry.preferProfile; + let profile: string | undefined; + if ( + prefer !== null && + !restriction.disallowed.has(prefer) && + (yield* usableAndPresent(prefer)) + ) { + profile = prefer; + } else { + if (!anyRoom) return undefined; + // Counted as pending only once the thread is known to take it, below. + const placed = yield* placeWith(threadId, entry.avoid ? [entry.avoid] : [], { + purpose: "revive", + requirePlaced: true, + commit: false, + }).pipe(Effect.result); + if (placed._tag === "Failure") { + const { expectedAt, expectedProfile } = placed.failure; + const next: WaitEntry = { + queuedAt: entry.queuedAt, + wasActive: entry.wasActive, + preferProfile: entry.preferProfile, + notBefore: entry.notBefore, + restricted: isRestricted(restriction), + ...(entry.avoid !== undefined ? { avoid: entry.avoid } : {}), + ...(expectedAt !== undefined ? { expectedAt } : {}), + ...(expectedProfile !== undefined ? { expectedProfile } : {}), + }; + waits.set(threadId, next); + if ( + next.expectedAt !== entry.expectedAt || + next.expectedProfile !== entry.expectedProfile || + next.restricted !== entry.restricted + ) { + yield* announce(threadId); + } + return undefined; + } + profile = placed.success?.profile; + } + if (profile === undefined) return undefined; + // The reads and the switcher call above yield: a message, a mode + // change or "Leave queue" may have taken the thread out meanwhile. + // `dequeue` removes it before its first yield, so only one side wins. + if (!(yield* dequeueQuietly(threadId, "revived"))) return undefined; + pendingPlacements.push({ profile, atMs: yield* Clock.currentTimeMillis }); + yield* persistResolved(threadId, profile).pipe( + Effect.catchCause((cause) => + Effect.logWarning("claude.account.persist-failed", { threadId, cause }), + ), + ); + // Like a move: a rejection before the next turn starts is a stale copy. + movedAwaitingTurn.add(threadId); + yield* Effect.logInfo("claude.account.revived", { + threadId, + profile, + wasActive: entry.wasActive, + }); + return { continueTurn: entry.wasActive } as const; + }), + ); + + /** + * Places the waiting threads again, in queue order. Each placement counts as + * pending for the next, so a wave of resets spreads over the accounts. A + * thread that still finds no room stays queued; the next one is tried anyway + * (owners may use different accounts). + */ + const revivePass: Effect.Effect = Semaphore.withPermit(reviveLock)( + Effect.gen(function* () { + if (waits.size === 0) return; + if (!(yield* readSettings).enabled) return; + yield* pruneExhausted; + const nowMs = yield* Clock.currentTimeMillis; + // While the last status shows no account that may take a new session, + // asking the switcher per waiting thread only spends processes. + const anyRoom = + lastStatus === undefined || + lastStatus.profiles.some( + (profile) => + profile.eligible === true && + profile.auth?.state !== "logged_out" && + !exhausted.has(profile.name), + ); + for (const [threadId] of queueOrder(nowMs)) { + const outcome = yield* reviveOne(threadId, nowMs, anyRoom).pipe( + Effect.catchCause((cause) => + Effect.as( + Effect.logWarning("claude.account.revive-failed", { threadId, cause }), + undefined, + ), + ), + ); + if (outcome === undefined) continue; + // A message accepted while this pass waited on a lock is already the + // next turn; a continue on top of it would be a second one. The shell + // marks a turn active as soon as its command is accepted. + if (outcome.continueTurn && !(yield* shellTurnInFlight(threadId))) { + yield* dispatchContinueTurn(threadId, CONTINUE_TEXT); + } + yield* announce(threadId); + } + }), + ); + + // --------------------------------------------------------------- rebalance + + /** + * At a turn boundary, moves an idle Auto thread off an account that carries + * `REBALANCE_EXCESS` or more sessions above its fair share, or that can no + * longer take new sessions, to where the switcher would place it now. The + * thread loses its prompt cache once, so this runs at most once an hour per + * thread, and an account gives up at most one thread per status read (its + * session count is stale until the next one). The switcher is asked only + * when the last status shows another account with room the owner may use, + * and not again for five minutes after it found nothing better. Never + * fails: on any trouble the thread stays where it is. + */ + const maybeRebalance = (threadId: ThreadId, settled?: OrchestrationSession) => + Effect.gen(function* () { + if (waits.has(threadId) || pendingRestart.has(threadId)) return; + if (movedAwaitingTurn.has(threadId)) return; + const current = threadProfiles.get(threadId); + if (current === undefined || rebalancedFrom.has(current)) return; + const status = lastStatus; + const known = status?.profiles.find((profile) => profile.name === current); + // Only a switcher that reports fair shares spreads load; with an older one this never runs. + if (status === undefined || known === undefined || known.fair_share === undefined) return; + const sessions = typeof known.sessions === "number" ? known.sessions : 0; + const overShare = + typeof known.fair_share === "number" && sessions - known.fair_share >= REBALANCE_EXCESS; + const notPlaceable = known.fair_share === null || known.guard_ok === false; + if (!overShare && !notPlaceable) return; + const nowMs = yield* Clock.currentTimeMillis; + const lastMove = lastRebalancedAt.get(threadId); + if (lastMove !== undefined && nowMs - lastMove < REBALANCE_COOLDOWN_MS) return; + const lastProbe = lastFailedProbeAt.get(threadId); + if (lastProbe !== undefined && nowMs - lastProbe < REBALANCE_PROBE_BACKOFF_MS) return; + if (!(yield* readSettings).enabled) return; + const row = yield* repository.get(threadId); + if (row.mode.kind !== "auto" || row.resolvedProfile !== current) return; + const live = (yield* providers.listSessions()).some( + (session) => session.threadId === threadId && session.status !== "closed", + ); + if (!live || !(yield* threadIsIdle(threadId, settled))) return; + // Without a subprocess: is there anywhere better at all? + yield* pruneExhausted; + const { disallowed } = yield* disallowedFor(threadId); + const pending = yield* pendingCounts; + const roomElsewhere = status.profiles.some( + (profile) => + profile.name !== current && + !disallowed.has(profile.name) && + !exhausted.has(profile.name) && + profile.auth?.state !== "logged_out" && + typeof profile.fair_share === "number" && + profile.fair_share > + (typeof profile.sessions === "number" ? profile.sessions : 0) + + (pending[profile.name] ?? 0), + ); + if (!roomElsewhere) return; + + const target = yield* Semaphore.withPermit(placementLock)( + Effect.gen(function* () { + if (rebalancedFrom.has(current)) return undefined; + // A probe: nothing is written or counted unless the thread moves. + const probe = yield* placeWith(threadId, [current], { + purpose: "rebalance", + requirePlaced: true, + commit: false, + }).pipe(Effect.option); + const to = + Option.isSome(probe) && probe.value !== undefined ? probe.value.profile : undefined; + if (to === undefined || to === current) { + lastFailedProbeAt.set(threadId, nowMs); + return undefined; + } + // The probe took time: a turn may have reached the provider since. A + // message the shell accepted but has not sent yet is caught later: + // `restartForNewAccount` re-reads the shell and defers the restart. + if (!(yield* threadIsIdle(threadId, settled))) return undefined; + pendingPlacements.push({ profile: to, atMs: yield* Clock.currentTimeMillis }); + lastRebalancedAt.set(threadId, nowMs); + lastFailedProbeAt.delete(threadId); + rebalancedFrom.add(current); + // Only the row: the thread still runs on `current` until its session + // stops, and the sticky branch adopts the row at the next spawn. + yield* repository.setResolved({ threadId, profile: to, resolvedAt: yield* nowIso }); + return to; + }), + ); + if (target === undefined) return; + yield* Effect.logInfo("claude.account.rebalance", { + threadId, + from: current, + to: target, + reason: overShare ? "over-fair-share" : "not-placeable", + }); + yield* restartForNewAccount(threadId, "rebalance"); + yield* announce(threadId); + yield* announceSnapshot; + }).pipe( + Effect.catchCause((cause) => + Effect.logWarning("claude.account.rebalance-failed", { threadId, cause }), + ), + ); + // ------------------------------------------------------------------ events const overlayFromEvent = (event: ProviderRuntimeEvent) => @@ -1485,12 +2064,17 @@ export const make = Effect.gen(function* () { case "turn.started": { activeTurns.add(event.threadId); // Someone sent a message during the wait: their turn replaces the continue. - if (yield* cancelResetWait(event.threadId)) { - notices.delete(event.threadId); + if (yield* dequeueQuietly(event.threadId, "turn-started")) { yield* announce(event.threadId); } + // Copies of one rejection all arrive before the next turn starts, so a + // new turn forgets them; a key kept after a failed park would + // otherwise swallow every later rejection with the same reset. + forgetConditions(event.threadId); // The first turn on the new account: from here a rejection is real. - if (movedAwaitingTurn.delete(event.threadId)) yield* announce(event.threadId); + if (movedAwaitingTurn.delete(event.threadId)) { + yield* announce(event.threadId); + } return; } case "turn.completed": @@ -1517,35 +2101,52 @@ export const make = Effect.gen(function* () { // A message accepted during a wait replaces the continue, even before // its turn reaches the provider. if (event.type === "thread.turn-start-requested") { - if (yield* cancelResetWait(event.payload.threadId)) { - notices.delete(event.payload.threadId); + if (yield* dequeueQuietly(event.payload.threadId, "message-sent")) { yield* announce(event.payload.threadId); } return; } if (event.type === "thread.session-set") { const session = event.payload.session; + const running = session.status === "running" && session.activeTurnId !== null; // T3-CUSTOM(expbkt3): native run/session projections drive account lifecycle. - if (session.status === "running" && session.activeTurnId !== null) { + if (running) { activeTurns.add(event.payload.threadId); - if (movedAwaitingTurn.delete(event.payload.threadId)) + projectedRunning.add(event.payload.threadId); + if (movedAwaitingTurn.delete(event.payload.threadId)) { + forgetConditions(event.payload.threadId); yield* announce(event.payload.threadId); + } } else if (session.status !== "starting") { activeTurns.delete(event.payload.threadId); } - if ( - pendingRestart.has(event.payload.threadId) && + const settled = session.activeTurnId === null && session.status !== "running" && - session.status !== "starting" - ) { + session.status !== "starting"; + // Only the settle that ends a turn this projection saw running is a + // turn boundary; the many other session updates are not. + const turnEnded = settled && projectedRunning.delete(event.payload.threadId); + if (settled && pendingRestart.has(event.payload.threadId)) { yield* completePendingRestart(event.payload.threadId, session); + } else if (turnEnded) { + yield* maybeRebalance(event.payload.threadId, session); + } + return; + } + if (event.type === "thread.archived") { + // An archived thread is not continued; its account row stays for an unarchive. + if (yield* dequeueQuietly(event.payload.threadId, "thread-archived")) { + yield* announce(event.payload.threadId); } return; } if (event.type !== "thread.deleted") return; const threadId = event.payload.threadId; - yield* cancelResetWait(threadId); + yield* dequeueQuietly(threadId, "thread-deleted"); + lastRebalancedAt.delete(threadId); + lastFailedProbeAt.delete(threadId); + projectedRunning.delete(threadId); threadProfiles.delete(threadId); notices.delete(threadId); pendingRestart.delete(threadId); @@ -1600,7 +2201,7 @@ export const make = Effect.gen(function* () { if (profile === null) continue; const owner = yield* threadOwner(threadId); if (mayUse(access, owner, profile)) continue; - if (row.mode.kind === "profile") notices.set(threadId, noAccessDetail(profile)); + if (row.mode.kind === "profile") setNotice(threadId, noAccessDetail(profile)); yield* Effect.logInfo("claude.account.access-revoked", { threadId, profile }); yield* restartForNewAccount(threadId); yield* announce(threadId); @@ -1696,6 +2297,7 @@ export const make = Effect.gen(function* () { resolveForSession, refreshStatus, markExhausted, + leaveQueue, reassign, handleHardLimit, }); diff --git a/apps/server/src/claudeAccounts/ClaudeAutoswitchClient.test.ts b/apps/server/src/claudeAccounts/ClaudeAutoswitchClient.test.ts index 478eaeae969cc..5c93723631d89 100644 --- a/apps/server/src/claudeAccounts/ClaudeAutoswitchClient.test.ts +++ b/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", () => diff --git a/apps/server/src/claudeAccounts/ClaudeAutoswitchClient.ts b/apps/server/src/claudeAccounts/ClaudeAutoswitchClient.ts index 6413f158ee961..496be8d613259 100644 --- a/apps/server/src/claudeAccounts/ClaudeAutoswitchClient.ts +++ b/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; diff --git a/apps/server/src/claudeAccounts/applyClaudeAccountProfile.ts b/apps/server/src/claudeAccounts/applyClaudeAccountProfile.ts index 1a19e2e72896c..69701b120814c 100644 --- a/apps/server/src/claudeAccounts/applyClaudeAccountProfile.ts +++ b/apps/server/src/claudeAccounts/applyClaudeAccountProfile.ts @@ -23,6 +23,9 @@ export class ClaudeAccountPolicyError extends Schema.TaggedError { // 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( diff --git a/apps/server/src/mcp/toolkits/webUi/registration.test.ts b/apps/server/src/mcp/toolkits/webUi/registration.test.ts index 5230767cd41a0..966942d216c19 100644 --- a/apps/server/src/mcp/toolkits/webUi/registration.test.ts +++ b/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( diff --git a/apps/server/src/orchestration-v2/Layers/OrchestrationEngine.ts b/apps/server/src/orchestration-v2/Layers/OrchestrationEngine.ts index 64c451480f4ef..2dc101fa10af8 100644 --- a/apps/server/src/orchestration-v2/Layers/OrchestrationEngine.ts +++ b/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); } diff --git a/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts b/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts index c68fc0ace5e93..f1b9859a40fd2 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts +++ b/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. diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 001512652c861..527e7ab30c729 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -465,6 +465,8 @@ function commandThreadId(command: OrchestrationV2ServerCommand): ThreadId { case "thread.visit": case "thread.mark-unread": case "thread.activity.append": + // T3-CUSTOM(expbkt3): a decided plan review retires its plan. + case "thread.proposed-plan.retire": case "thread.metadata.update": case "thread.pull-request.link": case "thread.pull-request.unlink": @@ -1737,6 +1739,24 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio session.status !== "stopped" && session.status !== "error", ); + // T3-CUSTOM(expbkt3): a queued continuation ends the thread's open plan when it starts, not + // when it was queued, so cancelling it leaves the plan open. Wakes and + // maintenance commands are not continuations. + const supersededPlans = + queuedMessage.delegatedCompletion === undefined && + queuedMessage.notification === undefined && + !isNativeMaintenanceCommand(queuedMessage) + ? (yield* projectionStore.getActiveProposedPlans(threadId).pipe( + Effect.mapError( + (cause) => + new OrchestratorDispatchError({ + commandId, + commandType: "message.dispatch", + cause, + }), + ), + )).filter((plan) => plan.id !== queuedRun.sourcePlanRef?.planId) + : []; yield* writeSystemEvents( [ ...(selectionChanged @@ -1857,6 +1877,16 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio occurredAt: now, payload: notificationTurnItem(userTurnItem, queuedMessage, projection.subagents), }, + // T3-CUSTOM(expbkt3): retire the plans this queued continuation ends. + ...supersededPlans.map((plan) => ({ + type: "plan.updated" as const, + threadId: plan.threadId, + ...(plan.runId === null ? {} : { runId: plan.runId }), + nodeId: plan.nodeId, + providerInstanceId: queuedRun.providerInstanceId, + occurredAt: now, + payload: { ...plan, status: "superseded" as const }, + })), { type: "run.updated", threadId, @@ -4891,27 +4921,62 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio cause: `Proposed plan ${command.sourcePlanRef?.planId} belongs to a different project.`, }); } - if (sourcePlan !== null && sourcePlan.status !== "active") { + // A superseded plan may still be implemented when the turn names it: a + // reviewer can approve the exact version after a plan-mode question. + if ( + sourcePlan !== null && + sourcePlan.status !== "active" && + sourcePlan.status !== "superseded" + ) { return yield* new OrchestratorDispatchError({ commandId: command.commandId, commandType: command.type, cause: `Proposed plan ${sourcePlan.id} is not active.`, }); } - const completeSourcePlan = (occurredAt: DateTime.Utc) => - sourcePlan === null - ? Effect.void - : emit( - events, - command, - )({ - type: "plan.updated", - threadId: sourcePlan.threadId, - ...(sourcePlan.runId === null ? {} : { runId: sourcePlan.runId }), - nodeId: sourcePlan.nodeId, - occurredAt, - payload: { ...sourcePlan, status: "completed" }, - }); + // Continuing a thread ends its open plan in every interaction mode: the + // plan this turn implements is completed, any other is superseded, so an + // earlier plan can never become actionable again. A background-task wake + // or a maintenance command (`/compact`) is not a continuation, and a + // queued message waits for its run, so those leave an open plan alone. + const continuesThread = + delegatedCompletion === undefined && + command.notification === undefined && + !isNativeMaintenanceCommand(command); + const completeSourcePlan = ( + occurredAt: DateTime.Utc, + options: { readonly supersedeOpenPlans: boolean } = { supersedeOpenPlans: true }, + ) => + Effect.gen(function* () { + const openPlans = + continuesThread && options.supersedeOpenPlans + ? yield* projectionStore + .getActiveProposedPlans(command.threadId) + .pipe(mapDispatchError(command)) + : []; + const retired = [ + ...openPlans + .filter((plan) => plan.id !== sourcePlan?.id) + .map((plan) => ({ ...plan, status: "superseded" as const })), + ...(sourcePlan === null ? [] : [{ ...sourcePlan, status: "completed" as const }]), + ]; + yield* Effect.forEach( + retired, + (plan) => + emit( + events, + command, + )({ + type: "plan.updated", + threadId: plan.threadId, + ...(plan.runId === null ? {} : { runId: plan.runId }), + nodeId: plan.nodeId, + occurredAt, + payload: plan, + }), + { discard: true }, + ); + }); if (dispatchMode.type === "steer_active" || dispatchMode.type === "restart_active") { yield* dispatchSteerIntoRun({ @@ -5157,7 +5222,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio occurredAt: now, payload: run, }); - yield* completeSourcePlan(now); + yield* completeSourcePlan(now, { supersedeOpenPlans: false }); yield* emitEvent({ type: "run-attempt.created", threadId: command.threadId, @@ -10394,6 +10459,26 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio readonly reason: string; } | undefined; + // T3-CUSTOM(expbkt3): a plan review decided outside a turn start ends its plan. + if (command.type === "thread.proposed-plan.retire") { + const plan = yield* projectionStore + .getPlan(command.threadId, command.planId) + .pipe(mapDispatchError(command)); + if (plan?.kind === "proposed_plan" && plan.status === "active") { + yield* emit( + events, + command, + )({ + type: "plan.updated", + threadId: plan.threadId, + ...(plan.runId === null ? {} : { runId: plan.runId }), + nodeId: plan.nodeId, + occurredAt: yield* DateTime.now, + payload: { ...plan, status: command.status }, + }); + } + return { events: yield* Ref.get(events), effects: [] }; + } // T3-CUSTOM(expbkt3): fork tool/activity payloads share the V2 event and receipt transaction. if (command.type === "thread.activity.append") { const projection = yield* projectionStore @@ -10899,10 +10984,12 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio Effect.flatMap((planned) => // A settle that finds the provider already ended everything, or a // stop that finds nothing running, has nothing to record. That is - // its expected outcome, not a failure. + // its expected outcome, not a failure. So is retiring a plan that a + // turn already completed or superseded. planned.events.length > 0 || command.type === "thread.background-work.settle" || - command.type === "thread.stop" + command.type === "thread.stop" || + command.type === "thread.proposed-plan.retire" ? Effect.succeed(planned) : Effect.fail( new OrchestratorDispatchError({ diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 5b3a82f1ea04a..682b7d64e5d12 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -418,6 +418,13 @@ export interface ProjectionStoreV2Shape { threadId: ThreadId, planId: PlanId, ) => Effect.Effect; + // T3-CUSTOM(expbkt3): proposed plans on the thread that still wait for a decision. + readonly getActiveProposedPlans: ( + threadId: ThreadId, + ) => Effect.Effect< + ReadonlyArray>, + ProjectionStoreV2Error + >; readonly hasUnpairedRunInterruptRequest: ( threadId: ThreadId, requestId: TurnItemId, @@ -1052,6 +1059,11 @@ const decodeMessagePayload = Schema.decodeUnknownEffect( ); const decodePlanArtifact = Schema.decodeUnknownEffect(OrchestrationV2PlanArtifactSchema); const decodePlanPayload = (json: string) => decodePlanArtifact(parseEncodedPayload(json)); +// T3-CUSTOM(expbkt3): narrows a plan artifact to a proposed plan. +const isProposedPlanArtifact = ( + plan: OrchestrationV2PlanArtifact, +): plan is Extract => + plan.kind === "proposed_plan"; const decodeTurnItemPayload = Schema.decodeUnknownEffect( Schema.fromJsonString(OrchestrationV2TurnItemJsonSchema), ); @@ -4310,6 +4322,17 @@ export const layer: Layer.Layer = ) .pipe(Effect.mapError(controlReadError(threadId))); + // T3-CUSTOM(expbkt3): one indexed read, so plan supersession never loads the thread. + const getActiveProposedPlans: ProjectionStoreV2Shape["getActiveProposedPlans"] = (threadId) => + Effect.gen(function* () { + const rows = yield* sql` + SELECT payload_json FROM orchestration_v2_projection_plans + WHERE thread_id = ${threadId} AND kind = 'proposed_plan' AND status = 'active' + `; + const plans = yield* decodeRows(decodePlanPayload, threadId)(rows); + return plans.filter(isProposedPlanArtifact); + }).pipe(Effect.mapError(controlReadError(threadId))); + const hasUnpairedRunInterruptRequest: ProjectionStoreV2Shape["hasUnpairedRunInterruptRequest"] = (threadId, requestId, resultId) => Effect.gen(function* () { @@ -5795,6 +5818,8 @@ export const layer: Layer.Layer = getThreadRecords, getRuntimeRequest, getPlan, + // T3-CUSTOM(expbkt3): plan supersession read. + getActiveProposedPlans, getProviderControlContext, getLimitRecoveryCandidates, getRecoveryThreadIds, @@ -6026,6 +6051,15 @@ export const layerMemory: Layer.Layer = Layer.effect( return yield* new ProjectionStoreThreadNotFoundError({ threadId }); return projection.plans.find((plan) => plan.id === planId); }), + // T3-CUSTOM(expbkt3): plan supersession read. + getActiveProposedPlans: (threadId) => + Effect.gen(function* () { + const projection = (yield* Ref.get(replayState)).projections.get(threadId); + if (projection === undefined) return []; + return projection.plans + .filter(isProposedPlanArtifact) + .filter((plan) => plan.status === "active"); + }), hasUnpairedRunInterruptRequest: (threadId, requestId, resultId) => Effect.gen(function* () { const projection = (yield* Ref.get(replayState)).projections.get(threadId); diff --git a/apps/server/src/orchestration-v2/ProviderEventIngestor.test.ts b/apps/server/src/orchestration-v2/ProviderEventIngestor.test.ts index c9740adfbf0c4..1f03e67498956 100644 --- a/apps/server/src/orchestration-v2/ProviderEventIngestor.test.ts +++ b/apps/server/src/orchestration-v2/ProviderEventIngestor.test.ts @@ -434,6 +434,56 @@ layer("ProviderEventIngestorV2", (it) => { }), ); + it.effect("keeps only the newest proposed plan actionable, for every provider", () => + Effect.gen(function* () { + const now = yield* DateTime.now; + const eventSink = yield* EventSink.EventSinkV2; + const projectionStore = yield* ProjectionStore.ProjectionStoreV2; + const ingestor = yield* ProviderEventIngestor.ProviderEventIngestorV2; + const idAllocator = yield* IdAllocator.IdAllocatorV2; + const threadEvent = yield* threadCreatedEvent(now); + const providerSessionId = yield* idAllocator.allocate.providerSession({ + providerInstanceId: modelSelection.instanceId, + threadId: threadEvent.threadId, + }); + const proposePlan = (revision: number) => + ingestor.ingestNormalized({ + providerSessionId, + providerInstanceId: modelSelection.instanceId, + threadId: threadEvent.threadId, + event: { + type: "plan.updated", + driver: CODEX_DRIVER, + plan: { + id: PlanId.make(`plan:supersede-r${revision}`), + threadId: threadEvent.threadId, + runId: null, + nodeId: NodeId.make(`node:supersede-r${revision}`), + kind: "proposed_plan", + status: "active", + markdown: `# Plan revision ${revision}`, + }, + }, + }); + + yield* eventSink.write({ events: [threadEvent] }); + yield* proposePlan(2); + const stored = yield* proposePlan(3); + // A streamed update of the newest plan retires nothing more. + const restreamed = yield* proposePlan(3); + + const statuses = (yield* projectionStore.getThreadProjection(threadEvent.threadId)).plans + .map((plan) => [plan.id, plan.status]) + .toSorted(([left], [right]) => String(left).localeCompare(String(right))); + assert.deepEqual(statuses, [ + ["plan:supersede-r2", "superseded"], + ["plan:supersede-r3", "active"], + ]); + assert.equal(stored.length, 2); + assert.equal(restreamed.length, 1); + }), + ); + it.effect( "treats successful provider terminal markers as non-persisted orchestration control signals", () => diff --git a/apps/server/src/orchestration-v2/ProviderEventIngestor.ts b/apps/server/src/orchestration-v2/ProviderEventIngestor.ts index ad5d3a7ddb1b2..ee7a8523ff4e0 100644 --- a/apps/server/src/orchestration-v2/ProviderEventIngestor.ts +++ b/apps/server/src/orchestration-v2/ProviderEventIngestor.ts @@ -500,7 +500,25 @@ export const layer: Layer.Layer< occurredAt, ) : plan; + // A thread has one actionable proposed plan: the newest. Retiring the + // others here covers every adapter and survives restarts. + const superseded = + plan.kind === "proposed_plan" && plan.status === "active" + ? (yield* projections.getActiveProposedPlans(plan.threadId)).filter( + (candidate) => candidate.id !== plan.id, + ) + : []; return [ + ...(yield* Effect.forEach(superseded, (previousPlan) => + makeDomainEvent(input, { + type: "plan.updated", + threadId: previousPlan.threadId, + payload: { ...previousPlan, status: "superseded" }, + runId: previousPlan.runId, + nodeId: previousPlan.nodeId, + occurredAt, + }), + )), yield* makeDomainEvent(input, { type: "plan.updated", threadId: plan.threadId, diff --git a/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts b/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts index 7918f819eee2a..92ae881ce3a50 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts @@ -222,6 +222,7 @@ it.effect( getTurnStartHistory: () => Effect.die("unused"), getRuntimeRecoveryProjection: () => Effect.die("unused getRuntimeRecoveryProjection"), getPlan: () => Effect.die("unused"), + getActiveProposedPlans: () => Effect.die("unused"), hasUnpairedRunInterruptRequest: () => Effect.die("unused interrupt read"), getThreadAttachmentIds: () => Effect.die("Unused attachment lookup"), getTimelinePage: () => Effect.die("Unused timeline read"), diff --git a/apps/server/src/orchestration-v2/testkit/OrchestratorScenario.ts b/apps/server/src/orchestration-v2/testkit/OrchestratorScenario.ts index 8136cf696aab6..b08c590ebb007 100644 --- a/apps/server/src/orchestration-v2/testkit/OrchestratorScenario.ts +++ b/apps/server/src/orchestration-v2/testkit/OrchestratorScenario.ts @@ -135,6 +135,7 @@ function commandThreadIds(command: OrchestrationV2Command): ReadonlyArray [id, name] as const); diff --git a/apps/server/src/persistence/Migrations/055_OrchestrationV2.test.ts b/apps/server/src/persistence/Migrations/055_OrchestrationV2.test.ts index 6cc9ebd641a6d..d51c0e7395a7c 100644 --- a/apps/server/src/persistence/Migrations/055_OrchestrationV2.test.ts +++ b/apps/server/src/persistence/Migrations/055_OrchestrationV2.test.ts @@ -19,8 +19,9 @@ layer("055_OrchestrationV2", (it) => { // upstream's scheduled-task webhook migrations (57-58) at 1046-1047, the // review comment send-once mark at 1048 and upstream's MCP app context and // snapshot indexes (59-60) at 1049-1050, scheduled task custom - // groups at 1051 and scheduled task options at 1052. - ...Array.from({ length: 53 }, (_, index) => 1000 + index), + // groups at 1051, scheduled task options at 1052 and the Claude account + // revive queue at 1053. + ...Array.from({ length: 54 }, (_, index) => 1000 + index), ], ); }), @@ -44,6 +45,7 @@ layer("055_OrchestrationV2", (it) => { [1050, "ThreadSnapshotWindowIndexes"], [1051, "ScheduledTaskCustomGroups"], // T3-CUSTOM(expbkt3): scheduled task custom groups. [1052, "ScheduledTaskForkOptions"], // T3-CUSTOM(expbkt3): scheduled task options. + [1053, "ThreadClaudeAccountWait"], // T3-CUSTOM(expbkt3): Claude account revive queue. ]); assert.deepStrictEqual(yield* runMigrations(), []); @@ -80,6 +82,7 @@ layer("055_OrchestrationV2", (it) => { { migration_id: 1050, name: "ThreadSnapshotWindowIndexes" }, { migration_id: 1051, name: "ScheduledTaskCustomGroups" }, // T3-CUSTOM(expbkt3) { migration_id: 1052, name: "ScheduledTaskForkOptions" }, // T3-CUSTOM(expbkt3) + { migration_id: 1053, name: "ThreadClaudeAccountWait" }, // T3-CUSTOM(expbkt3) ]); // T3-CUSTOM(expbkt3): verify callback destinations and durable deliveries after the full upgrade. diff --git a/apps/server/src/persistence/Migrations/1053_ThreadClaudeAccountWait.test.ts b/apps/server/src/persistence/Migrations/1053_ThreadClaudeAccountWait.test.ts new file mode 100644 index 0000000000000..d5f23ab9256b3 --- /dev/null +++ b/apps/server/src/persistence/Migrations/1053_ThreadClaudeAccountWait.test.ts @@ -0,0 +1,105 @@ +// T3-CUSTOM(expbkt3): the Claude account revive queue columns and the +// repository methods that write and read them. +import { assert, it } from "@effect/vitest"; +import { ThreadId } from "@t3tools/contracts"; +import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient"; +import * as Effect from "effect/Effect"; +import * as SqlClient from "effect/sql/SqlClient"; + +import { runMigrations } from "../Migrations.ts"; +import * as ThreadClaudeAccountRepo from "../ThreadClaudeAccount.ts"; + +const columnNames = Effect.flatMap( + SqlClient.SqlClient, + (sql) => sql<{ readonly name: string }>`PRAGMA table_info(thread_claude_account)`, +).pipe(Effect.map((columns) => columns.map((column) => column.name))); + +it.layer(NodeSqliteClient.layer({ filename: ":memory:" }))("1053_ThreadClaudeAccountWait", (it) => { + it.effect("adds the wait columns once, keeping existing rows as not waiting", () => + Effect.gen(function* () { + yield* runMigrations({ toMigrationInclusive: 1052 }); + const sql = yield* SqlClient.SqlClient; + yield* sql` + INSERT INTO thread_claude_account (thread_id, mode_json, resolved_profile, resolved_at, updated_at) + VALUES ('thread-old', '{"kind":"auto"}', 'sam', '2026-10-16T12:00:00.000Z', '2026-10-16T12:00:00.000Z') + `; + yield* runMigrations({ toMigrationInclusive: 1053 }); + yield* runMigrations({ toMigrationInclusive: 1053 }); + const names = yield* columnNames; + for (const expected of [ + "wait_queued_at", + "wait_was_active", + "wait_prefer_profile", + "wait_not_before", + ]) { + assert.strictEqual(names.filter((name) => name === expected).length, 1); + } + + const repository = yield* ThreadClaudeAccountRepo.make; + const old = yield* repository.get(ThreadId.make("thread-old")); + assert.strictEqual(old.resolvedProfile, "sam"); + assert.strictEqual(old.wait, null); + assert.deepStrictEqual(yield* repository.listWaiting(), []); + }), + ); + + it.effect("writes, lists and clears a wait without touching the mode or account", () => + Effect.gen(function* () { + yield* runMigrations(); + const repository = yield* ThreadClaudeAccountRepo.make; + const queued = ThreadId.make("thread-queued"); + const later = ThreadId.make("thread-queued-later"); + const draft = ThreadId.make("thread-queued-draft"); + yield* repository.setResolved({ + threadId: queued, + profile: "sam", + resolvedAt: "2026-10-16T12:19:02.000Z", + }); + yield* repository.setWait({ + threadId: later, + wait: { + queuedAt: "2026-10-16T12:20:00.000Z", + wasActive: false, + preferProfile: null, + notBefore: null, + }, + }); + yield* repository.setWait({ + threadId: queued, + wait: { + queuedAt: "2026-10-16T12:19:02.000Z", + wasActive: true, + preferProfile: "sam", + notBefore: "2026-10-16T12:21:00.000Z", + }, + }); + + const waiting = yield* repository.listWaiting(); + assert.deepStrictEqual( + waiting.map((row) => row.threadId), + [queued, later], + ); + assert.deepStrictEqual(waiting[0]!.wait, { + queuedAt: "2026-10-16T12:19:02.000Z", + wasActive: true, + preferProfile: "sam", + notBefore: "2026-10-16T12:21:00.000Z", + }); + assert.strictEqual(waiting[0]!.resolvedProfile, "sam"); + assert.strictEqual(waiting[1]!.wait?.wasActive, false); + + yield* repository.clearWait(queued); + // Clearing a thread with no row is a no-op. + yield* repository.clearWait(draft); + const cleared = yield* repository.get(queued); + assert.strictEqual(cleared.wait, null); + assert.strictEqual(cleared.resolvedProfile, "sam"); + assert.deepStrictEqual(cleared.mode, { kind: "auto" }); + assert.deepStrictEqual( + (yield* repository.listWaiting()).map((row) => row.threadId), + [later], + ); + assert.strictEqual((yield* repository.get(draft)).wait, null); + }), + ); +}); diff --git a/apps/server/src/persistence/Migrations/1053_ThreadClaudeAccountWait.ts b/apps/server/src/persistence/Migrations/1053_ThreadClaudeAccountWait.ts new file mode 100644 index 0000000000000..219ed4c2fa028 --- /dev/null +++ b/apps/server/src/persistence/Migrations/1053_ThreadClaudeAccountWait.ts @@ -0,0 +1,31 @@ +// T3-CUSTOM(expbkt3): the durable revive queue for Claude account placement. +// +// A thread that no account can take waits in `thread_claude_account` until +// the status poll finds room. All four columns are null for a thread that is +// not waiting: +// +// - `wait_queued_at`: when it joined the queue (oldest goes first); +// - `wait_was_active`: 1 when a turn was in flight, so the revive sends the +// continue turn; 0 when it wakes only once someone writes; +// - `wait_prefer_profile`: the account a short reset wait continues on; +// - `wait_not_before`: the time that wait ends. +import * as Effect from "effect/Effect"; +import * as SqlClient from "effect/sql/SqlClient"; + +export default Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + const columns = yield* sql<{ readonly name: string }>`PRAGMA table_info(thread_claude_account)`; + const present = new Set(columns.map((column) => column.name)); + if (!present.has("wait_queued_at")) { + yield* sql`ALTER TABLE thread_claude_account ADD COLUMN wait_queued_at TEXT`; + } + if (!present.has("wait_was_active")) { + yield* sql`ALTER TABLE thread_claude_account ADD COLUMN wait_was_active INTEGER`; + } + if (!present.has("wait_prefer_profile")) { + yield* sql`ALTER TABLE thread_claude_account ADD COLUMN wait_prefer_profile TEXT`; + } + if (!present.has("wait_not_before")) { + yield* sql`ALTER TABLE thread_claude_account ADD COLUMN wait_not_before TEXT`; + } +}); diff --git a/apps/server/src/persistence/ThreadClaudeAccount.ts b/apps/server/src/persistence/ThreadClaudeAccount.ts index f9a0f99a4b88b..e0859fc87e97d 100644 --- a/apps/server/src/persistence/ThreadClaudeAccount.ts +++ b/apps/server/src/persistence/ThreadClaudeAccount.ts @@ -1,9 +1,10 @@ /** * T3-CUSTOM(expbkt3): persistence for the Claude account a thread runs on. * - * One row per thread, rewritten whole: the mode the user chose (`auto` or a - * pinned profile) plus the account the thread last resolved to. An absent row - * means `auto` with nothing resolved yet. + * One row per thread: the mode the user chose (`auto` or a pinned profile), + * the account the thread last resolved to, and its place in the revive queue + * when no account could take it. An absent row means `auto` with nothing + * resolved and no wait. */ import { CLAUDE_ACCOUNT_MODE_AUTO, ClaudeAccountMode, ThreadId } from "@t3tools/contracts"; import * as Context from "effect/Context"; @@ -22,12 +23,23 @@ import { export type ThreadClaudeAccountRepositoryError = ProjectionRepositoryError; +/** A thread waiting in the revive queue (migration 1053). */ +export interface ThreadClaudeAccountWait { + readonly queuedAt: string; + /** A turn was in flight when it stopped, so the revive sends the continue turn. */ + readonly wasActive: boolean; + /** A short reset wait: continue on this account once `notBefore` passes. */ + readonly preferProfile: string | null; + readonly notBefore: string | null; +} + export interface ThreadClaudeAccountRow { readonly threadId: ThreadId; readonly mode: ClaudeAccountMode; readonly resolvedProfile: string | null; readonly resolvedAt: string | null; readonly updatedAt: string; + readonly wait: ThreadClaudeAccountWait | null; } const DbRow = Schema.Struct({ @@ -36,6 +48,27 @@ const DbRow = Schema.Struct({ resolvedProfile: Schema.NullOr(Schema.String), resolvedAt: Schema.NullOr(Schema.String), updatedAt: Schema.String, + waitQueuedAt: Schema.NullOr(Schema.String), + waitWasActive: Schema.NullOr(Schema.Number), + waitPreferProfile: Schema.NullOr(Schema.String), + waitNotBefore: Schema.NullOr(Schema.String), +}); + +const fromDbRow = (row: typeof DbRow.Type): ThreadClaudeAccountRow => ({ + threadId: row.threadId, + mode: row.mode, + resolvedProfile: row.resolvedProfile, + resolvedAt: row.resolvedAt, + updatedAt: row.updatedAt, + wait: + row.waitQueuedAt === null + ? null + : { + queuedAt: row.waitQueuedAt, + wasActive: row.waitWasActive === 1, + preferProfile: row.waitPreferProfile, + notBefore: row.waitNotBefore, + }, }); export class ThreadClaudeAccountRepository extends Context.Service< @@ -57,6 +90,20 @@ export class ThreadClaudeAccountRepository extends Context.Service< readonly profile: string; readonly resolvedAt: string; }) => Effect.Effect; + /** Puts the thread in the revive queue (or replaces its entry); mode and account are kept. */ + readonly setWait: (input: { + readonly threadId: ThreadId; + readonly wait: ThreadClaudeAccountWait; + }) => Effect.Effect; + /** Takes the thread out of the revive queue; a missing row is a no-op. */ + readonly clearWait: ( + threadId: ThreadId, + ) => Effect.Effect; + /** Every thread in the revive queue, oldest first. */ + readonly listWaiting: () => Effect.Effect< + ReadonlyArray, + ThreadClaudeAccountRepositoryError + >; /** Every thread whose last session ran on the given account. */ readonly listByProfile: ( profile: string, @@ -81,6 +128,7 @@ export const defaultRow = (threadId: ThreadId): ThreadClaudeAccountRow => ({ resolvedProfile: null, resolvedAt: null, updatedAt: "1970-01-01T00:00:00.000Z", + wait: null, }); export const make = Effect.gen(function* () { @@ -91,7 +139,11 @@ export const make = Effect.gen(function* () { mode_json AS "mode", resolved_profile AS "resolvedProfile", resolved_at AS "resolvedAt", - updated_at AS "updatedAt" + updated_at AS "updatedAt", + wait_queued_at AS "waitQueuedAt", + wait_was_active AS "waitWasActive", + wait_prefer_profile AS "waitPreferProfile", + wait_not_before AS "waitNotBefore" `; const getRow = SqlSchema.findOneOption({ @@ -112,6 +164,16 @@ export const make = Effect.gen(function* () { `, }); + const listWaitingRows = SqlSchema.findAll({ + Request: Schema.Void, + Result: DbRow, + execute: () => sql` + SELECT ${columns} FROM thread_claude_account + WHERE wait_queued_at IS NOT NULL + ORDER BY wait_queued_at ASC + `, + }); + const setModeRow = SqlSchema.void({ Request: Schema.Struct({ threadId: ThreadId, @@ -141,6 +203,44 @@ export const make = Effect.gen(function* () { `, }); + const setWaitRow = SqlSchema.void({ + Request: Schema.Struct({ + threadId: ThreadId, + queuedAt: Schema.String, + // SQLite has no boolean; node:sqlite refuses one as a parameter. + wasActive: Schema.Number, + preferProfile: Schema.NullOr(Schema.String), + notBefore: Schema.NullOr(Schema.String), + }), + execute: ({ threadId, queuedAt, wasActive, preferProfile, notBefore }) => sql` + INSERT INTO thread_claude_account ( + thread_id, mode_json, resolved_profile, resolved_at, updated_at, + wait_queued_at, wait_was_active, wait_prefer_profile, wait_not_before + ) + VALUES ( + ${threadId}, ${JSON.stringify(CLAUDE_ACCOUNT_MODE_AUTO)}, NULL, NULL, ${queuedAt}, + ${queuedAt}, ${wasActive}, ${preferProfile}, ${notBefore} + ) + ON CONFLICT(thread_id) DO UPDATE SET + wait_queued_at = ${queuedAt}, + wait_was_active = ${wasActive}, + wait_prefer_profile = ${preferProfile}, + wait_not_before = ${notBefore} + `, + }); + + const clearWaitRow = SqlSchema.void({ + Request: Schema.Struct({ threadId: ThreadId }), + execute: ({ threadId }) => sql` + UPDATE thread_claude_account SET + wait_queued_at = NULL, + wait_was_active = NULL, + wait_prefer_profile = NULL, + wait_not_before = NULL + WHERE thread_id = ${threadId} + `, + }); + const deleteRow = SqlSchema.void({ Request: Schema.Struct({ threadId: ThreadId }), execute: ({ threadId }) => sql` @@ -151,7 +251,7 @@ export const make = Effect.gen(function* () { return ThreadClaudeAccountRepository.of({ get: (threadId) => getRow({ threadId }).pipe( - Effect.map(Option.getOrElse(() => defaultRow(threadId))), + Effect.map(Option.match({ onNone: () => defaultRow(threadId), onSome: fromDbRow })), Effect.mapError(mapError("ThreadClaudeAccount.get")), ), setMode: ({ threadId, mode, updatedAt }) => @@ -160,8 +260,26 @@ export const make = Effect.gen(function* () { ), setResolved: (input) => setResolvedRow(input).pipe(Effect.mapError(mapError("ThreadClaudeAccount.setResolved"))), + setWait: ({ threadId, wait }) => + setWaitRow({ + threadId, + queuedAt: wait.queuedAt, + wasActive: wait.wasActive ? 1 : 0, + preferProfile: wait.preferProfile, + notBefore: wait.notBefore, + }).pipe(Effect.mapError(mapError("ThreadClaudeAccount.setWait"))), + clearWait: (threadId) => + clearWaitRow({ threadId }).pipe(Effect.mapError(mapError("ThreadClaudeAccount.clearWait"))), + listWaiting: () => + listWaitingRows(undefined).pipe( + Effect.map((rows) => rows.map(fromDbRow)), + Effect.mapError(mapError("ThreadClaudeAccount.listWaiting")), + ), listByProfile: (profile) => - listRows({ profile }).pipe(Effect.mapError(mapError("ThreadClaudeAccount.listByProfile"))), + listRows({ profile }).pipe( + Effect.map((rows) => rows.map(fromDbRow)), + Effect.mapError(mapError("ThreadClaudeAccount.listByProfile")), + ), delete: (threadId) => deleteRow({ threadId }).pipe(Effect.mapError(mapError("ThreadClaudeAccount.delete"))), }); diff --git a/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts b/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts index 3823f8466abe7..fa0b710df8f94 100644 --- a/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts +++ b/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts @@ -20,6 +20,7 @@ const pendingV2Migrations = [ [1050, "ThreadSnapshotWindowIndexes"], [1051, "ScheduledTaskCustomGroups"], [1052, "ScheduledTaskForkOptions"], + [1053, "ThreadClaudeAccountWait"], ] as const; describe("fork V2 ledger upgrade", () => { @@ -65,6 +66,7 @@ describe("fork V2 ledger upgrade", () => { pendingV2Migrations[7], pendingV2Migrations[8], pendingV2Migrations[9], + pendingV2Migrations[10], ]); assert.deepStrictEqual(yield* runMigrations(), []); assert.deepStrictEqual(yield* sql`SELECT * FROM orchestration_v2_legacy_imports`, imports); diff --git a/apps/server/src/planreview/PlanReviewContextPolicy.test.ts b/apps/server/src/planreview/PlanReviewContextPolicy.test.ts index 2467354f293ed..a43a7f7ab78b1 100644 --- a/apps/server/src/planreview/PlanReviewContextPolicy.test.ts +++ b/apps/server/src/planreview/PlanReviewContextPolicy.test.ts @@ -8,6 +8,7 @@ const baseSignals = { planThreadId: "thread-1", targetThreadId: "thread-1", providerSessionStatus: "running", + agentRevisionCount: 1, } as const; describe("decidePlanResend", () => { @@ -21,6 +22,12 @@ describe("decidePlanResend", () => { expect(decision.reason).toContain("different session"); }); + it("resends when the agent revised the plan", () => { + const decision = decidePlanResend({ ...baseSignals, agentRevisionCount: 3 }); + expect(decision.shouldResend).toBe(true); + expect(decision.reason).toContain("only this revision is approved"); + }); + it("resends when the thread compacted after the plan was written", () => { const decision = decidePlanResend({ ...baseSignals, diff --git a/apps/server/src/planreview/PlanReviewContextPolicy.ts b/apps/server/src/planreview/PlanReviewContextPolicy.ts index 784ecb865147d..f287b62171c78 100644 --- a/apps/server/src/planreview/PlanReviewContextPolicy.ts +++ b/apps/server/src/planreview/PlanReviewContextPolicy.ts @@ -20,6 +20,8 @@ export interface PlanResendSignals { readonly targetThreadId: string; /** Provider session status bound to the target thread, null when unbound. */ readonly providerSessionStatus: string | null; + /** How many versions of the plan the agent wrote. */ + readonly agentRevisionCount: number; } export interface PlanResendDecision { @@ -34,7 +36,7 @@ export interface PlanResendDecision { /** * Returns whether the approval prompt must carry the whole plan. * - * Three signals force a resend; anything else keeps the prompt to one line. + * Four signals force a resend; anything else keeps the prompt to one line. */ export function decidePlanResend(signals: PlanResendSignals): PlanResendDecision { if (signals.targetThreadId !== signals.planThreadId) { @@ -44,6 +46,15 @@ export function decidePlanResend(signals: PlanResendSignals): PlanResendDecision }; } + // Earlier revisions are still in context, and "the plan above" has been read + // as one of them. Naming the approved text removes the ambiguity. + if (signals.agentRevisionCount > 1) { + return { + shouldResend: true, + reason: "the plan was revised, and only this revision is approved", + }; + } + if (signals.latestCompactionAt !== null && signals.latestCompactionAt > signals.planCreatedAt) { return { shouldResend: true, diff --git a/apps/server/src/planreview/PlanReviewService.test.ts b/apps/server/src/planreview/PlanReviewService.test.ts index f55c3e8a31211..0e527ce70d660 100644 --- a/apps/server/src/planreview/PlanReviewService.test.ts +++ b/apps/server/src/planreview/PlanReviewService.test.ts @@ -379,7 +379,7 @@ describe("PlanReviewService approval", () => { expect(result.resentPlan).toBe(false); expect(result.prompt).toBe( - "Plan approved. Implement the plan you proposed above, exactly as written.", + "Plan approved (revision 1, plan plan:a). Implement the plan you proposed above, exactly as written.", ); const commands = yield* Ref.get(dispatched); @@ -456,13 +456,43 @@ describe("PlanReviewService approval", () => { expect(result.resentPlan).toBe(true); const text = turnText(yield* Ref.get(dispatched)); - expect(text).toContain("PLEASE IMPLEMENT THIS APPROVED PLAN:"); + expect(text).toContain("PLEASE IMPLEMENT THIS APPROVED PLAN (revision 1, plan plan:a):"); expect(text).toContain("3. Flip the flag"); expect(text).toContain("compacted its context"); }), ), ); + it.effect("implements the approved revision, not an earlier one", () => + runWithService({ sessionStatus: "running", compactionAt: null }, ({ service, dispatched }) => + Effect.gen(function* () { + const document = yield* capturePlan(service, "plan:a"); + yield* capturePlan(service, "plan:b", `${PLAN}\n4. Announce it`); + const result = yield* service.submit({ + documentId: document.documentId, + decision: "approved", + globalComment: "", + editedMarkdown: null, + actorUserId: reviewerId, + actorLabel: "Tushar", + }); + + // Earlier revisions are still in the agent's context, so the approved + // one is named and repeated in full. + expect(result.resentPlan).toBe(true); + const commands = yield* Ref.get(dispatched); + const text = turnText(commands); + expect(text).toContain("PLEASE IMPLEMENT THIS APPROVED PLAN (revision 2, plan plan:b):"); + expect(text).toContain("4. Announce it"); + expect(text).toContain("only this revision is approved"); + const turn = commands.find((command) => command.type === "thread.turn.start"); + expect(turn?.type === "thread.turn.start" && turn.sourceProposedPlan?.planId).toBe( + "plan:b", + ); + }), + ), + ); + it.effect("re-sends the plan when no provider session is bound", () => runWithService({ sessionStatus: null, compactionAt: null }, ({ service }) => Effect.gen(function* () { @@ -495,7 +525,7 @@ describe("PlanReviewService approval", () => { }); const text = turnText(yield* Ref.get(dispatched)); - expect(text).toContain("Plan approved with reviewer edits."); + expect(text).toContain("Plan approved with reviewer edits (revision 2, plan plan:a)."); expect(text).toContain("1. Add the migration\n2. Backfill the rows"); expect(text).toContain("3. Flip the flag behind a kill switch"); expect(text).not.toContain("```diff"); @@ -534,7 +564,9 @@ describe("PlanReviewService approval", () => { const text = turnText(yield* Ref.get(dispatched)); // T3-CUSTOM(expbkt3): approving with comments open means "start now, and // fold these in" — so the prompt must not also say "exactly as written". - expect(text).toContain("Plan approved — start implementing it now."); + expect(text).toContain( + "Plan approved (revision 1, plan plan:a) — start implementing it now.", + ); expect(text).toContain("apply them as you implement"); expect(text).not.toContain("exactly as written"); expect(text).toContain("Reviewer notes:\nStart with the migration."); diff --git a/apps/server/src/planreview/PlanReviewService.ts b/apps/server/src/planreview/PlanReviewService.ts index eec89e50dcbd2..7af8634be13aa 100644 --- a/apps/server/src/planreview/PlanReviewService.ts +++ b/apps/server/src/planreview/PlanReviewService.ts @@ -315,7 +315,8 @@ export const make = Effect.gen(function* () { }, createdAt: implementedAt, }); - }), + // One plan that cannot be closed must not keep the rest open. + }).pipe(Effect.ignore), { discard: true }, ); }).pipe(Effect.ignore); @@ -833,12 +834,14 @@ export const make = Effect.gen(function* () { .sort() .at(-1) ?? null; + const agentRevisions = snapshot.versions.filter(({ authorKind }) => authorKind === "agent"); const resend = decidePlanResend({ latestCompactionAt, planCreatedAt: agentBaseline.createdAt, planThreadId: document.threadId, targetThreadId: document.threadId, providerSessionStatus: thread.session?.status ?? null, + agentRevisionCount: agentRevisions.length, }); resentPlan = resend.shouldResend; @@ -851,6 +854,8 @@ export const make = Effect.gen(function* () { hasReviewerEdits || resend.shouldResend ? approvedVersion.contentMarkdown : null, fullPlanReason: hasReviewerEdits ? null : resend.reason, wasEdited: hasReviewerEdits, + revision: approvedVersion.revision, + planId: approvedVersion.sourcePlanId ?? agentBaseline.sourcePlanId, }); } else { prompt = buildPlanReviewFeedbackPrompt({ diff --git a/apps/server/src/wsForkHandlers.claudeAccounts.test.ts b/apps/server/src/wsForkHandlers.claudeAccounts.test.ts index 17ba9ee0cf70b..9373f4fe2bc4c 100644 --- a/apps/server/src/wsForkHandlers.claudeAccounts.test.ts +++ b/apps/server/src/wsForkHandlers.claudeAccounts.test.ts @@ -1,8 +1,15 @@ /** - * T3-CUSTOM(expbkt3): the admin gate on Claude account access RPCs, and the - * per-connection identity the account handlers hand the service. + * T3-CUSTOM(expbkt3): the admin gate on Claude account access RPCs, the + * per-connection identity the account handlers hand the service, and the + * thread access check in front of leaving the revive queue. */ -import { ClaudeAccountsError, ThreadId, UserId, WS_FORK_METHODS } from "@t3tools/contracts"; +import { + ClaudeAccountsError, + OrchestrationGetSnapshotError, + ThreadId, + UserId, + WS_FORK_METHODS, +} from "@t3tools/contracts"; import { describe, expect, it } from "@effect/vitest"; import * as Effect from "effect/Effect"; import * as Stream from "effect/Stream"; @@ -17,6 +24,7 @@ function makeHandlers( actorUserId: UserId | null, actorIsAdmin: boolean, directory: ReadonlyArray | null = [member], + threadAccess: "draft" | "allowed" | "denied" = "draft", ) { const calls: Array<{ readonly method: string; readonly input: unknown }> = []; const entries = { entries: [{ profile: "agent", userIds: [member] }] }; @@ -36,6 +44,11 @@ function makeHandlers( calls.push({ method: "setThreadMode", input }); return { threadId: ThreadId.make("thread-1"), mode: { kind: "auto" } } as const; }), + leaveQueue: (threadId: unknown) => + Effect.sync(() => { + calls.push({ method: "leaveQueue", input: threadId }); + return { threadId: ThreadId.make("thread-1"), mode: { kind: "auto" } } as const; + }), watchSnapshot: (viewer: unknown) => { calls.push({ method: "watchSnapshot", input: viewer }); return Stream.empty; @@ -63,8 +76,13 @@ function makeHandlers( observeRpcEffect: (_method, effect) => effect, observeRpcStream: (_method, stream) => stream, projectionSnapshotQuery: { - getThreadShellById: () => Effect.succeedNone, + getThreadShellById: () => + threadAccess === "draft" ? Effect.succeedNone : Effect.succeedSome({}), } as unknown as ForkWsHandlerDeps["projectionSnapshotQuery"], + requireThreadAccess: () => + threadAccess === "denied" + ? Effect.fail(new OrchestrationGetSnapshotError({ message: "Thread not found." })) + : Effect.void, } satisfies Partial; return { handlers: makeForkWsHandlers(deps as unknown as ForkWsHandlerDeps), calls }; } @@ -157,4 +175,20 @@ describe("Claude account access websocket handlers", () => { ]); }), ); + + it.effect("takes a thread out of the queue only after the thread access check", () => + Effect.gen(function* () { + const threadId = ThreadId.make("thread-queued"); + const allowed = makeHandlers(member, false, [member], "allowed"); + yield* allowed.handlers[WS_FORK_METHODS.claudeAccountsLeaveQueue]({ threadId }); + expect(allowed.calls).toEqual([{ method: "leaveQueue", input: threadId }]); + + const denied = makeHandlers(outsider, false, [member], "denied"); + const error = yield* denied.handlers[WS_FORK_METHODS.claudeAccountsLeaveQueue]({ + threadId, + }).pipe(Effect.flip); + expect(error._tag === "ClaudeAccountsError" ? error.reason : error._tag).toBe("not-found"); + expect(denied.calls).toEqual([]); + }), + ); }); diff --git a/apps/server/src/wsForkHandlers.ts b/apps/server/src/wsForkHandlers.ts index 0342257b424e7..107a24f1a42ad 100644 --- a/apps/server/src/wsForkHandlers.ts +++ b/apps/server/src/wsForkHandlers.ts @@ -298,6 +298,14 @@ export const makeForkWsHandlers = ({ ), { "rpc.aggregate": "claude-accounts" }, ), + [WS_FORK_METHODS.claudeAccountsLeaveQueue]: (input) => + observeRpcEffect( + WS_FORK_METHODS.claudeAccountsLeaveQueue, + guardClaudeAccountsThread("leaveQueue", input.threadId).pipe( + Effect.andThen(claudeAccounts.leaveQueue(input.threadId)), + ), + { "rpc.aggregate": "claude-accounts" }, + ), [WS_FORK_METHODS.subscribeClaudeAccounts]: () => observeRpcStream( WS_FORK_METHODS.subscribeClaudeAccounts, @@ -333,6 +341,7 @@ export const makeForkWsHandlers = ({ ForkWsHandlers, | typeof WS_FORK_METHODS.claudeAccountsGetThread | typeof WS_FORK_METHODS.claudeAccountsSetThreadMode + | typeof WS_FORK_METHODS.claudeAccountsLeaveQueue | typeof WS_FORK_METHODS.subscribeClaudeAccounts | typeof WS_FORK_METHODS.subscribeThreadClaudeAccount | typeof WS_FORK_METHODS.claudeAccountsAccessList diff --git a/apps/web/src/fork/claudeAccounts/ClaudeAccountPicker.tsx b/apps/web/src/fork/claudeAccounts/ClaudeAccountPicker.tsx index f7b3ce01b1c11..5c2ad209f8682 100644 --- a/apps/web/src/fork/claudeAccounts/ClaudeAccountPicker.tsx +++ b/apps/web/src/fork/claudeAccounts/ClaudeAccountPicker.tsx @@ -9,6 +9,8 @@ * Claude Accounts dashboard. Rows follow Auto's order for the next new * session (#1 = next pick). The chosen mode carries the theme's primary * colour; under Auto, the account this thread runs on gets a lighter tint. + * A thread waiting in the server's revive queue shows its place and a way + * out; picking an account continues it there at once. */ import { CLAUDE_ACCOUNT_MODE_AUTO, @@ -28,6 +30,7 @@ import { Badge } from "../../components/ui/badge"; import { Button } from "../../components/ui/button"; import { Menu, + MenuItem, MenuPopup, MenuRadioGroup, MenuRadioItem, @@ -50,8 +53,11 @@ import { AUTO_ACCOUNT_LABEL, autoRowDetail, isAutoMode, + LEAVE_QUEUE_LABEL, modeEquals, PENDING_RESTART_HINT, + QUEUE_PICK_HINT, + queueLabel, spaceLine, SWITCH_RESTARTS_SESSION_HINT, switchRestartsSession, @@ -344,7 +350,7 @@ function ClaudeAccountPickerMenu(props: { readonly hidden: boolean; }) { const { snapshot } = props; - const { account, setMode } = useThreadClaudeAccount(props.threadRef, true); + const { account, setMode, leaveQueue } = useThreadClaudeAccount(props.threadRef, true); const [open, setOpen] = useComposerMenuState(props.hidden); const [now, setNow] = useState(() => Date.now()); const [hoveredValue, setHoveredValue] = useState(null); @@ -356,7 +362,8 @@ function ClaudeAccountPickerMenu(props: { }, [open]); const view = triggerView(snapshot, account); - const tooltip = triggerTooltip(view); + const queued = queueLabel(account); + const tooltip = queued === null ? triggerTooltip(view) : `${triggerTooltip(view)} · ${queued}`; const mode = account?.mode ?? CLAUDE_ACCOUNT_MODE_AUTO; const rows = accountRows(snapshot, account, now); const auto = isAutoMode(account); @@ -406,6 +413,9 @@ function ClaudeAccountPickerMenu(props: { {account?.notice ? ( {account.notice} ) : null} + {account?.wait ? ( + {QUEUE_PICK_HINT} + ) : null} {account?.pendingRestart ? ( {PENDING_RESTART_HINT} ) : null} @@ -413,6 +423,11 @@ function ClaudeAccountPickerMenu(props: { {unavailable} ) : null} + {account?.wait ? ( + void leaveQueue()}> + {LEAVE_QUEUE_LABEL} + + ) : null} { diff --git a/apps/web/src/fork/claudeAccounts/hooks.ts b/apps/web/src/fork/claudeAccounts/hooks.ts index 2d92150b5f342..ec2391a20e06f 100644 --- a/apps/web/src/fork/claudeAccounts/hooks.ts +++ b/apps/web/src/fork/claudeAccounts/hooks.ts @@ -148,6 +148,8 @@ export interface ThreadClaudeAccountState { /** Null until anything is known, which reads as Auto. */ readonly account: ThreadClaudeAccount | null; readonly setMode: (mode: ClaudeAccountMode) => Promise; + /** Takes the thread out of the revive queue; the thread stream then shows it. */ + readonly leaveQueue: () => Promise; } /** The thread's account choice and its setter. `enabled: false` mounts no streams. */ @@ -174,6 +176,7 @@ export function useThreadClaudeAccount( key === null ? undefined : state.choices[key], ); const write = useSetThreadClaudeAccountMode(); + const leave = useAtomCommand(claudeAccountsEnvironment.leaveQueue); const isServerThread = target !== null; // A draft pick the server refused before the thread existed. @@ -191,6 +194,14 @@ export function useThreadClaudeAccount( [isServerThread, server, threadRef, write], ); + const leaveQueue = useCallback(async () => { + if (threadRef === null || !isServerThread) return; + await leave({ + environmentId: threadRef.environmentId, + input: { threadId: threadRef.threadId }, + }); + }, [isServerThread, leave, threadRef]); + let account = server; if ( threadRef !== null && @@ -200,7 +211,7 @@ export function useThreadClaudeAccount( const base = local.result ?? server; account = { ...base, threadId: threadRef.threadId, mode: local.mode }; } - return { account, setMode }; + return { account, setMode, leaveQueue }; } export interface ClaudeAccountAccessState { diff --git a/apps/web/src/fork/claudeAccounts/model.test.ts b/apps/web/src/fork/claudeAccounts/model.test.ts index c7c620c1308ea..7618a82b2f874 100644 --- a/apps/web/src/fork/claudeAccounts/model.test.ts +++ b/apps/web/src/fork/claudeAccounts/model.test.ts @@ -12,6 +12,7 @@ import { autoRowDetail, nextPick, orderAccounts, + queueLabel, recoversIn, resetsIn, ringTone, @@ -373,3 +374,31 @@ describe("ringTone", () => { expect(ringTone(100)).toBe("bad"); }); }); + +describe("queueLabel", () => { + it("names the thread's place in the revive queue, and nothing when it is not waiting", () => { + expect(queueLabel(null)).toBeNull(); + expect(queueLabel(thread({ resolvedProfile: "sam" }))).toBeNull(); + expect( + queueLabel( + thread({ + resolvedProfile: "sam", + wait: { + position: 2, + queuedAt: "2026-10-02T07:55:00.000Z", + expectedAt: "2026-10-02T08:31:00.000Z", + expectedProfile: "agent", + wasActive: true, + }, + }), + ), + ).toBe("Waiting for an account · number 2 in the queue"); + expect( + queueLabel( + thread({ + wait: { position: 1, queuedAt: "2026-10-02T07:55:00.000Z", wasActive: false }, + }), + ), + ).toBe("No account free · number 1 in the queue"); + }); +}); diff --git a/apps/web/src/fork/claudeAccounts/model.ts b/apps/web/src/fork/claudeAccounts/model.ts index 6ee591c06ec21..9ce037003a9aa 100644 --- a/apps/web/src/fork/claudeAccounts/model.ts +++ b/apps/web/src/fork/claudeAccounts/model.ts @@ -22,6 +22,9 @@ export const AUTO_ACCOUNT_RULE = "new sessions go to the account with the most r export const SWITCH_RESTARTS_SESSION_HINT = "Switching restarts this thread's session on the next message"; export const PENDING_RESTART_HINT = "Applies after this turn"; +export const LEAVE_QUEUE_LABEL = "Leave queue"; +/** Under the queue notice: picking an account is the way to continue at once. */ +export const QUEUE_PICK_HINT = "Pick an account below to continue on it now."; /** Bar bands, as on the dashboard: only the trip lines are policy. */ export const WARN_USED_PERCENT = 60; @@ -425,6 +428,18 @@ export function autoRowDetail( : `Would pick #1 ${next} by space to reset`; } +/** + * A short label for a thread waiting in the server's revive queue, for the + * trigger's tooltip; null when it is not waiting. The full sentence is the + * thread's notice. + */ +export function queueLabel(thread: ThreadClaudeAccount | null): string | null { + const wait = thread?.wait; + if (wait === undefined) return null; + const place = `number ${wait.position} in the queue`; + return wait.wasActive ? `Waiting for an account · ${place}` : `No account free · ${place}`; +} + export function isAutoMode(thread: ThreadClaudeAccount | null): boolean { return thread === null || thread.mode.kind === "auto"; } diff --git a/apps/web/src/fork/claudeAccounts/state.ts b/apps/web/src/fork/claudeAccounts/state.ts index 2a752b5253605..b0b0127681005 100644 --- a/apps/web/src/fork/claudeAccounts/state.ts +++ b/apps/web/src/fork/claudeAccounts/state.ts @@ -44,6 +44,16 @@ export const claudeAccountsEnvironment = { key: ({ environmentId, input }) => `${environmentId}:${input.threadId}`, }, }), + /** Takes a waiting thread out of the server's revive queue without moving it. */ + leaveQueue: createEnvironmentRpcCommand(connectionAtomRuntime, { + label: "environment-data:claude-accounts:leave-queue", + tag: WS_METHODS.claudeAccountsLeaveQueue, + scheduler, + concurrency: { + mode: "serial", + key: ({ environmentId, input }) => `${environmentId}:${input.threadId}`, + }, + }), /** Every account's allow list (admins only). */ access: createEnvironmentRpcQueryAtomFamily(connectionAtomRuntime, { label: "environment-data:claude-accounts:access", diff --git a/apps/web/src/session-logic.test.ts b/apps/web/src/session-logic.test.ts index f7d61b8f65a70..ec631c2bd1a5b 100644 --- a/apps/web/src/session-logic.test.ts +++ b/apps/web/src/session-logic.test.ts @@ -239,6 +239,25 @@ describe("V2 session presentation", () => { expect(plan?.planMarkdown).toBe("Plan"); }); + it("never resurfaces a closed plan from an earlier run", () => { + const baseProjection = makeThreadProjectionFixture(); + const earlierPlan = (name: string, status: "active" | "completed" | "superseded") => ({ + id: PlanId.make(`plan-${name}`), + threadId: baseProjection.thread.id, + nodeId: NodeId.make(`node-plan-${name}`), + kind: "proposed_plan" as const, + markdown: `Plan ${name}`, + status, + runId: RunId.make("run-1"), + }); + const latest = (plans: ReadonlyArray>) => + findLatestProposedPlan({ ...baseProjection, plans: [...plans] }, RunId.make("run-2")); + + expect(latest([earlierPlan("old", "superseded"), earlierPlan("done", "completed")])).toBeNull(); + // A background wake starts a run without ending the plan that waits for review. + expect(latest([earlierPlan("waiting", "active")])?.planMarkdown).toBe("Plan waiting"); + }); + it("assigns run rollback to the turn-start message instead of a later steer", () => { const runId = RunId.make("run-steered"); const turnStartMessageId = MessageId.make("message-turn-start"); diff --git a/apps/web/src/session-logic.ts b/apps/web/src/session-logic.ts index 241b13574bdeb..eb85501e8c141 100644 --- a/apps/web/src/session-logic.ts +++ b/apps/web/src/session-logic.ts @@ -311,10 +311,18 @@ export function findLatestProposedPlan( projection: OrchestrationV2ThreadProjection | null, latestRunId: RunId | string | null | undefined, ): LatestProposedPlanState | null { + // T3-CUSTOM(expbkt3): an earlier run's plan is current only while it is still + // active. Falling back to any earlier plan resurfaced a superseded one in the + // banner and implement prompt; the server keeps at most one plan active. if (projection === null) return null; - const plans = projection.plans.filter((plan) => plan.kind === "proposed_plan"); - const candidates = latestRunId ? plans.filter((plan) => plan.runId === latestRunId) : plans; - const plan = [...(candidates.length > 0 ? candidates : plans)] + const plans = projection.plans.filter( + (plan): plan is Extract => + plan.kind === "proposed_plan", + ); + const latestRunPlans = latestRunId ? plans.filter((plan) => plan.runId === latestRunId) : []; + const candidates = + latestRunPlans.length > 0 ? latestRunPlans : plans.filter((plan) => plan.status === "active"); + const plan = candidates .toSorted( (left, right) => planItemTime(projection, left.id).localeCompare(planItemTime(projection, right.id)) || diff --git a/docs/operations/claude-account-switching.md b/docs/operations/claude-account-switching.md index fa90043fc1650..0fd36f92299fa 100644 --- a/docs/operations/claude-account-switching.md +++ b/docs/operations/claude-account-switching.md @@ -213,6 +213,28 @@ How it works when the flag is on: rejection arriving before the first turn starts on the new account is a stale copy from the stopped session; a rejection from the new account, even with the same reset instant, moves the thread again. +- **Revive queue.** When no account can take the Auto thread, its session stops and the + thread waits in `thread_claude_account` (migration `1053_ThreadClaudeAccountWait`), so a + restart keeps it. A limit that resets within five minutes is the same kind of entry, + pinned to the account it hit and due one minute after the reset. After every status + poll the server places the waiting threads again, in order: due reset waits, then + threads that were mid-turn, then the oldest. Only a real `placed` answer counts, and + each placement counts as pending for the next, so a wave of resets spreads. A thread + that was mid-turn gets the continue turn; an idle one only gets its account. The + composer shows the thread's place and when the first account it may use should have + room (from the switcher's `recoveries`). A message to the thread, a mode change or + `claudeAccounts.leaveQueue` takes it out, and so does archiving it; picking an account while it waited mid-turn + continues it there at once. A message still fails at start when nothing has room, with + the account expected to free up first. +- **Rebalance.** When an idle Auto thread's turn ends on an account that carries two or + more sessions above its `fair_share`, or that the switcher reports as not placeable, + the server asks `--place` where the thread would go now (avoiding its account). On a + `placed` answer it moves the thread and restarts its idle session, at most once an hour + per thread (the thread loses its prompt cache once). It asks only when the projection + saw the turn run and settle, when the last status shows another account with room + under its `fair_share` that the owner may use, and not again for five minutes after a + probe found nothing better. An account gives up at most one thread per status read. + Pinned threads never move. An older switcher without `fair_share` disables this. - **Hard limit, pinned thread.** Nothing moves. The thread gets the notice " is at its usage limit until HH:MM UTC — switch to Auto or another account." (account-wide) or " has no allowance left until HH:MM UTC — switch the model or account." @@ -236,8 +258,10 @@ account `~/.claude-active` resolves to (realpath; `default` for `~/.claude`), st per thread. A switcher crash or 10 s timeout logs a warning and leaves the spawn environment unchanged, except for an owner restricted by account access (below), who is never left on `~/.claude-active` blindly. When no account has headroom the start fails -with "All Claude accounts are at their limit — first resets at HH:MM UTC." (without the -time when the switcher gives none). The binary path is the setting, else `~/.local/bin/claude-autoswitch` +with "All Claude accounts are at their limit — is expected to have room at +HH:MM UTC." (an older switcher without `recoveries`: "— first resets at HH:MM UTC."; +without a time when the switcher gives none). Each such answer logs +`claude.account.place-none`. The binary path is the setting, else `~/.local/bin/claude-autoswitch` (a leading `~` expands to the server user's home, so macOS hosts work too); `T3_CLAUDE_AUTOSWITCH_BIN` overrides it for tests. The hard-limit rotation runs the same resolved binary. Calls use an argv array, never a shell diff --git a/docs/operations/expbkt3-customizations.md b/docs/operations/expbkt3-customizations.md index 9f3b776e221a1..ed4306352d648 100644 --- a/docs/operations/expbkt3-customizations.md +++ b/docs/operations/expbkt3-customizations.md @@ -39,7 +39,8 @@ the async question indicator; a read-only environment badge; the BK desktop and mobile builds; the deploy, CI and marker tooling; Claude account profiles per thread (`experimental.claudeAccountProfiles`: Auto placement through `claude-autoswitch --place`, per-thread pinning, sticky `CLAUDE_CONFIG_DIR` per spawn, hard-limit move with a -continue turn; [claude-account-switching.md](./claude-account-switching.md#per-thread-account-placement)). +continue turn, a durable revive queue when no account has room, turn-boundary +rebalancing; [claude-account-switching.md](./claude-account-switching.md#per-thread-account-placement)). **Dropped in favour of upstream's own behaviour:** diff --git a/packages/contracts/src/claudeAccounts.ts b/packages/contracts/src/claudeAccounts.ts index d16e62818f3fc..68f2c426dede6 100644 --- a/packages/contracts/src/claudeAccounts.ts +++ b/packages/contracts/src/claudeAccounts.ts @@ -15,7 +15,14 @@ import * as Effect from "effect/Effect"; import * as Schema from "effect/Schema"; -import { IsoDateTime, NonNegativeInt, ThreadId, TrimmedString, UserId } from "./baseSchemas.ts"; +import { + IsoDateTime, + NonNegativeInt, + PositiveInt, + ThreadId, + TrimmedString, + UserId, +} from "./baseSchemas.ts"; export const ClaudeAccountProfileName = Schema.String.check( Schema.isNonEmpty(), @@ -110,6 +117,22 @@ export const ClaudeAccountsSnapshot = Schema.Struct({ }); export type ClaudeAccountsSnapshot = typeof ClaudeAccountsSnapshot.Type; +/** + * A thread no account could take, waiting in the server's revive queue. The + * server continues it on the first account with room, in queue order. + */ +export const ThreadClaudeAccountWait = Schema.Struct({ + /** 1-based place in the order the server revives waiting threads. */ + position: PositiveInt, + queuedAt: IsoDateTime, + /** When the first account this thread may use is expected to have room. */ + expectedAt: Schema.optionalKey(IsoDateTime), + expectedProfile: Schema.optionalKey(ClaudeAccountProfileName), + /** A turn was running when it stopped: the revive continues it on its own. */ + wasActive: Schema.Boolean, +}); +export type ThreadClaudeAccountWait = typeof ThreadClaudeAccountWait.Type; + export const ThreadClaudeAccount = Schema.Struct({ threadId: ThreadId, mode: ClaudeAccountMode, @@ -120,6 +143,8 @@ export const ThreadClaudeAccount = Schema.Struct({ notice: Schema.optionalKey(Schema.String), /** True when a mode change waits for the running turn to finish. */ pendingRestart: Schema.optionalKey(Schema.Boolean), + /** Present while the thread waits in the revive queue. */ + wait: Schema.optionalKey(ThreadClaudeAccountWait), }); export type ThreadClaudeAccount = typeof ThreadClaudeAccount.Type; diff --git a/packages/contracts/src/orchestrationFork.ts b/packages/contracts/src/orchestrationFork.ts index 6e20871bd6c2b..3e3003316af7b 100644 --- a/packages/contracts/src/orchestrationFork.ts +++ b/packages/contracts/src/orchestrationFork.ts @@ -5,6 +5,7 @@ import { CommandId, EnvironmentId, IsoDateTime, + PlanId, ProjectId, ThreadId, TrimmedNonEmptyString, @@ -121,6 +122,15 @@ export const ForkThreadCommands = [ threadId: ThreadId, activity: OrchestrationThreadActivity, }), + // Ends a proposed plan whose review was decided outside a turn start, such as + // a discard. Only an active plan changes. + Schema.Struct({ + type: Schema.Literal("thread.proposed-plan.retire"), + commandId: CommandId, + threadId: ThreadId, + planId: PlanId, + status: Schema.Literals(["completed", "superseded"]), + }), Schema.Struct({ type: Schema.Literal("thread.member.add"), commandId: CommandId, diff --git a/packages/contracts/src/rpcFork.ts b/packages/contracts/src/rpcFork.ts index f278eb29bd61c..63a143bdea1dc 100644 --- a/packages/contracts/src/rpcFork.ts +++ b/packages/contracts/src/rpcFork.ts @@ -151,6 +151,7 @@ export const WS_FORK_METHODS = { // T3-CUSTOM(expbkt3): Claude account profiles per thread. claudeAccountsGetThread: "claudeAccounts.getThread", claudeAccountsSetThreadMode: "claudeAccounts.setThreadMode", + claudeAccountsLeaveQueue: "claudeAccounts.leaveQueue", subscribeClaudeAccounts: "subscribeClaudeAccounts", subscribeThreadClaudeAccount: "subscribeThreadClaudeAccount", // T3-CUSTOM(expbkt3): Claude account access per user (admin only). @@ -502,6 +503,13 @@ export const WsClaudeAccountsSetThreadModeRpc = Rpc.make( }, ); +/** Takes a waiting thread out of the revive queue without moving it. */ +export const WsClaudeAccountsLeaveQueueRpc = Rpc.make(WS_FORK_METHODS.claudeAccountsLeaveQueue, { + payload: ClaudeAccountsThreadInput, + success: ThreadClaudeAccount, + error: claudeAccountsError, +}); + export const WsSubscribeClaudeAccountsRpc = Rpc.make(WS_FORK_METHODS.subscribeClaudeAccounts, { payload: Schema.Struct({}), success: ClaudeAccountsSnapshot, @@ -628,6 +636,7 @@ export const FORK_WS_RPCS = [ // T3-CUSTOM(expbkt3): Claude account profiles per thread. WsClaudeAccountsGetThreadRpc, WsClaudeAccountsSetThreadModeRpc, + WsClaudeAccountsLeaveQueueRpc, WsSubscribeClaudeAccountsRpc, WsSubscribeThreadClaudeAccountRpc, // T3-CUSTOM(expbkt3): Claude account access per user. diff --git a/packages/shared/src/planReview.test.ts b/packages/shared/src/planReview.test.ts index 6cdde6175f5e1..76cdafbb3d255 100644 --- a/packages/shared/src/planReview.test.ts +++ b/packages/shared/src/planReview.test.ts @@ -262,13 +262,15 @@ describe("buildPlanReviewApprovalPrompt", () => { fullPlanMarkdown: null, fullPlanReason: null, wasEdited: false, + revision: 3, + planId: "plan-3", }; it("sends one short line when the plan is still in context", () => { const prompt = buildPlanReviewApprovalPrompt(base); expect(prompt).toBe( - "Plan approved. Implement the plan you proposed above, exactly as written.", + "Plan approved (revision 3, plan plan-3). Implement the plan you proposed above, exactly as written.", ); }); @@ -289,7 +291,7 @@ describe("buildPlanReviewApprovalPrompt", () => { wasEdited: true, }); - expect(prompt).toContain("Plan approved with reviewer edits."); + expect(prompt).toContain("Plan approved with reviewer edits (revision 3, plan plan-3)."); expect(prompt).toContain("3. Flip the flag safely"); expect(prompt).not.toContain("```diff"); expect(prompt.split("# Auth rewrite")).toHaveLength(2); @@ -334,7 +336,9 @@ describe("buildPlanReviewApprovalPrompt", () => { it("tells the agent to implement now and apply the refinements", () => { const prompt = buildPlanReviewApprovalPrompt(withComment); - expect(prompt).toContain("Plan approved — start implementing it now."); + expect(prompt).toContain( + "Plan approved (revision 3, plan plan-3) — start implementing it now.", + ); expect(prompt).toContain("Treat them as"); expect(prompt).toContain("apply them as you implement"); expect(prompt).toContain("Do not go back"); @@ -348,7 +352,9 @@ describe("buildPlanReviewApprovalPrompt", () => { it("treats overall notes alone as refinements too", () => { const prompt = buildPlanReviewApprovalPrompt({ ...base, notes: "Ship behind a flag." }); - expect(prompt).toContain("Plan approved — start implementing it now."); + expect(prompt).toContain( + "Plan approved (revision 3, plan plan-3) — start implementing it now.", + ); expect(prompt).toContain("Reviewer notes:\nShip behind a flag."); expect(prompt).not.toContain("exactly as written"); }); @@ -357,7 +363,7 @@ describe("buildPlanReviewApprovalPrompt", () => { const prompt = buildPlanReviewApprovalPrompt(base); expect(prompt).toBe( - "Plan approved. Implement the plan you proposed above, exactly as written.", + "Plan approved (revision 3, plan plan-3). Implement the plan you proposed above, exactly as written.", ); }); @@ -368,12 +374,20 @@ describe("buildPlanReviewApprovalPrompt", () => { fullPlanReason: "this session compacted its context after the plan was written", }); - expect(prompt).toContain("PLEASE IMPLEMENT THIS APPROVED PLAN:"); + expect(prompt).toContain("PLEASE IMPLEMENT THIS APPROVED PLAN (revision 3, plan plan-3):"); expect(prompt).toContain("apply them as you implement"); expect(prompt).toContain("Batch this."); }); }); + it("names the approved revision even without a source plan", () => { + const prompt = buildPlanReviewApprovalPrompt({ ...base, revision: 4, planId: null }); + + expect(prompt).toBe( + "Plan approved (revision 4). Implement the plan you proposed above, exactly as written.", + ); + }); + it("repeats the plan and explains why when the policy demands it", () => { const prompt = buildPlanReviewApprovalPrompt({ ...base, @@ -381,7 +395,7 @@ describe("buildPlanReviewApprovalPrompt", () => { fullPlanReason: "this session compacted its context after the plan was written", }); - expect(prompt).toContain("PLEASE IMPLEMENT THIS APPROVED PLAN:"); + expect(prompt).toContain("PLEASE IMPLEMENT THIS APPROVED PLAN (revision 3, plan plan-3):"); expect(prompt).toContain("3. Flip the flag"); expect(prompt).toContain("(The full plan is repeated because this session compacted"); }); diff --git a/packages/shared/src/planReview.ts b/packages/shared/src/planReview.ts index 8423da33ef649..8d396feaefd7c 100644 --- a/packages/shared/src/planReview.ts +++ b/packages/shared/src/planReview.ts @@ -49,6 +49,11 @@ export interface PlanReviewApprovalInput { readonly fullPlanMarkdown: string | null; readonly fullPlanReason: string | null; readonly wasEdited: boolean; + /** The approved document revision, named in the prompt so the agent and the + * transcript both record exactly which version was approved. */ + readonly revision: number; + /** The proposed plan the approved revision came from, when one exists. */ + readonly planId: string | null; } /** @@ -226,12 +231,14 @@ export function buildPlanReviewApprovalPrompt(input: PlanReviewApprovalInput): s * a list of changes contradicts itself, so the instruction changes instead. */ const hasRefinements = input.comments.length > 0 || notes.length > 0; + const planLabel = input.planId === null ? "" : `, plan ${input.planId}`; + const version = `revision ${input.revision}${planLabel}`; if (input.fullPlanMarkdown !== null) { sections.push( input.wasEdited - ? "Plan approved with reviewer edits. Implement this complete revised plan:" - : "PLEASE IMPLEMENT THIS APPROVED PLAN:", + ? `Plan approved with reviewer edits (${version}). Implement this complete revised plan:` + : `PLEASE IMPLEMENT THIS APPROVED PLAN (${version}):`, ); sections.push(planReviewFence("markdown", input.fullPlanMarkdown)); if (input.fullPlanReason !== null) { @@ -240,8 +247,8 @@ export function buildPlanReviewApprovalPrompt(input: PlanReviewApprovalInput): s } else { sections.push( hasRefinements - ? "Plan approved — start implementing it now." - : "Plan approved. Implement the plan you proposed above, exactly as written.", + ? `Plan approved (${version}) — start implementing it now.` + : `Plan approved (${version}). Implement the plan you proposed above, exactly as written.`, ); }