diff --git a/apps/server/scripts/acp-mock-agent.ts b/apps/server/scripts/acp-mock-agent.ts index a97aab9cdf2b1..36c36e1d44f3f 100644 --- a/apps/server/scripts/acp-mock-agent.ts +++ b/apps/server/scripts/acp-mock-agent.ts @@ -71,6 +71,7 @@ const failLoadSessionAfterConfigReplay = process.env.T3_ACP_FAIL_LOAD_SESSION_AFTER_CONFIG_REPLAY === "1"; const emitLoadReplay = process.env.T3_ACP_EMIT_LOAD_REPLAY === "1"; const failPrompt = process.env.T3_ACP_FAIL_PROMPT === "1"; +const echoAcceptedPrompt = process.env.T3_ACP_ECHO_ACCEPTED_PROMPT === "1"; // JSON for the `data` of an "Internal error" response to `session/new`, as // Antigravity sends when its runtime cannot reach its local harness. const newSessionErrorData = @@ -868,6 +869,15 @@ const program = Effect.gen(function* () { yield* agent.handlePrompt((request) => Effect.gen(function* () { const requestedSessionId = String(request.sessionId ?? sessionId); + if (echoAcceptedPrompt) { + yield* agent.client.sessionUpdate({ + sessionId: requestedSessionId, + update: { + sessionUpdate: "user_message_chunk", + content: { type: "text", text: "Accepted test prompt" }, + }, + }); + } beginAcpMockPrompt(cancelledSessions, requestedSessionId); promptCount += 1; if (idleError === "grok-background" && promptCount === 1) { diff --git a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts index 28b1897e58d81..8ef3df8cc642b 100644 --- a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts @@ -24,6 +24,7 @@ import { ProviderDriverKind, ProviderInstanceId, ProviderSessionId, + ProviderThreadId, ThreadId, type OrchestrationV2ProviderThread, type OrchestrationV2TurnItem, @@ -763,6 +764,8 @@ const pollProtocolMethods = (events: Queue.Queue>, + followAgentWorkState = false, + drainPromptEvents = true, prepareAttachments = false, ) { const childProcessSpawner = yield* ChildProcessSpawner.ChildProcessSpawner; @@ -771,6 +774,7 @@ const makeInstructionCompactionHarness = Effect.fnUntraced(function* ( const protocolEvents = yield* Queue.unbounded(); const instanceId = ProviderInstanceId.make("acp-test-instruction-compaction"); const threadId = ThreadId.make("thread-acp-instruction-compaction"); + let rejectLoad = false; yield* (yield* McpProviderSessions.McpProviderSessions).set({ environmentId: EnvironmentId.make("environment-acp-instruction-compaction"), threadId, @@ -798,6 +802,7 @@ const makeInstructionCompactionHarness = Effect.fnUntraced(function* ( driver: ACP_TEST_DRIVER, capabilities: AcpProviderCapabilitiesV2, supportsCompaction: true, + followAgentWorkState, rewriteUserText: (text) => prepareAttachments ? Effect.succeed(text) : prepare.pipe(Effect.as(text)), ...(prepareAttachments @@ -820,6 +825,10 @@ const makeInstructionCompactionHarness = Effect.fnUntraced(function* ( Effect.sync(() => { sessionUpdate = next; }).pipe(Effect.andThen(runtime.handleSessionUpdate(next))), + loadSession: (sessionId, options) => + rejectLoad + ? Effect.fail(EffectAcpErrors.AcpRequestError.internalError("Mock load failure")) + : runtime.loadSession(sessionId, options), handleExtNotification: (method, payload, next) => Effect.sync(() => { if (method === "github.com/copilot/sessionEvent") { @@ -829,7 +838,7 @@ const makeInstructionCompactionHarness = Effect.fnUntraced(function* ( // Make compaction during prompt completion deterministic at the adapter boundary. prompt: (payload, options) => runtime.prompt(payload, options).pipe( - Effect.tap(() => runtime.drainEvents), + Effect.tap(() => (drainPromptEvents ? runtime.drainEvents : Effect.void)), // Grok's local cancel reports `cancelled` even when the agent never saw the prompt. Effect.map((result) => { if (nextPromptStopReason === undefined) return result; @@ -842,6 +851,7 @@ const makeInstructionCompactionHarness = Effect.fnUntraced(function* ( }), }, selfInvocation, + continuationRequests: { offer: () => Effect.void }, }); const runtimePolicy = ProviderAdapter.ProviderAdapterV2RuntimePolicy.make({ runtimeMode: "full-access", @@ -849,30 +859,156 @@ const makeInstructionCompactionHarness = Effect.fnUntraced(function* ( cwd: process.cwd(), }); const modelSelection = { instanceId, model: "default" } as const; - const runtime = yield* adapter.openSession({ + const open = Effect.fnUntraced(function* (initialNativeThreadId?: string) { + const scope = yield* Scope.make(); + yield* Effect.addFinalizer(() => Scope.close(scope, Exit.void)); + const runtime = yield* adapter + .openSession({ + threadId, + providerSessionId: ProviderSessionId.make("provider-session-acp-instruction-compaction"), + modelSelection, + runtimePolicy, + ...(initialNativeThreadId === undefined ? {} : { initialNativeThreadId }), + }) + .pipe(Effect.provideService(Scope.Scope, scope)); + return { runtime, scope }; + }); + let session = yield* open(); + let providerThread = yield* session.runtime.ensureThread({ threadId, - providerSessionId: ProviderSessionId.make("provider-session-acp-instruction-compaction"), modelSelection, runtimePolicy, }); - const providerThread = yield* runtime.ensureThread({ threadId, modelSelection, runtimePolicy }); const now = yield* DateTime.now; let ordinal = 0; const sessionId = "mock-session-1"; return { - runTurn: Effect.fnUntraced(function* (messageText: string, expectedStatus = "completed") { + runTurn: Effect.fnUntraced(function* ( + messageText: string, + expectedStatus = "completed", + reopen: + | boolean + | "rollback" + | "compact" + | "resume-compact-reopen" + | "resume-info-reopen" = false, + interactionMode: "default" | "plan" = "default", + ) { + const turnRuntimePolicy = { ...runtimePolicy, interactionMode }; + if (reopen === "compact") { + if (sessionUpdate === undefined) return yield* Effect.die("Missing session handler"); + const sessionId = providerThread.nativeThreadRef?.nativeId; + if (sessionId == null) return yield* Effect.die("Missing native session"); + yield* sessionUpdate({ + sessionId, + update: { + sessionUpdate: "compaction_update", + compactionId: "idle-compact", + status: "completed", + }, + }); + } + if (reopen === "rollback") { + const result = yield* session.runtime.rollbackThread({ + providerThread, + providerThreadTurns: [], + target: { + type: "thread_start", + checkpointId: CheckpointId.make("instruction-rollback"), + appRunOrdinal: 0, + }, + }); + providerThread = result.providerThread; + } + const resume = Effect.fnUntraced(function* () { + const nativeSessionId = providerThread.nativeThreadRef?.nativeId; + if (nativeSessionId == null) + return yield* Effect.die("Expected a persisted native session"); + yield* Scope.close(session.scope, Exit.void); + session = yield* open(nativeSessionId); + providerThread = yield* session.runtime.resumeThread({ + providerThread, + modelSelection, + runtimePolicy: turnRuntimePolicy, + }); + }); + if (reopen === true || reopen === "resume-compact-reopen" || reopen === "resume-info-reopen") + yield* resume(); + if (reopen === "resume-info-reopen") { + if (sessionUpdate === undefined) return yield* Effect.die("Missing session handler"); + const sessionId = providerThread.nativeThreadRef?.nativeId; + if (sessionId == null) return yield* Effect.die("Missing native session"); + yield* sessionUpdate({ + sessionId, + update: { sessionUpdate: "session_info_update", title: "Updated native title" }, + }); + const updated = yield* session.runtime.events.pipe( + Stream.filter((event) => event.type === "provider_thread.updated"), + Stream.runHead, + ); + providerThread = Option.getOrThrow(updated).providerThread; + yield* resume(); + } + if (reopen === "resume-compact-reopen") { + if (sessionUpdate === undefined) return yield* Effect.die("Missing session handler"); + const stored = providerThread; + const nativeRef = stored.nativeThreadRef; + if (nativeRef?.nativeId == null) return yield* Effect.die("Missing native session"); + const sessionId = nativeRef.nativeId; + yield* sessionUpdate({ + sessionId, + update: { + sessionUpdate: "compaction_update", + compactionId: "resumed-idle-compact", + status: "completed", + }, + }); + const pendingInvalidations = yield* session.runtime.readInstructionInvalidations!; + const updated = yield* session.runtime.events.pipe( + Stream.filter((event) => event.type === "provider_thread.updated"), + Stream.runHead, + ); + providerThread = Option.getOrThrow(updated).providerThread; + assert.isNull(providerThread.nativeMetadata?.acpInstructions); + assert.deepEqual(pendingInvalidations, [providerThread]); + yield* session.runtime.resumeThread({ + providerThread: { + ...stored, + id: ProviderThreadId.make("acp-alternate-thread"), + nativeThreadRef: { ...nativeRef, nativeId: "mock-session-2" }, + nativeMetadata: null, + }, + }); + rejectLoad = true; + const failedResume = yield* Effect.exit( + session.runtime.resumeThread({ providerThread: stored }), + ); + rejectLoad = false; + assert.isTrue(Exit.isFailure(failedResume)); + const afterFailure = yield* session.runtime.readInstructionInvalidations!; + assert.isNull( + afterFailure.find((thread) => thread.id === stored.id)?.nativeMetadata?.acpInstructions, + ); + yield* resume(); + } + const { runtime } = session; yield* runtime.startTurn( makeTurnInput({ threadId, providerThread, instanceId, - runtimePolicy, + runtimePolicy: turnRuntimePolicy, now, ordinal: ++ordinal, messageText, }), ); const terminal = yield* runtime.events.pipe( + Stream.tap((event) => + Effect.sync(() => { + if (event.type === "provider_thread.updated") providerThread = event.providerThread; + }), + ), Stream.filter((event) => event.type === "turn.terminal"), Stream.runHead, ); @@ -1105,6 +1241,32 @@ const optionValues = (question: OrchestrationV2UserInputQuestion | undefined) => describe("AcpAdapterV2", () => { beforeAll(() => bundleAgents); + it.live("retains delivered instructions across a resumed session info update", () => + Effect.gen(function* () { + const { runTurn } = yield* makeInstructionCompactionHarness({}); + yield* runTurn("Start work."); + assert.equal( + (yield* runTurn("Continue work.", "completed", "resume-info-reopen"))[0]?.text, + "Continue work.", + ); + }).pipe(Effect.provide(layerTest), Effect.scoped), + ); + it.live.each(["rollback", "compact"] as const)( + "reinjects instructions after idle %s with no continuation consumer", + (transition) => + Effect.gen(function* () { + const { runTurn } = yield* makeInstructionCompactionHarness( + {}, + false, + transition !== "rollback", + ); + assertReinjectedInstructions(yield* runTurn("Start work."), "Start work."); + assertReinjectedInstructions( + yield* runTurn("Continue after reset.", "completed", transition), + "Continue after reset.", + ); + }).pipe(Effect.provide(layerTest), Effect.scoped), + ); it.live("keeps a completed tool's status when its linked ACP terminal streams output", () => Effect.gen(function* () { const instanceId = ProviderInstanceId.make("acp-test-terminal-linked-tool"); @@ -2439,6 +2601,98 @@ describe("AcpAdapterV2", () => { }).pipe(Effect.provide(layerTest)), ); + it.live.each([false, true])( + "invalidates old instructions after a failed mode change (reopen: %s)", + (reopen) => + Effect.gen(function* () { + const { runTurn } = yield* makeInstructionCompactionHarness({ + T3_ACP_FAIL_PROMPT_ORDINAL: "2", + }); + assert.include( + (yield* runTurn("Plan first.", "completed", false, "plan"))[0]?.text ?? "", + "T3 Code interaction mode: Plan", + ); + yield* runTurn("Implement now.", "failed"); + assert.include( + (yield* runTurn("Plan again.", "completed", reopen, "plan"))[0]?.text ?? "", + "T3 Code interaction mode: Plan", + ); + }).pipe(Effect.provide(layerTest), Effect.scoped), + ); + + it.live("retries instruction delivery after a cancelled prompt before and after reopen", () => + Effect.gen(function* () { + const { runTurn } = yield* makeInstructionCompactionHarness({ + T3_ACP_VIBE_RETRY_OUTCOME: "cancelled", + }); + yield* runTurn("Start work.", "cancelled"); + assertReinjectedInstructions( + yield* runTurn("Continue after compact.", "cancelled", "compact"), + "Continue after compact.", + ); + assertReinjectedInstructions(yield* runTurn("Continue work.", "cancelled"), "Continue work."); + assertReinjectedInstructions( + yield* runTurn("Continue after reopen.", "cancelled", true), + "Continue after reopen.", + ); + }).pipe(Effect.provide(layerTest), Effect.scoped), + ); + + it.live("records instructions before ACP work-state completion ends the run", () => + Effect.gen(function* () { + const { runTurn } = yield* makeInstructionCompactionHarness( + { T3_ACP_ECHO_ACCEPTED_PROMPT: "1" }, + true, + ); + assertReinjectedInstructions(yield* runTurn("Start work."), "Start work."); + assert.equal( + (yield* runTurn("Continue work.", "completed", true))[0]?.text, + "Continue work.", + ); + }).pipe(Effect.provide(layerTest), Effect.scoped), + ); + + it.live("persists idle compaction after resume before another turn", () => + Effect.gen(function* () { + const { runTurn } = yield* makeInstructionCompactionHarness({}); + yield* runTurn("Start work."); + assertReinjectedInstructions( + yield* runTurn("Continue after idle compaction.", "completed", "resume-compact-reopen"), + "Continue after idle compaction.", + ); + }).pipe(Effect.provide(layerTest), Effect.scoped), + ); + + it.live.each(["native", "focused", "protocol"] as const)( + "retains ACP instructions across reopen but reinjects after %s compaction", + (compaction) => + Effect.gen(function* () { + const { runTurn } = yield* makeInstructionCompactionHarness( + compaction === "protocol" + ? { + T3_ACP_COMPACTION_PROMPT_ORDINALS: "2", + } + : {}, + ); + assertReinjectedInstructions(yield* runTurn("Start work."), "Start work."); + assert.equal( + (yield* runTurn("Continue work.", "completed", true))[0]?.text, + "Continue work.", + ); + yield* runTurn( + compaction === "native" + ? "/compact" + : compaction === "focused" + ? "/compact keep the plan" + : "Compact while working.", + ); + assertReinjectedInstructions( + yield* runTurn("Continue after compact.", "completed", true), + "Continue after compact.", + ); + }).pipe(Effect.provide(layerTest), Effect.scoped), + ); + it.live.each(["cancelled", "refusal", "error"] as const)( "reinjects ACP instructions after protocol compaction, during reinjection, prompt preparation, or while idle (%s)", (stopReason) => @@ -2454,6 +2708,8 @@ describe("AcpAdapterV2", () => { T3_ACP_COMPACTION_PROMPT_ORDINALS: "2,4", T3_ACP_OPEN_COMPACTION_PROMPT_ORDINALS: "13", }, + false, + true, stopReason === "error", ); const expectedStatus = @@ -7698,6 +7954,7 @@ describe("AcpAdapterV2", () => { const secondChildSessionId = "mock-child-session-finalize-window-second"; let firstSubagentStatus: "running" | "completed" = "running"; let secondSubagentStatus: "waiting" | "completed" = "waiting"; + let backgroundTurn = false; type RuntimeService = AcpSessionRuntime.AcpSessionRuntime["Service"]; let sessionUpdateHandler: | Parameters[0] @@ -7715,6 +7972,7 @@ describe("AcpAdapterV2", () => { deferFinalizeForBackgroundWork: true, enablePostSettleContinuation: true, extractSubagentUpdate: (toolCall) => { + if (!backgroundTurn) return undefined; if (toolCall.toolCallId === "tool-call-generic-1") { return { nativeTaskId: "task-finalize-window-first", @@ -7783,7 +8041,7 @@ describe("AcpAdapterV2", () => { Stream.runForEach((event) => Queue.offer(events, event)), Effect.forkScoped, ); - const providerThread = yield* runtime.ensureThread({ + let providerThread = yield* runtime.ensureThread({ threadId, modelSelection, runtimePolicy, @@ -7792,6 +8050,16 @@ describe("AcpAdapterV2", () => { yield* runtime.startTurn( makeTurnInput({ threadId, providerThread, instanceId, runtimePolicy, now }), ); + while (true) { + const event = yield* Queue.take(events); + if (event.type === "provider_thread.updated") providerThread = event.providerThread; + if (event.type === "turn.terminal") break; + } + assert.isNotNull(providerThread.nativeMetadata?.acpInstructions); + backgroundTurn = true; + yield* runtime.startTurn( + makeTurnInput({ threadId, providerThread, instanceId, runtimePolicy, now, ordinal: 2 }), + ); yield* Deferred.await(promptSettled); yield* TestClock.adjust("1 second"); assert.isDefined(sessionUpdateHandler, "session update handler must be wired"); @@ -7842,20 +8110,30 @@ describe("AcpAdapterV2", () => { rawOutput: { content: "SECOND_DONE" }, }, }); + const compactFiber = yield* sessionUpdateHandler!({ + sessionId: "mock-session-1", + update: { + sessionUpdate: "compaction_update", + compactionId: "finalization-window", + status: "completed", + }, + }).pipe(Effect.forkScoped({ startImmediately: true })); yield* Deferred.succeed(releaseFinalizationClockRead, undefined); yield* Fiber.join(adjustFiber); + yield* Fiber.join(compactFiber); yield* Effect.yieldNow; yield* Effect.yieldNow; const providerTurnId = idAllocator.derive.providerTurn({ driver: ACP_TEST_DRIVER, - nativeTurnId: acpScopedNativeId(instanceId, "mock-session-1:turn:1"), + nativeTurnId: acpScopedNativeId(instanceId, "mock-session-1:turn:2"), }); let completedSubagentUpdates = 0; let rootTerminalStatus: string | null = null; let polled = yield* Queue.poll(events); while (Option.isSome(polled)) { const event = polled.value; + if (event.type === "provider_thread.updated") providerThread = event.providerThread; if ( event.type === "turn_item.updated" && event.turnItem.type === "subagent" && @@ -7870,6 +8148,7 @@ describe("AcpAdapterV2", () => { polled = yield* Queue.poll(events); } assert.equal(rootTerminalStatus, "completed"); + assert.isNull(providerThread.nativeMetadata?.acpInstructions); assert.equal(completedSubagentUpdates, 1); assert.isFalse( yield* hasPendingBackgroundWork, diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts index 5740faf8f8757..a69d7e0fe0350 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts @@ -17,7 +17,7 @@ import { ProjectId, ProviderDriverKind, ProviderInstanceId, - type ProviderSessionId, + ProviderSessionId, RunId, ThreadId, TurnItemId, @@ -158,7 +158,7 @@ const layerPausingAttachEventSink = (pause: { ).pipe(Layer.provide(layerTestEventSink)); const layerBeforeUnchangedWriteEventSink = ( - before: Ref.Ref>, + before: Ref.Ref>, // One effect runs before each turn-item write, in order. beforeTurnItems: Ref.Ref>> | undefined, ) => @@ -445,6 +445,9 @@ const makeProviderAdapter = Effect.fnUntraced(function* ( state: Ref.Ref, options: { readonly failEventStream?: boolean; + readonly readInstructionInvalidations?: Effect.Effect< + ReadonlyArray + >; readonly normalizeThreadLoadModelSelection?: (selection: ModelSelection) => ModelSelection; readonly capabilities?: OrchestrationV2ProviderCapabilities; readonly extensionStatuses?: OrchestrationV2ProviderSession["extensionStatuses"]; @@ -538,6 +541,9 @@ const makeProviderAdapter = Effect.fnUntraced(function* ( instanceId: ProviderInstanceId.make("codex"), driver: CODEX_DRIVER, providerSessionId: input.providerSessionId, + ...(options.readInstructionInvalidations === undefined + ? {} + : { readInstructionInvalidations: options.readInstructionInvalidations }), ...(options.normalizeThreadLoadModelSelection === undefined ? {} : { normalizeThreadLoadModelSelection: options.normalizeThreadLoadModelSelection }), @@ -631,7 +637,10 @@ function layerTest(input: { readonly failReleaseEventWrites?: boolean; readonly flakyReleaseWrites?: FlakyReleaseWrites; readonly pauseAttachWrite?: Parameters[0]; - readonly beforeUnchangedWrite?: Ref.Ref>; + readonly beforeUnchangedWrite?: Ref.Ref>; + readonly readInstructionInvalidations?: Effect.Effect< + ReadonlyArray + >; readonly beforeTurnItemsWrite?: Ref.Ref>>; readonly pauseResolve?: { readonly armed: Ref.Ref; @@ -668,6 +677,9 @@ function layerTest(input: { const layerRegistry = Layer.unwrap( makeProviderAdapter(input.state, { failEventStream: input.failEventStream ?? false, + ...(input.readInstructionInvalidations === undefined + ? {} + : { readInstructionInvalidations: input.readInstructionInvalidations }), ...(input.normalizeThreadLoadModelSelection === undefined ? {} : { normalizeThreadLoadModelSelection: input.normalizeThreadLoadModelSelection }), @@ -3478,7 +3490,9 @@ it.effect.each(["a runtime whose tasks can outlive it", "server shutdown"] as co it.effect("ProviderSessionManagerV2 keeps provider updates that land while release clears", () => Effect.gen(function* () { const state = yield* Ref.make(emptyState); - const beforeUnchangedWrite = yield* Ref.make>(Effect.void); + const beforeUnchangedWrite = yield* Ref.make< + Effect.Effect + >(Effect.void); const beforeTurnItemsWrite = yield* Ref.make>>([]); const effect = Effect.gen(function* () { const manager = yield* ProviderSessionManager.ProviderSessionManagerV2; @@ -4130,6 +4144,279 @@ it.effect("ProviderSessionManagerV2 settles a request the event pump persists du }), ); +it.effect.each([ + "current", + "queued-completion", + "released-completion", + "repeated-close", + "contended-close", + "detached-close", + "failed-close", + "failed-event", + "release-after-failed-event", + "retry-after-failed-release", + "detach-after-failed-event", + "release-before-event", + "newer-run", + "newer-session", +] as const)( + "ProviderSessionManagerV2 persists instruction invalidation only for its owner (%s)", + (owner) => + Effect.gen(function* () { + const state = yield* Ref.make(emptyState); + const beforeUnchangedWrite = yield* Ref.make< + Effect.Effect + >(Effect.void); + const adapterInvalidations = yield* Ref.make>( + [], + ); + yield* Effect.gen(function* () { + const eventSink = yield* EventSink.EventSinkV2; + const idAllocator = yield* IdAllocator.IdAllocatorV2; + const manager = yield* ProviderSessionManager.ProviderSessionManagerV2; + const projectionStore = yield* ProjectionStore.ProjectionStoreV2; + const now = yield* DateTime.now; + const threadId = ThreadId.make("instruction-invalidation"); + const providerSessionId = yield* idAllocator.allocate.providerSession({ + providerInstanceId: modelSelection.instanceId, + threadId, + }); + const providerThread = { + ...makeProviderThread({ idAllocator, threadId, providerSessionId, now }), + lastRunOrdinal: 1, + nativeMetadata: { + title: "current title", + acpInstructions: { + nativeSessionId: "native-thread", + digest: "delivered-instructions", + }, + }, + }; + const stored = { + ...providerThread, + ...(owner === "newer-run" ? { lastRunOrdinal: 2 } : {}), + ...(owner === "newer-session" + ? { providerSessionId: ProviderSessionId.make("newer-session") } + : {}), + }; + yield* eventSink.write({ + events: [ + yield* makeThreadCreatedEvent({ idAllocator, threadId, now }), + { + id: yield* idAllocator.allocate.event({ threadId, providerSessionId }), + type: "provider-thread.updated", + threadId, + driver: CODEX_DRIVER, + occurredAt: now, + payload: stored, + }, + ], + }); + const before = (yield* projectionStore.getThreadRecords(threadId, ["providerThreads"])) + .providerThreads[0]!; + const runtime = yield* manager.open({ + threadId, + providerSessionId, + modelSelection, + runtimePolicy, + }); + const invalidation = { + ...providerThread, + nativeMetadata: { title: "stale title", acpInstructions: null }, + }; + if (owner === "release-before-event") { + // The adapter has recorded compaction, but the event pump has not seen it. + yield* Ref.set(adapterInvalidations, [invalidation]); + yield* manager.release({ providerSessionId, reason: "manual_shutdown" }); + const projection = yield* projectionStore.getThreadRecords(threadId, ["providerThreads"]); + assert.isNull(projection.providerThreads[0]?.nativeMetadata?.acpInstructions); + return; + } + const subscription = yield* runtime.subscribeEvents!; + const queue = (yield* Ref.get(state)).eventQueues.get(String(providerSessionId)); + assert.isDefined(queue); + const failedWrite = Effect.fail(new EventSink.EventSinkWriteError({ eventCount: 1 })); + if ( + owner === "failed-event" || + owner === "release-after-failed-event" || + owner === "retry-after-failed-release" || + owner === "detach-after-failed-event" + ) { + yield* Ref.set(beforeUnchangedWrite, failedWrite); + } + yield* Queue.offer(queue!, { + type: "provider_thread.updated", + driver: CODEX_DRIVER, + providerThread: invalidation, + }); + const completionEventId = yield* idAllocator.allocate.event({ + threadId, + providerSessionId, + }); + const restoreCompletion = eventSink.write({ + events: [ + { + id: completionEventId, + type: "provider-thread.updated", + threadId, + driver: CODEX_DRIVER, + occurredAt: now, + payload: before, + }, + ], + }); + let conflicts = 0; + let release: Fiber.Fiber | undefined; + const raceWithCleanup: Effect.Effect = Effect.gen(function* () { + conflicts++; + if (conflicts === 1) { + release = yield* manager + .release({ providerSessionId, reason: "manual_shutdown" }) + .pipe(Effect.orDie, Effect.forkChild({ startImmediately: true })); + } else { + assert.isFalse( + yield* manager.isLive(providerSessionId), + "release must enter between retries", + ); + } + yield* eventSink.write({ + events: [ + { + id: yield* idAllocator.allocate.event({ threadId, providerSessionId }), + type: "provider-thread.updated", + threadId, + driver: CODEX_DRIVER, + occurredAt: now, + payload: { + ...before, + nativeMetadata: { ...before.nativeMetadata, title: `conflict ${conflicts}` }, + }, + }, + ], + }); + if (conflicts < 4) yield* Ref.set(beforeUnchangedWrite, raceWithCleanup); + }).pipe(Effect.orDie); + yield* subscription.events.pipe( + Stream.tap(() => + Effect.gen(function* () { + if (owner === "failed-event") { + assert.isTrue(yield* manager.isLive(providerSessionId)); + assert.deepEqual( + (yield* projectionStore.getThreadRecords(threadId, ["providerThreads"])) + .providerThreads[0]?.nativeMetadata?.acpInstructions, + before.nativeMetadata?.acpInstructions, + ); + } + if (owner === "released-completion") + yield* manager.release({ providerSessionId, reason: "manual_shutdown" }); + if ( + owner === "release-after-failed-event" || + owner === "retry-after-failed-release" || + owner === "detach-after-failed-event" + ) { + const release = + owner === "detach-after-failed-event" + ? manager.detach({ providerSessionId, threadId }) + : manager.release({ providerSessionId, reason: "manual_shutdown" }); + if (owner === "retry-after-failed-release") { + yield* Ref.set(beforeUnchangedWrite, failedWrite); + assert.isTrue(Exit.isFailure(yield* Effect.exit(release))); + assert.isFalse(yield* manager.isLive(providerSessionId)); + assert.equal((yield* Ref.get(state)).closeCount, 1); + const invalidation = yield* eventSink + .stream({ + threadId, + afterSequence: yield* eventSink.latestSequence({ threadId }), + eventType: "provider-thread.updated", + }) + .pipe(Stream.runHead, Effect.forkScoped); + yield* TestClock.adjust("1 second"); + yield* Fiber.join(invalidation); + } else { + yield* release; + } + // The subscriber is still open, so it cannot have retried the write. + assert.isNull( + (yield* projectionStore.getThreadRecords(threadId, ["providerThreads"])) + .providerThreads[0]?.nativeMetadata?.acpInstructions, + ); + } + if (owner === "queued-completion" || owner === "released-completion") + yield* restoreCompletion; + if (owner === "detached-close") { + yield* restoreCompletion; + yield* manager.detach({ providerSessionId, threadId }); + } + if (owner === "failed-close") { + yield* restoreCompletion; + yield* Ref.set( + beforeUnchangedWrite, + Effect.fail(new EventSink.EventSinkWriteError({ eventCount: 1 })), + ); + } + if (owner === "contended-close") { + yield* restoreCompletion; + yield* Ref.set(beforeUnchangedWrite, raceWithCleanup); + } + }), + ), + Stream.runHead, + ); + if (release !== undefined) yield* Fiber.join(release); + if (owner === "failed-event") { + assert.isTrue(yield* manager.isLive(providerSessionId)); + assert.equal((yield* Ref.get(state)).closeCount, 0); + } + if (owner === "contended-close") assert.equal(conflicts, 4); + if (owner === "failed-close") yield* subscription.close; + if (owner === "repeated-close") { + yield* restoreCompletion; + yield* subscription.close; + } + const projection = yield* projectionStore.getThreadRecords(threadId, ["providerThreads"]); + const updated = projection.providerThreads.find( + (candidate) => candidate.id === providerThread.id, + ); + assert.deepEqual( + updated, + owner === "current" || + owner === "queued-completion" || + owner === "released-completion" || + owner === "repeated-close" || + owner === "contended-close" || + owner === "failed-close" || + owner === "failed-event" || + owner === "release-after-failed-event" || + owner === "retry-after-failed-release" || + owner === "detach-after-failed-event" + ? { + ...before, + nativeMetadata: { + ...before.nativeMetadata, + ...(owner === "contended-close" ? { title: "conflict 4" } : {}), + acpInstructions: null, + }, + } + : before, + ); + }).pipe( + Effect.provide( + layerTest({ + state, + idleTimeoutMs: 1000, + beforeUnchangedWrite, + ...(owner === "detach-after-failed-event" + ? { capabilities: ExclusiveCapabilities } + : {}), + ...(owner === "release-before-event" + ? { readInstructionInvalidations: Ref.get(adapterInvalidations) } + : {}), + }), + ), + ); + }), +); + it.effect.each(["manual_shutdown", "server_shutdown"] as const)( "ProviderSessionManagerV2 preserves questions only for server shutdown (%s)", (reason) => diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.ts index eb56df9d049ac..5990998469342 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.ts @@ -301,6 +301,7 @@ interface LiveSessionEntry { ReadonlyMap> >; readonly requestEventPermit: Semaphore.Semaphore; + readonly instructionInvalidations: Map; readonly scope: Scope.Closeable; readonly idleGeneration: number; readonly busyTurns: ReadonlySet; @@ -1082,6 +1083,7 @@ export const layerWithOptions = ( Effect.exit( writeReleasedBackgroundWorkEvents(input.releasedProviderThreads, input.releasedAt), ), + Effect.exit(flushInstructionInvalidations(input.entry, input)), ], { concurrency: 1 }, ).pipe(Effect.flatMap(Exit.asVoidAll)); @@ -2146,6 +2148,7 @@ export const layerWithOptions = ( const makeEventSubscription = Effect.fn("ProviderSessionManager.makeEventSubscription")( function* ( + providerSessionId: ProviderSessionId, subscribers: Ref.Ref< ReadonlyMap> >, @@ -2157,28 +2160,37 @@ export const layerWithOptions = ( updated.set(subscriberId, queue); return updated; }); - const close = Ref.modify(subscribers, (current) => { - if (!current.has(subscriberId)) { - return [false, current] as const; + const entry = (yield* Ref.get(sessions)).get(sessionKey(providerSessionId)); + const close = Effect.gen(function* () { + const release = Effect.gen(function* () { + const removed = yield* Ref.modify(subscribers, (current) => { + if (!current.has(subscriberId)) return [false, current] as const; + const updated = new Map(current); + updated.delete(subscriberId); + return [true, updated] as const; + }); + if (removed) { + yield* Queue.clear(queue); + yield* Queue.end(queue); + } + }); + yield* entry === undefined ? release : entry.requestEventPermit.withPermits(1)(release); + // Completion can restore its cached receipt after an idle invalidation was persisted. + // Retry outside the cleanup lock so release can stop a competing runtime writer. + if (entry !== undefined) { + yield* flushInstructionInvalidations(entry).pipe( + Effect.catch((cause) => + Effect.logWarning( + "orchestration-v2.provider-session.instruction-invalidation-failed", + { providerSessionId, cause }, + ), + ), + ); } - const updated = new Map(current); - updated.delete(subscriberId); - return [true, updated] as const; - }).pipe( - Effect.flatMap((removed) => - removed - ? Queue.clear(queue).pipe(Effect.andThen(Queue.end(queue)), Effect.asVoid) - : Effect.void, - ), - // Detached so the consumer's run failure is not ordered behind release cleanup. - Effect.andThen( - Effect.suspend(() => { - const cleanup = subscriberReleaseCleanup.get(queue); - subscriberReleaseCleanup.delete(queue); - return cleanup === undefined ? Effect.void : Effect.forkDetach(cleanup); - }), - ), - ); + const cleanup = subscriberReleaseCleanup.get(queue); + subscriberReleaseCleanup.delete(queue); + if (cleanup !== undefined) yield* Effect.forkDetach(cleanup); + }).pipe(Effect.orDie); const events = Stream.fromQueue(queue).pipe( Stream.mapEffect((signal) => signal.type === "event" @@ -2198,7 +2210,7 @@ export const layerWithOptions = ( >, ): ProviderAdapter.ProviderAdapterV2SessionRuntime => { const providerSessionId = runtime.providerSessionId; - const subscribeEvents = makeEventSubscription(eventSubscribers); + const subscribeEvents = makeEventSubscription(providerSessionId, eventSubscribers); const turnMetrics = (operation: string, model?: string) => withMetrics({ counter: providerTurnsTotal, @@ -2467,6 +2479,85 @@ export const layerWithOptions = ( ), ); + const persistInstructionInvalidation = Effect.fn( + "ProviderSessionManager.persistInstructionInvalidation", + )(function* ( + entry: LiveSessionEntry, + offered: OrchestrationV2ProviderThread, + detachedThreadId?: ThreadId, + ) { + const threadId = offered.appThreadId; + if (threadId === null || offered.nativeMetadata?.acpInstructions !== null) return; + // Idle updates outlive run subscribers; invalidate only the receipt, not newer thread state. + yield* Effect.gen(function* () { + if (entry.instructionInvalidations.get(offered.id) !== offered) return true; + const current = (yield* Ref.get(sessions)).get( + sessionKey(entry.runtime.providerSessionId), + ); + if (current !== undefined && current.runtime !== entry.runtime) return true; + if ( + !(current?.attachedThreadIds ?? entry.attachedThreadIds).has(threadId) && + (current !== undefined || detachedThreadId !== threadId) + ) { + return true; + } + const projection = yield* projectionStore.getThreadRecords(threadId, ["providerThreads"]); + const stored = projection.providerThreads.find( + (candidate) => candidate.id === offered.id, + ); + if ( + stored === undefined || + stored.providerSessionId !== entry.runtime.providerSessionId || + stored.lastRunOrdinal !== offered.lastRunOrdinal || + stored.nativeThreadRef?.nativeId !== offered.nativeThreadRef?.nativeId || + stored.nativeMetadata?.acpInstructions == null + ) + return true; + const events = yield* providerEventIngestor.normalize({ + providerSessionId: entry.runtime.providerSessionId, + providerInstanceId: entry.runtime.instanceId, + threadId, + event: { + type: "provider_thread.updated", + driver: entry.runtime.driver, + providerThread: { + ...stored, + nativeMetadata: { ...stored.nativeMetadata, acpInstructions: null }, + }, + }, + }); + return (yield* eventSink.writeIfProviderThreadUnchanged({ expected: stored, events })) + .committed; + }).pipe( + // Each comparison owns the permit; retries must let cleanup and newer receipts through. + entry.requestEventPermit.withPermits(1), + Effect.tap((committed) => (committed ? Effect.void : Effect.yieldNow)), + Effect.repeat({ until: (committed) => committed }), + ); + }); + + const flushInstructionInvalidations = Effect.fn( + "ProviderSessionManager.flushInstructionInvalidations", + )(function* (entry: LiveSessionEntry, release?: { readonly detachedThreadId?: ThreadId }) { + if (release !== undefined && entry.runtime.readInstructionInvalidations !== undefined) { + // The adapter can have invalidated a receipt before the event pump saw the update. + yield* entry.runtime.readInstructionInvalidations.pipe( + Effect.tap((invalidations) => + Effect.sync(() => { + entry.instructionInvalidations.clear(); + for (const invalidation of invalidations) { + entry.instructionInvalidations.set(invalidation.id, invalidation); + } + }), + ), + entry.requestEventPermit.withPermits(1), + ); + } + return yield* Effect.forEach([...entry.instructionInvalidations.values()], (offered) => + Effect.exit(persistInstructionInvalidation(entry, offered, release?.detachedThreadId)), + ).pipe(Effect.flatMap(Exit.asVoidAll)); + }); + const makeRetryStallWatch = (entry: LiveSessionEntry) => { const runningRetries = new Map(); const stalledRetries = new Map(); @@ -2725,6 +2816,41 @@ export const layerWithOptions = ( }); }); } + if ( + event.type === "provider_thread.updated" && + event.providerThread.nativeMetadata?.acpInstructions !== undefined + ) { + const recorded = yield* Effect.gen(function* () { + const current = (yield* Ref.get(sessions)).get( + sessionKey(entry.runtime.providerSessionId), + ); + if (current?.runtime !== entry.runtime) return false; + if (event.providerThread.nativeMetadata?.acpInstructions === null) { + entry.instructionInvalidations.set( + event.providerThread.id, + event.providerThread, + ); + } else { + entry.instructionInvalidations.delete(event.providerThread.id); + } + return true; + }).pipe(entry.requestEventPermit.withPermits(1)); + if (!recorded) return; + yield* persistInstructionInvalidation(entry, event.providerThread).pipe( + Effect.catch((cause) => + Effect.logWarning( + "orchestration-v2.provider-session.instruction-invalidation-failed", + { + providerSessionId: entry.runtime.providerSessionId, + providerThreadId: event.providerThread.id, + cause, + }, + ), + ), + ); + yield* publishToSubscribers(entry.eventSubscribers, { type: "event", event }); + return; + } // Requests and recovered background commands can arrive before a run subscribes. const threadId = sessionScopedArtifactThreadId(event); if (threadId !== undefined) { @@ -2982,6 +3108,7 @@ export const layerWithOptions = ( exposedRuntime, eventSubscribers, requestEventPermit: yield* Semaphore.make(1), + instructionInvalidations: new Map(), scope: sessionScope, idleGeneration: 0, busyTurns: new Set(), diff --git a/packages/contracts/src/orchestrationV2.ts b/packages/contracts/src/orchestrationV2.ts index 69338865e55a6..0bc2f7247eacb 100644 --- a/packages/contracts/src/orchestrationV2.ts +++ b/packages/contracts/src/orchestrationV2.ts @@ -1089,6 +1089,14 @@ export const OrchestrationV2PendingBackgroundTask = kindUnionWithFallback( export type OrchestrationV2PendingBackgroundTask = typeof OrchestrationV2PendingBackgroundTask.Type; export const OrchestrationV2ProviderThreadNativeMetadata = Schema.Struct({ + acpInstructions: Schema.optional( + Schema.NullOr( + Schema.Struct({ + nativeSessionId: TrimmedNonEmptyString, + digest: TrimmedNonEmptyString, + }), + ), + ), cursorStoreCwd: Schema.optional(TrimmedNonEmptyString), modelSelection: Schema.optional(ModelSelection), title: Schema.optional(Schema.NullOr(TrimmedNonEmptyString)), diff --git a/packages/provider-acp-registry/src/server/adapter.ts b/packages/provider-acp-registry/src/server/adapter.ts index 25656a9d808e1..ed4858f6d7fc5 100644 --- a/packages/provider-acp-registry/src/server/adapter.ts +++ b/packages/provider-acp-registry/src/server/adapter.ts @@ -1,3 +1,4 @@ +import type * as Digest from "effect/crypto/Digest"; import { extractClaudeSubagentUpdate, extractClaudeSubagentSessionEvent, @@ -351,6 +352,7 @@ export const makeAcpRegistryAdapterV2 = Effect.fn("makeAcpRegistryAdapterV2")(fu }); export type AcpRegistryAdapterV2DriverEnv = + | Digest.Digest | ChildProcessSpawner.ChildProcessSpawner | SecureRandom.SecureRandom | FileSystem.FileSystem diff --git a/packages/provider-acp/src/server/adapter.ts b/packages/provider-acp/src/server/adapter.ts index 24cde9fd967ba..e43400e9a45d4 100644 --- a/packages/provider-acp/src/server/adapter.ts +++ b/packages/provider-acp/src/server/adapter.ts @@ -47,6 +47,7 @@ import { import { FILE_HEADERS_ONLY, formatPatch, structuredPatch } from "diff"; import * as Cause from "effect/Cause"; import * as Clock from "effect/Clock"; +import * as Digest from "effect/crypto/Digest"; import * as SecureRandom from "effect/crypto/SecureRandom"; import * as DateTime from "effect/DateTime"; import * as Deferred from "effect/Deferred"; @@ -55,6 +56,7 @@ import * as Effect from "effect/Effect"; import * as Equal from "effect/Equal"; import * as Exit from "effect/Exit"; import * as FileSystem from "effect/FileSystem"; +import * as Hex from "effect/encoding/Hex"; import * as Fiber from "effect/Fiber"; import * as Option from "effect/Option"; import * as Queue from "effect/Queue"; @@ -1433,6 +1435,11 @@ interface AcpNativeBuildConfiguration { } interface ActiveAcpTurn { + instructionDelivery?: { + readonly state: T3AcpInstructionState; + readonly digest: string; + compactionMarker: symbol | undefined; + }; readonly input: ProviderAdapter.ProviderAdapterV2TurnInput; readonly providerTurnId: OrchestrationV2ProviderTurn["id"]; readonly nativeThreadId: string; @@ -1780,6 +1787,7 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( options: AcpAdapterV2Options, ) { const random = yield* SecureRandom.SecureRandom; + const crypto = yield* Digest.Digest; const fileSystem = yield* FileSystem.FileSystem; const childProcessSpawner = yield* ChildProcessSpawner.ChildProcessSpawner; const agentScope = yield* AgentScope; @@ -2060,6 +2068,7 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( const markSessionCompacted = Effect.fn("AcpAdapterV2.markSessionCompacted")(function* ( sessionId: string, compactionId?: string, + finishing?: ActiveAcpTurn, ) { if (compactionId !== undefined) { const key = `${sessionId}\u0000${compactionId}`; @@ -2073,6 +2082,24 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( yield* Ref.update(compactionMarkers, (current) => new Map(current).set(sessionId, Symbol()), ); + const context = finishing ?? (yield* Ref.get(activeTurn)); + if ( + context?.nativeThreadId === sessionId && + (finishing !== undefined || !context.finalized) + ) { + yield* invalidateInstructions(context); + } else { + if (context?.nativeThreadId === sessionId) yield* Deferred.await(context.completed); + const known = (yield* Ref.get(providerThreadByNativeSessionId)).get(sessionId); + if (known === undefined) return; + yield* invalidateInstructions({ + nativeThreadId: sessionId, + nativeMetadata: + (yield* Ref.get(nativeMetadataBySessionId)).get(sessionId) ?? + known.nativeMetadata ?? + null, + }); + } }); const advertisedCommandNames = yield* Ref.make(new Map>()); const compactsContext = @@ -2412,6 +2439,28 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( const emitProviderEvent = (event: ProviderAdapter.ProviderAdapterV2Event) => Queue.offer(events, event).pipe(Effect.asVoid); + const invalidateInstructions = Effect.fn("AcpAdapterV2.invalidateInstructions")(function* ( + context: Pick, + ) { + const metadata = { ...context.nativeMetadata, acpInstructions: null }; + context.nativeMetadata = metadata; + yield* Ref.update(nativeMetadataBySessionId, (current) => + new Map(current).set(context.nativeThreadId, metadata), + ); + const known = (yield* Ref.get(providerThreadByNativeSessionId)).get( + context.nativeThreadId, + ); + if (known === undefined) return; + const providerThread = { + ...known, + nativeMetadata: metadata, + updatedAt: yield* DateTime.now, + }; + yield* Ref.update(providerThreadByNativeSessionId, (current) => + new Map(current).set(context.nativeThreadId, providerThread), + ); + yield* emitProviderEvent({ type: "provider_thread.updated", driver, providerThread }); + }); let scheduleDeferredFinalize: (context: ActiveAcpTurn) => Effect.Effect = () => Effect.void; @@ -4244,6 +4293,7 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( projectedStatus === undefined ? toolCall.toolCallId.slice("acp-compaction:".length) : undefined, + context, ); } } else if (mcpIdentity !== undefined) { @@ -6505,6 +6555,8 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( normalized.update.status === "completed" ) { yield* markSessionCompacted(normalized.sessionId, normalized.update.compactionId); + const active = yield* Ref.get(activeTurn); + if (active === null || active.finalized) return; } yield* handleSessionUpdate(normalized); }, @@ -7295,6 +7347,12 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( yield* Ref.set(activeSelection, null); yield* Ref.set(activeInteractionMode, null); yield* Ref.set(promptInstructionStates, new Map()); + yield* Ref.update(nativeMetadataBySessionId, (current) => + new Map(current).set(candidate.sessionId, { + ...current.get(candidate.sessionId), + acpInstructions: null, + }), + ); yield* Ref.set(promptRuntimeInstructions, new Map()); yield* Ref.set(compactionMarkers, new Map()); yield* Ref.set(completedCompactionIds, new Set()); @@ -7853,14 +7911,34 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( if (settledStatus !== "failed") { yield* emitProviderRetry(context, settledStatus); } + const now = yield* DateTime.now; + const delivery = context.instructionDelivery; + if ( + settledStatus === "completed" && + delivery !== undefined && + (yield* Ref.get(compactionMarkers)).get(context.nativeThreadId) === + delivery.compactionMarker + ) { + context.nativeMetadata = { + ...context.nativeMetadata, + acpInstructions: { nativeSessionId: context.nativeThreadId, digest: delivery.digest }, + }; + yield* Ref.update(promptInstructionStates, (current) => + new Map(current).set(context.nativeThreadId, delivery.state), + ); + yield* Ref.update(compactionMarkers, (current) => { + const updated = new Map(current); + updated.delete(context.nativeThreadId); + return updated; + }); + } if ( compactsContext && isCompactCommand(context.input.message) && settledStatus === "completed" ) { - yield* markSessionCompacted(context.nativeThreadId); + yield* markSessionCompacted(context.nativeThreadId, undefined, context); } - const now = yield* DateTime.now; if ( compactsContext && isCompactionMessage(context.input.message) && @@ -7909,6 +7987,12 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( threadId: context.input.threadId, providerTurn: turn, }); + const nativeMetadata = context.nativeMetadata; + if (nativeMetadata !== null) { + yield* Ref.update(nativeMetadataBySessionId, (current) => + new Map(current).set(context.nativeThreadId, nativeMetadata), + ); + } const updatedProviderThread: OrchestrationV2ProviderThread = { ...context.input.providerThread, providerSessionId: input.providerSessionId, @@ -8220,7 +8304,27 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( } // Read delivery state after the awaits above, so a compaction that lands // while the prompt is prepared still resends the notes in this prompt. - const previousInstructionState = (yield* Ref.get(promptInstructionStates)).get(sessionId); + const instructionDigest = yield* crypto + .digest( + Digest.sha256, + new TextEncoder().encode( + t3AcpPromptWithInstructions({ prompt: "", state: instructionState }), + ), + ) + .pipe(Effect.map(Hex.encode), Effect.orDie); + const rememberedInstructions = (yield* Ref.get(nativeMetadataBySessionId)).get( + sessionId, + )?.acpInstructions; + const recordedInstructions = + rememberedInstructions === undefined + ? turnInput.providerThread.nativeMetadata?.acpInstructions + : rememberedInstructions; + const previousInstructionState = + (yield* Ref.get(promptInstructionStates)).get(sessionId) ?? + (recordedInstructions?.nativeSessionId === sessionId && + recordedInstructions.digest === instructionDigest + ? instructionState + : undefined); const compactionMarker = (yield* Ref.get(compactionMarkers)).get(sessionId); const text = t3AcpPromptWithInstructions({ prompt: messageText, @@ -8258,6 +8362,7 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( return { prompt, instructionState: text === messageText ? undefined : instructionState, + instructionDigest, runtimeInstructions: sendRuntimeInstructions ? runtimeInstructions : undefined, compactionMarker, }; @@ -8412,6 +8517,15 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( ); } const context: ActiveAcpTurn = { + ...(promptParts?.instructionState === undefined + ? {} + : { + instructionDelivery: { + state: promptParts.instructionState, + digest: promptParts.instructionDigest, + compactionMarker: promptParts.compactionMarker, + }, + }), input: turnInput, providerTurnId, nativeThreadId: requestedSessionId, @@ -8596,6 +8710,14 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( ? undefined : yield* flavor.watchPromptOutcome(requestedSessionId); const promptGeneration = yield* Ref.get(runtimeCallbackGeneration); + if (promptParts?.instructionState !== undefined) { + yield* invalidateInstructions(context); + yield* Ref.update(promptInstructionStates, (current) => { + const updated = new Map(current); + updated.delete(requestedSessionId); + return updated; + }); + } const promptOptions = { dispatched: context.promptDispatched, released: context.foregroundIdle, @@ -8626,6 +8748,12 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( // response must not flip a settled turn. Effect.tap((result) => Effect.sync(() => { + if ( + result.stopReason === "refusal" && + context.instructionDelivery !== undefined + ) { + delete context.instructionDelivery; + } if (flavor.reportsPerTurnPromptUsage === true && !context.finalized) { context.promptUsage = result.usage ?? undefined; } @@ -8662,18 +8790,6 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( updated.set(requestedSessionId, promptParts.runtimeInstructions); return updated; }); - yield* Ref.update(compactionMarkers, (current) => { - if (promptParts?.instructionState === undefined) return current; - const observedMarker = promptParts.compactionMarker; - if ( - observedMarker === undefined || - current.get(requestedSessionId) !== observedMarker - ) - return current; - const updated = new Map(current); - updated.delete(requestedSessionId); - return updated; - }); } if (context.finalized || followsAgentWorkState) return; // ACP v2 reports a failed prompt as an idle `error`, not an RPC error. @@ -8860,6 +8976,13 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( providerSession, canResumeAfterRestart: canLoadSession || canResumeSession, events: Stream.fromEffectRepeat(Queue.take(events)), + readInstructionInvalidations: Ref.get(providerThreadByNativeSessionId).pipe( + Effect.map((threads) => + [...threads.values()].filter( + (thread) => thread.nativeMetadata?.acpInstructions === null, + ), + ), + ), ...(flavor.getModelContextWindow === undefined ? {} : { @@ -8988,6 +9111,20 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( Effect.gen(function* () { yield* awaitRuntimeTeardown(); const sessionId = yield* nativeThreadId(driver, threadInput.providerThread); + const nativeMetadata = yield* Ref.modify(nativeMetadataBySessionId, (current) => { + const metadata = { + ...threadInput.providerThread.nativeMetadata, + ...current.get(sessionId), + }; + return [metadata, new Map(current).set(sessionId, metadata)] as const; + }); + yield* Ref.update(providerThreadByNativeSessionId, (current) => + new Map(current).set(sessionId, { + ...threadInput.providerThread, + providerSessionId: input.providerSessionId, + nativeMetadata, + }), + ); const previousItemIdentityVersion = itemIdentityVersion; useProviderThreadIdentity(threadInput.providerThread); const restorePreviousItemIdentity = Effect.sync(() => { @@ -9011,12 +9148,27 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( yield* Ref.set(activeInteractionMode, nextRuntimePolicy.interactionMode); } const now = yield* DateTime.now; - return { + const rememberedMetadata = (yield* Ref.get(nativeMetadataBySessionId)).get( + sessionId, + ); + const providerThread = { ...threadInput.providerThread, providerSessionId: input.providerSessionId, + ...(rememberedMetadata === undefined + ? {} + : { + nativeMetadata: { + ...threadInput.providerThread.nativeMetadata, + ...rememberedMetadata, + }, + }), status: "idle" as const, updatedAt: now, }; + yield* Ref.update(providerThreadByNativeSessionId, (current) => + new Map(current).set(sessionId, providerThread), + ); + return providerThread; }), ); }, @@ -9582,6 +9734,7 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( nativeConversationHeadRef: null, nativeMetadata: { ...rollbackInput.providerThread.nativeMetadata, + acpInstructions: null, itemIdentityVersion: 2 as const, }, status: "idle" as const, diff --git a/packages/provider-core/src/server/ProviderAdapter.ts b/packages/provider-core/src/server/ProviderAdapter.ts index 77e13b2ce0216..e7b9c3c719390 100644 --- a/packages/provider-core/src/server/ProviderAdapter.ts +++ b/packages/provider-core/src/server/ProviderAdapter.ts @@ -670,6 +670,10 @@ export interface ProviderAdapterV2SessionRuntime { readonly events: Stream.Stream; /** Manager runtimes broadcast independent subscriptions; adapters may expose only a single-consumer stream. */ readonly subscribeEvents?: Effect.Effect; + /** Instruction receipts invalidated by the adapter, including updates still queued in events. */ + readonly readInstructionInvalidations?: Effect.Effect< + ReadonlyArray + >; readonly hasPendingBackgroundWork?: Effect.Effect; /** Root-subscription probe; replacement checks opt in to subagents owned by the live process. */ readonly hasPendingBackgroundWorkForThread?: ( diff --git a/packages/provider-fx/src/server/adapter.ts b/packages/provider-fx/src/server/adapter.ts index b48293f4fb986..05fbad54a52a3 100644 --- a/packages/provider-fx/src/server/adapter.ts +++ b/packages/provider-fx/src/server/adapter.ts @@ -1,3 +1,4 @@ +import type * as Digest from "effect/crypto/Digest"; import * as HostProcess from "@t3tools/shared/HostProcess"; import { resolveSelfInvocation, type SelfInvocation } from "@t3tools/shared/nodeRuntime"; import { type OrchestrationV2ProviderCapabilities } from "@t3tools/contracts"; @@ -205,6 +206,7 @@ export const makeFxAdapterV2 = Effect.fn("makeFxAdapterV2")(function* ( }); export type FxAdapterV2DriverEnv = + | Digest.Digest | ChildProcessSpawner.ChildProcessSpawner | SecureRandom.SecureRandom | FileSystem.FileSystem diff --git a/packages/provider-grok/src/server/adapter.test.ts b/packages/provider-grok/src/server/adapter.test.ts index 74ac3e1ca441d..2587f94804a62 100644 --- a/packages/provider-grok/src/server/adapter.test.ts +++ b/packages/provider-grok/src/server/adapter.test.ts @@ -16,6 +16,7 @@ import { } from "@t3tools/contracts"; import * as HostProcess from "@t3tools/shared/HostProcess"; import { resolveSelfInvocation } from "@t3tools/shared/nodeRuntime"; +import * as Digest from "effect/crypto/Digest"; import * as SecureRandom from "effect/crypto/SecureRandom"; import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; @@ -38,6 +39,7 @@ const layerTest = Layer.mergeAll( IdAllocator.layer, McpProviderSessions.layer, SecureRandom.layerWebCrypto, + Digest.layerWebCrypto, TestProviderHost.layer().pipe(Layer.provide(NodeServices.layer)), ); diff --git a/packages/provider-grok/src/server/adapter.ts b/packages/provider-grok/src/server/adapter.ts index 0e8d9b6e04714..6a1a01ccdaa62 100644 --- a/packages/provider-grok/src/server/adapter.ts +++ b/packages/provider-grok/src/server/adapter.ts @@ -1,3 +1,4 @@ +import type * as Digest from "effect/crypto/Digest"; import { makeProviderFailure } from "@t3tools/provider-core/server/failure"; import { XAiPromptFailureText, @@ -432,6 +433,7 @@ export const makeGrokAdapterV2 = Effect.fn("makeGrokAdapterV2")(function* ( }); export type GrokAdapterV2DriverEnv = + | Digest.Digest | ChildProcessSpawner.ChildProcessSpawner | SecureRandom.SecureRandom | FileSystem.FileSystem diff --git a/packages/provider-kiro/src/server/adapter.ts b/packages/provider-kiro/src/server/adapter.ts index 67e837a35b354..2cb85c3308885 100644 --- a/packages/provider-kiro/src/server/adapter.ts +++ b/packages/provider-kiro/src/server/adapter.ts @@ -1,3 +1,4 @@ +import type * as Digest from "effect/crypto/Digest"; import * as HostProcess from "@t3tools/shared/HostProcess"; import { resolveSelfInvocation, type SelfInvocation } from "@t3tools/shared/nodeRuntime"; import { @@ -601,6 +602,7 @@ export const makeKiroAdapterV2 = Effect.fn("makeKiroAdapterV2")(function* ( }); export type KiroAdapterV2DriverEnv = + | Digest.Digest | ChildProcessSpawner.ChildProcessSpawner | SecureRandom.SecureRandom | FileSystem.FileSystem