diff --git a/apps/coder-gateway/src/server.ts b/apps/coder-gateway/src/server.ts index bfce6727a2222..5f39317225b15 100644 --- a/apps/coder-gateway/src/server.ts +++ b/apps/coder-gateway/src/server.ts @@ -106,7 +106,7 @@ function sendText( "Cache-Control": "no-store", "Content-Type": contentType, "Content-Security-Policy": - "default-src 'self'; base-uri 'none'; form-action 'none'; frame-ancestors 'none'; style-src 'self' 'unsafe-inline'; script-src 'self' 'wasm-unsafe-eval' 'sha256-+xUlzIW26WOhLd/WV5CUp/KDzSwFacmCLHCiJxm72Yw='; connect-src 'self' blob: https: http: ws://127.0.0.1:*; img-src 'self' data: blob: https: http:; media-src 'self' blob: https: http:; worker-src 'self' blob:; object-src 'self' blob:; frame-src 'self' blob:", + "default-src 'self'; base-uri 'none'; form-action 'none'; frame-ancestors 'none'; style-src 'self' 'unsafe-inline'; script-src 'self' 'wasm-unsafe-eval' 'sha256-n0OPhj7RDHbzPYmNVtzBFHdU5YSjXCbQBVoFS8BcmDQ='; connect-src 'self' blob: https: http: ws://127.0.0.1:*; img-src 'self' data: blob: https: http:; media-src 'self' blob: https: http:; worker-src 'self' blob:; object-src 'self' blob:; frame-src 'self' blob:", "X-Content-Type-Options": "nosniff", "X-Frame-Options": "DENY", "Cross-Origin-Resource-Policy": "same-origin", diff --git a/apps/server/integration/SqlStatementCounter.integration.ts b/apps/server/integration/SqlStatementCounter.integration.ts deleted file mode 100644 index 0c215ba79b69d..0000000000000 --- a/apps/server/integration/SqlStatementCounter.integration.ts +++ /dev/null @@ -1,18 +0,0 @@ -import * as Tracer from "effect/Tracer"; - -export interface SqlStatementCounter { - readonly tracer: Tracer.Tracer; - readonly count: () => number; -} - -/** Install on every test runtime to include background reactor statements. */ -export function makeSqlStatementCounter(): SqlStatementCounter { - let statements = 0; - const tracer = Tracer.make({ - span: (options) => { - if (options.kind === "client" && options.name === "sqlite") statements += 1; - return new Tracer.NativeSpan(options); - }, - }); - return { tracer, count: () => statements }; -} diff --git a/apps/server/integration/transferBudgetV2.integration.test.ts b/apps/server/integration/transferBudgetV2.integration.test.ts deleted file mode 100644 index 8417d852acfdf..0000000000000 --- a/apps/server/integration/transferBudgetV2.integration.test.ts +++ /dev/null @@ -1,409 +0,0 @@ -// @effect-diagnostics nodeBuiltinImport:off - Measures real local HTTP/WebSocket sockets. -import * as NodeHttp from "node:http"; -import * as NodeUtil from "node:util"; -import { assert, it } from "@effect/vitest"; -import * as NodeHttpServer from "@effect/platform-node/NodeHttpServer"; -import * as NodeHttpPlatform from "@effect/platform-node/NodeHttpPlatform"; -import * as NodeServices from "@effect/platform-node/NodeServices"; -import { - ORCHESTRATION_PROTOCOL_HEADER, - ORCHESTRATION_PROTOCOL_VERSION, - AuthSessionId, - AuthOrchestrationReadScope, - EnvironmentHttpApi, - EnvironmentId, - EnvironmentAuthenticatedAuth, - EnvironmentAuthenticatedPrincipal, - ORCHESTRATION_V2_WS_METHODS, - OrchestrationV2RpcSchemas, - OrchestrationV2ThreadBoundedSnapshot, - OrchestrationV2ThreadDetailSnapshot, - OrchestrationV2ThreadHistoryPage, - OrchestrationV2ShellSnapshot, - OrchestrationV2GetThreadProjectionError, - OrchestrationV2GetShellSnapshotError, - ProviderDriverKind, - type OrchestrationV2ThreadStreamItem, - type OrchestrationV2ShellStreamItem, -} from "@t3tools/contracts"; -import * as Context from "effect/Context"; -import * as Effect from "effect/Effect"; -import * as FileSystem from "effect/FileSystem"; -import * as Layer from "effect/Layer"; -import * as PubSub from "effect/PubSub"; -import * as Queue from "effect/Queue"; -import * as Schema from "effect/Schema"; -import * as Scope from "effect/Scope"; -import * as Stream from "effect/Stream"; -import * as HttpApi from "effect/http-api/HttpApi"; -import * as HttpApiBuilder from "effect/http-api/HttpApiBuilder"; -import { HttpRouter, HttpServer } from "effect/http"; -import { Rpc, RpcGroup, RpcServer, RpcSerialization } from "effect/rpc"; -import * as ServerEnvironment from "../src/environment/ServerEnvironment.ts"; -import * as SqlitePersistence from "../src/persistence/Sqlite.ts"; -import * as OrchestrationEventStore from "../src/persistence/OrchestrationEventStore.ts"; -import * as EventStore from "../src/orchestration-v2/EventStore.ts"; -import * as EventSink from "../src/orchestration-v2/EventSink.ts"; -import * as ProjectionStore from "../src/orchestration-v2/ProjectionStore.ts"; -import * as ThreadManagementService from "../src/orchestration-v2/ThreadManagementService.ts"; -import * as ProjectStore from "../src/orchestration-v2/ProjectStore.ts"; -import * as ProjectService from "../src/project/ProjectService.ts"; -import * as ProjectEnrichmentService from "../src/project/ProjectEnrichmentService.ts"; -import * as OrchestrationHttp from "../src/orchestration-v2/http.ts"; -import * as ServerHttp from "../src/http.ts"; -import { subscribeOrchestrationV2Thread, subscribeOrchestrationV2Shell } from "../src/ws.ts"; -import { - measureHttpGet, - openMeasuredWsClient, - type MeasuredWsClient, -} from "./NetworkTransferMeasurement.integration.ts"; -import { makeSqlStatementCounter } from "./SqlStatementCounter.integration.ts"; -import { - formatTransferBudgetReport, - formatTransferBudgetResult, - transferBudgetViolations, - type TransferBudgetRun, -} from "./TransferBudgetReport.integration.ts"; -import { TRANSFER_HISTORY_TURN_COUNT } from "./fixtures/transferBudget.ts"; -import { - LONG_TURN_THREAD_ID, - longTurnThreadBatches, - THREAD_ID, - threadCreated, - turnEvents, -} from "./TransferBudgetV2Fixture.integration.ts"; - -const decodeThreadSnapshot = Schema.decodeUnknownEffect( - Schema.fromJsonString(Schema.toCodecJson(OrchestrationV2ThreadDetailSnapshot)), -); -const decodeBoundedSnapshot = Schema.decodeUnknownEffect( - Schema.fromJsonString(Schema.toCodecJson(OrchestrationV2ThreadBoundedSnapshot)), -); -const decodeHistoryPage = Schema.decodeUnknownEffect( - Schema.fromJsonString(Schema.toCodecJson(OrchestrationV2ThreadHistoryPage)), -); -const decodeShellSnapshot = Schema.decodeUnknownEffect( - Schema.fromJsonString(Schema.toCodecJson(OrchestrationV2ShellSnapshot)), -); - -const layerPersistence = EventSink.layer.pipe( - Layer.provideMerge(EventStore.layerFromOrchestrationEventStore), - Layer.provideMerge(ProjectionStore.layer), - Layer.provideMerge(OrchestrationEventStore.layer), - Layer.provideMerge(SqlitePersistence.layerMemory), -); -const layerManagement = Layer.unwrap( - Effect.gen(function* () { - const projections = yield* ProjectionStore.ProjectionStoreV2; - const sink = yield* EventSink.EventSinkV2; - return Layer.mock(ThreadManagementService.ThreadManagementService)({ - ensureLegacyTranscript: () => Effect.void, - getThreadSnapshot: (id) => projections.getThreadSnapshot(id).pipe(Effect.orDie), - getThreadSnapshotWindow: (id, options) => - projections.getThreadSnapshotWindow(id, options).pipe(Effect.orDie), - getThreadHistoryPage: (id, cursor, throughEntryId, conversationOnly) => - projections - .getThreadHistoryPage(id, cursor, throughEntryId, conversationOnly) - .pipe(Effect.orDie), - getThreadShell: (id) => projections.getThreadShell(id).pipe(Effect.orDie), - getShellSnapshot: (options) => projections.getShellSnapshot(options).pipe(Effect.orDie), - listActiveSubagentThreadIds: () => - projections.listActiveSubagentThreadIds().pipe(Effect.orDie), - readShellSnapshot: (options) => - projections.readShellSnapshot(options).pipe(Effect.map(Effect.orDie), Effect.orDie), - streamStoredEventsFrom: (input) => - sink.stream({ ...input, bounded: true }).pipe(Stream.orDie), - }); - }), -); -const layerEnrichment = Layer.unwrap( - Effect.gen(function* () { - const changes = yield* PubSub.unbounded(); - return Layer.mock(ProjectEnrichmentService.ProjectEnrichmentService)({ - getAvailable: () => - Effect.succeed({ - repositoryIdentity: null, - faviconPath: null, - repositoryIdentityResolved: true, - }), - subscribeChanges: PubSub.subscribe(changes), - }); - }), -); -const layerServices = layerManagement.pipe( - Layer.provideMerge( - Layer.succeed(ServerEnvironment.ServerEnvironmentIdentity, { - getEnvironmentId: Effect.succeed(EnvironmentId.make("transfer-test")), - }), - ), - Layer.provideMerge(ProjectStore.layer), - Layer.provideMerge(Layer.mock(ProjectService.ProjectService)({})), - Layer.provideMerge(layerEnrichment), - Layer.provideMerge(layerPersistence), -); -class TransferApi extends HttpApi.make("environment").add( - EnvironmentHttpApi.groups.orchestration, -) {} -const layerAuth = Layer.succeed(EnvironmentAuthenticatedAuth, (effect) => - effect.pipe( - Effect.provideService(EnvironmentAuthenticatedPrincipal, { - sessionId: AuthSessionId.make("transfer-session"), - subject: "transfer-test", - method: "browser-session-cookie", - scopes: new Set([AuthOrchestrationReadScope]), - }), - ), -); -const group = RpcGroup.make( - Rpc.make(ORCHESTRATION_V2_WS_METHODS.subscribeThread, { - payload: OrchestrationV2RpcSchemas.subscribeThread.input, - success: OrchestrationV2RpcSchemas.subscribeThread.output, - error: OrchestrationV2GetThreadProjectionError, - stream: true, - }), - Rpc.make(ORCHESTRATION_V2_WS_METHODS.subscribeShell, { - payload: OrchestrationV2RpcSchemas.subscribeShell.input, - success: OrchestrationV2RpcSchemas.subscribeShell.output, - error: OrchestrationV2GetShellSnapshotError, - stream: true, - }), -); -const layerHandlers = group.toLayer({ - [ORCHESTRATION_V2_WS_METHODS.subscribeThread]: (input) => - Stream.unwrap(subscribeOrchestrationV2Thread(input)), - [ORCHESTRATION_V2_WS_METHODS.subscribeShell]: (input) => - Stream.unwrap(subscribeOrchestrationV2Shell(input)), -}); -const layerRoutes = Layer.mergeAll( - HttpApiBuilder.layer(TransferApi).pipe( - Layer.provide(OrchestrationHttp.layer), - Layer.provide(layerAuth), - ), - RpcServer.layerHttp({ group, path: "/ws", protocol: "websocket" }).pipe( - Layer.provide(layerHandlers), - Layer.provide(RpcSerialization.layerJson), - ), -).pipe(Layer.provide(ServerHttp.layerHttpCompression), Layer.provide(NodeHttpPlatform.layer)); - -function difference( - after: { wireBytes: number; decodedBytes: number; messages: number }, - before: typeof after, -) { - return { - wireBytes: after.wireBytes - before.wireBytes, - decodedBytes: after.decodedBytes - before.decodedBytes, - messages: after.messages - before.messages, - }; -} -type StreamItem = OrchestrationV2ThreadStreamItem | OrchestrationV2ShellStreamItem; -const collectUntil = Effect.fn("transfer.collectUntil")(function* ( - queue: Queue.Queue, - done: (item: T) => boolean, -) { - const items: T[] = []; - while (true) { - const item = yield* Queue.take(queue); - items.push(item); - if (done(item)) return items; - } -}); -const subscribe = Effect.fn("transfer.subscribe")(function* ( - client: MeasuredWsClient, - kind: "thread" | "shell", - afterSequence: number, -) { - const queue = yield* Queue.unbounded(); - const stream: Stream.Stream = - kind === "thread" - ? client.client[ORCHESTRATION_V2_WS_METHODS.subscribeThread]({ - threadId: THREAD_ID, - afterSequence, - requestCompletionMarker: true, - acceptBoundedSnapshot: true, - }) - : client.client[ORCHESTRATION_V2_WS_METHODS.subscribeShell]({ - afterSequence, - requestCompletionMarker: true, - }); - yield* stream.pipe( - Stream.runForEach((item) => Queue.offer(queue, item)), - Scope.provide(client.scope), - Effect.forkIn(client.scope), - ); - return queue; -}); -const synchronized = (queue: Queue.Queue) => - collectUntil(queue, (item) => item.kind === "synchronized").pipe( - Effect.map((items) => - items.some((item) => item.kind === "snapshot") ? ("snapshot" as const) : ("replay" as const), - ), - ); - -it.live("reports V2 thread HTTP and WebSocket transfer budgets", () => - Effect.gen(function* () { - const runs: TransferBudgetRun[] = []; - for (const provider of [ - ProviderDriverKind.make("codex"), - ProviderDriverKind.make("claudeAgent"), - ]) { - const counter = makeSqlStatementCounter(); - const run = yield* Effect.scoped( - Effect.gen(function* () { - const sink = yield* EventSink.EventSinkV2; - const projection = yield* ProjectionStore.ProjectionStoreV2; - yield* sink.write({ events: [threadCreated(provider)] }); - for (let index = 0; index < TRANSFER_HISTORY_TURN_COUNT; index++) - yield* sink.write({ events: turnEvents(provider, index, false) }); - const context = yield* Effect.context>(); - const server = yield* Layer.build( - HttpRouter.serve(layerRoutes, { disableListenLog: true }).pipe( - Layer.provide(Layer.succeedContext(context)), - Layer.provideMerge( - NodeHttpServer.layer(NodeHttp.createServer, { - host: "127.0.0.1", - port: 0, - websocket: { perMessageDeflate: true }, - }), - ), - ), - ); - const address = Context.get(server, HttpServer.HttpServer).address; - if (!("port" in address)) return yield* Effect.die("Expected TCP server"); - const base = `http://127.0.0.1:${address.port}`; - const threadSnapshot = yield* measureHttpGet({ - url: `${base}/api/orchestration/threads/${THREAD_ID}`, - headers: { [ORCHESTRATION_PROTOCOL_HEADER]: String(ORCHESTRATION_PROTOCOL_VERSION) }, - }); - assert.equal( - threadSnapshot.status, - 200, - Buffer.from(threadSnapshot.decodedBody).toString(), - ); - assert.equal(threadSnapshot.contentEncoding, "gzip"); - const decodedThread = yield* decodeThreadSnapshot( - Buffer.from(threadSnapshot.decodedBody).toString(), - ); - assert.equal(decodedThread.projection.messages.length, TRANSFER_HISTORY_TURN_COUNT * 2); - assert.equal(decodedThread.projection.turnItems.length, TRANSFER_HISTORY_TURN_COUNT * 6); - assert.notInclude(Buffer.from(threadSnapshot.decodedBody).toString(), "digest="); - const shellSnapshot = yield* measureHttpGet({ - url: `${base}/api/orchestration/shell`, - headers: { [ORCHESTRATION_PROTOCOL_HEADER]: String(ORCHESTRATION_PROTOCOL_VERSION) }, - }); - assert.equal(shellSnapshot.status, 200); - const decodedShell = yield* decodeShellSnapshot( - Buffer.from(shellSnapshot.decodedBody).toString(), - ); - assert.equal(decodedShell.threads.length, 1); - const sequence = decodedThread.snapshotSequence; - assert.equal(decodedShell.snapshotSequence, sequence); - const url = `${base.replace("http:", "ws:")}/ws`; - const thread = yield* openMeasuredWsClient({ url, cookie: "" }); - const shell = yield* openMeasuredWsClient({ url, cookie: "" }); - const second = yield* openMeasuredWsClient({ url, cookie: "" }); - assert.include(thread.recorder.negotiatedExtensions(), "permessage-deflate"); - const tq = yield* subscribe(thread, "thread", sequence); - const sq = yield* subscribe(shell, "shell", sequence); - const ttq = yield* subscribe(second, "thread", sequence); - const ssq = yield* subscribe(second, "shell", sequence); - for (const queue of [tq, sq, ttq, ssq]) - assert.equal(yield* synchronized(queue), "replay"); - const threadBefore = thread.recorder.totals(); - const shellBefore = shell.recorder.totals(); - const secondBefore = second.recorder.totals(); - const sqlBefore = counter.count(); - const events = turnEvents(provider, TRANSFER_HISTORY_TURN_COUNT, true); - const stored = yield* sink.write({ events }); - const last = stored.at(-1)!; - const threadTerminal = (item: StreamItem) => - "sequence" in item && item.sequence >= last.sequence; - // Shell items carry the shell read when their batch is projected, and unchanged - // shells are skipped, so a turn split across batches may never yield an item at - // the last sequence. Wait for the final shell instead. - const finalShell = yield* projection.getThreadShell(THREAD_ID); - const shellTerminal = (item: StreamItem) => - item.kind === "thread.updated" && NodeUtil.isDeepStrictEqual(item.thread, finalShell); - for (const queue of [tq, ttq]) yield* collectUntil(queue, threadTerminal); - for (const queue of [sq, ssq]) yield* collectUntil(queue, shellTerminal); - const measuredTurnWebSocket = difference(thread.recorder.totals(), threadBefore); - const measuredTurnShellWebSocket = difference(shell.recorder.totals(), shellBefore); - const measuredTurnSecondClientWebSocket = difference( - second.recorder.totals(), - secondBefore, - ); - const measuredTurnSqlStatements = counter.count() - sqlBefore; - const durable = yield* projection.getThreadProjection(THREAD_ID); - assert.isTrue( - durable.turnItems.some( - (item) => - item.type === "dynamic_tool" && JSON.stringify(item.output).length > 1_000_000, - ), - ); - yield* second.close; - const reconnectSqlStart = counter.count(); - const resumedThread = yield* openMeasuredWsClient({ url, cookie: "" }); - const resumedShell = yield* openMeasuredWsClient({ url, cookie: "" }); - const rq = yield* subscribe(resumedThread, "thread", sequence); - const rsq = yield* subscribe(resumedShell, "shell", sequence); - const threadMode = yield* synchronized(rq); - const shellMode = yield* synchronized(rsq); - const reconnectThread = { ...resumedThread.recorder.totals(), mode: threadMode }; - const reconnectShell = { ...resumedShell.recorder.totals(), mode: shellMode }; - const reconnectSqlStatements = counter.count() - reconnectSqlStart; - assert.equal(threadMode, "snapshot"); - - for (const events of longTurnThreadBatches(provider)) yield* sink.write({ events }); - const longTurnColdOpen = yield* measureHttpGet({ - url: `${base}/api/orchestration/threads/${LONG_TURN_THREAD_ID}/bounded?compactTurnItems=1`, - headers: { [ORCHESTRATION_PROTOCOL_HEADER]: String(ORCHESTRATION_PROTOCOL_VERSION) }, - }); - assert.equal(longTurnColdOpen.status, 200); - const coldOpen = yield* decodeBoundedSnapshot( - Buffer.from(longTurnColdOpen.decodedBody).toString(), - ); - const cursor = coldOpen.historyCursor; - if (cursor === null) return yield* Effect.die("Expected older long-turn history"); - const longTurnHistoryPage = yield* measureHttpGet({ - url: `${base}/api/orchestration/threads/${LONG_TURN_THREAD_ID}/history?cursor=${encodeURIComponent(cursor)}`, - headers: { [ORCHESTRATION_PROTOCOL_HEADER]: String(ORCHESTRATION_PROTOCOL_VERSION) }, - }); - assert.equal(longTurnHistoryPage.status, 200); - const historyPage = yield* decodeHistoryPage( - Buffer.from(longTurnHistoryPage.decodedBody).toString(), - ); - assert.isTrue(historyPage.hasMoreHistory); - return { - provider, - threadSnapshot, - shellSnapshot, - measuredTurnWebSocket, - measuredTurnShellWebSocket, - measuredTurnSecondClientWebSocket, - measuredTurnSqlStatements, - reconnectThread, - reconnectShell, - reconnectSqlStatements, - longTurnColdOpen: { - http: longTurnColdOpen, - rows: coldOpen.projection.visibleTurnItems.length, - }, - longTurnHistoryPage: { http: longTurnHistoryPage, rows: historyPage.items.length }, - } satisfies TransferBudgetRun; - }).pipe( - Effect.provide(Layer.fresh(layerServices).pipe(Layer.provideMerge(NodeServices.layer))), - Effect.withTracer(counter.tracer), - ), - ); - runs.push(run); - } - const fs = yield* FileSystem.FileSystem; - const report = formatTransferBudgetReport(runs); - for (const [path, contents] of [ - [process.env.T3CODE_TRANSFER_BUDGET_REPORT_PATH, report], - [process.env.T3CODE_TRANSFER_BUDGET_RESULT_PATH, formatTransferBudgetResult(runs)], - ]) - if (path && contents) yield* fs.writeFileString(path, contents); - yield* Effect.log(report); - assert.deepEqual(transferBudgetViolations(runs), []); - }).pipe(Effect.provide(NodeServices.layer)), -); diff --git a/apps/server/package.json b/apps/server/package.json index 2a309f13d3b48..cb224f187d257 100644 --- a/apps/server/package.json +++ b/apps/server/package.json @@ -1,6 +1,6 @@ { "name": "t3", - "version": "0.1.2", + "version": "0.1.3", "license": "MIT", "type": "module", "scripts": { diff --git a/apps/server/scripts/acp-mock-agent.ts b/apps/server/scripts/acp-mock-agent.ts index e237cbe4ed10d..397c8e85fa13b 100644 --- a/apps/server/scripts/acp-mock-agent.ts +++ b/apps/server/scripts/acp-mock-agent.ts @@ -24,6 +24,7 @@ const emitToolCalls = process.env.T3_ACP_EMIT_TOOL_CALLS === "1"; const emitInterleavedAssistantToolCalls = process.env.T3_ACP_EMIT_INTERLEAVED_ASSISTANT_TOOL_CALLS === "1"; const emitV2Fidelity = process.env.T3_ACP_EMIT_V2_FIDELITY === "1"; +const emitNotice = process.env.T3_ACP_EMIT_NOTICE === "1"; const vibeRetryOutcome = process.env.T3_ACP_VIBE_RETRY_OUTCOME; const emitGenericToolPlaceholders = process.env.T3_ACP_EMIT_GENERIC_TOOL_PLACEHOLDERS === "1"; const emitPostSettleMonitorFlow = process.env.T3_ACP_EMIT_POST_SETTLE_MONITOR_FLOW === "1"; @@ -1170,6 +1171,30 @@ const program = Effect.gen(function* () { return yield* finishPrompt(requestedSessionId, "end_turn"); } + if (emitNotice) { + // A notice in fx 0.0.14's shape for a skipped skill, between two chunks of one message. + const chunk = (text: string) => + writeJsonRpcNotification("session/update", { + sessionId: requestedSessionId, + update: { + sessionUpdate: "agent_message_chunk", + messageId: "mock-notice-message", + content: { type: "text", text }, + }, + }); + chunk("before the notice, "); + writeJsonRpcNotification("session/update", { + sessionId: requestedSessionId, + update: { + sessionUpdate: "notice", + severity: "info", + title: "skill discovery warning: candidate was skipped because its metadata is invalid", + }, + }); + chunk("after the notice"); + return yield* finishPrompt(requestedSessionId, "end_turn"); + } + if (emitV2Fidelity) { yield* agent.client.sessionUpdate({ sessionId: `${requestedSessionId}-child`, diff --git a/apps/server/src/mcp/toolkits/html/tools.ts b/apps/server/src/mcp/toolkits/html/tools.ts index a8010fab4f522..0175b4021071f 100644 --- a/apps/server/src/mcp/toolkits/html/tools.ts +++ b/apps/server/src/mcp/toolkits/html/tools.ts @@ -25,7 +25,7 @@ const Html = Schema.String.check(Schema.isMinLength(1), Schema.isMaxLength(512_0 // This affects only the caller's thread, not its workspace, so read-only sandboxes and plan mode may use it. const HtmlRenderTool = Tool.make(HTML_RENDER_TOOL_NAME, { - description: `Show a finished HTML page inline in this thread, above your final reply. Do not announce or restate the page. Frame fits height to ${HTML_RENDER_MAX_FIT_HEIGHT}px. ${HTML_RENDER_LAYOUT_GUIDE} ${HTML_RENDER_THEME_GUIDE}`, + description: `Show a finished HTML page (chart, diagram, mockup) inline in this thread, above your final reply. Do not announce or restate the page. Frame fits height to ${HTML_RENDER_MAX_FIT_HEIGHT}px. ${HTML_RENDER_LAYOUT_GUIDE} ${HTML_RENDER_THEME_GUIDE}`, parameters: Schema.Struct({ html: Html, title: Schema.String.check( diff --git a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.testkit.ts b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.testkit.ts index bb74f75bd3463..d043ac09a6aee 100644 --- a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.testkit.ts +++ b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.testkit.ts @@ -194,6 +194,7 @@ export function makeAcpReplayRuntime(input: { readonly fileSystem: FileSystem.FileSystem; readonly customModels?: AcpSessionRuntime.AcpSessionRuntimeOptions["customModels"]; readonly cancelMeta?: AcpSessionRuntime.AcpSessionRuntimeOptions["cancelMeta"]; + readonly cancelBehavior?: AcpSessionRuntime.AcpSessionRuntimeOptions["cancelBehavior"]; readonly initializeMeta?: AcpSessionRuntime.AcpSessionRuntimeOptions["initializeMeta"]; readonly replayGate?: ProviderReplayGate; }): ( @@ -235,6 +236,7 @@ export function makeAcpReplayRuntime(input: { authMethodId: "replay", ...(input.customModels === undefined ? {} : { customModels: input.customModels }), ...(input.cancelMeta === undefined ? {} : { cancelMeta: input.cancelMeta }), + ...(input.cancelBehavior === undefined ? {} : { cancelBehavior: input.cancelBehavior }), ...(input.initializeMeta === undefined ? {} : { initializeMeta: input.initializeMeta }), ...(input.replayGate === undefined ? {} diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts index ff4a9c63f39d0..077e5f237ad8b 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts @@ -5023,6 +5023,138 @@ describe("ClaudeAdapterV2 background wake turns", () => { ), ); + it.effect("an interrupted turn after a rollback stops resuming at the rolled-back head", () => + Effect.gen(function* () { + const harness = yield* makeWakeHarnessWithOptions({ + close: (sdkMessages) => Queue.shutdown(sdkMessages), + freshQueueOnReopen: true, + }); + const now = yield* DateTime.now; + const pinnedHead = "00000000-0000-4000-8000-000000025510"; + const pinnedThread = { + ...harness.providerThread, + nativeConversationHeadRef: { + driver: ClaudeAdapterV2.CLAUDE_PROVIDER, + nativeId: pinnedHead, + strength: "weak" as const, + }, + }; + const interruptedInput = makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: pinnedThread, + now, + attemptId: RunAttemptId.make("attempt-claude-interrupt-after-rollback"), + text: "The secret word is apricot-25510. Then run sleep 60.", + attachments: [], + providerTurnOrdinal: 2, + nativeThreadHasTurns: true, + }); + yield* harness.runtime.startTurn(interruptedInput); + assert.equal(harness.getOpenedOptions()?.resumeSessionAt, pinnedHead); + yield* harness.offerAndWait( + makeAssistantTextFrame({ + uuid: "00000000-0000-4000-8000-000000025511", + text: "Noted apricot-25510. Running sleep.", + }), + ); + yield* harness.runtime.interruptTurn({ + providerThread: pinnedThread, + providerTurnId: (yield* IdAllocator.IdAllocatorV2).derive.providerTurn({ + driver: ClaudeAdapterV2.CLAUDE_PROVIDER, + nativeTurnId: `turn:${interruptedInput.attemptId}`, + }), + }); + assert.equal((yield* Queue.take(harness.terminalReceipts)).status, "interrupted"); + const afterInterrupt = harness.events.findLast( + (event) => event.type === "provider_thread.updated", + )?.providerThread; + assert.isDefined(afterInterrupt); + assert.isNull(afterInterrupt.nativeConversationHeadRef); + + yield* harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: afterInterrupt, + now, + attemptId: RunAttemptId.make("attempt-claude-after-interrupt"), + text: "What is the secret word?", + attachments: [], + providerTurnOrdinal: 3, + nativeThreadHasTurns: true, + }), + ); + assert.equal(harness.processQueues.length, 2); + assert.isUndefined(harness.getOpenedOptions()?.resumeSessionAt); + }).pipe( + Effect.provide( + Layer.mergeAll(IdAllocator.layer, McpProviderSessions.layer, NodeServices.layer), + ), + ), + ); + + it.effect("an interrupted turn with no assistant output keeps the rolled-back head", () => + Effect.gen(function* () { + const harness = yield* makeWakeHarnessWithOptions({ + close: (sdkMessages) => Queue.shutdown(sdkMessages), + freshQueueOnReopen: true, + }); + const now = yield* DateTime.now; + const pinnedHead = "00000000-0000-4000-8000-000000025510"; + const pinnedThread = { + ...harness.providerThread, + nativeConversationHeadRef: { + driver: ClaudeAdapterV2.CLAUDE_PROVIDER, + nativeId: pinnedHead, + strength: "weak" as const, + }, + }; + const interruptedInput = makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: pinnedThread, + now, + attemptId: RunAttemptId.make("attempt-claude-silent-interrupt-after-rollback"), + text: "The secret word is apricot-25510. Then run sleep 60.", + attachments: [], + providerTurnOrdinal: 2, + nativeThreadHasTurns: true, + }); + yield* harness.runtime.startTurn(interruptedInput); + assert.equal(harness.getOpenedOptions()?.resumeSessionAt, pinnedHead); + yield* harness.runtime.interruptTurn({ + providerThread: pinnedThread, + providerTurnId: (yield* IdAllocator.IdAllocatorV2).derive.providerTurn({ + driver: ClaudeAdapterV2.CLAUDE_PROVIDER, + nativeTurnId: `turn:${interruptedInput.attemptId}`, + }), + }); + assert.equal((yield* Queue.take(harness.terminalReceipts)).status, "interrupted"); + const afterInterrupt = harness.events.findLast( + (event) => event.type === "provider_thread.updated", + )?.providerThread; + assert.isDefined(afterInterrupt); + assert.equal(afterInterrupt.nativeConversationHeadRef?.nativeId, pinnedHead); + + yield* harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: afterInterrupt, + now, + attemptId: RunAttemptId.make("attempt-claude-after-silent-interrupt"), + text: "What is the secret word?", + attachments: [], + providerTurnOrdinal: 3, + nativeThreadHasTurns: true, + }), + ); + assert.equal(harness.processQueues.length, 2); + assert.equal(harness.getOpenedOptions()?.resumeSessionAt, pinnedHead); + }).pipe( + Effect.provide( + Layer.mergeAll(IdAllocator.layer, McpProviderSessions.layer, NodeServices.layer), + ), + ), + ); + it.effect("never records a subagent uuid for a turn with no main-thread assistant message", () => Effect.gen(function* () { const harness = yield* makeWakeHarnessWithOptions({ diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts index b3ab6b4448529..1c0bd3ceb1f35 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts @@ -7180,8 +7180,9 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( yield* Ref.get(awaitedBackgroundTasksByNativeThread), nativeThreadId, ); + // A stopped turn that wrote to the transcript moved past the pinned head. const clearConversationHead = - input.status === "completed" && + (input.status === "completed" || input.context.nativeMessageCursor !== null) && input.context.input.providerThread.nativeConversationHeadRef !== null; const goal = nativeThreadId === null ? undefined : goalsByNativeThread.get(nativeThreadId); diff --git a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts index 3bf563c1f794a..78bf5dcce18d7 100644 --- a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts +++ b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts @@ -974,6 +974,144 @@ it.layer(layerTest)("delegated completion delivery repairs", (it) => { ); }), ); + + it.effect.each([ + { command: "acknowledge", before: "running", after: "failed", capture: false }, + { command: "acknowledge", before: "waiting", after: "completed", capture: true }, + { command: "accept", before: "waiting", after: "completed", capture: true }, + ] as const)( + "keeps a $after run that settled between $command's read and commit", + ({ command, before, after, capture }) => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const sink = yield* EventSink.EventSinkV2; + const now = yield* DateTime.now; + const race = `${command}-settle-race-${after}`; + const threadId = ThreadId.make(`thread:${race}`); + const runId = RunId.make(`run:${race}`); + const taskId = NodeId.make(`node:${race}-task`); + const checkpointId = CheckpointId.make(`checkpoint:${race}`); + const messageId = MessageId.make(`message:delegated-delivery:${threadId}`); + yield* seedParentWithTerminalTask({ + threadId, + projectId: ProjectId.make(`project:${race}`), + runId, + rootNodeId: NodeId.make(`node:${race}-root`), + taskId, + deliveryState: "claimed", + deliveryTaskIds: [taskId], + now, + }); + const parent = (yield* orchestrator.getThreadProjection(threadId)).runs[0]!; + yield* sink.write({ + events: [ + { + id: EventId.make(`event:${race}-before`), + type: "run.updated", + threadId, + runId, + occurredAt: now, + payload: { ...parent, status: before }, + }, + { + id: EventId.make(`event:${race}-message`), + type: "message.updated", + threadId, + runId, + occurredAt: now, + payload: { + id: messageId, + threadId, + runId, + nodeId: parent.rootNodeId, + role: "user", + text: "Background task finished", + attachments: [], + streaming: false, + createdBy: "agent", + creationSource: "server", + createdAt: now, + updatedAt: now, + delegatedCompletion: { parentRunId: runId, generation: 1, taskIds: [taskId] }, + }, + }, + ], + }); + const { delegatedCompletion: _cohort, ...settledRun } = parent; + const commit = sink.commitCommand; + let settled = false; + yield* Effect.acquireUseRelease( + Effect.sync(() => + vi.spyOn(sink, "commitCommand").mockImplementation((input) => + Effect.gen(function* () { + if ( + !settled && + input.commandType === + (command === "accept" + ? "notification.delivery.accept" + : "delegated_task.completion-delivery.acknowledge") + ) { + settled = true; + // The provider turn ends (or its checkpoint lands) after the command read the run. + yield* commit({ + commandId: CommandId.make(`command:${race}`), + threadId, + commandType: capture ? "checkpoint.capture" : "provider.event", + acceptedAt: now, + effects: [], + events: [ + { + id: EventId.make(`event:${race}-settled`), + type: "run.updated", + threadId, + runId, + occurredAt: now, + payload: { + ...settledRun, + status: after, + completedAt: now, + checkpointId: capture ? checkpointId : null, + }, + }, + ], + }); + } + return yield* commit(input); + }), + ), + ), + () => + orchestrator.dispatch( + command === "accept" + ? { + type: "notification.delivery.accept", + commandId: CommandId.make(`command:${race}:accept`), + threadId, + messageId, + } + : { + type: "delegated_task.completion-delivery.acknowledge", + commandId: CommandId.make(`command:${race}:task-status`), + parentThreadId: threadId, + taskId, + observedByRunId: runId, + }, + ), + (spy) => Effect.sync(() => spy.mockRestore()), + ); + assert.isTrue(settled); + const finished = yield* orchestrator.getThreadProjection(threadId); + assert.equal( + finished.subagents.find((row) => row.id === taskId)?.completionDelivery?.state, + command === "accept" ? "delivered" : "acknowledged", + ); + const run = finished.runs.find((candidate) => candidate.id === runId)!; + assert.deepEqual( + { status: run.status, completedAt: run.completedAt, checkpointId: run.checkpointId }, + { status: after, completedAt: now, checkpointId: capture ? checkpointId : null }, + ); + }), + ); }); const reconcileCommandId = (name: string) => CommandId.make(`command:runtime-reconcile:${name}`); diff --git a/apps/server/src/orchestration-v2/EventSink.ts b/apps/server/src/orchestration-v2/EventSink.ts index 9b81ef66e86bb..b40d280e17c3c 100644 --- a/apps/server/src/orchestration-v2/EventSink.ts +++ b/apps/server/src/orchestration-v2/EventSink.ts @@ -145,6 +145,8 @@ export interface EventSinkV2Shape { readonly effectTypes: ReadonlyArray; readonly reason: string; }; + /** Like writeWithEffects' updateDelegatedCompletionOnly, for this run's run.updated only. */ + readonly updateDelegatedCompletionOnlyRunId?: RunId; }) => Effect.Effect< { readonly receipt: CommandReceiptStore.CommandReceiptV2; @@ -366,6 +368,28 @@ const layerBase: Layer.Layer< } }); + // The enclosing write or commit span already traces this per-event projection merge. + const updateDelegatedCompletionOnly = Effect.fnUntraced(function* ( + events: ReadonlyArray, + includesRun: (runId: RunId) => boolean, + ) { + return yield* Effect.forEach(events, (event) => + Effect.gen(function* () { + if (event.type !== "run.updated" || !includesRun(event.payload.id)) return event; + const { runs } = yield* projectionStore.getThreadRecords(event.threadId, ["runs"], { + runIds: [event.payload.id], + }); + const current = runs[0]; + return current === undefined + ? event + : { + ...event, + payload: { ...current, delegatedCompletion: event.payload.delegatedCompletion }, + }; + }), + ); + }); + const writeEffect = Effect.fn("orchestrationV2.EventSink.write")(function* ( input: Parameters[0], ) { @@ -394,26 +418,7 @@ const layerBase: Layer.Layer< // Checkpoint capture can finish after a cohort writer reads its parent run. // Merge in this transaction so the event log and projection keep that completion. const events = input.updateDelegatedCompletionOnly - ? yield* Effect.forEach(input.events, (event) => - Effect.gen(function* () { - if (event.type !== "run.updated") return event; - const { runs } = yield* projectionStore.getThreadRecords( - event.threadId, - ["runs"], - { runIds: [event.payload.id] }, - ); - const current = runs[0]; - return current === undefined - ? event - : { - ...event, - payload: { - ...current, - delegatedCompletion: event.payload.delegatedCompletion, - }, - }; - }), - ) + ? yield* updateDelegatedCompletionOnly(input.events, () => true) : input.events; const normalized = yield* normalizeEvents( input.guardPendingUserInputCancellations === true @@ -622,7 +627,12 @@ const layerBase: Layer.Layer< return { ...existing, committed: false as const, cancelledEffectIds: [] }; } - const normalized = yield* normalizeEvents(input.events); + const runId = input.updateDelegatedCompletionOnlyRunId; + const normalized = yield* normalizeEvents( + runId === undefined + ? input.events + : yield* updateDelegatedCompletionOnly(input.events, (id) => id === runId), + ); const storedEvents = yield* eventStore.append({ commandId: input.commandId, events: normalized, diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index dbf51b67b00af..fc4cf75c8cd34 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -23,6 +23,7 @@ import { type ChatAttachment, CommandId, type GuestMessageId, + isCompactionCommandMessage, isOrchestrationV2WorkActive, isProviderNativeSubagentThread, MessageId, @@ -2357,7 +2358,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio run: deliveryRun, now, }); - return; + return parentRun.id; } const message = completionDeliveryMessage(projection, delivery); if (message !== undefined) { @@ -2381,6 +2382,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio }); } } + return parentRun.id; }); const disposeDelegatedCompletionCohort = Effect.fn( @@ -5772,13 +5774,14 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio }); } if (goalStopRunId !== undefined) { - return yield* stopRunForGoalCommand({ + return yield* stopRunForQueuedCommand({ command, events, effects, projection: yield* getProjectionWithPendingEvents(command.threadId, events), queuedRunId: runId, targetRunId: goalStopRunId, + reason: "Stopped to apply /goal", now, }); } @@ -8353,15 +8356,46 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio (yield* goalDriverFor(command, queuedRun.providerInstanceId)) === undefined ? null : parsedGoalCommand; + const target = projection.runs.find((run) => run.id === command.targetRunId); + const targetMessage = projection.messages.find( + (message) => message.id === target?.userMessageId, + ); + // A compacting turn cannot take steering input, so steering skips the compaction instead. + if ( + target !== undefined && + targetMessage !== undefined && + isCompactionCommandMessage(targetMessage) + ) { + if (goalCommand !== null) { + yield* ensureGoalCommandSupported(command, queuedMessage, queuedRun.providerInstanceId); + } + yield* emitEvent({ + type: "run.updated", + threadId: command.threadId, + runId: queuedRun.id, + nodeId: queuedRootNode.id, + providerInstanceId: queuedRun.providerInstanceId, + occurredAt: now, + payload: { ...queuedRun, queuePosition: 1 }, + }); + const stops = + isLiveRun(target) && !(yield* stopReachedRun(command, command.threadId, target.id)); + return yield* stopRunForQueuedCommand({ + command, + events, + effects, + projection, + queuedRunId: queuedRun.id, + targetRunId: stops ? target.id : undefined, + reason: "Skipped compaction to send a queued message", + now, + }); + } if (goalCommand !== null) { yield* ensureGoalCommandSupported(command, queuedMessage, queuedRun.providerInstanceId); - const target = projection.runs.find((run) => run.id === command.targetRunId); const goal = projection.providerThreads.find( (candidate) => candidate.id === target?.providerThreadId, )?.goal; - const targetMessage = projection.messages.find( - (message) => message.id === target?.userMessageId, - ); yield* emitEvent({ type: "run.updated", threadId: command.threadId, @@ -8377,13 +8411,14 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio goalCommandStopsActiveRun(goalCommand, goal?.status ?? null) && (targetMessage === undefined || !isNativeMaintenanceCommand(targetMessage)) && !(yield* stopReachedRun(command, command.threadId, target.id)); - return yield* stopRunForGoalCommand({ + return yield* stopRunForQueuedCommand({ command, events, effects, projection, queuedRunId: queuedRun.id, targetRunId: stops ? target.id : undefined, + reason: "Stopped to apply /goal", now, }); } @@ -10494,71 +10529,74 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ); /** Puts a queued goal command first and, given a target, stops the run in its way. */ - const stopRunForGoalCommand = Effect.fn("Orchestrator.stopRunForGoalCommand")(function* (input: { - readonly command: Extract< - OrchestrationV2Command, - { readonly type: "message.dispatch" | "queued-message.promote-to-steer" } - >; - readonly events: Ref.Ref>; - readonly effects: Ref.Ref>; - readonly projection: Pick; - readonly queuedRunId: RunId; - readonly targetRunId: RunId | undefined; - readonly now: DateTime.Utc; - }) { - const { command, events, effects, projection, now, targetRunId } = input; - const others = queuedRunsInDeliveryOrder(projection).filter( - (run) => run.id !== input.queuedRunId && !isAutomaticCompletionRun(projection, run), - ); - for (const [index, run] of others.entries()) { - const queuePosition = index + 2; - if (run.queuePosition === queuePosition) continue; - yield* emit( - events, - command, - )({ - type: "run.updated", - threadId: command.threadId, - runId: run.id, - ...(run.rootNodeId === null ? {} : { nodeId: run.rootNodeId }), - providerInstanceId: run.providerInstanceId, - occurredAt: now, - payload: { ...run, queuePosition }, - }); - } - if (targetRunId === undefined) return undefined; - const interruptEvents = yield* Ref.make>([]); - const interruptEffects = yield* Ref.make>([]); - const interrupted = yield* Effect.exit( - dispatchRunInterrupt( - { - type: "run.interrupt", - commandId: command.commandId, + const stopRunForQueuedCommand = Effect.fn("Orchestrator.stopRunForQueuedCommand")( + function* (input: { + readonly command: Extract< + OrchestrationV2Command, + { readonly type: "message.dispatch" | "queued-message.promote-to-steer" } + >; + readonly events: Ref.Ref>; + readonly effects: Ref.Ref>; + readonly projection: Pick; + readonly queuedRunId: RunId; + readonly targetRunId: RunId | undefined; + readonly reason: string; + readonly now: DateTime.Utc; + }) { + const { command, events, effects, projection, now, targetRunId } = input; + const others = queuedRunsInDeliveryOrder(projection).filter( + (run) => run.id !== input.queuedRunId && !isAutomaticCompletionRun(projection, run), + ); + for (const [index, run] of others.entries()) { + const queuePosition = index + 2; + if (run.queuePosition === queuePosition) continue; + yield* emit( + events, + command, + )({ + type: "run.updated", + threadId: command.threadId, + runId: run.id, + ...(run.rootNodeId === null ? {} : { nodeId: run.rootNodeId }), + providerInstanceId: run.providerInstanceId, + occurredAt: now, + payload: { ...run, queuePosition }, + }); + } + if (targetRunId === undefined) return undefined; + const interruptEvents = yield* Ref.make>([]); + const interruptEffects = yield* Ref.make>([]); + const interrupted = yield* Effect.exit( + dispatchRunInterrupt( + { + type: "run.interrupt", + commandId: command.commandId, + threadId: command.threadId, + runId: targetRunId, + reason: input.reason, + }, + interruptEvents, + interruptEffects, + ), + ); + // A failed interrupt may have written part of its events, so they are kept apart. + // The queued command then waits for the turn to end. + if (Exit.isFailure(interrupted)) { + yield* Effect.logWarning("queued command could not stop the running turn", { threadId: command.threadId, runId: targetRunId, - reason: "Stopped to apply /goal", - }, - interruptEvents, - interruptEffects, - ), - ); - // A failed interrupt may have written part of its events, so they are kept apart. - // The goal command then waits for the turn to end. - if (Exit.isFailure(interrupted)) { - yield* Effect.logWarning("goal command could not stop the running turn", { - threadId: command.threadId, - runId: targetRunId, - cause: interrupted.cause, - }); - return undefined; - } - const written = yield* Ref.get(interruptEvents); - const enqueued = yield* Ref.get(interruptEffects); - yield* Ref.update(events, (existing) => [...existing, ...written]); - yield* Ref.update(effects, (existing) => [...existing, ...enqueued]); - // Like Stop: a run stopped before its provider started also abandons that start. - return interrupted.value; - }); + cause: interrupted.cause, + }); + return undefined; + } + const written = yield* Ref.get(interruptEvents); + const enqueued = yield* Ref.get(interruptEffects); + yield* Ref.update(events, (existing) => [...existing, ...written]); + yield* Ref.update(effects, (existing) => [...existing, ...enqueued]); + // Like Stop: a run stopped before its provider started also abandons that start. + return interrupted.value; + }, + ); const dispatchThreadStop = Effect.fn("Orchestrator.dispatchThreadStop")(function* ( command: Extract, @@ -11731,6 +11769,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio }, }, }); + return parentRun.id; }); const dispatchUnsupported = (command: OrchestrationV2ServerCommand) => @@ -11751,6 +11790,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio readonly effectTypes: ReadonlyArray; readonly reason: string; }; + readonly updateDelegatedCompletionOnlyRunId?: RunId; }, OrchestratorV2Error > { @@ -11768,6 +11808,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio readonly reason: string; } | undefined; + let updateDelegatedCompletionOnlyRunId: RunId | undefined; switch (command.type) { case "thread.create": yield* dispatchThreadCreate(command, events); @@ -11903,7 +11944,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio break; } case "notification.delivery.accept": - yield* dispatchNotificationAccepted(command, events); + updateDelegatedCompletionOnlyRunId = yield* dispatchNotificationAccepted(command, events); break; case "prepared-run.release": yield* dispatchPreparedRunRelease(command, events, effects); @@ -12079,7 +12120,8 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio break; case "delegated_task.completion-delivery.acknowledge": case "delegated_task.completion-delivery.dispose": - yield* dispatchDelegatedTaskCompletionDeliveryResolution(command, events); + updateDelegatedCompletionOnlyRunId = + yield* dispatchDelegatedTaskCompletionDeliveryResolution(command, events); break; case "thread.created.record": yield* dispatchCreatedThreadRecord(command, events); @@ -12094,6 +12136,9 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio events: yield* Ref.get(events), effects: yield* Ref.get(effects), ...(cancelUnsettledEffects === undefined ? {} : { cancelUnsettledEffects }), + ...(updateDelegatedCompletionOnlyRunId === undefined + ? {} + : { updateDelegatedCompletionOnlyRunId }), }; }); @@ -12270,6 +12315,9 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ...(plan.cancelUnsettledEffects === undefined ? {} : { cancelUnsettledEffects: plan.cancelUnsettledEffects }), + ...(plan.updateDelegatedCompletionOnlyRunId === undefined + ? {} + : { updateDelegatedCompletionOnlyRunId: plan.updateDelegatedCompletionOnlyRunId }), }) .pipe( Effect.mapError( diff --git a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts index 9d9ae2e646cd5..049f8e9a692d4 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts @@ -93,12 +93,14 @@ const isProviderTurnStartError = Schema.is(ProviderTurnStartError); /** * Claude refuses these before it reads the prompt: replacing a process running background work, - * or launch arguments whose appended system prompt cannot be read. + * launch arguments whose appended system prompt cannot be read, or MCP registration that never + * answers while the query opens. */ const PROMPT_REFUSAL_TAGS = [ "ClaudeBackgroundWorkBlocksQueryReplacementError", "ClaudeAppendSystemPromptFileError", "ClaudeAppendSystemPromptMissingValueError", + "ClaudeMcpRegistrationTimeoutError", ] as const; const refusedBeforePrompt = (error: unknown): boolean => diff --git a/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts b/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts index 2cc6ee6f825b2..4c2d6712c5ad7 100644 --- a/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts +++ b/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts @@ -429,6 +429,29 @@ const nextTurnSelectionHarness = Effect.fn("nextTurnSelectionHarness")(function* openSession: (input) => Effect.gen(function* () { const now = yield* DateTime.now; + const startTurn = (turn: ProviderAdapter.ProviderAdapterV2TurnInput) => + Effect.gen(function* () { + started.push(turn); + yield* Queue.offer(events, { + type: "provider_turn.updated", + driver, + providerTurn: { + id: ProviderTurnId.make(`provider-turn:${turn.attemptId}`), + providerThreadId: turn.providerThread.id, + nodeId: turn.rootNodeId, + runAttemptId: turn.attemptId, + nativeTurnRef: { + driver, + nativeId: `native:${turn.attemptId}`, + strength: "strong", + }, + ordinal: turn.providerTurnOrdinal, + status: "running", + startedAt: now, + completedAt: null, + }, + }); + }); return { instanceId, driver, @@ -465,29 +488,8 @@ const nextTurnSelectionHarness = Effect.fn("nextTurnSelectionHarness")(function* updatedAt: now, }), resumeThread: ({ providerThread }) => Effect.succeed(providerThread), - startTurn: (turn) => - Effect.gen(function* () { - started.push(turn); - yield* Queue.offer(events, { - type: "provider_turn.updated", - driver, - providerTurn: { - id: ProviderTurnId.make(`provider-turn:${turn.attemptId}`), - providerThreadId: turn.providerThread.id, - nodeId: turn.rootNodeId, - runAttemptId: turn.attemptId, - nativeTurnRef: { - driver, - nativeId: `native:${turn.attemptId}`, - strength: "strong", - }, - ordinal: turn.providerTurnOrdinal, - status: "running", - startedAt: now, - completedAt: null, - }, - }); - }), + startTurn, + compactThread: startTurn, steerTurn: (turn) => Effect.sync(() => { steered.push(turn.message.text); @@ -1167,6 +1169,96 @@ it.effect("promotes a queued goal by stopping the running turn", () => ), ); +it.effect.each(["next", "/goal ship it"])( + "promoting %s into a compaction stops it and sends it next", + (promoted) => + Effect.scoped( + Effect.gen(function* () { + const { events, layer, send, read, started, steered, startFirstTurn } = yield* goalHarness( + `compaction-promote-${promoted.length}`, + ); + yield* Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; + const threadId = yield* startFirstTurn; + yield* send("compact", "/compact", "queue"); + yield* send("queued", "queued", "queue"); + yield* send("next", promoted, "queue"); + const settled = yield* orchestrator.streamDomainEvents.pipe( + Stream.filter( + (event) => + event.type === "run.updated" && + event.payload.id === started[0]!.runId && + event.payload.status === "waiting", + ), + Stream.take(1), + Stream.runDrain, + Effect.forkScoped, + ); + const compacting = yield* orchestrator.streamDomainEvents.pipe( + Stream.filter( + (event) => + event.type === "provider-turn.updated" && + event.payload.status === "running" && + event.payload.runAttemptId !== started[0]!.attemptId, + ), + Stream.take(1), + Stream.runDrain, + Effect.forkScoped, + ); + const turn = (yield* orchestrator.getThreadProjection(threadId)).providerTurns[0]!; + yield* Queue.offer(events, { + type: "provider_turn.updated", + driver, + providerTurn: { ...turn, status: "completed", completedAt: yield* DateTime.now }, + }); + yield* Queue.offer(events, { + type: "turn.terminal", + driver, + providerThreadId: turn.providerThreadId, + providerTurnId: turn.id, + runOrdinal: started[0]!.runOrdinal, + status: "completed", + failure: null, + threadDisposition: "reusable", + }); + yield* Fiber.join(settled); + yield* worker.drain(); + yield* orchestrator.resumeQueuedRuns; + yield* worker.drain(); + yield* Fiber.join(compacting); + const compactRun = (yield* orchestrator.getThreadProjection(threadId)).runs.find( + (run) => run.userMessageId === MessageId.make("message:compact"), + )!; + const nextRun = (yield* orchestrator.getThreadProjection(threadId)).runs.find( + (run) => run.userMessageId === MessageId.make("message:next"), + )!; + yield* orchestrator.dispatch({ + type: "queued-message.promote-to-steer", + commandId: CommandId.make("promote"), + threadId, + queuedRunId: nextRun.id, + targetRunId: compactRun.id, + }); + yield* worker.drain(); + + const projection = yield* orchestrator.getThreadProjection(threadId); + assert.deepEqual(steered, []); + assert.isTrue( + projection.turnItems.some( + (item) => item.type === "run_interrupt_request" && item.runId === compactRun.id, + ), + ); + assert.deepEqual((yield* read(threadId)).queue, ["queued"]); + assert.deepEqual( + started.map((turn) => turn.message.text), + ["first", "/compact", promoted], + ); + }).pipe(Effect.provide(layer)); + }), + ), +); + it.effect("stops a turn that is still starting when a goal is sent into it", () => Effect.scoped( Effect.gen(function* () { diff --git a/apps/server/src/orchestration-v2/testkit/ProviderReplayFixtures.ts b/apps/server/src/orchestration-v2/testkit/ProviderReplayFixtures.ts index 7b4007256a222..bc829b611a8b5 100644 --- a/apps/server/src/orchestration-v2/testkit/ProviderReplayFixtures.ts +++ b/apps/server/src/orchestration-v2/testkit/ProviderReplayFixtures.ts @@ -25,7 +25,7 @@ import { materializeReplayTranscriptRuntimeInstructions } from "./ReplayRuntimeI // Each driver's fixtures run from its own ReplayFixtures.integration.test.ts, // so a focused run loads one adapter. A fixture for a driver missing here would // silently never run, so loading this module fails instead. -// Coder: the cut drivers' fixtures (cursor, grok, muse, opencode) are not carried. +// Coder: the cut drivers' fixtures (cursor, fx, grok, muse, opencode) are not carried. const DRIVERS_WITH_REPLAY_FIXTURE_FILES = new Set(["acpRegistry", "claudeAgent", "codex", "pi"]); for (const fixture of ORCHESTRATOR_REPLAY_FIXTURES) { for (const provider of fixture.providers) { diff --git a/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts b/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts index ab1a74337e5c0..658c2b364e724 100644 --- a/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts +++ b/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts @@ -36,6 +36,7 @@ import * as SqlClient from "effect/sql/SqlClient"; import * as SqlitePersistence from "../../persistence/Sqlite.ts"; import { CommandPolicyCapabilityUnsupportedError } from "../CommandPolicy.ts"; import { + ClaudeAgentSdkQueryRunnerError, ClaudeAppendSystemPromptFileError, ClaudeBackgroundWorkBlocksQueryReplacementError, ClaudeProviderCapabilitiesV2, @@ -404,6 +405,12 @@ const waitForIdle = Effect.fn("ProviderSwitchTest.waitForIdle")(function* ( return idle; }); +// Same tag as the adapter's unexported error raised when Claude never answers setMcpServers. +class ClaudeMcpRegistrationTimeoutError extends Schema.TaggedError()( + "ClaudeMcpRegistrationTimeoutError", + {}, +) {} + describe("orchestration v2 provider switching", () => { it.live.each([ "compact-native", @@ -1311,6 +1318,13 @@ describe("orchestration v2 provider switching", () => { it.live.each([ ["background work", new ClaudeBackgroundWorkBlocksQueryReplacementError()], ["an unreadable append file", new ClaudeAppendSystemPromptFileError({ path: "missing.md" })], + [ + "an MCP registration timeout", + new ClaudeAgentSdkQueryRunnerError({ + method: "setMcpServers", + cause: new ClaudeMcpRegistrationTimeoutError(), + }), + ], ] as const)("keeps the native session after turns refused for %s", ([, refusalCause]) => Effect.scoped( Effect.gen(function* () { diff --git a/apps/server/src/orchestration-v2/testkit/ReplayRuntimeInstructions.ts b/apps/server/src/orchestration-v2/testkit/ReplayRuntimeInstructions.ts index 7099d68a19220..661856720fc27 100644 --- a/apps/server/src/orchestration-v2/testkit/ReplayRuntimeInstructions.ts +++ b/apps/server/src/orchestration-v2/testkit/ReplayRuntimeInstructions.ts @@ -13,7 +13,9 @@ export function materializeReplayTranscriptRuntimeInstructions( ? "Grok" : runtime.driver === "acpRegistry" ? "acpRegistry" - : undefined; + : runtime.driver === "fx" + ? "fx" + : undefined; if (harness === undefined) return transcript; const instructions = buildRuntimeInstructions({ harness, model: runtime.model }); diff --git a/apps/server/src/provider/model-manifest.json b/apps/server/src/provider/model-manifest.json index ce0a47fe56f48..9d2f5ae7256bc 100644 --- a/apps/server/src/provider/model-manifest.json +++ b/apps/server/src/provider/model-manifest.json @@ -95,6 +95,15 @@ { "range": ">=1.4.2", "status": "supported" }, { "range": "<1.4.2", "status": "unknown" } ] + }, + { + "driver": "fx", + "t3CodeRange": ">=0.0.46", + "recommendedRange": ">=0.0.13", + "ranges": [ + { "range": ">=0.0.13", "status": "supported" }, + { "range": "<0.0.13", "status": "unsupported" } + ] } ], "currentModels": { diff --git a/apps/server/src/provider/providerStatusCache.ts b/apps/server/src/provider/providerStatusCache.ts index a4f0a9a385546..6566748349f6a 100644 --- a/apps/server/src/provider/providerStatusCache.ts +++ b/apps/server/src/provider/providerStatusCache.ts @@ -33,6 +33,7 @@ const BUILT_IN_DRIVER_ORDER: ReadonlyArray = [ "grok", "opencode", "antigravity", + "fx", ]; const driverRank = (driver: string): number => { diff --git a/apps/web/index.html b/apps/web/index.html index 13518deb6e5d1..a7a5e7450b3de 100644 --- a/apps/web/index.html +++ b/apps/web/index.html @@ -6,14 +6,14 @@ name="viewport" content="width=device-width, initial-scale=1.0, viewport-fit=cover, interactive-widget=resizes-content" /> - +