diff --git a/apps/server/src/jones/goals/GoalControl.test.ts b/apps/server/src/jones/goals/GoalControl.test.ts index e14e313c84..e5dc468319 100644 --- a/apps/server/src/jones/goals/GoalControl.test.ts +++ b/apps/server/src/jones/goals/GoalControl.test.ts @@ -13,7 +13,9 @@ import { type OrchestrationV2ProviderGoal, type ThreadGoalInput, } from "@t3tools/contracts"; +import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; +import * as Fiber from "effect/Fiber"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; import type { ProviderAdapterV2SessionRuntime } from "@t3tools/provider-core/server/ProviderAdapter"; @@ -105,6 +107,9 @@ function fixture( let starts = 0; let active = options.active ?? false; let startVisible = !options.unknownStart; + let sessionStopped = false; + let controlCalls = 0; + let hold: { entered: Deferred.Deferred; release: Deferred.Deferred } | undefined; const startResult = () => startVisible ? { @@ -118,6 +123,13 @@ function fixture( providerSessionId: sessionId, controlGoal: (input: ProviderGoalControlInput) => Effect.gen(function* () { + controlCalls++; + const held = hold; + hold = undefined; + if (held !== undefined) { + yield* Deferred.succeed(held.entered, undefined); + yield* Deferred.await(held.release); + } if (!(yield* input.revalidate)) return { outcome: "stale" as const, goal: null, reason: "stale" }; if (input.action !== "read") { @@ -140,8 +152,9 @@ function fixture( }; }), } as ProviderAdapterV2SessionRuntime; - const thread = () => ({ - id: threadId, + // Ids prefixed "other-" stand for a second thread bound to its own copy of this session. + const thread = (id: string = threadId) => ({ + id, projectId: "project", activeProviderThreadId: provider.id, providerInstanceId: instanceId, @@ -166,8 +179,8 @@ function fixture( getThreadShell: (id) => Effect.sync( () => - (id === threadId - ? thread() + (id === threadId || id.startsWith("other-") + ? thread(id) : { ...thread(), id: callerId, @@ -182,10 +195,11 @@ function fixture( providerThreads: [options.caller === "self" ? provider : callerProvider], }) as never, ), - getProjectThreadRecords: () => Effect.sync(() => records() as never), + getProjectThreadRecords: (input) => + Effect.sync(() => ({ ...records(), thread: thread(input.threadId) }) as never), }), Layer.mock(Sessions.ProviderSessionManagerV2)({ - get: () => Effect.sync(() => Option.some(runtime)), + get: () => Effect.sync(() => (sessionStopped ? Option.none() : Option.some(runtime))), isMcpCallerAttached: (input) => Effect.sync( () => @@ -236,6 +250,18 @@ function fixture( } }, counts: () => ({ mutations: mutationCalls, starts }), + controlCalls: () => controlCalls, + stopSession: (stopped: boolean) => { + sessionStopped = stopped; + }, + holdNextControl: Effect.gen(function* () { + const next = { + entered: yield* Deferred.make(), + release: yield* Deferred.make(), + }; + hold = next; + return next; + }), resumeState: (running: boolean, visible = true) => { active = running; startVisible = visible; @@ -657,6 +683,85 @@ describe("agent goal controls", () => { expect(f.counts().mutations).toBe(0); }).pipe(Effect.provide(f.layer)); }); + it.effect("locks per thread: same-thread calls queue, other threads do not", () => { + const f = fixture(); + const other = ThreadId.make("other-thread"); + const settled = (fiber: Fiber.Fiber) => + Effect.gen(function* () { + for (let i = 0; i < 50; i++) yield* Effect.yieldNow; + return fiber.pollUnsafe() !== undefined; + }); + return Effect.gen(function* () { + const service = yield* GoalControl.GoalControl; + const held = yield* f.holdNextControl; + const slow = yield* Effect.forkChild(read(service)); + yield* Deferred.await(held.entered); + const entered = f.controlCalls(); + const sameThread = yield* Effect.forkChild(read(service)); + const otherThread = yield* Effect.forkChild( + service.execute(scope, { threadId: other, action: "read" }), + ); + expect(yield* settled(otherThread)).toBe(true); + expect(yield* settled(sameThread)).toBe(false); + expect(f.controlCalls()).toBe(entered + 1); + yield* Deferred.succeed(held.release, undefined); + yield* Fiber.join(slow); + yield* Fiber.join(sameThread); + expect(f.controlCalls()).toBe(entered + 2); + }).pipe(Effect.provide(f.layer)); + }); + it.effect("a stopped session releases its binding when the thread is next called", () => { + const f = fixture(); + return Effect.gen(function* () { + const service = yield* GoalControl.GoalControl; + const before = yield* read(service); + f.stopSession(true); + expect(yield* read(service)).toMatchObject({ reason: "session_stopped" }); + f.stopSession(false); + // Same runtime object returns, yet the old control token no longer applies. + expect(yield* mutate(service, before.controlToken!, "pause")).toMatchObject({ + outcome: "stale", + reason: "read_current_goal_before_mutation", + }); + expect(f.counts().mutations).toBe(0); + }).pipe(Effect.provide(f.layer)); + }); + it.effect("another thread's call releases a stopped session's binding", () => { + const f = fixture(); + const other = ThreadId.make("other-thread"); + return Effect.gen(function* () { + const service = yield* GoalControl.GoalControl; + const before = yield* read(service); + f.stopSession(true); + yield* service.execute(scope, { threadId: other, action: "read" }); + f.stopSession(false); + expect(yield* mutate(service, before.controlToken!, "pause")).toMatchObject({ + outcome: "stale", + reason: "read_current_goal_before_mutation", + }); + expect(f.counts().mutations).toBe(0); + }).pipe(Effect.provide(f.layer)); + }); + it.effect("keeps a released binding's request keys from becoming new mutations", () => { + const f = fixture(); + const other = ThreadId.make("other-thread"); + return Effect.gen(function* () { + const service = yield* GoalControl.GoalControl; + const before = yield* read(service); + expect(yield* mutate(service, before.controlToken!, "pause", "kept")).toMatchObject({ + outcome: "applied", + }); + f.stopSession(true); + yield* service.execute(scope, { threadId: other, action: "read" }); + f.stopSession(false); + const fresh = yield* read(service); + expect(yield* mutate(service, fresh.controlToken!, "pause", "kept")).toMatchObject({ + outcome: "stale", + reason: "goal_binding_changed", + }); + expect(f.counts().mutations).toBe(1); + }).pipe(Effect.provide(f.layer)); + }); it.effect.each(["claudeAgent", "opencode"])("%s never simulates native controls", (driver) => { const f = fixture({ driver }); return Effect.gen(function* () { diff --git a/apps/server/src/jones/goals/GoalControl.ts b/apps/server/src/jones/goals/GoalControl.ts index 88e9888377..4f18121e42 100644 --- a/apps/server/src/jones/goals/GoalControl.ts +++ b/apps/server/src/jones/goals/GoalControl.ts @@ -67,7 +67,9 @@ interface Receipt { interface Binding { readonly token: string; readonly provider: OrchestrationV2ProviderThread; - readonly runtime: ProviderAdapterV2SessionRuntime; + // Undefined once the session stopped: the runtime reference is released, but the + // receipts stay so a retried key still resolves to a stale answer, never a new mutation. + readonly runtime: ProviderAdapterV2SessionRuntime | undefined; readonly receipts: Map; uncertain: boolean; } @@ -96,311 +98,367 @@ const make = Effect.gen(function* () { const sessions = yield* Effect.serviceOption(Sessions.ProviderSessionManagerV2); const starts = yield* Effect.serviceOption(GoalStart.GoalStartDispatch); const crypto = yield* Crypto.Crypto; - const lock = yield* Semaphore.make(1); const bindings = new Map(); + // One lock per target thread, so a slow native call on one thread never queues another. + // An entry exists only while a call holds or awaits it. + const locks = new Map(); + // Release the runtime reference held for a stopped or replaced session. Receipts are kept + // so every reused clientRequestId still resolves as stale after the next read rebinds. + const retire = (key: string) => { + const binding = bindings.get(key); + if (binding === undefined) return; + if (binding.receipts.size === 0) bindings.delete(key); + else bindings.set(key, { ...binding, runtime: undefined }); + }; + // ProviderSessionManagerV2 publishes no stop event, so release stopped sessions' runtime + // references on the next call. A thread with a call in flight is left to that call. + const retireStopped = Effect.fn("GoalControl.retireStopped")(function* (currentKey: string) { + if (Option.isNone(sessions)) return; + for (const [key, binding] of bindings) { + const sessionId = binding.provider.providerSessionId; + if ( + sessionId === null || + key === currentKey || + binding.runtime === undefined || + locks.has(key) + ) + continue; + const found = yield* sessions.value + .get(sessionId, { touchActivity: false }) + .pipe(Effect.option); + if (Option.isNone(found)) continue; // lookup failed: keep the binding + // Re-check after the lookup: a call may have started or rebound this thread. + if ( + Option.getOrUndefined(found.value) !== binding.runtime && + bindings.get(key) === binding && + !locks.has(key) + ) + retire(key); + } + }); + const perThread = + (key: string) => + (effect: Effect.Effect) => + retireStopped(key).pipe( + Effect.andThen( + Effect.acquireUseRelease( + Effect.sync(() => { + let entry = locks.get(key); + if (entry === undefined) { + entry = { semaphore: Semaphore.makeUnsafe(1), holders: 0 }; + locks.set(key, entry); + } + entry.holders += 1; + return entry; + }), + (entry) => entry.semaphore.withPermits(1)(effect), + (entry) => + Effect.sync(() => { + entry.holders -= 1; + if (entry.holders === 0 && locks.get(key) === entry) locks.delete(key); + }), + ), + ), + ); const execute = (scope: Invocation.McpInvocationScope, input: ThreadGoalInput) => - lock - .withPermits(1)( - Effect.gen(function* () { - const context = yield* readThread(input.threadId, ["providerThreads", "runs"]); - const { thread } = context.projection; - const now = DateTime.formatIso(yield* DateTime.now); - const answer = ( - outcome: ThreadGoalResult["outcome"], - reason: string, - fields: Partial = {}, - ): ThreadGoalResult => ({ - threadId: thread.id, - action: input.action, - outcome, - reason, - source: "none", - goal: null, - observedAt: now, - ...fields, - }); - const mutation = input.action !== "read"; - if (mutation) { - if (scope.client?.access === "read-only") - return yield* new OrchestratorMcpFailure({ - code: "capability_denied", - message: "This MCP client has read-only access.", - }); - yield* assertLiveCaller(context); - yield* assertTargetWithinLimits(context.limits, thread); - if ( - !input.clientRequestId || - !input.controlToken || - (input.action === "replace" - ? !input.objective?.trim() - : input.objective !== undefined) - ) - return yield* new OrchestratorMcpFailure({ - code: "invalid_request", - message: - "Goal mutations require the controlToken from read and a stable clientRequestId; only replace accepts and requires objective.", - }); - } else if (input.objective !== undefined) + perThread(input.threadId ?? scope.thread?.threadId ?? "")( + Effect.gen(function* () { + const context = yield* readThread(input.threadId, ["providerThreads", "runs"]); + const { thread } = context.projection; + const now = DateTime.formatIso(yield* DateTime.now); + const answer = ( + outcome: ThreadGoalResult["outcome"], + reason: string, + fields: Partial = {}, + ): ThreadGoalResult => ({ + threadId: thread.id, + action: input.action, + outcome, + reason, + source: "none", + goal: null, + observedAt: now, + ...fields, + }); + const mutation = input.action !== "read"; + if (mutation) { + if (scope.client?.access === "read-only") return yield* new OrchestratorMcpFailure({ - code: "invalid_request", - message: "Only replace accepts objective.", + code: "capability_denied", + message: "This MCP client has read-only access.", }); - const provider = context.projection.providerThreads.find( - (p) => p.id === thread.activeProviderThreadId, - ); - if (provider?.driver === "claudeAgent") - return answer( - mutation ? "unsupported" : "observed", - "claude_sdk_controls_unavailable", - { source: "claude_transcript", goal: provider.goal ?? null }, - ); - if (provider?.driver !== "codex") - return answer("unsupported", "provider_goal_controls_unsupported"); + yield* assertLiveCaller(context); + yield* assertTargetWithinLimits(context.limits, thread); if ( - thread.archivedAt !== null || - provider.providerSessionId === null || - Option.isNone(sessions) + !input.clientRequestId || + !input.controlToken || + (input.action === "replace" ? !input.objective?.trim() : input.objective !== undefined) ) - return answer("unavailable", "session_unavailable"); - if (provider.nativeThreadRef?.driver !== "codex" || !provider.nativeThreadRef.nativeId) - return answer("unavailable", "native_binding_missing"); - const runtimeOption = yield* sessions.value - .get(provider.providerSessionId) - .pipe(Effect.mapError(unavailable)); - if (Option.isNone(runtimeOption)) return answer("unavailable", "session_stopped"); - const runtime = runtimeOption.value; - if ( - runtime.instanceId !== thread.providerInstanceId || - runtime.instanceId !== provider.providerInstanceId - ) - return answer("stale", "instance_changed"); - if (runtime.controlGoal === undefined) - return answer("unsupported", "runtime_controls_unsupported"); - const bindingKey = thread.id; - let binding = bindings.get(bindingKey); + return yield* new OrchestratorMcpFailure({ + code: "invalid_request", + message: + "Goal mutations require the controlToken from read and a stable clientRequestId; only replace accepts and requires objective.", + }); + } else if (input.objective !== undefined) + return yield* new OrchestratorMcpFailure({ + code: "invalid_request", + message: "Only replace accepts objective.", + }); + const provider = context.projection.providerThreads.find( + (p) => p.id === thread.activeProviderThreadId, + ); + if (provider?.driver === "claudeAgent") + return answer(mutation ? "unsupported" : "observed", "claude_sdk_controls_unavailable", { + source: "claude_transcript", + goal: provider.goal ?? null, + }); + if (provider?.driver !== "codex") + return answer("unsupported", "provider_goal_controls_unsupported"); + if ( + thread.archivedAt !== null || + provider.providerSessionId === null || + Option.isNone(sessions) + ) { + retire(thread.id); + return answer("unavailable", "session_unavailable"); + } + if (provider.nativeThreadRef?.driver !== "codex" || !provider.nativeThreadRef.nativeId) + return answer("unavailable", "native_binding_missing"); + const runtimeOption = yield* sessions.value + .get(provider.providerSessionId) + .pipe(Effect.mapError(unavailable)); + if (Option.isNone(runtimeOption)) { + retire(thread.id); + return answer("unavailable", "session_stopped"); + } + const runtime = runtimeOption.value; + if ( + runtime.instanceId !== thread.providerInstanceId || + runtime.instanceId !== provider.providerInstanceId + ) { + retire(thread.id); + return answer("stale", "instance_changed"); + } + if (runtime.controlGoal === undefined) + return answer("unsupported", "runtime_controls_unsupported"); + const bindingKey = thread.id; + let binding = bindings.get(bindingKey); + if ( + binding === undefined || + binding.runtime !== runtime || + !sameBinding(binding.provider, provider) + ) { + if (mutation) return answer("stale", "read_current_goal_before_mutation"); + const token = yield* crypto.randomUUIDv4().pipe(Effect.orDie); + const retain = binding !== undefined && sameConversation(binding, provider, runtime); + // Preserve rejected keys across an ownership change so durable command dedupe + // cannot attribute an old conversation's run to its replacement. + const receipts = retain + ? binding!.receipts + : new Map( + [...(binding?.receipts ?? [])].map(([key, receipt]) => [ + key, + { + fingerprint: receipt.fingerprint, + result: answer("stale", "goal_binding_changed", { + action: receipt.result.action, + controlToken: token, + }), + }, + ]), + ); + binding = { + token, + provider, + runtime, + receipts, + uncertain: retain && binding!.uncertain, + }; + bindings.set(bindingKey, binding); + } + if (mutation && input.controlToken !== binding.token) + return answer("stale", "control_token_expired"); + const fingerprint = [input.action, input.objective ?? ""].map(encodeURIComponent).join(":"); + const key = [scope.requestNamespace, input.clientRequestId ?? ""] + .map(encodeURIComponent) + .join(":"); + const prior = mutation ? binding.receipts.get(key) : undefined; + if (prior !== undefined) { + if (prior.fingerprint !== fingerprint) + return yield* new OrchestratorMcpFailure({ + code: "invalid_request", + message: "clientRequestId was already used for a different goal operation.", + }); + return { ...prior.result, controlToken: binding.token, replayed: true }; + } + if (mutation && binding.uncertain) + return answer("unknown", "prior_mutation_unresolved", { controlToken: binding.token }); + const revalidate = Effect.gen(function* () { + const fresh = yield* readThread(thread.id, ["providerThreads"]); if ( - binding === undefined || - binding.runtime !== runtime || - !sameBinding(binding.provider, provider) - ) { - if (mutation) return answer("stale", "read_current_goal_before_mutation"); - const token = yield* crypto.randomUUIDv4().pipe(Effect.orDie); - const retain = binding !== undefined && sameConversation(binding, provider, runtime); - // Preserve rejected keys across an ownership change so durable command dedupe - // cannot attribute an old conversation's run to its replacement. - const receipts = retain - ? binding!.receipts - : new Map( - [...(binding?.receipts ?? [])].map(([key, receipt]) => [ - key, - { - fingerprint: receipt.fingerprint, - result: answer("stale", "goal_binding_changed", { - action: receipt.result.action, - controlToken: token, - }), - }, - ]), - ); - binding = { - token, + fresh.projection.thread.archivedAt !== null || + fresh.projection.thread.providerInstanceId !== thread.providerInstanceId || + !sameBinding( provider, - runtime, - receipts, - uncertain: retain && binding!.uncertain, - }; - bindings.set(bindingKey, binding); - } - if (mutation && input.controlToken !== binding.token) - return answer("stale", "control_token_expired"); - const fingerprint = [input.action, input.objective ?? ""] - .map(encodeURIComponent) - .join(":"); - const key = [scope.requestNamespace, input.clientRequestId ?? ""] - .map(encodeURIComponent) - .join(":"); - const prior = mutation ? binding.receipts.get(key) : undefined; - if (prior !== undefined) { - if (prior.fingerprint !== fingerprint) - return yield* new OrchestratorMcpFailure({ - code: "invalid_request", - message: "clientRequestId was already used for a different goal operation.", - }); - return { ...prior.result, controlToken: binding.token, replayed: true }; - } - if (mutation && binding.uncertain) - return answer("unknown", "prior_mutation_unresolved", { controlToken: binding.token }); - const revalidate = Effect.gen(function* () { - const fresh = yield* readThread(thread.id, ["providerThreads"]); - if ( - fresh.projection.thread.archivedAt !== null || - fresh.projection.thread.providerInstanceId !== thread.providerInstanceId || - !sameBinding( - provider, - fresh.projection.providerThreads.find( - (p) => p.id === fresh.projection.thread.activeProviderThreadId, - ), - ) + fresh.projection.providerThreads.find( + (p) => p.id === fresh.projection.thread.activeProviderThreadId, + ), ) - return false; - if (mutation) { - yield* assertLiveCaller(fresh); - if (scope.thread !== undefined) { - const callerRecords = yield* threads.getThreadRecords(scope.thread.threadId, [ - "providerThreads", - ]); - const callerProvider = callerRecords.providerThreads.find( - (p) => p.id === callerRecords.thread.activeProviderThreadId, - ); - if ( - callerProvider === undefined || - callerProvider.providerSessionId === null || - callerProvider.providerInstanceId !== scope.thread.providerInstanceId - ) - return false; - if ( - !(yield* sessions.value.isMcpCallerAttached({ - threadId: scope.thread.threadId, - providerSessionId: callerProvider.providerSessionId, - providerInstanceId: scope.thread.providerInstanceId, - mcpCredentialId: scope.thread.providerSessionId, - })) - ) - return false; - } - yield* assertTargetWithinLimits(fresh.limits, fresh.projection.thread); + ) + return false; + if (mutation) { + yield* assertLiveCaller(fresh); + if (scope.thread !== undefined) { + const callerRecords = yield* threads.getThreadRecords(scope.thread.threadId, [ + "providerThreads", + ]); + const callerProvider = callerRecords.providerThreads.find( + (p) => p.id === callerRecords.thread.activeProviderThreadId, + ); + if ( + callerProvider === undefined || + callerProvider.providerSessionId === null || + callerProvider.providerInstanceId !== scope.thread.providerInstanceId + ) + return false; + if ( + !(yield* sessions.value.isMcpCallerAttached({ + threadId: scope.thread.threadId, + providerSessionId: callerProvider.providerSessionId, + providerInstanceId: scope.thread.providerInstanceId, + mcpCredentialId: scope.thread.providerSessionId, + })) + ) + return false; } - const current = yield* sessions.value.get(provider.providerSessionId!); - return Option.isSome(current) && current.value === runtime; - }).pipe( - Effect.orElseSucceed(() => false), - Effect.provideService(Invocation.McpInvocationContext, scope), - Effect.provideService(ThreadManagement.ThreadManagementService, threads), - ); - const observed = yield* runtime.controlGoal({ - providerThread: provider, - action: "read", - revalidate, - }); - const base = { - source: "native_rpc" as const, - goal: observed.goal, - observedAt: DateTime.formatIso(yield* DateTime.now), - controlToken: binding.token, - }; - if (observed.outcome !== "observed") - return answer(observed.outcome, observed.reason, base); - if (!mutation) { - for (const receipt of binding.receipts.values()) { - if (receipt.result.outcome !== "unknown") continue; - if (receipt.result.action === "resume") { - if (receipt.startRequestId !== undefined && Option.isSome(starts)) { - const reconciled = yield* starts.value.observe(thread.id, receipt.startRequestId); - receipt.result = { - ...receipt.result, - ...reconciled, - observedAt: base.observedAt, - }; - } - continue; - } - const confirmed = - receipt.result.action === "clear" - ? observed.goal === null - : receipt.result.action === "pause" - ? observed.goal?.status === "paused" - : receipt.result.action === "replace" && - observed.goal?.objective === receipt.objective && - observed.goal?.status === receipt.previousStatus; - if (confirmed) + yield* assertTargetWithinLimits(fresh.limits, fresh.projection.thread); + } + const current = yield* sessions.value.get(provider.providerSessionId!); + return Option.isSome(current) && current.value === runtime; + }).pipe( + Effect.orElseSucceed(() => false), + Effect.provideService(Invocation.McpInvocationContext, scope), + Effect.provideService(ThreadManagement.ThreadManagementService, threads), + ); + const observed = yield* runtime.controlGoal({ + providerThread: provider, + action: "read", + revalidate, + }); + const base = { + source: "native_rpc" as const, + goal: observed.goal, + observedAt: DateTime.formatIso(yield* DateTime.now), + controlToken: binding.token, + }; + if (observed.outcome !== "observed") return answer(observed.outcome, observed.reason, base); + if (!mutation) { + for (const receipt of binding.receipts.values()) { + if (receipt.result.outcome !== "unknown") continue; + if (receipt.result.action === "resume") { + if (receipt.startRequestId !== undefined && Option.isSome(starts)) { + const reconciled = yield* starts.value.observe(thread.id, receipt.startRequestId); receipt.result = { ...receipt.result, - outcome: "applied", - reason: "reconciled_by_native_read", - goal: observed.goal, - observedAt: now, + ...reconciled, + observedAt: base.observedAt, }; + } + continue; } - binding.uncertain = [...binding.receipts.values()].some( - (receipt) => receipt.result.outcome === "unknown", - ); - return answer( - "observed", - binding.uncertain ? "native_read_prior_mutation_unresolved" : observed.reason, - base, - ); + const confirmed = + receipt.result.action === "clear" + ? observed.goal === null + : receipt.result.action === "pause" + ? observed.goal?.status === "paused" + : receipt.result.action === "replace" && + observed.goal?.objective === receipt.objective && + observed.goal?.status === receipt.previousStatus; + if (confirmed) + receipt.result = { + ...receipt.result, + outcome: "applied", + reason: "reconciled_by_native_read", + goal: observed.goal, + observedAt: now, + }; } - if (!(yield* revalidate)) return answer("stale", "binding_or_access_changed", base); - const active = context.projection.runs.some((run) => - ["preparing", "queued", "starting", "running", "waiting"].includes(run.status), + binding.uncertain = [...binding.receipts.values()].some( + (receipt) => receipt.result.outcome === "unknown", ); - if (input.action === "resume" && active) - return answer("busy", "run_already_active", base); - if (input.action === "resume" && observed.goal === null) - return answer("unavailable", "goal_missing", base); - const receipt: Receipt = { - fingerprint, - ...(input.action === "resume" - ? { - startRequestId: [thread.id, key].map(encodeURIComponent).join(":"), - } - : {}), - ...(input.objective === undefined ? {} : { objective: input.objective }), - ...(observed.goal === null ? {} : { previousStatus: observed.goal.status }), - result: answer("unknown", "mutation_in_progress_or_acknowledgement_lost", base), - }; - // Receipts survive caller cancellation; every rebind still expires old control tokens. - binding.receipts.set(key, receipt); - binding.uncertain = true; - if (input.action === "resume" && !active) { - if (Option.isNone(starts)) - receipt.result = answer("unavailable", "owned_start_unavailable", base); - else { - const captured = yield* starts.value.capture(thread.id).pipe(Effect.result); - if (captured._tag === "Failure") - receipt.result = answer("unavailable", "start_guard_unavailable", base); - else if (captured.success === null || !(yield* revalidate)) - receipt.result = answer("busy", "target_changed_before_start", base); - else { - const refused = yield* Ref.make(undefined); - const started = yield* starts.value - .start({ - thread, - guard: captured.success, - requestId: receipt.startRequestId!, - ...(scope.thread === undefined - ? {} - : { senderThreadId: scope.thread.threadId }), - text: "/goal resume", - }) - .pipe(Effect.provideService(DispatchModeLimit, { ...context.limits, refused })); - receipt.result = answer(started.outcome, started.reason, { - ...base, - ...(started.runId === undefined ? {} : { runId: started.runId }), - }); + return answer( + "observed", + binding.uncertain ? "native_read_prior_mutation_unresolved" : observed.reason, + base, + ); + } + if (!(yield* revalidate)) return answer("stale", "binding_or_access_changed", base); + const active = context.projection.runs.some((run) => + ["preparing", "queued", "starting", "running", "waiting"].includes(run.status), + ); + if (input.action === "resume" && active) return answer("busy", "run_already_active", base); + if (input.action === "resume" && observed.goal === null) + return answer("unavailable", "goal_missing", base); + const receipt: Receipt = { + fingerprint, + ...(input.action === "resume" + ? { + startRequestId: [thread.id, key].map(encodeURIComponent).join(":"), } + : {}), + ...(input.objective === undefined ? {} : { objective: input.objective }), + ...(observed.goal === null ? {} : { previousStatus: observed.goal.status }), + result: answer("unknown", "mutation_in_progress_or_acknowledgement_lost", base), + }; + // Receipts survive caller cancellation; every rebind still expires old control tokens. + binding.receipts.set(key, receipt); + binding.uncertain = true; + if (input.action === "resume" && !active) { + if (Option.isNone(starts)) + receipt.result = answer("unavailable", "owned_start_unavailable", base); + else { + const captured = yield* starts.value.capture(thread.id).pipe(Effect.result); + if (captured._tag === "Failure") + receipt.result = answer("unavailable", "start_guard_unavailable", base); + else if (captured.success === null || !(yield* revalidate)) + receipt.result = answer("busy", "target_changed_before_start", base); + else { + const refused = yield* Ref.make(undefined); + const started = yield* starts.value + .start({ + thread, + guard: captured.success, + requestId: receipt.startRequestId!, + ...(scope.thread === undefined ? {} : { senderThreadId: scope.thread.threadId }), + text: "/goal resume", + }) + .pipe(Effect.provideService(DispatchModeLimit, { ...context.limits, refused })); + receipt.result = answer(started.outcome, started.reason, { + ...base, + ...(started.runId === undefined ? {} : { runId: started.runId }), + }); } - } else { - const controlled = yield* runtime.controlGoal({ - providerThread: provider, - action: input.action as "pause" | "clear" | "replace", - ...(input.objective === undefined ? {} : { objective: input.objective }), - revalidate, - }); - receipt.result = answer(controlled.outcome, controlled.reason, { - ...base, - observedAt: DateTime.formatIso(yield* DateTime.now), - goal: controlled.goal, - }); } - binding.uncertain = receipt.result.outcome === "unknown"; - return receipt.result; - }), - ) - .pipe( - Effect.provideService(Invocation.McpInvocationContext, scope), - Effect.provideService(ThreadManagement.ThreadManagementService, threads), - ); + } else { + const controlled = yield* runtime.controlGoal({ + providerThread: provider, + action: input.action as "pause" | "clear" | "replace", + ...(input.objective === undefined ? {} : { objective: input.objective }), + revalidate, + }); + receipt.result = answer(controlled.outcome, controlled.reason, { + ...base, + observedAt: DateTime.formatIso(yield* DateTime.now), + goal: controlled.goal, + }); + } + binding.uncertain = receipt.result.outcome === "unknown"; + return receipt.result; + }), + ).pipe( + Effect.provideService(Invocation.McpInvocationContext, scope), + Effect.provideService(ThreadManagement.ThreadManagementService, threads), + ); return GoalControl.of({ execute }); }); export const layer = Layer.effect(GoalControl, make); diff --git a/apps/server/src/jones/threadResultDependencies/DependencyReactor.test.ts b/apps/server/src/jones/threadResultDependencies/DependencyReactor.test.ts index c474e630ef..79c8d177e8 100644 --- a/apps/server/src/jones/threadResultDependencies/DependencyReactor.test.ts +++ b/apps/server/src/jones/threadResultDependencies/DependencyReactor.test.ts @@ -6,6 +6,7 @@ import * as Deferred from "effect/Deferred"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; import * as Stream from "effect/Stream"; +import * as TestClock from "effect/testing/TestClock"; import * as Orchestrator from "../../orchestration-v2/Orchestrator.ts"; import * as Forwarding from "../../peer/PeerForwarding.ts"; import { DependencyWakeCommit } from "./DependencyCommit.ts"; @@ -347,3 +348,182 @@ it.effect("the acquired domain consumer drains when its owning startup scope clo assert.isTrue(drained); }).pipe(Effect.provide(Fixture.layer)), ); + +// Let ready fibers and in-memory SQL settle without advancing the virtual clock. +const settle = Effect.forEach(Array.from({ length: 300 }), () => Effect.yieldNow, { + discard: true, +}); +const answered = { + token: "done", + state: "answered", + request: null, + admissionId: null, + observation: "receiver_working", + outcome: "fulfilled", + result: "Done", +} as never; + +it.effect("paces a peer failure that carries no delay of its own with local backoff", () => + Effect.gen(function* () { + const f = yield* Fixture.seed; + yield* prepare(f); + let calls = 0; + const peer = Layer.mock(Forwarding.PeerForwarding)({ + call: () => + Effect.suspend(() => { + calls++; + return Effect.fail( + new OrchestratorMcpFailure({ code: "provider_unavailable", message: "Down" }), + ); + }), + }); + yield* Effect.gen(function* () { + const reactor = yield* Reactor.DependencyReactor; + yield* reactor.start(); + yield* settle; + assert.equal(calls, 1); + yield* TestClock.adjust(999); + yield* settle; + assert.equal(calls, 1); + yield* TestClock.adjust(1); + yield* settle; + assert.equal(calls, 2); + yield* TestClock.adjust(1_999); + yield* settle; + assert.equal(calls, 2); + yield* TestClock.adjust(1); + yield* settle; + assert.equal(calls, 3); + }).pipe( + Effect.provide( + Reactor.layer.pipe( + Layer.provide(Watch.layer), + Layer.provide(ports(f)), + Layer.provide(peer), + Layer.provide(NodeCrypto.layer), + ), + ), + ); + }).pipe(Effect.provide(Fixture.layer)), +); + +it.effect("a follower that fails unexpectedly retries instead of waiting for another change", () => + Effect.gen(function* () { + const f = yield* Fixture.seed; + const changes = yield* Changes.DependencyChanges; + yield* prepare(f); + const subscription = yield* changes.subscribe; + let calls = 0; + const peer = Layer.mock(Forwarding.PeerForwarding)({ + call: () => + Effect.suspend(() => { + calls++; + return calls === 1 ? Effect.die("Synthetic follower defect") : Effect.succeed(answered); + }), + }); + yield* Effect.gen(function* () { + const reactor = yield* Reactor.DependencyReactor; + yield* reactor.start(); + yield* settle; + assert.equal(calls, 1); + yield* TestClock.adjust(1_000); + const delivered = yield* Stream.fromSubscription(subscription).pipe( + Stream.mapEffect(() => f.store.get("recovered-remote")), + Stream.filter((dependency) => dependency?.state === "delivered"), + Stream.runHead, + ); + assert.isTrue(Option.isSome(delivered)); + assert.equal(calls, 2); + }).pipe( + Effect.provide( + Reactor.layer.pipe( + Layer.provide(Watch.layer), + Layer.provide(ports(f)), + Layer.provide(peer), + Layer.provide(NodeCrypto.layer), + ), + ), + ); + }).pipe(Effect.provide(Fixture.layer)), +); + +it.effect("a domain-event subscriber that stops is reacquired after a delay", () => + Effect.gen(function* () { + const f = yield* Fixture.seed; + const reacquired = yield* Deferred.make(); + let acquisitions = 0; + const acquisition: Orchestrator.OrchestratorV2Shape["acquireDomainEvents"] = Effect.suspend( + () => { + acquisitions++; + if (acquisitions === 1) + return Effect.succeed( + Stream.fail( + new Orchestrator.OrchestratorDomainEventStreamError({ + cause: "synthetic stream loss", + }), + ), + ); + return Deferred.succeed(reacquired, undefined).pipe(Effect.as(Stream.never)); + }, + ); + yield* Effect.gen(function* () { + const reactor = yield* Reactor.DependencyReactor; + yield* reactor.start(); + yield* settle; + assert.equal(acquisitions, 1); + yield* TestClock.adjust(999); + yield* settle; + assert.equal(acquisitions, 1); + yield* TestClock.adjust(1); + yield* Deferred.await(reacquired); + assert.equal(acquisitions, 2); + }).pipe( + Effect.provide( + Reactor.layer.pipe( + Layer.provide(Watch.layer), + Layer.provide(ports(f, acquisition)), + Layer.provide(NodeCrypto.layer), + Layer.provide( + Layer.mock(Forwarding.PeerForwarding)({ + call: () => Effect.die("No dependency is open in this test"), + }), + ), + ), + ), + ); + }).pipe(Effect.provide(Fixture.layer)), +); + +it.effect("retention runs again while the reactor stays started", () => + Effect.gen(function* () { + const f = yield* Fixture.seed; + yield* Effect.gen(function* () { + const reactor = yield* Reactor.DependencyReactor; + yield* reactor.start(); // The startup pass finds nothing to retire. + yield* prepare(f); + yield* f.sql.withTransaction(f.store.release(f.waiter.id, "recovered-remote")); + // Older than the replay window, so the next pass may retire it. + yield* f.sql`UPDATE jones_thread_result_dependencies SET changed_sequence = -5000 + WHERE dependency_id = 'recovered-remote'`; + yield* TestClock.adjust(60 * 60 * 1_000 - 1); + yield* settle; + assert.equal((yield* f.store.get("recovered-remote"))?.state, "released"); + yield* TestClock.adjust(1); + yield* settle; + assert.isUndefined(yield* f.store.get("recovered-remote")); + }).pipe( + Effect.provide( + Reactor.layer.pipe( + Layer.provide(Watch.layer), + Layer.provide(ports(f)), + Layer.provide(NodeCrypto.layer), + Layer.provide( + Layer.mock(Forwarding.PeerForwarding)({ + call: () => Effect.die("A released dependency must not call a peer"), + }), + ), + ), + ), + ); + }).pipe(Effect.provide(Fixture.layer)), +); diff --git a/apps/server/src/jones/threadResultDependencies/DependencyReactor.ts b/apps/server/src/jones/threadResultDependencies/DependencyReactor.ts index 99200fb5f4..ddea8b4189 100644 --- a/apps/server/src/jones/threadResultDependencies/DependencyReactor.ts +++ b/apps/server/src/jones/threadResultDependencies/DependencyReactor.ts @@ -6,12 +6,15 @@ import { type ThreadId, type ThreadRequestObservation, } from "@t3tools/contracts"; +import * as Cause from "effect/Cause"; import * as Context from "effect/Context"; import * as Crypto from "effect/Crypto"; import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; +import * as Exit from "effect/Exit"; import * as Fiber from "effect/Fiber"; import * as Layer from "effect/Layer"; +import type * as Scope from "effect/Scope"; import * as Stream from "effect/Stream"; import * as SqlClient from "effect/sql/SqlClient"; import * as Orchestrator from "../../orchestration-v2/Orchestrator.ts"; @@ -26,6 +29,12 @@ import { ThreadRequestObserveTool } from "./tools.ts"; import { DependencyWakeCommit } from "./DependencyCommit.ts"; import { notification } from "./Wake.ts"; +// Local pacing for every follower, subscriber and retention retry: 1s, 2s, 4s ... capped at 60s. +// This never replaces PeerMonitor's shared per-peer delay; it only bounds how fast this module +// can loop when a failure carries no delay of its own. +const retryDelayMs = (attempt: number) => Math.min(60_000, 1_000 * 2 ** Math.min(attempt, 6)); +const PRUNE_INTERVAL_MS = 60 * 60 * 1_000; + export class DependencyReactor extends Context.Service< DependencyReactor, { @@ -74,6 +83,7 @@ const make = Effect.gen(function* () { }); const follow = Effect.fn("DependencyReactor.follow")(function* (initial: Dependency) { let token: string | undefined; + let unreachable = 0; while (true) { const current = yield* store.get(initial.dependencyId); if ( @@ -140,10 +150,14 @@ const make = Effect.gen(function* () { continue; } yield* mutate(store.observe(current.dependencyId, current.generation, "unreachable")); + // Every other failure code carries no delay of its own; never spin on it. + yield* Effect.sleep(retryDelayMs(unreachable)); + unreachable += 1; continue; } observation = observed.success; } + unreachable = 0; token = observation.token; yield* mutate( Effect.gen(function* () { @@ -164,6 +178,24 @@ const make = Effect.gen(function* () { ); } }); + // A failed follower retries itself; follow() re-reads the dependency and exits once it is + // released, delivered or replaced, so a retry never outlives its work. + const followWithRetry = Effect.fn("DependencyReactor.followWithRetry")(function* ( + dependency: Dependency, + ) { + for (let attempt = 0; ; attempt++) { + const exit = yield* Effect.exit(follow(dependency)); + if (Exit.isSuccess(exit)) return; + if (Cause.hasInterrupts(exit.cause)) return yield* Effect.interrupt; + const delayMs = retryDelayMs(attempt); + yield* Effect.logWarning("Result dependency follower failed; retrying", { + dependencyId: dependency.dependencyId, + delayMs, + cause: exit.cause, + }); + yield* Effect.sleep(delayMs); + } + }); const start = Effect.fn("DependencyReactor.start")(function* () { const scope = yield* Effect.scope; // A completed acquisition fences publication even if stream execution starts later. @@ -199,13 +231,7 @@ const make = Effect.gen(function* () { generation: dependency.generation, }; running.set(dependency.dependencyId, entry); - const fiber = yield* follow(dependency).pipe( - Effect.catchCause((cause) => - Effect.logWarning("Result dependency follower deferred until the next change", { - dependencyId: dependency.dependencyId, - cause, - }), - ), + const fiber = yield* followWithRetry(dependency).pipe( Effect.ensuring( Effect.sync(() => { if (running.get(dependency.dependencyId) === entry) @@ -221,38 +247,78 @@ const make = Effect.gen(function* () { yield* orchestrator.finalizeDelegatedTask(threadId); } }); - yield* Stream.fromSubscription(subscription).pipe( - Stream.runForEach((change) => - reconcile(change.threadIds).pipe( - Effect.catchCause((cause) => - Effect.logWarning("Dependency change reconciliation failed", { cause }), + // Run a subscriber for the life of the startup scope. If it ends or fails, back off and + // resubscribe, then reconcile everything once so changes made while it was down are seen. + // The first run uses the baseline acquired above, which fences startup. + const supervise = Effect.fn("DependencyReactor.supervise")(function* ( + name: string, + first: Effect.Effect, + resubscribe: Effect.Effect< + Effect.Effect, + Orchestrator.OrchestratorV2Error, + Scope.Scope + >, + ) { + let run = first; + for (let attempt = 0; ; attempt++) { + const exit = yield* Effect.exit(run); + if (Exit.isFailure(exit) && Cause.hasInterrupts(exit.cause)) return yield* Effect.interrupt; + const delayMs = retryDelayMs(attempt); + yield* Effect.logWarning(`Dependency ${name} subscriber stopped; resubscribing`, { + delayMs, + ...(Exit.isFailure(exit) ? { cause: exit.cause } : {}), + }); + yield* Effect.sleep(delayMs); + run = Effect.scoped( + resubscribe.pipe( + Effect.flatMap((next) => + reconcile().pipe( + Effect.catchCause((cause) => + Effect.logWarning("Dependency recovery failed", { cause }), + ), + Effect.andThen(next), + ), + ), + ), + ); + } + }); + const consumeChanges = (source: typeof subscription) => + Stream.fromSubscription(source).pipe( + Stream.runForEach((change) => + reconcile(change.threadIds).pipe( + Effect.catchCause((cause) => + Effect.logWarning("Dependency change reconciliation failed", { cause }), + ), ), ), - ), - Effect.catchCause((cause) => - Effect.logWarning("Dependency changes subscriber failed", { cause }), - ), - Effect.forkScoped, - ); - yield* domainEvents.pipe( - Stream.filter( - (event) => - event.type === "run.updated" || - event.type === "thread.unarchived" || - event.type === "thread.unsettled", - ), - Stream.runForEach((event) => - reconcile([event.threadId]).pipe( - Effect.catchCause((cause) => - Effect.logWarning("Dependency event reconciliation failed", { cause }), + ); + const consumeEvents = (source: typeof domainEvents) => + source.pipe( + Stream.filter( + (event) => + event.type === "run.updated" || + event.type === "thread.unarchived" || + event.type === "thread.unsettled", + ), + Stream.runForEach((event) => + reconcile([event.threadId]).pipe( + Effect.catchCause((cause) => + Effect.logWarning("Dependency event reconciliation failed", { cause }), + ), ), ), - ), - Effect.catchCause((cause) => - Effect.logWarning("Dependency event subscriber failed", { cause }), - ), - Effect.forkScoped, - ); + ); + yield* supervise( + "changes", + consumeChanges(subscription), + changes.subscribe.pipe(Effect.map(consumeChanges)), + ).pipe(Effect.forkScoped); + yield* supervise( + "event", + consumeEvents(domainEvents), + orchestrator.acquireDomainEvents.pipe(Effect.map(consumeEvents)), + ).pipe(Effect.forkScoped); yield* sql .withTransaction(store.prune) .pipe( @@ -261,6 +327,20 @@ const make = Effect.gen(function* () { yield* reconcile().pipe( Effect.catchCause((cause) => Effect.logWarning("Dependency recovery failed", { cause })), ); + // Retention also runs while the server stays up; the startup pass alone never repeats. + yield* Effect.sleep(PRUNE_INTERVAL_MS).pipe( + Effect.andThen( + sql + .withTransaction(store.prune) + .pipe( + Effect.catchCause((cause) => + Effect.logWarning("Dependency retention deferred", { cause }), + ), + ), + ), + Effect.forever, + Effect.forkScoped, + ); }); return DependencyReactor.of({ start }); }); diff --git a/apps/server/src/jones/updates/PlannedUpdateContinuity.test.ts b/apps/server/src/jones/updates/PlannedUpdateContinuity.test.ts index 55267414b5..fceda83948 100644 --- a/apps/server/src/jones/updates/PlannedUpdateContinuity.test.ts +++ b/apps/server/src/jones/updates/PlannedUpdateContinuity.test.ts @@ -257,3 +257,18 @@ it.effect("does not recapture a completed live conversation after a preexisting expect(yield* continuity.hasWork(state.projection.thread.id)).toBe(false); }).pipe(Effect.provide(state.layer)); }); + +it.effect( + "does not recapture a completed live conversation after a preexisting run.interrupt", + () => { + const state = harness(); + return Effect.gen(function* () { + // `run.interrupt` is the real accepted command type (packages/contracts orchestrationV2). + yield* receipt("previous-interrupt", "run.interrupt", state.projection.thread.id); + const continuity = yield* Planned.make; + yield* continuity.capture({ operationId, binding, continueRunningThreads: true }); + yield* continuity.activate(proof()); + expect(yield* continuity.hasWork(state.projection.thread.id)).toBe(false); + }).pipe(Effect.provide(state.layer)); + }, +); diff --git a/apps/server/src/jones/updates/PlannedUpdateContinuity.ts b/apps/server/src/jones/updates/PlannedUpdateContinuity.ts index 2e60b8b2f2..5641456873 100644 --- a/apps/server/src/jones/updates/PlannedUpdateContinuity.ts +++ b/apps/server/src/jones/updates/PlannedUpdateContinuity.ts @@ -194,7 +194,7 @@ export const make = Effect.gen(function* () { const latestControl = (yield* sql<{ command_type: string }>` SELECT command_type FROM orchestration_command_receipts WHERE aggregate_kind='thread' AND aggregate_id=${shell.id} AND status='accepted' - AND command_type IN ('message.dispatch','queue.resume','turn.interrupt','thread.stop','provider-session.detach') + AND command_type IN ('message.dispatch','queue.resume','run.interrupt','thread.stop','provider-session.detach') ORDER BY rowid DESC LIMIT 1`)[0]?.command_type; if ( latestControl !== undefined && diff --git a/apps/server/src/jones/updates/plannedContinuityPolicy.test.ts b/apps/server/src/jones/updates/plannedContinuityPolicy.test.ts index 4782c6bfd6..fd22a33e25 100644 --- a/apps/server/src/jones/updates/plannedContinuityPolicy.test.ts +++ b/apps/server/src/jones/updates/plannedContinuityPolicy.test.ts @@ -122,7 +122,7 @@ it("preserves preexisting holds and recognizes later Stop or hold even without n const ids = continuationRunIds(snapshot, projection); for (const command_type of [ "queue.hold", - "turn.interrupt", + "run.interrupt", "thread.runtime.stop", "thread.metadata.update", "message.dispatch", @@ -244,7 +244,7 @@ it("ignores a recovery-cancelled provider wake without ignoring a user Stop or h const snapshot = capture(recovered)!; expect(snapshot.sourceRunId).toBe(root.id); expect(canReleasePlannedQueue(snapshot, recovered)).toBe(true); - for (const command_type of ["turn.interrupt", "thread.runtime.stop", "queue.hold"]) + for (const command_type of ["run.interrupt", "thread.runtime.stop", "queue.hold"]) expect( changesPlannedControl( { command_id: "explicit-user-control", command_type },