diff --git a/apps/server/scripts/acp-mock-agent.ts b/apps/server/scripts/acp-mock-agent.ts index 342f2b0fa937d..e7d85f8b39799 100644 --- a/apps/server/scripts/acp-mock-agent.ts +++ b/apps/server/scripts/acp-mock-agent.ts @@ -64,6 +64,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 = @@ -829,6 +830,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 f63a220fb5191..53627e22298e7 100644 --- a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts @@ -668,6 +668,8 @@ const pollProtocolMethods = (events: Queue.Queue>, + followAgentWorkState = false, + drainPromptEvents = true, ) { const childProcessSpawner = yield* ChildProcessSpawner.ChildProcessSpawner; const path = yield* Path.Path; @@ -678,6 +680,9 @@ 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 sessionUpdate: + | Parameters[0] + | undefined; yield* (yield* McpProviderSessions.McpProviderSessions).set({ environmentId: EnvironmentId.make("environment-acp-instruction-compaction"), threadId, @@ -693,6 +698,7 @@ const makeInstructionCompactionHarness = Effect.fnUntraced(function* ( driver: ACP_TEST_DRIVER, capabilities: AcpProviderCapabilitiesV2, supportsCompaction: true, + followAgentWorkState, makeRuntime: makeMockRuntime({ childProcessSpawner, mockAgentPath, @@ -700,13 +706,20 @@ const makeInstructionCompactionHarness = Effect.fnUntraced(function* ( environment, wrapRuntime: (runtime) => ({ ...runtime, + handleSessionUpdate: (handler) => { + sessionUpdate = handler; + return runtime.handleSessionUpdate(handler); + }, // Make compaction during prompt completion deterministic at the adapter boundary. prompt: (payload, options) => - runtime.prompt(payload, options).pipe(Effect.tap(() => runtime.drainEvents)), + runtime + .prompt(payload, options) + .pipe(Effect.tap(() => (drainPromptEvents ? runtime.drainEvents : Effect.void))), }), }), }, selfInvocation, + continuationRequests: { offer: () => Effect.void }, }); const runtimePolicy = ProviderAdapter.ProviderAdapterV2RuntimePolicy.make({ runtimeMode: "full-access", @@ -714,28 +727,131 @@ 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; - return Effect.fnUntraced(function* (messageText: string, expectedStatus = "completed") { + return 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 sessionId = providerThread.nativeThreadRef?.nativeId; + if (sessionId == null) return yield* Effect.die("Missing native session"); + yield* sessionUpdate({ + sessionId, + update: { + sessionUpdate: "compaction_update", + compactionId: "resumed-idle-compact", + status: "completed", + }, + }); + 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); + 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, ); @@ -855,6 +971,32 @@ const optionValues = (question: OrchestrationV2UserInputQuestion | undefined) => question?.options.map((option) => option.value ?? option.label); describe("AcpAdapterV2", () => { + 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("reports per-turn usage unavailable on a continuation turn that never prompts", () => Effect.gen(function* () { const path = yield* Path.Path; @@ -2277,6 +2419,94 @@ describe("AcpAdapterV2", () => { }).pipe(Effect.provide(layerTest), Effect.scoped), ); + 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( + "acknowledges reinjected instructions on a cancelled prompt without a durable receipt", + () => + 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.", + ); + assert.equal((yield* runTurn("Continue work.", "cancelled"))[0]?.text, "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", "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" : "Compact while working."); + assertReinjectedInstructions( + yield* runTurn("Continue after compact.", "completed", true), + "Continue after compact.", + ); + }).pipe(Effect.provide(layerTest), Effect.scoped), + ); + it.live( "reinjects ACP instructions after protocol compaction, including during reinjection", () => @@ -7023,6 +7253,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] @@ -7040,6 +7271,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", @@ -7108,7 +7340,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, @@ -7117,6 +7349,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"); @@ -7167,20 +7409,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" && @@ -7195,6 +7447,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 896de803a38f6..5aced3f7facda 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts @@ -15,7 +15,7 @@ import { ProjectId, ProviderDriverKind, ProviderInstanceId, - type ProviderSessionId, + ProviderSessionId, ThreadId, } from "@t3tools/contracts"; import * as Cause from "effect/Cause"; @@ -145,7 +145,9 @@ const layerPausingAttachEventSink = (pause: { }), ).pipe(Layer.provide(layerTestEventSink)); -const layerBeforeUnchangedWriteEventSink = (before: Ref.Ref>) => +const layerBeforeUnchangedWriteEventSink = ( + before: Ref.Ref>, +) => Layer.effect( EventSink.EventSinkV2, Effect.gen(function* () { @@ -588,7 +590,7 @@ function layerTest(input: { readonly failReleaseEventWrites?: boolean; readonly flakyReleaseWrites?: FlakyReleaseWrites; readonly pauseAttachWrite?: Parameters[0]; - readonly beforeUnchangedWrite?: Ref.Ref>; + readonly beforeUnchangedWrite?: Ref.Ref>; readonly pauseResolve?: { readonly armed: Ref.Ref; readonly paused: Deferred.Deferred; @@ -2866,7 +2868,9 @@ it.effect("ProviderSessionManagerV2 keeps the roster of a runtime whose tasks ca it.effect("ProviderSessionManagerV2 keeps a provider update that lands 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 effect = Effect.gen(function* () { const manager = yield* ProviderSessionManager.ProviderSessionManagerV2; const { providerSessionId, providerThreads, writeProviderThread, rosters } = @@ -3455,6 +3459,211 @@ 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", + "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); + 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 subscription = yield* runtime.subscribeEvents!; + const queue = (yield* Ref.get(state)).eventQueues.get(String(providerSessionId)); + assert.isDefined(queue); + if (owner === "failed-event") { + yield* Ref.set( + beforeUnchangedWrite, + Effect.fail(new EventSink.EventSinkWriteError({ eventCount: 1 })), + ); + } + yield* Queue.offer(queue!, { + type: "provider_thread.updated", + driver: CODEX_DRIVER, + providerThread: { + ...providerThread, + nativeMetadata: { title: "stale title", acpInstructions: null }, + }, + }); + 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 === "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" + ? { + ...before, + nativeMetadata: { + ...before.nativeMetadata, + ...(owner === "contended-close" ? { title: "conflict 4" } : {}), + acpInstructions: null, + }, + } + : before, + ); + }).pipe(Effect.provide(layerTest({ state, idleTimeoutMs: 1000, beforeUnchangedWrite }))); + }), +); + 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 39f383666bae1..989c37e857ffd 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.ts @@ -227,6 +227,7 @@ interface LiveSessionEntry { ReadonlyMap> >; readonly requestEventPermit: Semaphore.Semaphore; + readonly instructionInvalidations: Map; readonly scope: Scope.Closeable; readonly idleGeneration: number; readonly busyTurns: ReadonlySet; @@ -1666,6 +1667,7 @@ export const layerWithOptions = ( const makeEventSubscription = Effect.fn("ProviderSessionManager.makeEventSubscription")( function* ( + providerSessionId: ProviderSessionId, subscribers: Ref.Ref< ReadonlyMap> >, @@ -1677,20 +1679,39 @@ 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* Effect.forEach( + [...entry.instructionInvalidations.values()], + (offered) => + persistInstructionInvalidation(entry, offered).pipe( + Effect.catch((cause) => + Effect.logWarning( + "orchestration-v2.provider-session.instruction-invalidation-failed", + { providerSessionId, providerThreadId: offered.id, cause }, + ), + ), + ), + { discard: true }, + ); } - 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, - ), - ); + }).pipe(Effect.orDie); const events = Stream.fromQueue(queue).pipe( Stream.mapEffect((signal) => signal.type === "event" @@ -1710,7 +1731,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, @@ -1957,6 +1978,54 @@ export const layerWithOptions = ( ), ); + const persistInstructionInvalidation = Effect.fn( + "ProviderSessionManager.persistInstructionInvalidation", + )(function* (entry: LiveSessionEntry, offered: OrchestrationV2ProviderThread) { + 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)) 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 startEventPump = (entry: LiveSessionEntry) => { let stoppedByProvider = false; let lastSessionStatus = entry.runtime.providerSession.status; @@ -2022,6 +2091,35 @@ export const layerWithOptions = ( }); }); } + if ( + event.type === "provider_thread.updated" && + event.providerThread.nativeMetadata?.acpInstructions !== undefined + ) { + yield* Effect.sync(() => { + if (event.providerThread.nativeMetadata?.acpInstructions === null) { + entry.instructionInvalidations.set( + event.providerThread.id, + event.providerThread, + ); + } else { + entry.instructionInvalidations.delete(event.providerThread.id); + } + }).pipe(entry.requestEventPermit.withPermits(1)); + 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; + } // Some providers can block before a run subscriber exists // (project trust, login, or session-switch hooks). Persist // their runless request artifacts directly so the normal T3 @@ -2269,6 +2367,7 @@ export const layerWithOptions = ( exposedRuntime, eventSubscribers, requestEventPermit: yield* Semaphore.make(1), + instructionInvalidations: new Map(), scope: sessionScope, idleGeneration: 0, busyTurns: new Set(), diff --git a/apps/server/src/orchestration-v2/testkit/fixtures/provider_thread_resume/kiro_transcript.ndjson b/apps/server/src/orchestration-v2/testkit/fixtures/provider_thread_resume/kiro_transcript.ndjson index 2f5d9833e73c7..3fc95702aa121 100644 --- a/apps/server/src/orchestration-v2/testkit/fixtures/provider_thread_resume/kiro_transcript.ndjson +++ b/apps/server/src/orchestration-v2/testkit/fixtures/provider_thread_resume/kiro_transcript.ndjson @@ -44,7 +44,7 @@ {"type":"emit_inbound","label":"notification:session/update:available_commands_update","frame":{"kind":"notification","method":"session/update","params":{"sessionId":"sess_00000000-0000-4000-8000-000000000001","update":{"sessionUpdate":"available_commands_update","availableCommands":[{"name":"architecture-selection","description":"Steering: architecture-selection","input":{"hint":"optional context"},"_meta":{"kiro":{"type":"steering","contextQuery":"global:architecture-selection","commandId":"architecture-selection","resource":{"resourceType":"steering","source":{"origin":"bundled"}},"originalName":"architecture-selection"}}},{"name":"quick-spec","description":"Steering: quick-spec","input":{"hint":"optional context"},"_meta":{"kiro":{"type":"steering","contextQuery":"global:quick-spec","commandId":"quick-spec","resource":{"resourceType":"steering","source":{"origin":"bundled"}},"originalName":"quick-spec"}}},{"name":"bug-fix","description":"Steering: bug-fix","input":{"hint":"optional context"},"_meta":{"kiro":{"type":"steering","contextQuery":"global:bug-fix","commandId":"bug-fix","resource":{"resourceType":"steering","source":{"origin":"bundled"}},"originalName":"bug-fix"}}},{"name":"general-task-execution","description":"General-purpose sub-agent with access to all tools for executing arbitrary tasks","input":{"hint":"task to delegate"},"_meta":{"kiro":{"type":"custom-agent","commandId":"general-task-execution","resource":{"resourceType":"agent","source":{"origin":"bundled"}},"originalName":"general-task-execution"}}},{"name":"context-gatherer","description":"Investigates the codebase to answer a specific question about it. Deeply analyzes existing code by tracing execution paths, mapping architecture and abstraction layers, understanding patterns, and documenting dependencies, returning a thorough, direct written analysis that explains how the relevant code works and points to the specific files and lines involved. Use it to understand code before you change it: before designing or implementing a change to code you don't already fully understand, start with context-gatherer so your edits are grounded in how the code actually works. When a task requires understanding code you don't already know, dispatch context-gatherer to do that exploration before running your own searches or reads. Rules: 1. Provide a clear, specific prompt that includes your intent: a detailed task description of the context you need and what you're trying to accomplish. 2. The subagent explores the codebase and returns an analysis of your query 3. Trust the returned results. You may still inspect additional code yourself when you need context beyond what was returned. 4. Reuse existing context-gatherer output for related follow-up questions; do not re-invoke on the same question with different phrasing. 5. For multiple distinct questions, dispatch several context-gatherers in parallel, each targeting a different aspect of the codebase.","input":{"hint":"task to delegate"},"_meta":{"kiro":{"type":"custom-agent","commandId":"context-gatherer","resource":{"resourceType":"agent","source":{"origin":"bundled"}},"originalName":"context-gatherer"}}}]}}}} {"type":"emit_inbound","label":"response:session/set_config_option","frame":{"kind":"response","method":"session/set_config_option","result":{"configOptions":[{"type":"select","id":"mode","name":"Mode","category":"mode","currentValue":"vibe","options":[{"value":"vibe","name":"Default","description":"General coding assistance","_meta":{"kiro":{"source":"bundled","resource":{"resourceType":"agent","source":{"origin":"bundled"}}}}},{"value":"spec","name":"Spec","description":"Structured feature development","_meta":{"kiro":{"source":"bundled","resource":{"resourceType":"agent","source":{"origin":"bundled"}}}}},{"value":"quick-spec","name":"Quick Spec","description":"Fast spec workflow: clarify, then auto-generate requirements, design, and tasks","_meta":{"kiro":{"source":"bundled","resource":{"resourceType":"agent","source":{"origin":"bundled"}}}}},{"value":"bug-fix","name":"Bug Fix","description":"Structured bug-fix workflow: investigate, diagnose, and resolve bugs","_meta":{"kiro":{"source":"bundled","resource":{"resourceType":"agent","source":{"origin":"bundled"}}}}},{"value":"plan","name":"Plan","description":"Plan-only mode that helps break ideas down into an implementation plan without making any changes","_meta":{"kiro":{"welcomeMessage":"Transform any idea into fully working code. What do you want to build today?","source":"bundled","resource":{"resourceType":"agent","source":{"origin":"bundled"}}}}},{"value":"autonomous","name":"Autonomous","description":"Autonomous agent execution","_meta":{"kiro":{"source":"bundled","resource":{"resourceType":"agent","source":{"origin":"bundled"}}}}},{"value":"semantic_reviewer","name":"semantic_reviewer","description":"Behavioral, narrative code review of a local or PR diff. Writes a review document and a verdict.","_meta":{"kiro":{"source":"bundled","resource":{"resourceType":"agent","source":{"origin":"bundled"}}}}}]},{"type":"select","id":"memoryReflection","name":"Memory reflection","description":"Allow this session to reflect on new memory","currentValue":"on","options":[{"value":"on","name":"On"},{"value":"off","name":"Off"}]},{"type":"select","id":"autopilot","name":"Autopilot","currentValue":"on","options":[{"value":"on","name":"Autopilot","description":"Agent executes tools without confirmation"},{"value":"off","name":"Supervised","description":"Agent asks for approval before file changes"}]},{"type":"select","id":"contentCollection","name":"Content Collection","currentValue":"enabled","options":[{"value":"enabled","name":"Enabled","description":"Allow content to be used for service improvement"},{"value":"disabled","name":"Disabled","description":"Opt out of content collection"}]}]}}} {"type":"emit_inbound","label":"notification:session/update:config_option_update","frame":{"kind":"notification","method":"session/update","params":{"sessionId":"sess_00000000-0000-4000-8000-000000000001","update":{"sessionUpdate":"config_option_update","configOptions":[{"type":"select","id":"mode","name":"Mode","category":"mode","currentValue":"vibe","options":[{"value":"vibe","name":"Default","description":"General coding assistance","_meta":{"kiro":{"source":"bundled","resource":{"resourceType":"agent","source":{"origin":"bundled"}}}}},{"value":"spec","name":"Spec","description":"Structured feature development","_meta":{"kiro":{"source":"bundled","resource":{"resourceType":"agent","source":{"origin":"bundled"}}}}},{"value":"quick-spec","name":"Quick Spec","description":"Fast spec workflow: clarify, then auto-generate requirements, design, and tasks","_meta":{"kiro":{"source":"bundled","resource":{"resourceType":"agent","source":{"origin":"bundled"}}}}},{"value":"bug-fix","name":"Bug Fix","description":"Structured bug-fix workflow: investigate, diagnose, and resolve bugs","_meta":{"kiro":{"source":"bundled","resource":{"resourceType":"agent","source":{"origin":"bundled"}}}}},{"value":"plan","name":"Plan","description":"Plan-only mode that helps break ideas down into an implementation plan without making any changes","_meta":{"kiro":{"welcomeMessage":"Transform any idea into fully working code. What do you want to build today?","source":"bundled","resource":{"resourceType":"agent","source":{"origin":"bundled"}}}}},{"value":"autonomous","name":"Autonomous","description":"Autonomous agent execution","_meta":{"kiro":{"source":"bundled","resource":{"resourceType":"agent","source":{"origin":"bundled"}}}}},{"value":"semantic_reviewer","name":"semantic_reviewer","description":"Behavioral, narrative code review of a local or PR diff. Writes a review document and a verdict.","_meta":{"kiro":{"source":"bundled","resource":{"resourceType":"agent","source":{"origin":"bundled"}}}}}]},{"type":"select","id":"model","name":"Model","category":"model","currentValue":"claude-haiku-4.5","options":[{"value":"auto","name":"Auto","description":"Models chosen by task for optimal usage and consistent quality","_meta":{"kiro":{"rateMultiplier":1,"rateUnit":"Credit","hasEffort":false,"thinkingToggleable":false}}},{"value":"claude-opus-5.5","name":"Claude Opus 5.5","description":"Experimental preview of Claude Opus 5.5 model with 1M context window","_meta":{"kiro":{"rateMultiplier":2,"rateUnit":"Credit","hasEffort":true,"effortSchemaPath":"output_config","effortLevels":["low","medium","high","xhigh","max"],"defaultEffortLevel":"medium","thinkingToggleable":false}}},{"value":"claude-sonnet-5.5","name":"Claude Sonnet 5.5","description":"Claude Sonnet 5.5 model with 1M context window","_meta":{"kiro":{"rateMultiplier":1.3,"rateUnit":"Credit","hasEffort":true,"effortSchemaPath":"output_config","effortLevels":["low","medium","high","xhigh","max"],"defaultEffortLevel":"high","thinkingToggleable":false}}},{"value":"claude-opus-5","name":"Claude Opus 5","description":"Claude Opus 5 model with 1M context window","_meta":{"kiro":{"rateMultiplier":2.2,"rateUnit":"Credit","hasEffort":true,"effortSchemaPath":"output_config","effortLevels":["low","medium","high","xhigh","max"],"defaultEffortLevel":"high","thinkingToggleable":true,"defaultThinkingEnabled":true}}},{"value":"claude-sonnet-5","name":"Claude Sonnet 5","description":"Claude Sonnet 5 model with 1M context window","_meta":{"kiro":{"rateMultiplier":1.3,"rateUnit":"Credit","hasEffort":true,"effortSchemaPath":"output_config","effortLevels":["low","medium","high","xhigh","max"],"defaultEffortLevel":"high","thinkingToggleable":true,"defaultThinkingEnabled":true}}},{"value":"claude-opus-4.8","name":"Claude Opus 4.8","description":"Claude Opus 4.8 model with 1M context window","_meta":{"kiro":{"rateMultiplier":2.2,"rateUnit":"Credit","hasEffort":true,"effortSchemaPath":"output_config","effortLevels":["low","medium","high","xhigh","max"],"defaultEffortLevel":"high","thinkingToggleable":true,"defaultThinkingEnabled":true}}},{"value":"gpt-5.6-sol","name":"GPT 5.6 Sol","description":"Experimental preview of OpenAI GPT 5.6 Sol with 1M context window","_meta":{"kiro":{"rateMultiplier":4.4,"rateUnit":"Credit","hasEffort":true,"effortSchemaPath":"reasoning","effortLevels":["none","low","medium","high","xhigh","max"],"defaultEffortLevel":"high","thinkingToggleable":false}}},{"value":"gpt-5.6-terra","name":"GPT 5.6 Terra","description":"Experimental preview of OpenAI GPT 5.6 Terra with 1M context window","_meta":{"kiro":{"rateMultiplier":2.2,"rateUnit":"Credit","hasEffort":true,"effortSchemaPath":"reasoning","effortLevels":["none","low","medium","high","xhigh","max"],"defaultEffortLevel":"high","thinkingToggleable":false}}},{"value":"gpt-5.6-luna","name":"GPT 5.6 Luna","description":"Experimental preview of OpenAI GPT 5.6 Luna with 1M context window","_meta":{"kiro":{"rateMultiplier":0.6,"rateUnit":"Credit","hasEffort":true,"effortSchemaPath":"reasoning","effortLevels":["none","low","medium","high","xhigh","max"],"defaultEffortLevel":"high","thinkingToggleable":false}}},{"value":"claude-opus-4.7","name":"Claude Opus 4.7","description":"Claude Opus 4.7 model with 1M context window","_meta":{"kiro":{"rateMultiplier":2.2,"rateUnit":"Credit","hasEffort":true,"effortSchemaPath":"output_config","effortLevels":["low","medium","high","xhigh","max"],"defaultEffortLevel":"xhigh","thinkingToggleable":true,"defaultThinkingEnabled":true}}},{"value":"claude-opus-4.6","name":"Claude Opus 4.6","description":"Claude Opus 4.6 model with 1M context window","_meta":{"kiro":{"rateMultiplier":2.2,"rateUnit":"Credit","hasEffort":true,"effortSchemaPath":"output_config","effortLevels":["low","medium","high","max"],"defaultEffortLevel":"high","thinkingToggleable":true,"defaultThinkingEnabled":true}}},{"value":"claude-sonnet-4.6","name":"Claude Sonnet 4.6","description":"Claude Sonnet 4.6 model with 1M context window","_meta":{"kiro":{"rateMultiplier":1.3,"rateUnit":"Credit","hasEffort":true,"effortSchemaPath":"output_config","effortLevels":["low","medium","high","max"],"defaultEffortLevel":"high","thinkingToggleable":true,"defaultThinkingEnabled":true}}},{"value":"claude-opus-4.5","name":"Claude Opus 4.5","description":"Claude Opus 4.5 model","_meta":{"kiro":{"rateMultiplier":2.2,"rateUnit":"Credit","hasEffort":false,"thinkingToggleable":false}}},{"value":"claude-sonnet-4.5","name":"Claude Sonnet 4.5","description":"Claude Sonnet 4.5 model","_meta":{"kiro":{"rateMultiplier":1.3,"rateUnit":"Credit","hasEffort":false,"thinkingToggleable":false}}},{"value":"claude-haiku-4.5","name":"Claude Haiku 4.5","description":"The latest Claude Haiku model","_meta":{"kiro":{"rateMultiplier":0.4,"rateUnit":"Credit","hasEffort":false,"thinkingToggleable":false}}},{"value":"deepseek-3.2","name":"Deepseek v3.2","description":"Experimental preview of DeepSeek V3.2","_meta":{"kiro":{"rateMultiplier":0.25,"rateUnit":"Credit","hasEffort":false,"thinkingToggleable":false}}},{"value":"minimax-m2.5","name":"MiniMax M2.5","description":"MiniMax M2.5 model","_meta":{"kiro":{"rateMultiplier":0.25,"rateUnit":"Credit","hasEffort":false,"thinkingToggleable":false}}},{"value":"minimax-m2.1","name":"MiniMax M2.1","description":"Experimental preview of MiniMax M2.1","_meta":{"kiro":{"rateMultiplier":0.15,"rateUnit":"Credit","hasEffort":false,"thinkingToggleable":false}}},{"value":"glm-5","name":"GLM 5","description":"GLM-5 model","_meta":{"kiro":{"rateMultiplier":0.5,"rateUnit":"Credit","hasEffort":false,"thinkingToggleable":false}}},{"value":"qwen3-coder-next","name":"Qwen3 Coder Next","description":"Experimental preview of Qwen3 Coder Next","_meta":{"kiro":{"rateMultiplier":0.05,"rateUnit":"Credit","hasEffort":false,"thinkingToggleable":false}}},{"value":"kimi-k3","name":"Kimi K3","description":"Experimental preview of Kimi K3. Kimi K3 is a Kiro Excluded Model. For more info, see documentation.","_meta":{"kiro":{"rateMultiplier":2.4,"rateUnit":"Credit","hasEffort":true,"effortSchemaPath":"reasoning","effortLevels":["low","high","max"],"defaultEffortLevel":"max","thinkingToggleable":false}}}]},{"type":"select","id":"memoryReflection","name":"Memory reflection","description":"Allow this session to reflect on new memory","currentValue":"on","options":[{"value":"on","name":"On"},{"value":"off","name":"Off"}]},{"type":"select","id":"autopilot","name":"Autopilot","currentValue":"on","options":[{"value":"on","name":"Autopilot","description":"Agent executes tools without confirmation"},{"value":"off","name":"Supervised","description":"Agent asks for approval before file changes"}]},{"type":"select","id":"contentCollection","name":"Content Collection","currentValue":"enabled","options":[{"value":"enabled","name":"Enabled","description":"Allow content to be used for service improvement"},{"value":"disabled","name":"Disabled","description":"Opt out of content collection"}]}]}}}} -{"type":"expect_outbound","label":"request:session/prompt","frame":{"kind":"request","method":"session/prompt","params":{"sessionId":"sess_00000000-0000-4000-8000-000000000001","prompt":[{"type":"text","text":"\n\n\n\n\nUsing the conversation history available in this resumed thread, first repeat the exact final answer you gave in the previous turn. Then on a new line write exactly: provider thread resume fixture second turn complete\n"}]}}} +{"type":"expect_outbound","label":"request:session/prompt","frame":{"kind":"request","method":"session/prompt","params":{"sessionId":"sess_00000000-0000-4000-8000-000000000001","prompt":[{"type":"text","text":"Using the conversation history available in this resumed thread, first repeat the exact final answer you gave in the previous turn. Then on a new line write exactly: provider thread resume fixture second turn complete"}]}}} {"type":"emit_inbound","label":"notification:session/update:session_info_update","frame":{"kind":"notification","method":"session/update","params":{"sessionId":"sess_00000000-0000-4000-8000-000000000001","update":{"sessionUpdate":"session_info_update","_meta":{"kiro":{"userMessageId":"00000000-0000-4000-9000-000000000006","kind":"user_message_id_assigned"}}}}}} {"type":"emit_inbound","label":"notification:session/update:session_info_update","frame":{"kind":"notification","method":"session/update","params":{"sessionId":"sess_00000000-0000-4000-8000-000000000001","update":{"sessionUpdate":"session_info_update","_meta":{"kiro":{"focus":{"status":"in_progress"},"kind":"focus_update","status":"in_progress","activity":{"turnActive":true,"runningWorkflows":0,"pausedWorkflows":0}}}}}}} {"type":"emit_inbound","label":"notification:session/update:session_info_update","frame":{"kind":"notification","method":"session/update","params":{"sessionId":"sess_00000000-0000-4000-8000-000000000001","update":{"sessionUpdate":"session_info_update","_meta":{"kiro":{"turnStart":true,"kind":"turn_start","messageId":"00000000-0000-4000-9000-000000000007-turn-start"}}}}}} diff --git a/packages/contracts/src/orchestrationV2.ts b/packages/contracts/src/orchestrationV2.ts index 5c430ef75581a..5bfa72dd89a02 100644 --- a/packages/contracts/src/orchestrationV2.ts +++ b/packages/contracts/src/orchestrationV2.ts @@ -1049,6 +1049,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, + }), + ), + ), modelSelection: Schema.optional(ModelSelection), title: Schema.optional(Schema.NullOr(TrimmedNonEmptyString)), updatedAt: 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 1b8b4240fd00e..506b02e9f11d6 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, @@ -309,6 +310,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 c4fc2f092644a..7849a3e59813e 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"; @@ -1344,6 +1346,11 @@ interface AcpNativeBuildConfiguration { } interface ActiveAcpTurn { + readonly instructionDelivery?: { + readonly state: T3AcpInstructionState; + readonly digest: string; + compactionMarker: symbol | undefined; + }; readonly input: ProviderAdapter.ProviderAdapterV2TurnInput; readonly providerTurnId: OrchestrationV2ProviderTurn["id"]; readonly nativeThreadId: string; @@ -1678,6 +1685,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; @@ -2274,6 +2282,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; @@ -4100,6 +4130,7 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( : { beforeTokenCount: context.contextUsage.usedTokens }), }; if (status === "completed") { + yield* invalidateInstructions(context); context.contextUsage = null; yield* Ref.update(contextUsageBySessionId, (current) => { const updated = new Map(current); @@ -4974,6 +5005,33 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( ) { const context = yield* Ref.get(activeTurn); const update = notification.update; + if ( + (context === null || context.finalized) && + notification.sessionId === (yield* Ref.get(activeSessionId)) && + update.sessionUpdate === "compaction_update" && + update.status === "completed" + ) { + // Deferred finalization can still publish a receipt after marking itself finalized. + if (context !== null) yield* Deferred.await(context.completed); + const known = (yield* Ref.get(providerThreadByNativeSessionId)).get( + notification.sessionId, + ); + yield* invalidateInstructions({ + nativeThreadId: notification.sessionId, + nativeMetadata: + (yield* Ref.get(nativeMetadataBySessionId)).get(notification.sessionId) ?? + known?.nativeMetadata ?? + null, + }); + yield* Ref.update(promptInstructionStates, (current) => { + const updated = new Map(current); + updated.delete(notification.sessionId); + return updated; + }); + yield* Ref.update(compactionMarkers, (current) => + new Map(current).set(notification.sessionId, Symbol()), + ); + } if ( followsAgentWorkState && flavor.extractSubagentSessionEvent !== undefined && @@ -7103,6 +7161,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(compactionMarkers, new Map()); yield* Ref.set(providerTurns, new Map()); yield* Ref.set(snapshot, { @@ -7645,12 +7709,33 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( 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 ( (flavor.supportsCompaction === true || flavor.compactSession !== undefined) && isCompactionMessage(context.input.message) && settledStatus === "completed" ) { const nativeItemId = `${context.nativeTurnId}:compaction`; + yield* invalidateInstructions(context); yield* Ref.update(compactionMarkers, (current) => { const updated = new Map(current); updated.set(context.nativeThreadId, Symbol()); @@ -7691,6 +7776,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, @@ -7915,7 +8006,27 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( hasT3Mcp: mcpSession !== undefined, hasDeviceTools: mcpSession?.capabilities?.has("device") === true, } satisfies T3AcpInstructionState; - 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 userText = flavor.rewriteUserText === undefined @@ -8034,6 +8145,7 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( return { prompt, instructionState: text === messageText ? undefined : instructionState, + instructionDigest, compactionMarker, }; }); @@ -8172,6 +8284,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, @@ -8355,6 +8476,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 }; context.promptSentAt = yield* Clock.currentTimeMillis; if (promptInactivityTimeoutMillis !== undefined) { @@ -8387,13 +8516,13 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( ), Effect.tap(() => Ref.update(compactionMarkers, (current) => { - if (promptParts?.instructionState === undefined) return current; - const observedMarker = promptParts.compactionMarker; + const delivery = context.instructionDelivery; if ( - observedMarker === undefined || - current.get(requestedSessionId) !== observedMarker + delivery?.compactionMarker === undefined || + current.get(requestedSessionId) !== delivery.compactionMarker ) return current; + delivery.compactionMarker = undefined; const updated = new Map(current); updated.delete(requestedSessionId); return updated; @@ -8723,6 +8852,18 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( Effect.gen(function* () { yield* awaitRuntimeTeardown(); const sessionId = yield* nativeThreadId(driver, threadInput.providerThread); + yield* Ref.update(nativeMetadataBySessionId, (current) => + new Map(current).set(sessionId, { + ...threadInput.providerThread.nativeMetadata, + ...current.get(sessionId), + }), + ); + yield* Ref.update(providerThreadByNativeSessionId, (current) => + new Map(current).set(sessionId, { + ...threadInput.providerThread, + providerSessionId: input.providerSessionId, + }), + ); const previousItemIdentityVersion = itemIdentityVersion; useProviderThreadIdentity(threadInput.providerThread); const restorePreviousItemIdentity = Effect.sync(() => { @@ -8752,12 +8893,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; }), ); }, @@ -9325,6 +9481,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-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.ts b/packages/provider-grok/src/server/adapter.ts index 9ed1368f16984..6912930216faf 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, @@ -357,6 +358,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 6815b60e8f227..8f64cfdda5f73 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 { @@ -589,6 +590,7 @@ export const makeKiroAdapterV2 = Effect.fn("makeKiroAdapterV2")(function* ( }); export type KiroAdapterV2DriverEnv = + | Digest.Digest | ChildProcessSpawner.ChildProcessSpawner | SecureRandom.SecureRandom | FileSystem.FileSystem