diff --git a/apps/server/src/device/DeviceHost.ts b/apps/server/src/device/DeviceHost.ts index 14537985b9b8..f8bb80b5b4e8 100644 --- a/apps/server/src/device/DeviceHost.ts +++ b/apps/server/src/device/DeviceHost.ts @@ -39,6 +39,39 @@ export class DeviceHostTimeoutError extends Schema.TaggedError()( + "AgentDeviceStartError", + { + hostId: Schema.String, + stateDir: Schema.String, + exitCode: Schema.NullOr(Schema.Int), + reason: Schema.Literals([ + "daemon_lock_not_directory", + "daemon_registration_unproven", + "daemon_retirement_unconfirmed", + "command_failed", + ]), + cause: Schema.Defect(), + }, +) { + override get message(): string { + const exit = this.exitCode === null ? "" : ` (exit code ${this.exitCode})`; + const detail = + this.reason === "daemon_lock_not_directory" + ? "Daemon registration ownership could not be verified: daemon.lock is not a directory." + : this.reason === "daemon_registration_unproven" + ? "Daemon registration ownership could not be verified." + : this.reason === "daemon_retirement_unconfirmed" + ? "Daemon replacement could not be confirmed." + : "The AgentDevice startup command failed."; + const recovery = + this.reason === "command_failed" + ? `Inspect daemon diagnostics in ${this.stateDir}.` + : `Inspect daemon.lock and daemon diagnostics in ${this.stateDir}. Stop older clients and daemons with their original CLI before manually repairing retained state.`; + return `Agent tools failed to start${exit}. ${detail} ${recovery}`; + } +} + export interface DeviceHubEndpoint { /** Loopback origin of expo-device-hub, e.g. `http://127.0.0.1:3400`. */ readonly origin: string; @@ -96,7 +129,7 @@ export class DeviceHost extends Context.Service< onPhase: (phase: "installing" | "starting", detail?: string) => Effect.Effect, ) => Effect.Effect< DeviceHostAgentReady, - DeviceHostError | DeviceHostTimeoutError | NodeRuntimeUnavailableError + DeviceHostError | DeviceHostTimeoutError | AgentDeviceStartError | NodeRuntimeUnavailableError >; /** Current endpoints when already running, without starting anything. */ readonly current: Effect.Effect; diff --git a/apps/server/src/device/DeviceService.test.ts b/apps/server/src/device/DeviceService.test.ts index 55a7ac28a67c..82ff8cec8978 100644 --- a/apps/server/src/device/DeviceService.test.ts +++ b/apps/server/src/device/DeviceService.test.ts @@ -66,7 +66,10 @@ const fixture = Effect.fn("fixture")(function* ( onBoot: Effect.Effect = Effect.void, bootError?: string, failListAfterShutdown = false, - runtimeFailure?: NodeRuntimeUnavailableError | DeviceHost.DeviceHostError, + runtimeFailure?: + | NodeRuntimeUnavailableError + | DeviceHost.DeviceHostError + | DeviceHost.AgentDeviceStartError, inspectError = false, installTool?: Parameters[3], ) { @@ -107,7 +110,8 @@ const fixture = Effect.fn("fixture")(function* ( platformAvailability: (platform) => Effect.succeed({ platform, available: true }), ensureReady: (onPhase) => Effect.gen(function* () { - if (runtimeFailure) return yield* runtimeFailure; + if (runtimeFailure && runtimeFailure._tag !== "AgentDeviceStartError") + return yield* runtimeFailure; starts.push("start"); yield* onPhase("installing", "Updating device hub from 0.9.0 to 0.10.1…"); return ready; @@ -217,6 +221,36 @@ const fixture = Effect.fn("fixture")(function* ( }); describe("device setup consent", () => { + it.effect("exposes daemon startup diagnostics through agent readiness and host status", () => + Effect.gen(function* () { + const startupError = new DeviceHost.AgentDeviceStartError({ + hostId: LOCAL_DEVICE_HOST_ID, + stateDir: "/test/device/agent-device", + exitCode: 1, + reason: "daemon_lock_not_directory", + cause: { stdout: "private startup diagnostics", code: 1 }, + }); + const { service, settings } = yield* fixture(Effect.void, undefined, false, startupError); + yield* Ref.update(settings, (current) => ({ + ...current, + enableDeviceSupport: true, + enableAgentDeviceAccess: true, + })); + const error = yield* service.agentReadinessIfSupported().pipe(Effect.flip); + expect(error).toMatchObject({ + _tag: "DeviceHostUnavailableError", + reason: startupError.message, + cause: startupError, + }); + expect(error.message).toContain("daemon.lock is not a directory"); + expect(error.message).not.toContain("private startup diagnostics"); + expect((yield* service.state).hostStatuses[LOCAL_DEVICE_HOST_ID]).toEqual({ + status: "failed", + detail: startupError.message, + }); + }).pipe(Effect.scoped), + ); + it.effect( "preserves missing-runtime guidance and causes through manual and agent readiness", () => diff --git a/apps/server/src/device/DeviceService.ts b/apps/server/src/device/DeviceService.ts index 4226755d48eb..3d1da7baa49a 100644 --- a/apps/server/src/device/DeviceService.ts +++ b/apps/server/src/device/DeviceService.ts @@ -337,9 +337,11 @@ export const makeWithHosts = Effect.fn("DeviceService.makeWithHosts")(function* ? nodeRuntimeUnavailableMessage("Local device support") : error._tag === "DeviceHostTimeoutError" ? `Agent tools did not start within ${error.timeoutMs} ms.` - : error.step === "probe" - ? "Could not connect to this host over SSH." - : `Device support failed during ${error.step}.`, + : error._tag === "AgentDeviceStartError" + ? error.message + : error.step === "probe" + ? "Could not connect to this host over SSH." + : `Device support failed during ${error.step}.`, cause: error, }), ), diff --git a/apps/server/src/device/LocalDeviceHost.test.ts b/apps/server/src/device/LocalDeviceHost.test.ts index 288b70763a9a..eef93e4bdaec 100644 --- a/apps/server/src/device/LocalDeviceHost.test.ts +++ b/apps/server/src/device/LocalDeviceHost.test.ts @@ -6,14 +6,236 @@ import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; import * as Path from "effect/Path"; import * as FileSystem from "effect/FileSystem"; +import * as Sink from "effect/Sink"; +import * as Stream from "effect/Stream"; import * as LocalDeviceHost from "./LocalDeviceHost.ts"; import * as NodeServices from "@effect/platform-node/NodeServices"; import * as ChildProcessSpawner from "effect/process/ChildProcessSpawner"; -import { HttpClient } from "effect/http"; +import { HttpClient, HttpClientResponse } from "effect/http"; import * as NetService from "@t3tools/shared/Net"; import * as ServerConfig from "../config.ts"; import * as ProcessRunner from "../processRunner.ts"; +import { + AGENT_DEVICE_VERSION, + DEVICE_HUB_VERSION, + agentDeviceStateDir, +} from "./DeviceToolchain.ts"; + +const processOutput = (code: number, stdout = "", stderr = "", timedOut = false) => ({ + code: ChildProcessSpawner.ExitCode(code), + stdout, + stderr, + timedOut, + stdoutTruncated: false, + stderrTruncated: false, + stdoutInvalidUtf8: false, + stderrInvalidUtf8: false, +}); + +const agentFixture = Effect.fn("agentFixture")(function* ( + startup: ( + stateDir: string, + ) => Effect.Effect, +) { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const baseDir = yield* fs.makeTempDirectoryScoped({ prefix: "t3-device-agent-start-" }); + const stateDir = agentDeviceStateDir(path, path.join(baseDir, "userdata")); + for (const [name, version, entry] of [ + ["expo-device-hub", DEVICE_HUB_VERSION, "dist/server/cli.mjs"], + ["agent-device", AGENT_DEVICE_VERSION, "bin/agent-device.mjs"], + ]) { + const installDir = path.join(baseDir, "tools", name!, version!); + const entryPath = path.join(installDir, "node_modules", name!, entry!); + yield* fs.makeDirectory(path.dirname(entryPath), { recursive: true }); + yield* fs.writeFileString(entryPath, ""); + yield* fs.writeFileString(path.join(installDir, ".install-complete"), version!); + } + const host = yield* LocalDeviceHost.make().pipe( + Effect.provide(Layer.mergeAll(ServerConfig.layerTest(baseDir, baseDir), NetService.layer)), + Effect.provideService(HostProcess.Environment, { HOME: baseDir, PATH: "" }), + Effect.provideService(HostProcess.IsExecutable, false), + Effect.provideService(HostProcess.Platform, "darwin"), + Effect.provideService( + ChildProcessSpawner.ChildProcessSpawner, + ChildProcessSpawner.make(() => + Effect.succeed( + ChildProcessSpawner.makeHandle({ + pid: ChildProcessSpawner.ProcessId(123), + stdout: Stream.empty, + stderr: Stream.empty, + all: Stream.empty, + exitCode: Effect.never, + isRunning: Effect.succeed(true), + kill: () => Effect.void, + stdin: Sink.drain, + getInputFd: () => Sink.drain, + getOutputFd: () => Stream.empty, + unref: Effect.succeed(Effect.void), + }), + ), + ), + ), + Effect.provideService(ProcessRunner.ProcessRunner, { + run: (input) => + input.args.includes("devices") ? startup(stateDir) : Effect.succeed(processOutput(0)), + }), + Effect.provideService( + HttpClient.HttpClient, + HttpClient.make((request) => + Effect.succeed(HttpClientResponse.fromWeb(request, new Response("ok"))), + ), + ), + ); + yield* host.ensureReady(() => Effect.void); + return { host, fs, stateDir }; +}); + +describe("AgentDevice startup", () => { + it.effect("reports an incompatible lock immediately and retains it for manual repair", () => + Effect.gen(function* () { + const result = processOutput( + 1, + JSON.stringify({ + success: false, + error: { + message: "private command output", + details: { + reason: "daemon_registration_unproven", + inspection: { state: "unproven", reason: "non-directory" }, + }, + }, + }), + ); + const { host, fs, stateDir } = yield* agentFixture(() => Effect.succeed(result)); + const lock = `${stateDir}/daemon.lock`; + const legacyOwner = JSON.stringify({ pid: 85669, processStartTime: "old start time" }); + yield* fs.writeFileString(lock, legacyOwner); + const error = yield* host.ensureAgentReady(() => Effect.void).pipe(Effect.flip); + expect(error).toMatchObject({ + _tag: "AgentDeviceStartError", + reason: "daemon_lock_not_directory", + exitCode: 1, + cause: result, + }); + expect(error.message).toContain("Daemon registration ownership could not be verified"); + expect(error.message).toContain("daemon.lock is not a directory"); + expect(error.message).toContain(stateDir); + expect(error.message).toContain("original CLI"); + expect(error.message).not.toContain("private command output"); + expect(yield* fs.readFileString(lock)).toBe(legacyOwner); + expect(yield* host.current).not.toBeNull(); + }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)), + ); + + it.effect("reports unknown nonzero exits without exposing arbitrary command output", () => + Effect.gen(function* () { + const result = processOutput(2, "invalid JSON", "private credentials"); + const { host, stateDir } = yield* agentFixture(() => Effect.succeed(result)); + const error = yield* host.ensureAgentReady(() => Effect.void).pipe(Effect.flip); + expect(error).toMatchObject({ + _tag: "AgentDeviceStartError", + reason: "command_failed", + exitCode: 2, + cause: result, + }); + expect(error.message).toContain("exit code 2"); + expect(error.message).toContain(stateDir); + expect(error.message).not.toContain("private credentials"); + }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)), + ); + + it.effect("preserves process execution failures instead of waiting for a daemon file", () => + Effect.gen(function* () { + const cause = new ProcessRunner.ProcessSpawnError({ + command: "/missing/node", + argumentCount: 3, + cause: new Error("missing executable"), + }); + const { host } = yield* agentFixture(() => Effect.fail(cause)); + const error = yield* host.ensureAgentReady(() => Effect.void).pipe(Effect.flip); + expect(error).toMatchObject({ + _tag: "DeviceHostError", + step: "starting agent tools", + cause, + }); + }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)), + ); + + it.effect("returns a command timeout without starting another readiness wait", () => + Effect.gen(function* () { + const { host } = yield* agentFixture(() => Effect.succeed(processOutput(124, "", "", true))); + const error = yield* host.ensureAgentReady(() => Effect.void).pipe(Effect.flip); + expect(error).toMatchObject({ _tag: "DeviceHostTimeoutError", timeoutMs: 30_000 }); + }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)), + ); + + it.effect("uses a registered daemon when reading startup command output fails", () => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const cause = new ProcessRunner.ProcessReadError({ + command: process.execPath, + argumentCount: 3, + stream: "stdout", + cause: new Error("output read failed after registration"), + }); + let attempts = 0; + const { host } = yield* agentFixture((stateDir) => + Effect.gen(function* () { + attempts++; + yield* fs + .writeFileString( + `${stateDir}/daemon.json`, + JSON.stringify({ httpPort: 12345, token: "test-token" }), + ) + .pipe(Effect.orDie); + return yield* cause; + }), + ); + const ready = yield* host.ensureAgentReady(() => Effect.void); + expect(ready.agentDevice).toMatchObject({ + baseUrl: "http://127.0.0.1:12345", + token: "test-token", + }); + expect((yield* host.ensureAgentReady(() => Effect.void)).agentDevice).toEqual( + ready.agentDevice, + ); + expect(attempts).toBe(1); + }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)), + ); + + it.effect("allows retry and uses a registered daemon even when discovery exits nonzero", () => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + let attempts = 0; + const { host } = yield* agentFixture((stateDir) => + Effect.gen(function* () { + attempts++; + if (attempts > 1) { + yield* fs + .writeFileString( + `${stateDir}/daemon.json`, + JSON.stringify({ httpPort: 12345, token: "test-token" }), + ) + .pipe(Effect.orDie); + } + return processOutput(1); + }), + ); + yield* host.ensureAgentReady(() => Effect.void).pipe(Effect.flip); + const ready = yield* host.ensureAgentReady(() => Effect.void); + expect(ready.agentDevice).toMatchObject({ + baseUrl: "http://127.0.0.1:12345", + token: "test-token", + }); + expect((yield* host.ensureAgentReady(() => Effect.void)).agentDevice).toEqual( + ready.agentDevice, + ); + expect(attempts).toBe(2); + }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)), + ); +}); const diagnose = ( files: ReadonlyArray, diff --git a/apps/server/src/device/LocalDeviceHost.ts b/apps/server/src/device/LocalDeviceHost.ts index 779221ab93cf..63245da93d2d 100644 --- a/apps/server/src/device/LocalDeviceHost.ts +++ b/apps/server/src/device/LocalDeviceHost.ts @@ -86,6 +86,19 @@ const AgentDeviceDaemonFile = Schema.Struct({ }); const decodeDaemonFile = Schema.decodeUnknownEffect(Schema.fromJsonString(AgentDeviceDaemonFile)); +const decodeAgentDeviceFailure = Schema.decodeUnknownOption( + Schema.fromJsonString( + Schema.Struct({ + error: Schema.Struct({ + details: Schema.Struct({ + reason: Schema.String, + inspection: Schema.optional(Schema.Struct({ reason: Schema.String })), + }), + }), + }), + ), +); + interface HubProcess { readonly nodePath: string; readonly child: ChildProcessSpawner.ChildProcessHandle; @@ -498,7 +511,12 @@ export const make = Effect.fn("LocalDeviceHost.make")(function* () { const startAgentDeviceDaemon = Effect.fn("LocalDeviceHost.startAgentDeviceDaemon")(function* ( agentTool: DeviceToolPaths, nodePath: string, - ): Effect.fn.Return { + ): Effect.fn.Return< + DeviceHost.AgentDeviceEndpoint, + | DeviceHost.DeviceHostError + | DeviceHost.DeviceHostTimeoutError + | DeviceHost.AgentDeviceStartError + > { const stateDir = agentDeviceStateDir(path, config.stateDir); yield* fs.makeDirectory(stateDir, { recursive: true }).pipe(Effect.ignore); const existing = yield* readDaemonFile().pipe(Effect.option); @@ -536,7 +554,7 @@ export const make = Effect.fn("LocalDeviceHost.make")(function* () { } // There is no `daemon start`; the first command in a state dir spawns the // daemon and blocks until it answers. `devices` is the cheapest one. - yield* runner + const runResult = yield* runner .run({ command: nodePath, args: [agentTool.entryPath, "devices", "--json"], @@ -544,7 +562,44 @@ export const make = Effect.fn("LocalDeviceHost.make")(function* () { timeout: Duration.millis(DAEMON_READY_TIMEOUT_MS), timeoutBehavior: "timedOutResult", }) - .pipe(Effect.ignore); + .pipe(Effect.result); + // Discovery can fail after the daemon starts. A published endpoint still + // lets agents connect; absent one, preserve the CLI failure instead of polling. + const started = yield* readDaemonFile().pipe(Effect.option); + if (started._tag === "Some") return toEndpoint(started.value); + if (runResult._tag === "Failure") { + return yield* new DeviceHost.DeviceHostError({ + hostId, + step: "starting agent tools", + cause: runResult.failure, + }); + } + const result = runResult.success; + if (result.timedOut) { + return yield* new DeviceHost.DeviceHostTimeoutError({ + hostId, + timeoutMs: DAEMON_READY_TIMEOUT_MS, + }); + } + if (result.code !== 0) { + const failure = decodeAgentDeviceFailure(result.stdout); + const details = Option.isSome(failure) ? failure.value.error.details : undefined; + const reason = + details?.reason === "daemon_registration_unproven" + ? details.inspection?.reason === "non-directory" + ? "daemon_lock_not_directory" + : "daemon_registration_unproven" + : details?.reason === "daemon_retirement_unconfirmed" + ? "daemon_retirement_unconfirmed" + : "command_failed"; + return yield* new DeviceHost.AgentDeviceStartError({ + hostId, + stateDir, + exitCode: result.code === null ? null : Number(result.code), + reason, + cause: result, + }); + } const deadline = (yield* Clock.currentTimeMillis) + DAEMON_READY_TIMEOUT_MS; while (true) { const file = yield* readDaemonFile().pipe(Effect.option); @@ -658,6 +713,7 @@ export const make = Effect.fn("LocalDeviceHost.make")(function* () { unknown, | DeviceHost.DeviceHostError | DeviceHost.DeviceHostTimeoutError + | DeviceHost.AgentDeviceStartError | NodeRuntimeUnavailableError >, DeviceHost.DeviceHostAgentReady diff --git a/apps/server/src/orchestration-v2/NotificationMailbox.ts b/apps/server/src/orchestration-v2/NotificationMailbox.ts index e7045e28d9e1..cbf76bf28426 100644 --- a/apps/server/src/orchestration-v2/NotificationMailbox.ts +++ b/apps/server/src/orchestration-v2/NotificationMailbox.ts @@ -1,9 +1,22 @@ import { latestProviderTurnForAttempt, type MessageId, + type OrchestrationV2ConversationMessage, type OrchestrationV2ThreadProjection, } from "@t3tools/contracts"; +/** A wake attaches to output held in provider memory, which Stop or process loss discards. */ +export function isBufferedProviderNotification( + message: OrchestrationV2ConversationMessage, +): boolean { + return ( + message.createdBy === "agent" && + message.creationSource === "provider" && + message.notification !== undefined && + message.delegatedCompletion === undefined + ); +} + /** * A persisted steer without an acceptance receipt can be retried as a continuation. * Delivery is at least once: provider acceptance and our receipt cannot commit diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 5b921266ceaa..8dc89c7a33d7 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -89,7 +89,10 @@ import { CommandReceiptStoreV2 } from "./CommandReceiptStore.ts"; import { ContextHandoffServiceV2 } from "./ContextHandoffService.ts"; import { notificationTurnItem } from "@t3tools/provider-core/server/notification"; import { isRestartNoteSource } from "./RestartBackgroundNote.ts"; -import { isUndeliveredMailboxSteer } from "./NotificationMailbox.ts"; +import { + isBufferedProviderNotification, + isUndeliveredMailboxSteer, +} from "./NotificationMailbox.ts"; import { EventSinkV2 } from "./EventSink.ts"; import * as EffectOutbox from "./EffectOutbox.ts"; import type { OrchestrationEffectRequestV2, PendingOrchestrationEffectV2 } from "./EffectOutbox.ts"; @@ -2799,7 +2802,24 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio cause: `Thread ${command.threadId} has a pending approval or user-input request and cannot be snoozed.`, }); } - if (projection.runs.some((run) => run.status === "queued")) { + const queuedRuns = projection.runs.filter((run) => run.status === "queued"); + const queuedMessages = + queuedRuns.length === 0 + ? [] + : (yield* loadProjectionForCommand(command, ["messages"], { + messageIds: queuedRuns.map((run) => run.userMessageId), + })).messages; + // Automatic wakes are hidden from the user queue and must not block snooze. + // Snooze only changes visibility, so keep those deliveries intact. + const automaticMessageIds = new Set( + queuedMessages + .filter( + (message) => + message.notification !== undefined || message.delegatedCompletion !== undefined, + ) + .map((message) => message.id), + ); + if (queuedRuns.some((run) => !automaticMessageIds.has(run.userMessageId))) { return yield* new OrchestratorDispatchError({ commandId: command.commandId, commandType: command.type, @@ -8811,9 +8831,9 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio }); /** - * What Stop holds besides the run it interrupts: queued runs wait for the user, the - * thread's pull request watches end, and every wake its delegated tasks still owe is - * dropped, since Stop stops those tasks too. Nothing automatic starts the thread again. + * What Stop holds besides the run it interrupts: durable queued messages wait for the + * user, buffered provider wakes are cancelled, and the thread's pull request watches end. + * Every wake its delegated tasks still owe is dropped, since Stop stops those tasks too. * `cohortRunIds` are dropped even when they owe nothing, like a plain interrupt's. */ const holdStoppedThread = (input: { @@ -8829,6 +8849,22 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio Effect.gen(function* () { const emitEvent = emit(input.events, input.command); const { thread, runs, subagents } = input.projection; + const queuedRuns = runs.filter((run) => run.status === "queued"); + const queuedMessages = + queuedRuns.length === 0 + ? [] + : (yield* projectionStore + .getThreadRecords(thread.id, ["messages"], { + messageIds: queuedRuns.map((run) => run.userMessageId), + }) + .pipe( + Effect.mapError( + (cause) => new OrchestratorProjectionError({ threadId: thread.id, cause }), + ), + )).messages; + const bufferedMessageIds = new Set( + queuedMessages.filter(isBufferedProviderNotification).map((message) => message.id), + ); for (const run of runs) { // A resumed queue can have a lower ordinal than a pending continuation source. // Stop must reach those sources even when interrupting the resumed run succeeds. @@ -8838,7 +8874,21 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ) { yield* markStoppedRun({ ...input, thread, run }); } - if (run.status !== "queued" || run.queueHeld === true) continue; + if (run.status !== "queued") continue; + // Stop discards the provider's wake buffer, so its queued placeholders cannot resume. + if (bufferedMessageIds.has(run.userMessageId)) { + yield* dispatchQueuedRunCancel( + { + type: "queued-run.cancel", + commandId: input.command.commandId, + threadId: thread.id, + runId: run.id, + }, + input.events, + ); + continue; + } + if (run.queueHeld === true) continue; yield* emitEvent({ type: "run.updated", threadId: thread.id, diff --git a/apps/server/src/orchestration-v2/ProjectionRecovery.test.ts b/apps/server/src/orchestration-v2/ProjectionRecovery.test.ts index 95efab067f80..e616b1d4f3b5 100644 --- a/apps/server/src/orchestration-v2/ProjectionRecovery.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionRecovery.test.ts @@ -23,14 +23,34 @@ import * as Layer from "effect/Layer"; import * as SqlClient from "effect/sql/SqlClient"; import * as SqlitePersistence from "../persistence/Sqlite.ts"; +import * as ServerSettings from "../serverSettings.ts"; import { CodexProviderCapabilitiesV2 } from "./Adapters/CodexAdapterV2.ts"; import * as EffectOutbox from "./EffectOutbox.ts"; import * as ProjectionStore from "./ProjectionStore.ts"; +import * as EventSink from "./EventSink.ts"; +import * as EventStore from "./EventStore.ts"; +import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator"; +import * as ProviderRuntimeRecovery from "./ProviderRuntimeRecoveryService.ts"; import { restartContinuationRun } from "./RestartContinuation.ts"; const layerTest = Layer.mergeAll(ProjectionStore.layer, EffectOutbox.layer).pipe( Layer.provideMerge(SqlitePersistence.layerMemory), ); +const layerRecoveryTest = ProviderRuntimeRecovery.layer.pipe( + Layer.provideMerge( + EventSink.layer.pipe( + Layer.provideMerge( + Layer.mergeAll( + EventStore.layer, + ProjectionStore.layer, + EffectOutbox.layer, + IdAllocator.layer, + ).pipe(Layer.provideMerge(SqlitePersistence.layerMemory)), + ), + ), + ), + Layer.provide(ServerSettings.layerTest()), +); const providerInstanceId = ProviderInstanceId.make("codex"); const modelSelection = { instanceId: providerInstanceId, model: "gpt-5.4" }; const driver = ProviderDriverKind.make("codex"); @@ -113,6 +133,82 @@ const createRun = Effect.fn(function* ( return runId; }); +it.effect.each(["startup", "shutdown"] as const)( + "%s cancels queued buffered wakes and holds durable messages", + (trigger) => + Effect.gen(function* () { + const projections = yield* ProjectionStore.ProjectionStoreV2; + const recovery = yield* ProviderRuntimeRecovery.ProviderRuntimeRecoveryService; + const now = yield* DateTime.now; + const bufferedThreads = [] as ThreadId[]; + const durableThreads = [] as ThreadId[]; + for (const kind of ["buffered", "user", "notification", "delegated"] as const) { + const threadId = yield* createThread(`wake:${trigger}:${kind}`); + const buffered = kind === "buffered"; + (buffered ? bufferedThreads : durableThreads).push(threadId); + const messageId = MessageId.make(`message:${threadId}`); + const runId = yield* createRun(threadId, "queued", { + userMessageId: messageId, + startedAt: null, + queuePosition: 1, + }); + yield* createRun(threadId, "completed", { ordinal: 2 }); + yield* projections.apply({ + id: EventId.make(`message:${threadId}`), + type: "message.updated", + threadId, + runId, + occurredAt: now, + payload: { + id: messageId, + threadId, + runId, + nodeId: null, + role: "user", + text: "Queued work", + attachments: [], + streaming: false, + createdAt: now, + updatedAt: now, + createdBy: kind === "user" ? "user" : "agent", + creationSource: buffered ? "provider" : "server", + ...(kind === "user" || kind === "delegated" + ? {} + : { + notification: { + source: { kind: "background_task" as const }, + outcome: "updated" as const, + summary: "Background activity updated", + }, + }), + ...(kind === "delegated" + ? { delegatedCompletion: { parentRunId: runId, generation: 1, taskIds: [] } } + : {}), + }, + }); + } + assert.deepEqual( + new Set(yield* projections.getRecoveryThreadIds("runtime")), + new Set([...bufferedThreads, ...durableThreads]), + ); + const summary = yield* recovery.reconcile(trigger); + assert.equal(summary.terminalizedRuns, 1); + for (const threadId of bufferedThreads) { + const projection = yield* projections.getThreadProjection(threadId); + assert.equal(projection.runs[0]?.status, "cancelled"); + assert.isNotNull(projection.runs[0]?.completedAt); + assert.isNull(projection.runs[0]?.queuePosition); + assert.equal(projection.runs[1]?.status, "completed"); + } + for (const threadId of durableThreads) { + const projection = yield* projections.getThreadProjection(threadId); + assert.equal(projection.runs[0]?.status, "queued"); + assert.isTrue(projection.runs[0]?.queueHeld); + } + assert.deepEqual(yield* projections.getRecoveryThreadIds("runtime"), []); + }).pipe(Effect.provide(layerRecoveryTest)), +); + it.effect("selects unfinished recovery work without reading settled thread histories", () => Effect.gen(function* () { const projections = yield* ProjectionStore.ProjectionStoreV2; diff --git a/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.test.ts b/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.test.ts index 05c2b41c0116..1e2fcc44248d 100644 --- a/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.test.ts +++ b/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.test.ts @@ -690,6 +690,7 @@ it.effect("holds accepted queued work without cancelling its execution state aft Layer.mock(ProjectionStore.ProjectionStoreV2)({ getRecoveryThreadIds: () => Effect.succeed([threadId]), getRuntimeRecoveryProjection: () => Effect.succeed(projection), + getThreadRecords: () => Effect.succeed(projection), }), Layer.mock(EventSink.EventSinkV2)({ commitCommand: (input) => { diff --git a/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.ts b/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.ts index ad57247e0384..27cacf50da94 100644 --- a/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.ts +++ b/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.ts @@ -22,6 +22,7 @@ import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator"; import * as ProjectionStore from "./ProjectionStore.ts"; import * as ServerSettings from "../serverSettings.ts"; import { restartContinuationRun } from "./RestartContinuation.ts"; +import { isBufferedProviderNotification } from "./NotificationMailbox.ts"; import { endOrphanedNativeSubagent, endRunlessRootTurns, @@ -284,10 +285,35 @@ export const make = Effect.gen(function* () { cancelledBackgroundNativeIds.add(item.nativeItemRef.nativeId); } }; - // Queued runs have not started provider work. Preserve their execution - // identities and order, but require explicit consent before draining them. - for (const run of projection.runs) { - if (run.status !== "queued" || run.queueHeld === true) continue; + const queuedRuns = projection.runs.filter((run) => run.status === "queued"); + const queuedMessages = + queuedRuns.length === 0 + ? [] + : (yield* projections + .getThreadRecords(projection.thread.id, ["messages"], { + messageIds: queuedRuns.map((run) => run.userMessageId), + }) + .pipe( + Effect.mapError( + (cause) => + new ProviderRuntimeRecoveryError({ + operation: "read-projections", + threadId: projection.thread.id, + cause, + }), + ), + )).messages; + const bufferedMessageIds = new Set( + queuedMessages.filter(isBufferedProviderNotification).map((message) => message.id), + ); + // Buffered wakes lost their output with the process. Durable queued messages + // keep their execution identities and order until the user resumes them. + for (const run of queuedRuns) { + if (bufferedMessageIds.has(run.userMessageId)) { + runs.push(run); + continue; + } + if (run.queueHeld === true) continue; events.push({ id: yield* allocateEventId(), type: "run.updated", diff --git a/apps/server/src/orchestration-v2/ThreadStop.test.ts b/apps/server/src/orchestration-v2/ThreadStop.test.ts index 73fc29a05569..2357c1490901 100644 --- a/apps/server/src/orchestration-v2/ThreadStop.test.ts +++ b/apps/server/src/orchestration-v2/ThreadStop.test.ts @@ -225,6 +225,62 @@ it.effect("Stop ends watches, holds queues, and stops the delegated tasks under }).pipe(Effect.provide(layerTest)), ); +it.effect.each(["thread.stop", "run.interrupt"] as const)( + "%s cancels buffered wakes and preserves durable queued messages", + (type) => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const threadId = ThreadId.make(`thread:stop-buffered:${type}`); + yield* createWatchingThread(threadId, 5); + yield* send(threadId, "active", "start_immediately"); + yield* send(threadId, "user follow-up", "queue_after_active"); + for (const creationSource of ["provider", "server"] as const) { + yield* orchestrator.dispatch({ + type: "message.dispatch", + commandId: CommandId.make(`notification:${type}:${creationSource}`), + threadId, + messageId: MessageId.make(`notification:${type}:${creationSource}`), + text: "Background activity updated", + attachments: [], + createdBy: "agent", + creationSource, + notification: { + source: { kind: "background_task" }, + outcome: "updated", + summary: "Background activity updated", + }, + dispatchMode: { type: "queue_after_active" }, + }); + } + const before = yield* orchestrator.getThreadProjection(threadId); + const buffered = before.runs.find((run) => + before.messages.some( + (message) => message.id === run.userMessageId && message.creationSource === "provider", + ), + )!; + const stop = { commandId: CommandId.make(`stop:${type}`), threadId }; + yield* orchestrator.dispatch( + type === "run.interrupt" + ? { ...stop, type, runId: before.runs[0]!.id, holdQueue: true } + : { ...stop, type }, + ); + const stopped = yield* orchestrator.getThreadProjection(threadId); + assert.equal(stopped.runs.find((run) => run.id === buffered.id)?.status, "cancelled"); + assert.equal( + stopped.attempts.find((attempt) => attempt.runId === buffered.id)?.status, + "cancelled", + ); + assert.equal( + stopped.nodes.find((node) => node.id === buffered.rootNodeId)?.status, + "cancelled", + ); + assert.deepEqual( + stopped.runs.filter((run) => run.status === "queued").map((run) => run.queueHeld), + [true, true], + ); + }).pipe(Effect.provide(layerTest)), +); + it.effect( "thread.stop ends an idle thread's watches and accepts a thread with nothing to stop", () => diff --git a/apps/server/src/orchestration-v2/runtimeLayer.test.ts b/apps/server/src/orchestration-v2/runtimeLayer.test.ts index 3758ec41c032..1e624e90c2bd 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.test.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.test.ts @@ -4572,6 +4572,149 @@ it.layer(layerSharedApplicationDataPlaneTest)("pending provider interruption", ( ); }); +it.layer(layerTest)("snooze queued runs", (it) => { + it.effect.each(["notification", "delegated-completion", "user"] as const)( + "only user messages block snooze with a held %s run", + (kind) => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const eventSink = yield* EventSink.EventSinkV2; + const threadId = ThreadId.make(`snooze-queued:${kind}`); + const snoozedUntil = "2099-07-25T09:00:00.000Z"; + yield* orchestrator.dispatch({ + type: "thread.create", + commandId: CommandId.make(`create:${threadId}`), + threadId, + projectId: ProjectId.make("snooze-queued-project"), + title: "Snooze with held queued work", + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: "/tmp/snooze-queued", + createdBy: "user", + creationSource: "web", + }); + yield* orchestrator.dispatch({ + type: "message.dispatch", + commandId: CommandId.make(`active:${threadId}`), + threadId, + messageId: MessageId.make(`active:${threadId}`), + text: "Active run", + attachments: [], + modelSelection, + dispatchMode: { type: "start_immediately" }, + createdBy: "user", + creationSource: "web", + }); + const activeRun = (yield* orchestrator.getThreadProjection(threadId)).runs[0]!; + const now = yield* DateTime.now; + if (kind === "delegated-completion") { + yield* eventSink.write({ + events: [ + { + id: EventId.make(`completion:${threadId}`), + type: "run.updated", + threadId, + runId: activeRun.id, + occurredAt: now, + payload: { + ...activeRun, + delegatedCompletion: { + disposition: "open", + nextGeneration: 2, + delivery: { + messageId: MessageId.make(`queued:${threadId}`), + generation: 1, + taskIds: [NodeId.make("snooze-queued-task")], + }, + }, + }, + }, + ], + }); + } + yield* orchestrator.dispatch({ + type: "message.dispatch", + commandId: CommandId.make(`queued:${threadId}`), + threadId, + messageId: MessageId.make(`queued:${threadId}`), + text: "Queued work", + attachments: [], + modelSelection, + dispatchMode: { type: "queue_after_active" }, + createdBy: kind === "user" ? "user" : "agent", + creationSource: kind === "user" ? "web" : "server", + ...(kind === "notification" + ? { + notification: { + source: { kind: "background_task" as const }, + outcome: "updated" as const, + summary: "Background activity updated", + }, + } + : kind === "delegated-completion" + ? { + delegatedCompletion: { + parentRunId: activeRun.id, + generation: 1, + taskIds: [NodeId.make("snooze-queued-task")], + }, + } + : {}), + }); + // Recovery holds the queue while the thread's active run ends. + const beforeRecovery = yield* orchestrator.getThreadProjection(threadId); + yield* eventSink.commitCommand({ + commandId: CommandId.make(`recovery:${threadId}`), + threadId, + commandType: "provider-runtime.reconcile", + acceptedAt: now, + events: beforeRecovery.runs.map((run) => ({ + id: EventId.make(`recovery:${run.id}`), + type: "run.updated" as const, + threadId, + runId: run.id, + occurredAt: now, + payload: + run.status === "queued" + ? { ...run, queueHeld: true } + : { ...run, status: "completed" as const, completedAt: now }, + })), + effects: [], + }); + const beforeSnooze = yield* orchestrator.getThreadProjection(threadId); + assert.isTrue(beforeSnooze.runs.some((run) => run.status === "queued" && run.queueHeld)); + const snooze = orchestrator.dispatch({ + type: "thread.snooze", + commandId: CommandId.make(`snooze:${threadId}`), + threadId, + snoozedUntil, + }); + if (kind === "user") { + const error = yield* snooze.pipe(Effect.flip); + assert.equal(error._tag, "OrchestratorDispatchError"); + assert.equal(error.cause, `Thread ${threadId} has a queued run and cannot be snoozed.`); + assert.isNull((yield* orchestrator.getThreadProjection(threadId)).thread.snoozedUntil); + } else { + yield* snooze; + const snoozed = yield* orchestrator.getThreadProjection(threadId); + assert.equal(DateTime.formatIso(snoozed.thread.snoozedUntil!), snoozedUntil); + assert.deepEqual(snoozed.runs, beforeSnooze.runs); + yield* orchestrator.dispatch({ + type: "thread.unsnooze", + commandId: CommandId.make(`unsnooze:${threadId}`), + threadId, + reason: "user", + }); + const awake = yield* orchestrator.getThreadProjection(threadId); + assert.isNull(awake.thread.snoozedUntil); + assert.deepEqual(awake.runs, beforeSnooze.runs); + } + }), + ); +}); + it.layer(layerSharedApplicationDataPlaneTest)("snooze projection", (it) => { it.effect("carries snooze state through the V2 shell projection", () => Effect.gen(function* () { diff --git a/apps/server/src/orchestration-v2/testkit/fixtures/message_steering/pi_output.ts b/apps/server/src/orchestration-v2/testkit/fixtures/message_steering/pi_output.ts index cb48f1364384..1ccd33eb00d1 100644 --- a/apps/server/src/orchestration-v2/testkit/fixtures/message_steering/pi_output.ts +++ b/apps/server/src/orchestration-v2/testkit/fixtures/message_steering/pi_output.ts @@ -36,7 +36,7 @@ export function assertPiMessageSteeringOutput( assert.deepEqual( prompts.map((frame) => [frameField(frame, "message"), frameField(frame, "streamingBehavior")]), [ - [MESSAGE_STEERING_INITIAL_PROMPT, undefined], + [MESSAGE_STEERING_INITIAL_PROMPT, "steer"], [MESSAGE_STEERING_STEER_PROMPT, "steer"], ], ); diff --git a/apps/server/src/orchestration-v2/testkit/fixtures/message_steering/pi_transcript.ndjson b/apps/server/src/orchestration-v2/testkit/fixtures/message_steering/pi_transcript.ndjson index 32ffdd3987c4..907bf6618b41 100644 --- a/apps/server/src/orchestration-v2/testkit/fixtures/message_steering/pi_transcript.ndjson +++ b/apps/server/src/orchestration-v2/testkit/fixtures/message_steering/pi_transcript.ndjson @@ -12,7 +12,7 @@ {"type":"expect_outbound","label":"set_session_name","frame":{"type":"set_session_name","name":"Replay fixture: message_steering","id":"t3-5"}} {"type":"emit_inbound","label":"session_info_changed","frame":{"type":"session_info_changed","name":"Replay fixture: message_steering"}} {"type":"emit_inbound","label":"response:set_session_name","frame":{"id":"t3-5","type":"response","command":"set_session_name","success":true}} -{"type":"expect_outbound","label":"prompt","frame":{"type":"prompt","message":"Respond with exactly: steering fixture initial response"}} +{"type":"expect_outbound","label":"prompt","frame":{"type":"prompt","message":"Respond with exactly: steering fixture initial response","streamingBehavior":"steer"}} {"type":"emit_inbound","label":"response:prompt","frame":{"type":"response","command":"prompt","success":true,"data":{"disposition":"started"}}} {"type":"emit_inbound","label":"agent_start","frame":{"type":"agent_start"}} {"type":"emit_inbound","label":"turn_start","frame":{"type":"turn_start"}} diff --git a/apps/server/src/orchestration-v2/testkit/fixtures/multi_turn/pi_transcript.ndjson b/apps/server/src/orchestration-v2/testkit/fixtures/multi_turn/pi_transcript.ndjson index 7e32afc0a5b2..5841abd415c1 100644 --- a/apps/server/src/orchestration-v2/testkit/fixtures/multi_turn/pi_transcript.ndjson +++ b/apps/server/src/orchestration-v2/testkit/fixtures/multi_turn/pi_transcript.ndjson @@ -12,7 +12,7 @@ {"type":"expect_outbound","label":"set_session_name","frame":{"type":"set_session_name","name":"Replay fixture: multi_turn","id":"t3-5"}} {"type":"emit_inbound","label":"session_info_changed","frame":{"type":"session_info_changed","name":"Replay fixture: multi_turn"}} {"type":"emit_inbound","label":"response:set_session_name","frame":{"id":"t3-5","type":"response","command":"set_session_name","success":true}} -{"type":"expect_outbound","label":"prompt","frame":{"type":"prompt","message":"Respond with exactly: first fixture turn complete"}} +{"type":"expect_outbound","label":"prompt","frame":{"type":"prompt","message":"Respond with exactly: first fixture turn complete","streamingBehavior":"steer"}} {"type":"emit_inbound","label":"response:prompt","frame":{"type":"response","command":"prompt","success":true,"data":{"disposition":"started"}}} {"type":"emit_inbound","label":"agent_start","frame":{"type":"agent_start"}} {"type":"emit_inbound","label":"turn_start","frame":{"type":"turn_start"}} @@ -41,7 +41,7 @@ {"type":"emit_inbound","label":"response:get_entries","frame":{"id":"t3-7","type":"response","command":"get_entries","success":true,"data":{"entries":[{"type":"model_change","id":"5e37f535","parentId":"f3bce6ef","timestamp":"2026-01-01T00:00:00.437Z","provider":"openrouter","modelId":"deepseek/deepseek-v4-flash"},{"type":"session_info","id":"16d089a9","parentId":"5e37f535","timestamp":"2026-01-01T00:00:00.440Z","name":"Replay fixture: multi_turn"},{"type":"message","id":"ea6f5949","parentId":"16d089a9","timestamp":"2026-01-01T00:00:00.448Z","message":{"role":"system","content":"","sections":{"preamble":"","tools":"","rules":"","docs":"","cwd":""},"timestamp":1790898591750,"toolsAdded":[{"name":"read","description":"Read the contents of a file. Supports text files and images (jpg, png, gif, webp, bmp). Images are sent as attachments. For text files, output is truncated to 2000 lines or 50KB (whichever is hit first). Use offset/limit for large files. When you need the full file, continue with offset until complete.","parameters":{"type":"object","required":["path"],"properties":{"path":{"type":"string","description":"Path to the file to read (relative or absolute)"},"offset":{"type":"number","description":"Line number to start reading from (1-indexed)"},"limit":{"type":"number","description":"Maximum number of lines to read"}}},"constrainedSampling":{"type":"json_schema","strict":"prefer"}},{"name":"bash","description":"Execute a bash command in the current working directory. Returns stdout and stderr. Output is truncated to last 2000 lines or 50KB (whichever is hit first). If truncated, full output is saved to a temp file. Optionally provide a timeout in seconds.","parameters":{"type":"object","required":["command"],"properties":{"command":{"type":"string","description":"Shell command to execute"},"timeout":{"type":"number","description":"Timeout in seconds (optional, no default timeout)"}}},"constrainedSampling":{"type":"json_schema","strict":"prefer"}},{"name":"edit","description":"Edit a single file using exact text replacement. Every edits[].oldText must match a unique, non-overlapping region of the original file. If two changes affect the same block or nearby lines, merge them into one edit instead of emitting overlapping edits. Do not include large unchanged regions just to connect distant changes.","parameters":{"type":"object","required":["path","edits"],"properties":{"path":{"type":"string","description":"Path to the file to edit (relative or absolute)"},"edits":{"type":"array","items":{"type":"object","required":["oldText","newText"],"properties":{"oldText":{"type":"string","description":"Exact text for one targeted replacement. It must be unique in the original file and must not overlap with any other edits[].oldText in the same call."},"newText":{"type":"string","description":"Replacement text for this targeted edit."}}},"description":"One or more targeted replacements. Each edit is matched against the original file, not incrementally. Do not include overlapping or nested edits. If two changes touch the same block or nearby lines, merge them into one edit instead."}}},"constrainedSampling":{"type":"json_schema","strict":"prefer"}},{"name":"write","description":"Write content to a file. Creates the file if it doesn't exist, overwrites if it does. Automatically creates parent directories.","parameters":{"type":"object","required":["path","content"],"properties":{"path":{"type":"string","description":"Path to the file to write (relative or absolute)"},"content":{"type":"string","description":"Content to write to the file"}}},"constrainedSampling":{"type":"json_schema","strict":"prefer"}}]}},{"type":"message","id":"ff895990","parentId":"ea6f5949","timestamp":"2026-01-01T00:00:00.448Z","message":{"role":"user","content":[{"type":"text","text":"Respond with exactly: first fixture turn complete"}],"timestamp":1767225600444}},{"type":"message","id":"5c0d3865","parentId":"ff895990","timestamp":"2026-01-01T00:00:01.723Z","message":{"role":"assistant","content":[{"type":"thinking","thinking":"The user wants me to respond with exactly: \"first fixture turn complete\"","thinkingSignature":""},{"type":"text","text":"first fixture turn complete"}],"api":"openai-completions","provider":"openrouter","model":"deepseek/deepseek-v4-flash","usage":{"input":1787,"output":21,"cacheRead":0,"cacheWrite":0,"reasoning":18,"totalTokens":1808,"cost":{"input":0.00007480382,"output":0.0000017581199999999999,"cacheRead":0,"cacheWrite":0,"total":0.00007656194}},"stopReason":"stop","timestamp":1767225600465,"responseId":"gen-1790898592-RtbvUBJEB0AHaUCPMqrg","rawStopReason":"stop","thinkingLevel":"high"}}],"leafId":"5c0d3865"}}} {"type":"expect_outbound","label":"get_session_stats","frame":{"type":"get_session_stats","id":"t3-8"}} {"type":"emit_inbound","label":"response:get_session_stats","frame":{"id":"t3-8","type":"response","command":"get_session_stats","success":true,"data":{"sessionFile":"/pi-sessions/session-1.jsonl","sessionId":"00000000-0000-4000-8000-000000000002","userMessages":1,"assistantMessages":1,"toolCalls":0,"toolResults":0,"totalMessages":3,"tokens":{"input":1787,"output":21,"cacheRead":0,"cacheWrite":0,"total":1808},"cost":0.00007656194,"contextUsage":{"tokens":1808,"contextWindow":1048576,"percent":0.17242431640625}}}} -{"type":"expect_outbound","label":"prompt","frame":{"type":"prompt","message":"Respond with exactly: second fixture turn complete"}} +{"type":"expect_outbound","label":"prompt","frame":{"type":"prompt","message":"Respond with exactly: second fixture turn complete","streamingBehavior":"steer"}} {"type":"emit_inbound","label":"response:prompt","frame":{"type":"response","command":"prompt","success":true,"data":{"disposition":"started"}}} {"type":"emit_inbound","label":"agent_start","frame":{"type":"agent_start"}} {"type":"emit_inbound","label":"turn_start","frame":{"type":"turn_start"}} diff --git a/apps/server/src/orchestration-v2/testkit/fixtures/pi_compaction/pi_transcript.ndjson b/apps/server/src/orchestration-v2/testkit/fixtures/pi_compaction/pi_transcript.ndjson index 43aadfd2d881..39243d970f6d 100644 --- a/apps/server/src/orchestration-v2/testkit/fixtures/pi_compaction/pi_transcript.ndjson +++ b/apps/server/src/orchestration-v2/testkit/fixtures/pi_compaction/pi_transcript.ndjson @@ -12,7 +12,7 @@ {"type":"expect_outbound","label":"set_session_name","frame":{"type":"set_session_name","name":"Replay fixture: pi_compaction","id":"t3-5"}} {"type":"emit_inbound","label":"session_info_changed","frame":{"type":"session_info_changed","name":"Replay fixture: pi_compaction"}} {"type":"emit_inbound","label":"response:set_session_name","frame":{"id":"t3-5","type":"response","command":"set_session_name","success":true}} -{"type":"expect_outbound","label":"prompt","frame":{"type":"prompt","message":"Remember the opaque marker compaction-marker-4Z7Q for later. Respond with exactly: compaction marker stored"}} +{"type":"expect_outbound","label":"prompt","frame":{"type":"prompt","message":"Remember the opaque marker compaction-marker-4Z7Q for later. Respond with exactly: compaction marker stored","streamingBehavior":"steer"}} {"type":"emit_inbound","label":"response:prompt","frame":{"type":"response","command":"prompt","success":true,"data":{"disposition":"started"}}} {"type":"emit_inbound","label":"agent_start","frame":{"type":"agent_start"}} {"type":"emit_inbound","label":"turn_start","frame":{"type":"turn_start"}} @@ -56,7 +56,7 @@ {"type":"emit_inbound","label":"response:get_entries","frame":{"id":"t3-11","type":"response","command":"get_entries","success":true,"data":{"entries":[],"leafId":"9923aaa9"}}} {"type":"expect_outbound","label":"get_session_stats","frame":{"type":"get_session_stats","id":"t3-12"}} {"type":"emit_inbound","label":"response:get_session_stats","frame":{"id":"t3-12","type":"response","command":"get_session_stats","success":true,"data":{"sessionFile":"/pi-sessions/session-1.jsonl","sessionId":"00000000-0000-4000-8000-000000000002","userMessages":1,"assistantMessages":1,"toolCalls":0,"toolResults":0,"totalMessages":3,"tokens":{"input":1803,"output":36,"cacheRead":0,"cacheWrite":0,"total":1839},"cost":0.0000784875,"contextUsage":{"tokens":1839,"contextWindow":1048576,"percent":0.17538070678710938}}}} -{"type":"expect_outbound","label":"prompt","frame":{"type":"prompt","message":"Respond with exactly: second compaction fixture turn"}} +{"type":"expect_outbound","label":"prompt","frame":{"type":"prompt","message":"Respond with exactly: second compaction fixture turn","streamingBehavior":"steer"}} {"type":"emit_inbound","label":"response:prompt","frame":{"type":"response","command":"prompt","success":true,"data":{"disposition":"started"}}} {"type":"emit_inbound","label":"agent_start","frame":{"type":"agent_start"}} {"type":"emit_inbound","label":"turn_start","frame":{"type":"turn_start"}} @@ -82,7 +82,7 @@ {"type":"emit_inbound","label":"response:get_entries","frame":{"id":"t3-14","type":"response","command":"get_entries","success":true,"data":{"entries":[{"type":"message","id":"de423e78","parentId":"9923aaa9","timestamp":"2026-01-01T00:00:02.437Z","message":{"role":"user","content":[{"type":"text","text":"Respond with exactly: second compaction fixture turn"}],"timestamp":1767225602435}},{"type":"message","id":"70c802be","parentId":"de423e78","timestamp":"2026-01-01T00:00:03.192Z","message":{"role":"assistant","content":[{"type":"thinking","thinking":"The user wants me to respond with exactly \"second compaction fixture turn\".","thinkingSignature":""},{"type":"text","text":"second compaction fixture turn"}],"api":"openai-completions","provider":"openrouter","model":"deepseek/deepseek-v4-flash","usage":{"input":58,"output":20,"cacheRead":1792,"cacheWrite":0,"reasoning":19,"totalTokens":1870,"cost":{"input":0.00000242788,"output":0.0000016744,"cacheRead":0.000015002623999999998,"cacheWrite":0,"total":0.000019104903999999998}},"stopReason":"stop","timestamp":1767225602438,"responseId":"gen-1790898593-ijLmDH6S1P8UCX8xelRy","rawStopReason":"stop","thinkingLevel":"high"}}],"leafId":"70c802be"}}} {"type":"expect_outbound","label":"get_session_stats","frame":{"type":"get_session_stats","id":"t3-15"}} {"type":"emit_inbound","label":"response:get_session_stats","frame":{"id":"t3-15","type":"response","command":"get_session_stats","success":true,"data":{"sessionFile":"/pi-sessions/session-1.jsonl","sessionId":"00000000-0000-4000-8000-000000000002","userMessages":2,"assistantMessages":2,"toolCalls":0,"toolResults":0,"totalMessages":5,"tokens":{"input":1861,"output":56,"cacheRead":1792,"cacheWrite":0,"total":3709},"cost":0.00009759240400000001,"contextUsage":{"tokens":1870,"contextWindow":1048576,"percent":0.17833709716796875}}}} -{"type":"expect_outbound","label":"prompt","frame":{"type":"prompt","message":"Respond with exactly: third compaction fixture turn"}} +{"type":"expect_outbound","label":"prompt","frame":{"type":"prompt","message":"Respond with exactly: third compaction fixture turn","streamingBehavior":"steer"}} {"type":"emit_inbound","label":"response:prompt","frame":{"type":"response","command":"prompt","success":true,"data":{"disposition":"started"}}} {"type":"emit_inbound","label":"agent_start","frame":{"type":"agent_start"}} {"type":"emit_inbound","label":"turn_start","frame":{"type":"turn_start"}} @@ -120,7 +120,7 @@ {"type":"emit_inbound","label":"response:get_entries","frame":{"id":"t3-21","type":"response","command":"get_entries","success":true,"data":{"entries":[{"type":"compaction","id":"f5058450","parentId":"705e1e5c","timestamp":"2026-01-01T00:00:34.035Z","summary":"## Goal\nThe user asked the assistant to remember the opaque marker \"compaction-marker-4Z7Q\" for later use in the conversation.\n\n## Constraints & Preferences\n- The assistant was required to respond with exactly \"compaction marker stored\" (which it did).\n\n## Progress\n### Done\n- [x] Acknowledged and stored the marker \"compaction-marker-4Z7Q\"\n- [x] Responded with exactly \"compaction marker stored\"\n\n### In Progress\n- (none)\n\n### Blocked\n- (none)\n\n## Key Decisions\n- **Remember the marker**: The value \"compaction-marker-4Z7Q\" is to be retained as the opaque marker for later reference.\n\n## Next Steps\n1. When the user later references \"the marker\" or asks to use it, recall that the marker value is \"compaction-marker-4Z7Q\".\n2. Use this marker in whatever context the user requests (e.g., compaction operations).\n\n## Critical Context\n- Opaque marker value: **compaction-marker-4Z7Q**\n\n---\n\n**Turn Context (split turn):**\n\n# Conversation Summary\n\n## Original Request\nThe user asked the assistant to \"Respond with exactly: second compaction fixture turn\" — i.e., to reply using only that exact phrase.\n\n## Progress So Far\n- The user issued a single directive requiring a verbatim response of the phrase \"second compaction fixture turn.\"\n- No additional work, decisions, or content has been provided in this segment.\n\n## Context Needed to Continue\n- The required exact response text is: **second compaction fixture turn**\n- The user's instruction is to output nothing beyond that exact phrase.","firstKeptEntryId":"70c802be","tokensBefore":1901,"details":{"readFiles":[],"modifiedFiles":[]},"usage":{"input":0,"output":1003,"cacheRead":693,"cacheWrite":0,"reasoning":660,"totalTokens":1696,"cost":{"input":0,"output":0.00008397116,"cacheRead":0.000005801795999999999,"cacheWrite":0,"total":0.000089772956}},"fromHook":false,"systemMessage":{"role":"system","content":"","sections":{"preamble":"","tools":"","rules":"","docs":"","cwd":""},"toolsAdded":[{"name":"read","description":"Read the contents of a file. Supports text files and images (jpg, png, gif, webp, bmp). Images are sent as attachments. For text files, output is truncated to 2000 lines or 50KB (whichever is hit first). Use offset/limit for large files. When you need the full file, continue with offset until complete.","parameters":{"type":"object","required":["path"],"properties":{"path":{"type":"string","description":"Path to the file to read (relative or absolute)"},"offset":{"type":"number","description":"Line number to start reading from (1-indexed)"},"limit":{"type":"number","description":"Maximum number of lines to read"}}},"constrainedSampling":{"type":"json_schema","strict":"prefer"}},{"name":"bash","description":"Execute a bash command in the current working directory. Returns stdout and stderr. Output is truncated to last 2000 lines or 50KB (whichever is hit first). If truncated, full output is saved to a temp file. Optionally provide a timeout in seconds.","parameters":{"type":"object","required":["command"],"properties":{"command":{"type":"string","description":"Shell command to execute"},"timeout":{"type":"number","description":"Timeout in seconds (optional, no default timeout)"}}},"constrainedSampling":{"type":"json_schema","strict":"prefer"}},{"name":"edit","description":"Edit a single file using exact text replacement. Every edits[].oldText must match a unique, non-overlapping region of the original file. If two changes affect the same block or nearby lines, merge them into one edit instead of emitting overlapping edits. Do not include large unchanged regions just to connect distant changes.","parameters":{"type":"object","required":["path","edits"],"properties":{"path":{"type":"string","description":"Path to the file to edit (relative or absolute)"},"edits":{"type":"array","items":{"type":"object","required":["oldText","newText"],"properties":{"oldText":{"type":"string","description":"Exact text for one targeted replacement. It must be unique in the original file and must not overlap with any other edits[].oldText in the same call."},"newText":{"type":"string","description":"Replacement text for this targeted edit."}}},"description":"One or more targeted replacements. Each edit is matched against the original file, not incrementally. Do not include overlapping or nested edits. If two changes touch the same block or nearby lines, merge them into one edit instead."}}},"constrainedSampling":{"type":"json_schema","strict":"prefer"}},{"name":"write","description":"Write content to a file. Creates the file if it doesn't exist, overwrites if it does. Automatically creates parent directories.","parameters":{"type":"object","required":["path","content"],"properties":{"path":{"type":"string","description":"Path to the file to write (relative or absolute)"},"content":{"type":"string","description":"Content to write to the file"}}},"constrainedSampling":{"type":"json_schema","strict":"prefer"}}],"timestamp":1790898625260}}],"leafId":"f5058450"}}} {"type":"expect_outbound","label":"get_session_stats","frame":{"type":"get_session_stats","id":"t3-22"}} {"type":"emit_inbound","label":"response:get_session_stats","frame":{"id":"t3-22","type":"response","command":"get_session_stats","success":true,"data":{"sessionFile":"/pi-sessions/session-1.jsonl","sessionId":"00000000-0000-4000-8000-000000000002","userMessages":3,"assistantMessages":3,"toolCalls":0,"toolResults":0,"totalMessages":7,"tokens":{"input":1950,"output":1079,"cacheRead":4277,"cacheWrite":0,"total":7306},"cost":0.000207767924,"contextUsage":{"tokens":null,"contextWindow":1048576,"percent":null}}}} -{"type":"expect_outbound","label":"prompt","frame":{"type":"prompt","message":"Return the opaque marker I asked you to remember. Respond with only the marker."}} +{"type":"expect_outbound","label":"prompt","frame":{"type":"prompt","message":"Return the opaque marker I asked you to remember. Respond with only the marker.","streamingBehavior":"steer"}} {"type":"emit_inbound","label":"response:prompt","frame":{"type":"response","command":"prompt","success":true,"data":{"disposition":"started"}}} {"type":"emit_inbound","label":"agent_start","frame":{"type":"agent_start"}} {"type":"emit_inbound","label":"turn_start","frame":{"type":"turn_start"}} diff --git a/apps/server/src/orchestration-v2/testkit/fixtures/provider_thread_resume/pi_transcript.ndjson b/apps/server/src/orchestration-v2/testkit/fixtures/provider_thread_resume/pi_transcript.ndjson index cba3bd8a396d..c86c6f198f58 100644 --- a/apps/server/src/orchestration-v2/testkit/fixtures/provider_thread_resume/pi_transcript.ndjson +++ b/apps/server/src/orchestration-v2/testkit/fixtures/provider_thread_resume/pi_transcript.ndjson @@ -12,7 +12,7 @@ {"type":"expect_outbound","label":"set_session_name","frame":{"type":"set_session_name","name":"Replay fixture: provider_thread_resume","id":"t3-5"}} {"type":"emit_inbound","label":"session_info_changed","frame":{"type":"session_info_changed","name":"Replay fixture: provider_thread_resume"}} {"type":"emit_inbound","label":"response:set_session_name","frame":{"id":"t3-5","type":"response","command":"set_session_name","success":true}} -{"type":"expect_outbound","label":"prompt","frame":{"type":"prompt","message":"Respond with exactly: provider thread resume fixture first turn complete"}} +{"type":"expect_outbound","label":"prompt","frame":{"type":"prompt","message":"Respond with exactly: provider thread resume fixture first turn complete","streamingBehavior":"steer"}} {"type":"emit_inbound","label":"response:prompt","frame":{"type":"response","command":"prompt","success":true,"data":{"disposition":"started"}}} {"type":"emit_inbound","label":"agent_start","frame":{"type":"agent_start"}} {"type":"emit_inbound","label":"turn_start","frame":{"type":"turn_start"}} @@ -74,7 +74,7 @@ {"type":"expect_outbound","label":"set_session_name@p2","frame":{"type":"set_session_name","name":"Replay fixture: provider_thread_resume","id":"t3-6"}} {"type":"emit_inbound","label":"session_info_changed@p2","frame":{"type":"session_info_changed","name":"Replay fixture: provider_thread_resume"}} {"type":"emit_inbound","label":"response:set_session_name@p2","frame":{"id":"t3-6","type":"response","command":"set_session_name","success":true}} -{"type":"expect_outbound","label":"prompt@p2","frame":{"type":"prompt","message":"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":"expect_outbound","label":"prompt@p2","frame":{"type":"prompt","message":"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","streamingBehavior":"steer"}} {"type":"emit_inbound","label":"response:prompt@p2","frame":{"type":"response","command":"prompt","success":true,"data":{"disposition":"started"}}} {"type":"emit_inbound","label":"agent_start@p2","frame":{"type":"agent_start"}} {"type":"emit_inbound","label":"turn_start@p2","frame":{"type":"turn_start"}} diff --git a/apps/server/src/orchestration-v2/testkit/fixtures/simple/pi_transcript.ndjson b/apps/server/src/orchestration-v2/testkit/fixtures/simple/pi_transcript.ndjson index ba7f84de2878..61c3eb45d7ca 100644 --- a/apps/server/src/orchestration-v2/testkit/fixtures/simple/pi_transcript.ndjson +++ b/apps/server/src/orchestration-v2/testkit/fixtures/simple/pi_transcript.ndjson @@ -12,7 +12,7 @@ {"type":"expect_outbound","label":"set_session_name","frame":{"type":"set_session_name","name":"Replay fixture: simple","id":"t3-5"}} {"type":"emit_inbound","label":"session_info_changed","frame":{"type":"session_info_changed","name":"Replay fixture: simple"}} {"type":"emit_inbound","label":"response:set_session_name","frame":{"id":"t3-5","type":"response","command":"set_session_name","success":true}} -{"type":"expect_outbound","label":"prompt","frame":{"type":"prompt","message":"Respond with the following text: fixture simple ok"}} +{"type":"expect_outbound","label":"prompt","frame":{"type":"prompt","message":"Respond with the following text: fixture simple ok","streamingBehavior":"steer"}} {"type":"emit_inbound","label":"response:prompt","frame":{"type":"response","command":"prompt","success":true,"data":{"disposition":"started"}}} {"type":"emit_inbound","label":"agent_start","frame":{"type":"agent_start"}} {"type":"emit_inbound","label":"turn_start","frame":{"type":"turn_start"}} diff --git a/apps/server/src/orchestration-v2/testkit/fixtures/thread_rollback/pi_transcript.ndjson b/apps/server/src/orchestration-v2/testkit/fixtures/thread_rollback/pi_transcript.ndjson index 2fc3def568e6..44aeb73c4cee 100644 --- a/apps/server/src/orchestration-v2/testkit/fixtures/thread_rollback/pi_transcript.ndjson +++ b/apps/server/src/orchestration-v2/testkit/fixtures/thread_rollback/pi_transcript.ndjson @@ -12,7 +12,7 @@ {"type":"expect_outbound","label":"set_session_name","frame":{"type":"set_session_name","name":"Replay fixture: thread_rollback","id":"t3-5"}} {"type":"emit_inbound","label":"session_info_changed","frame":{"type":"session_info_changed","name":"Replay fixture: thread_rollback"}} {"type":"emit_inbound","label":"response:set_session_name","frame":{"id":"t3-5","type":"response","command":"set_session_name","success":true}} -{"type":"expect_outbound","label":"prompt","frame":{"type":"prompt","message":"Respond with exactly: rollback fixture first turn complete"}} +{"type":"expect_outbound","label":"prompt","frame":{"type":"prompt","message":"Respond with exactly: rollback fixture first turn complete","streamingBehavior":"steer"}} {"type":"emit_inbound","label":"response:prompt","frame":{"type":"response","command":"prompt","success":true,"data":{"disposition":"started"}}} {"type":"emit_inbound","label":"agent_start","frame":{"type":"agent_start"}} {"type":"emit_inbound","label":"turn_start","frame":{"type":"turn_start"}} @@ -52,7 +52,7 @@ {"type":"emit_inbound","label":"response:get_entries","frame":{"id":"t3-7","type":"response","command":"get_entries","success":true,"data":{"entries":[{"type":"model_change","id":"cc15a6b4","parentId":"00c31b78","timestamp":"2026-01-01T00:00:00.344Z","provider":"openrouter","modelId":"deepseek/deepseek-v4-flash"},{"type":"session_info","id":"2c1dcf96","parentId":"cc15a6b4","timestamp":"2026-01-01T00:00:00.349Z","name":"Replay fixture: thread_rollback"},{"type":"message","id":"aca5528d","parentId":"2c1dcf96","timestamp":"2026-01-01T00:00:00.357Z","message":{"role":"system","content":"","sections":{"preamble":"","tools":"","rules":"","docs":"","cwd":""},"timestamp":1790898591831,"toolsAdded":[{"name":"read","description":"Read the contents of a file. Supports text files and images (jpg, png, gif, webp, bmp). Images are sent as attachments. For text files, output is truncated to 2000 lines or 50KB (whichever is hit first). Use offset/limit for large files. When you need the full file, continue with offset until complete.","parameters":{"type":"object","required":["path"],"properties":{"path":{"type":"string","description":"Path to the file to read (relative or absolute)"},"offset":{"type":"number","description":"Line number to start reading from (1-indexed)"},"limit":{"type":"number","description":"Maximum number of lines to read"}}},"constrainedSampling":{"type":"json_schema","strict":"prefer"}},{"name":"bash","description":"Execute a bash command in the current working directory. Returns stdout and stderr. Output is truncated to last 2000 lines or 50KB (whichever is hit first). If truncated, full output is saved to a temp file. Optionally provide a timeout in seconds.","parameters":{"type":"object","required":["command"],"properties":{"command":{"type":"string","description":"Shell command to execute"},"timeout":{"type":"number","description":"Timeout in seconds (optional, no default timeout)"}}},"constrainedSampling":{"type":"json_schema","strict":"prefer"}},{"name":"edit","description":"Edit a single file using exact text replacement. Every edits[].oldText must match a unique, non-overlapping region of the original file. If two changes affect the same block or nearby lines, merge them into one edit instead of emitting overlapping edits. Do not include large unchanged regions just to connect distant changes.","parameters":{"type":"object","required":["path","edits"],"properties":{"path":{"type":"string","description":"Path to the file to edit (relative or absolute)"},"edits":{"type":"array","items":{"type":"object","required":["oldText","newText"],"properties":{"oldText":{"type":"string","description":"Exact text for one targeted replacement. It must be unique in the original file and must not overlap with any other edits[].oldText in the same call."},"newText":{"type":"string","description":"Replacement text for this targeted edit."}}},"description":"One or more targeted replacements. Each edit is matched against the original file, not incrementally. Do not include overlapping or nested edits. If two changes touch the same block or nearby lines, merge them into one edit instead."}}},"constrainedSampling":{"type":"json_schema","strict":"prefer"}},{"name":"write","description":"Write content to a file. Creates the file if it doesn't exist, overwrites if it does. Automatically creates parent directories.","parameters":{"type":"object","required":["path","content"],"properties":{"path":{"type":"string","description":"Path to the file to write (relative or absolute)"},"content":{"type":"string","description":"Content to write to the file"}}},"constrainedSampling":{"type":"json_schema","strict":"prefer"}}]}},{"type":"message","id":"70559bb2","parentId":"aca5528d","timestamp":"2026-01-01T00:00:00.357Z","message":{"role":"user","content":[{"type":"text","text":"Respond with exactly: rollback fixture first turn complete"}],"timestamp":1767225600354}},{"type":"message","id":"c906b4e2","parentId":"70559bb2","timestamp":"2026-01-01T00:00:02.034Z","message":{"role":"assistant","content":[{"type":"thinking","thinking":"The user is asking me to respond with exactly: \"rollback fixture first turn complete\"\n\nLet me check if there are any rollback-related fixtures or tests in the workspace first, as this seems like a test instruction.","thinkingSignature":""},{"type":"text","text":"rollback fixture first turn complete"}],"api":"openai-completions","provider":"openrouter","model":"deepseek/deepseek-v4-flash","usage":{"input":1791,"output":52,"cacheRead":0,"cacheWrite":0,"reasoning":54,"totalTokens":1843,"cost":{"input":0.00007497126,"output":0.00000435344,"cacheRead":0,"cacheWrite":0,"total":0.0000793247}},"stopReason":"stop","timestamp":1767225600369,"responseId":"gen-1790898592-a7T4i6GUUneRyLxy0gtp","rawStopReason":"stop","thinkingLevel":"high"}}],"leafId":"c906b4e2"}}} {"type":"expect_outbound","label":"get_session_stats","frame":{"type":"get_session_stats","id":"t3-8"}} {"type":"emit_inbound","label":"response:get_session_stats","frame":{"id":"t3-8","type":"response","command":"get_session_stats","success":true,"data":{"sessionFile":"/pi-sessions/session-1.jsonl","sessionId":"00000000-0000-4000-8000-000000000002","userMessages":1,"assistantMessages":1,"toolCalls":0,"toolResults":0,"totalMessages":3,"tokens":{"input":1791,"output":52,"cacheRead":0,"cacheWrite":0,"total":1843},"cost":0.0000793247,"contextUsage":{"tokens":1843,"contextWindow":1048576,"percent":0.17576217651367188}}}} -{"type":"expect_outbound","label":"prompt","frame":{"type":"prompt","message":"Respond with exactly: rollback fixture second turn complete"}} +{"type":"expect_outbound","label":"prompt","frame":{"type":"prompt","message":"Respond with exactly: rollback fixture second turn complete","streamingBehavior":"steer"}} {"type":"emit_inbound","label":"response:prompt","frame":{"type":"response","command":"prompt","success":true,"data":{"disposition":"started"}}} {"type":"emit_inbound","label":"agent_start","frame":{"type":"agent_start"}} {"type":"emit_inbound","label":"turn_start","frame":{"type":"turn_start"}} @@ -100,7 +100,7 @@ {"type":"expect_outbound","label":"set_session_name","frame":{"type":"set_session_name","name":"Replay fixture: thread_rollback","id":"t3-19"}} {"type":"emit_inbound","label":"session_info_changed","frame":{"type":"session_info_changed","name":"Replay fixture: thread_rollback"}} {"type":"emit_inbound","label":"response:set_session_name","frame":{"id":"t3-19","type":"response","command":"set_session_name","success":true}} -{"type":"expect_outbound","label":"prompt","frame":{"type":"prompt","message":"Repeat the conversation verbatim."}} +{"type":"expect_outbound","label":"prompt","frame":{"type":"prompt","message":"Repeat the conversation verbatim.","streamingBehavior":"steer"}} {"type":"emit_inbound","label":"response:prompt","frame":{"type":"response","command":"prompt","success":true,"data":{"disposition":"started"}}} {"type":"emit_inbound","label":"agent_start","frame":{"type":"agent_start"}} {"type":"emit_inbound","label":"turn_start","frame":{"type":"turn_start"}} diff --git a/apps/server/src/orchestration-v2/testkit/fixtures/thread_rollback_after_stop/pi_transcript.ndjson b/apps/server/src/orchestration-v2/testkit/fixtures/thread_rollback_after_stop/pi_transcript.ndjson index 2d92c45b2adb..ffdecab232a5 100644 --- a/apps/server/src/orchestration-v2/testkit/fixtures/thread_rollback_after_stop/pi_transcript.ndjson +++ b/apps/server/src/orchestration-v2/testkit/fixtures/thread_rollback_after_stop/pi_transcript.ndjson @@ -12,7 +12,7 @@ {"type":"expect_outbound","label":"set_session_name","frame":{"type":"set_session_name","name":"Replay fixture: thread_rollback_after_stop","id":"t3-5"}} {"type":"emit_inbound","label":"session_info_changed","frame":{"type":"session_info_changed","name":"Replay fixture: thread_rollback_after_stop"}} {"type":"emit_inbound","label":"response:set_session_name","frame":{"id":"t3-5","type":"response","command":"set_session_name","success":true}} -{"type":"expect_outbound","label":"prompt","frame":{"type":"prompt","message":"Respond with exactly: rollback fixture first turn complete"}} +{"type":"expect_outbound","label":"prompt","frame":{"type":"prompt","message":"Respond with exactly: rollback fixture first turn complete","streamingBehavior":"steer"}} {"type":"emit_inbound","label":"response:prompt","frame":{"type":"response","command":"prompt","success":true,"data":{"disposition":"started"}}} {"type":"emit_inbound","label":"agent_start","frame":{"type":"agent_start"}} {"type":"emit_inbound","label":"turn_start","frame":{"type":"turn_start"}} @@ -42,7 +42,7 @@ {"type":"emit_inbound","label":"response:get_entries","frame":{"id":"t3-7","type":"response","command":"get_entries","success":true,"data":{"entries":[{"type":"model_change","id":"7aecf566","parentId":"10e2239d","timestamp":"2026-01-01T00:00:00.359Z","provider":"openrouter","modelId":"deepseek/deepseek-v4-flash"},{"type":"session_info","id":"5e1135b5","parentId":"7aecf566","timestamp":"2026-01-01T00:00:00.363Z","name":"Replay fixture: thread_rollback_after_stop"},{"type":"message","id":"66a686a0","parentId":"5e1135b5","timestamp":"2026-01-01T00:00:00.371Z","message":{"role":"system","content":"","sections":{"preamble":"","tools":"","rules":"","docs":"","cwd":""},"timestamp":1790898591965,"toolsAdded":[{"name":"read","description":"Read the contents of a file. Supports text files and images (jpg, png, gif, webp, bmp). Images are sent as attachments. For text files, output is truncated to 2000 lines or 50KB (whichever is hit first). Use offset/limit for large files. When you need the full file, continue with offset until complete.","parameters":{"type":"object","required":["path"],"properties":{"path":{"type":"string","description":"Path to the file to read (relative or absolute)"},"offset":{"type":"number","description":"Line number to start reading from (1-indexed)"},"limit":{"type":"number","description":"Maximum number of lines to read"}}},"constrainedSampling":{"type":"json_schema","strict":"prefer"}},{"name":"bash","description":"Execute a bash command in the current working directory. Returns stdout and stderr. Output is truncated to last 2000 lines or 50KB (whichever is hit first). If truncated, full output is saved to a temp file. Optionally provide a timeout in seconds.","parameters":{"type":"object","required":["command"],"properties":{"command":{"type":"string","description":"Shell command to execute"},"timeout":{"type":"number","description":"Timeout in seconds (optional, no default timeout)"}}},"constrainedSampling":{"type":"json_schema","strict":"prefer"}},{"name":"edit","description":"Edit a single file using exact text replacement. Every edits[].oldText must match a unique, non-overlapping region of the original file. If two changes affect the same block or nearby lines, merge them into one edit instead of emitting overlapping edits. Do not include large unchanged regions just to connect distant changes.","parameters":{"type":"object","required":["path","edits"],"properties":{"path":{"type":"string","description":"Path to the file to edit (relative or absolute)"},"edits":{"type":"array","items":{"type":"object","required":["oldText","newText"],"properties":{"oldText":{"type":"string","description":"Exact text for one targeted replacement. It must be unique in the original file and must not overlap with any other edits[].oldText in the same call."},"newText":{"type":"string","description":"Replacement text for this targeted edit."}}},"description":"One or more targeted replacements. Each edit is matched against the original file, not incrementally. Do not include overlapping or nested edits. If two changes touch the same block or nearby lines, merge them into one edit instead."}}},"constrainedSampling":{"type":"json_schema","strict":"prefer"}},{"name":"write","description":"Write content to a file. Creates the file if it doesn't exist, overwrites if it does. Automatically creates parent directories.","parameters":{"type":"object","required":["path","content"],"properties":{"path":{"type":"string","description":"Path to the file to write (relative or absolute)"},"content":{"type":"string","description":"Content to write to the file"}}},"constrainedSampling":{"type":"json_schema","strict":"prefer"}}]}},{"type":"message","id":"0d296cbc","parentId":"66a686a0","timestamp":"2026-01-01T00:00:00.372Z","message":{"role":"user","content":[{"type":"text","text":"Respond with exactly: rollback fixture first turn complete"}],"timestamp":1767225600369}},{"type":"message","id":"b184b5a8","parentId":"0d296cbc","timestamp":"2026-01-01T00:00:01.622Z","message":{"role":"assistant","content":[{"type":"thinking","thinking":"The user wants me to respond with exactly \"rollback fixture first turn complete\".","thinkingSignature":""},{"type":"text","text":"rollback fixture first turn complete"}],"api":"openai-completions","provider":"openrouter","model":"deepseek/deepseek-v4-flash","usage":{"input":1795,"output":24,"cacheRead":0,"cacheWrite":0,"reasoning":21,"totalTokens":1819,"cost":{"input":0.00007513869999999999,"output":0.00000200928,"cacheRead":0,"cacheWrite":0,"total":0.00007714798}},"stopReason":"stop","timestamp":1767225600386,"responseId":"gen-1790898592-vpyqfefcvhCaD41XnfZc","rawStopReason":"stop","thinkingLevel":"high"}}],"leafId":"b184b5a8"}}} {"type":"expect_outbound","label":"get_session_stats","frame":{"type":"get_session_stats","id":"t3-8"}} {"type":"emit_inbound","label":"response:get_session_stats","frame":{"id":"t3-8","type":"response","command":"get_session_stats","success":true,"data":{"sessionFile":"/pi-sessions/session-1.jsonl","sessionId":"00000000-0000-4000-8000-000000000002","userMessages":1,"assistantMessages":1,"toolCalls":0,"toolResults":0,"totalMessages":3,"tokens":{"input":1795,"output":24,"cacheRead":0,"cacheWrite":0,"total":1819},"cost":0.00007714798,"contextUsage":{"tokens":1819,"contextWindow":1048576,"percent":0.17347335815429688}}}} -{"type":"expect_outbound","label":"prompt","frame":{"type":"prompt","message":"Run this exact local command: `node -e \"console.log('interrupt fixture tool started'); setTimeout(() => {}, 30000)\"`. Do not answer until it completes, then respond exactly: interrupt fixture should not finish naturally."}} +{"type":"expect_outbound","label":"prompt","frame":{"type":"prompt","message":"Run this exact local command: `node -e \"console.log('interrupt fixture tool started'); setTimeout(() => {}, 30000)\"`. Do not answer until it completes, then respond exactly: interrupt fixture should not finish naturally.","streamingBehavior":"steer"}} {"type":"emit_inbound","label":"response:prompt","frame":{"type":"response","command":"prompt","success":true,"data":{"disposition":"started"}}} {"type":"emit_inbound","label":"agent_start","frame":{"type":"agent_start"}} {"type":"emit_inbound","label":"turn_start","frame":{"type":"turn_start"}} @@ -164,7 +164,7 @@ {"type":"expect_outbound","label":"set_session_name@p2","frame":{"type":"set_session_name","name":"Replay fixture: thread_rollback_after_stop","id":"t3-6"}} {"type":"emit_inbound","label":"session_info_changed@p2","frame":{"type":"session_info_changed","name":"Replay fixture: thread_rollback_after_stop"}} {"type":"emit_inbound","label":"response:set_session_name@p2","frame":{"id":"t3-6","type":"response","command":"set_session_name","success":true}} -{"type":"expect_outbound","label":"prompt@p2","frame":{"type":"prompt","message":"Respond with exactly: rollback fixture second turn complete"}} +{"type":"expect_outbound","label":"prompt@p2","frame":{"type":"prompt","message":"Respond with exactly: rollback fixture second turn complete","streamingBehavior":"steer"}} {"type":"emit_inbound","label":"response:prompt@p2","frame":{"type":"response","command":"prompt","success":true,"data":{"disposition":"started"}}} {"type":"emit_inbound","label":"agent_start@p2","frame":{"type":"agent_start"}} {"type":"emit_inbound","label":"turn_start@p2","frame":{"type":"turn_start"}} @@ -223,7 +223,7 @@ {"type":"expect_outbound","label":"set_session_name@p2","frame":{"type":"set_session_name","name":"Replay fixture: thread_rollback_after_stop","id":"t3-17"}} {"type":"emit_inbound","label":"session_info_changed@p2","frame":{"type":"session_info_changed","name":"Replay fixture: thread_rollback_after_stop"}} {"type":"emit_inbound","label":"response:set_session_name@p2","frame":{"id":"t3-17","type":"response","command":"set_session_name","success":true}} -{"type":"expect_outbound","label":"prompt@p2","frame":{"type":"prompt","message":"Repeat the conversation verbatim."}} +{"type":"expect_outbound","label":"prompt@p2","frame":{"type":"prompt","message":"Repeat the conversation verbatim.","streamingBehavior":"steer"}} {"type":"emit_inbound","label":"response:prompt@p2","frame":{"type":"response","command":"prompt","success":true,"data":{"disposition":"started"}}} {"type":"emit_inbound","label":"agent_start@p2","frame":{"type":"agent_start"}} {"type":"emit_inbound","label":"turn_start@p2","frame":{"type":"turn_start"}} diff --git a/apps/server/src/orchestration-v2/testkit/fixtures/turn_interrupt_mid_tool/pi_transcript.ndjson b/apps/server/src/orchestration-v2/testkit/fixtures/turn_interrupt_mid_tool/pi_transcript.ndjson index 14e710c1ff9d..d4db23f4fd8e 100644 --- a/apps/server/src/orchestration-v2/testkit/fixtures/turn_interrupt_mid_tool/pi_transcript.ndjson +++ b/apps/server/src/orchestration-v2/testkit/fixtures/turn_interrupt_mid_tool/pi_transcript.ndjson @@ -12,7 +12,7 @@ {"type":"expect_outbound","label":"set_session_name","frame":{"type":"set_session_name","name":"Replay fixture: turn_interrupt_mid_tool","id":"t3-5"}} {"type":"emit_inbound","label":"session_info_changed","frame":{"type":"session_info_changed","name":"Replay fixture: turn_interrupt_mid_tool"}} {"type":"emit_inbound","label":"response:set_session_name","frame":{"id":"t3-5","type":"response","command":"set_session_name","success":true}} -{"type":"expect_outbound","label":"prompt","frame":{"type":"prompt","message":"Run this exact local command: `node -e \"console.log('interrupt fixture tool started'); setTimeout(() => {}, 30000)\"`. Do not answer until it completes, then respond exactly: interrupt fixture should not finish naturally."}} +{"type":"expect_outbound","label":"prompt","frame":{"type":"prompt","message":"Run this exact local command: `node -e \"console.log('interrupt fixture tool started'); setTimeout(() => {}, 30000)\"`. Do not answer until it completes, then respond exactly: interrupt fixture should not finish naturally.","streamingBehavior":"steer"}} {"type":"emit_inbound","label":"response:prompt","frame":{"type":"response","command":"prompt","success":true,"data":{"disposition":"started"}}} {"type":"emit_inbound","label":"agent_start","frame":{"type":"agent_start"}} {"type":"emit_inbound","label":"turn_start","frame":{"type":"turn_start"}} diff --git a/apps/web/src/components/AnimatedWidth.tsx b/apps/web/src/components/AnimatedWidth.tsx new file mode 100644 index 000000000000..d893f2b6aff6 --- /dev/null +++ b/apps/web/src/components/AnimatedWidth.tsx @@ -0,0 +1,39 @@ +import { type ReactNode, useLayoutEffect, useRef, useState } from "react"; + +import { observeResize } from "~/lib/observeResize"; +import { cn } from "~/lib/utils"; + +export function AnimatedWidth({ + children, + className, +}: { + readonly children: ReactNode; + readonly className?: string; +}) { + const contentRef = useRef(null); + const [width, setWidth] = useState(null); + + useLayoutEffect(() => { + const element = contentRef.current; + if (!element) return; + return observeResize(element, () => { + const nextWidth = Math.ceil(element.getBoundingClientRect().width); + if (nextWidth > 0) setWidth(nextWidth); + }); + }, []); + + return ( + + + {children} + + + ); +} diff --git a/apps/web/src/components/CommandPalette.tsx b/apps/web/src/components/CommandPalette.tsx index c423110dd3f2..5d86c9c171c0 100644 --- a/apps/web/src/components/CommandPalette.tsx +++ b/apps/web/src/components/CommandPalette.tsx @@ -3134,6 +3134,10 @@ function OpenCommandPaletteDialog(props: { ? "Create & Add" : "Add"; const addShortcutLabel = hasHighlightedBrowseItem ? `${submitModifierLabel} Enter` : "Enter"; + const hasHighlightedNewProjectItem = newProjectFlow !== null && highlightedItemValue !== null; + const createShortcutLabel = hasHighlightedNewProjectItem + ? `${submitModifierLabel} Enter` + : "Enter"; const remoteProjectButtonLabel = addProjectCloneFlow ? addProjectCloneFlow.source === "url" ? "Continue" @@ -3227,15 +3231,17 @@ function OpenCommandPaletteDialog(props: { } // Enter creates the project unless an item below the name is - // highlighted, in which case it runs that item. + // highlighted, in which case it runs that item. Mod+Enter always creates. if ( newProjectFlow !== null && event.key === "Enter" && - highlightedItemValue === null && + (!hasHighlightedNewProjectItem || isPrimaryModifierPressed(event)) && // Enter that confirms an IME composition is part of typing the name. !event.nativeEvent.isComposing && event.keyCode !== 229 ) { + // Base UI would otherwise also click the highlighted row. + (event as typeof event & { preventBaseUIHandler?: () => void }).preventBaseUIHandler?.(); event.preventDefault(); void submitNewProject(); return; @@ -3445,7 +3451,7 @@ function OpenCommandPaletteDialog(props: { size="xs" tabIndex={-1} className="absolute inset-e-2.5 top-1/2 -translate-y-1/2" - aria-label="Create (Enter)" + aria-label={`Create (${createShortcutLabel})`} disabled={!canSubmitNewProject} onMouseDown={(event) => { event.preventDefault(); @@ -3458,10 +3464,10 @@ function OpenCommandPaletteDialog(props: { > {isCreatingNewProject ? "Creating" : "Create"} - Enter + {createShortcutLabel} - Create (Enter) + {`Create (${createShortcutLabel})`} ) : addProjectCloneFlow?.step === "repository" ? ( @@ -3574,8 +3580,9 @@ function OpenCommandPaletteDialog(props: { inputProps={{ // The submit button is absolutely positioned over the field, so the // inner input must reserve enough room for the full action label. - className: - addProjectCloneFlow?.step === "repository" || newProjectFlow !== null + className: hasHighlightedNewProjectItem + ? "*:data-[slot=autocomplete-input]:pe-38!" + : addProjectCloneFlow?.step === "repository" || newProjectFlow !== null ? "*:data-[slot=autocomplete-input]:pe-32!" : isBrowsing ? browseInputEndPaddingClass({ diff --git a/apps/web/src/components/pullRequest/PullRequestDetailPanel.tsx b/apps/web/src/components/pullRequest/PullRequestDetailPanel.tsx index 77bccacca464..009205404d10 100644 --- a/apps/web/src/components/pullRequest/PullRequestDetailPanel.tsx +++ b/apps/web/src/components/pullRequest/PullRequestDetailPanel.tsx @@ -2,6 +2,7 @@ import { parseChangeRequestUrl } from "@t3tools/shared/changeRequestUrl"; import { useAtomValue } from "@effect/atom-react"; import { usePullRequestStack } from "~/state/usePullRequestStack"; import { RefreshIcon } from "~/components/ui/refresh-icon"; +import { AnimatedWidth } from "~/components/AnimatedWidth"; import { scopedThreadKey, scopeProjectRef, @@ -2122,21 +2123,20 @@ export function PullRequestDetailPanel({ ) : ( )} - - - {selectedMergeMethodLabel} - - - Confirm merge? + + + + {selectedMergeMethodLabel} + + + {pendingAction === "merge" + ? "Merging..." + : inlineConfirm.isArmed("merge") + ? "Confirm merge?" + : selectedMergeMethodLabel} + - - {pendingAction === "merge" - ? "Merging..." - : inlineConfirm.isArmed("merge") - ? "Confirm merge?" - : selectedMergeMethodLabel} - - + } diff --git a/apps/web/src/components/settings/scopedSettings.test.ts b/apps/web/src/components/settings/scopedSettings.test.ts index e8e5d0a1243e..dc5b7d74fd9b 100644 --- a/apps/web/src/components/settings/scopedSettings.test.ts +++ b/apps/web/src/components/settings/scopedSettings.test.ts @@ -230,6 +230,89 @@ describe("scoped settings writes", () => { expect(persistClient).not.toHaveBeenCalled(); }); + describe("an environment that saves but drops keys it does not know", () => { + // Laptop knows every key; Server is an older build that keeps its settings as they were. + const upgradedLaptop = (saved: Partial) => + vi.fn(async ({ environmentId }: { environmentId: EnvironmentId }) => ({ + _tag: "Success" as const, + value: { + ...DEFAULT_SERVER_SETTINGS, + ...(environmentId === laptop.environmentId ? saved : {}), + }, + })); + const outcome = async ( + plan: ReturnType, + persistServer: ReturnType, + ) => { + const result = await persistScopedSettingsPatch(plan, persistServer, vi.fn()); + return { + saved: result.savedEnvironments.map(({ label }) => label), + failed: result.failedEnvironments.map(({ label, message }) => ({ label, message })), + }; + }; + const unsupported = [ + { + label: server.label, + message: "This environment does not support this setting. Update T3 Code on it.", + }, + ]; + + it("reports a dropped toggle", async () => { + expect( + await outcome( + planScopedSettingsPatch(all, environments, { reviewGuidesEnabled: true }), + upgradedLaptop({ reviewGuidesEnabled: true }), + ), + ).toEqual({ saved: [laptop.label], failed: unsupported }); + }); + + it("reports dropped text but accepts the server trimming it", async () => { + expect( + await outcome( + planScopedSettingsPatch(all, environments, { reviewGuideInstructions: " Read tests " }), + upgradedLaptop({ reviewGuideInstructions: "Read tests" }), + ), + ).toEqual({ saved: [laptop.label], failed: unsupported }); + }); + + it("reports a dropped key inside a project override", async () => { + expect( + await outcome( + planScopedSettingsPatch(project, environments, { defaultAutoPull: true }), + upgradedLaptop({ + projectSettingsOverrides: { [laptopProjectId]: { defaultAutoPull: true } }, + }), + ), + ).toEqual({ saved: [laptop.label], failed: unsupported }); + }); + + it("reports a nested key dropped beside one that saved", async () => { + const patch = { defaultAutoPull: true, enableAgentBrowserAccess: false }; + const persistServer = vi.fn(async ({ environmentId }: { environmentId: EnvironmentId }) => ({ + _tag: "Success" as const, + value: { + ...DEFAULT_SERVER_SETTINGS, + projectSettingsOverrides: + environmentId === laptop.environmentId + ? { [laptopProjectId]: patch } + : { [projectId]: { defaultAutoPull: true } }, + }, + })); + expect( + await outcome(planScopedSettingsPatch(project, environments, patch), persistServer), + ).toEqual({ saved: [laptop.label], failed: unsupported }); + }); + + it("accepts a write that matches what the environment already had", async () => { + expect( + await outcome( + planScopedSettingsPatch(all, environments, { reviewGuidesEnabled: false }), + upgradedLaptop({}), + ), + ).toEqual({ saved: [laptop.label, server.label], failed: [] }); + }); + }); + it("persists client keys locally at any scope alongside server keys", async () => { const persistServer = vi.fn().mockResolvedValue({ _tag: "Success" }); const persistClient = vi.fn(); @@ -262,6 +345,7 @@ describe("scoped settings writes", () => { { environmentId: server.environmentId, label: server.label, + previous: expect.anything(), patch: { projectSettingsOverrides: { [projectId]: { enableAgentBrowserAccess: false, defaultAutoPull: true }, @@ -271,6 +355,7 @@ describe("scoped settings writes", () => { { environmentId: laptop.environmentId, label: laptop.label, + previous: expect.anything(), patch: { projectSettingsOverrides: { [laptopProjectId]: { defaultAutoPull: true } } }, }, ]); @@ -391,6 +476,7 @@ describe("scoped settings writes", () => { { environmentId: server.environmentId, label: server.label, + previous: expect.anything(), patch: { projectSettingsOverrides: { [projectId]: { enableAgentBrowserAccess: false } } }, }, ]); @@ -527,6 +613,7 @@ describe("project overrides at environment scope", () => { { environmentId: laptop, label: laptop, + previous: expect.anything(), patch: { projectSettingsOverrides: { [fleet]: { defaultThreadEnvMode: "local" }, [t3]: null }, }, @@ -655,6 +742,11 @@ it("routes legacy model preference keys exclusively to the selected environment" expect(plan.clientPatch).toEqual({}); expect(plan.hasClientWrite).toBe(false); expect(plan.serverWrites).toEqual([ - { environmentId: server.environmentId, label: server.label, patch }, + { + environmentId: server.environmentId, + label: server.label, + previous: expect.anything(), + patch, + }, ]); }); diff --git a/apps/web/src/components/settings/scopedSettings.ts b/apps/web/src/components/settings/scopedSettings.ts index f4fe6e41905c..dcedf419caf2 100644 --- a/apps/web/src/components/settings/scopedSettings.ts +++ b/apps/web/src/components/settings/scopedSettings.ts @@ -166,6 +166,8 @@ interface ScopedServerWrite { readonly environmentId: EnvironmentId; readonly label: string; readonly patch: ServerSettingsPatch; + /** The environment's settings before the write, to tell a dropped key from a saved one. */ + readonly previous: ServerSettings; } function projectOverrideWrites( @@ -194,6 +196,7 @@ function projectOverrideWrites( writes.set(member.environmentId, { environmentId: member.environmentId, label: environment.label, + previous: settings, patch: { projectSettingsOverrides: { ...existing?.patch.projectSettingsOverrides, @@ -266,29 +269,38 @@ export function planScopedSettingsPatch( return next as ProjectSettingsOverrides; }) : scope.kind === "all" || scope.kind === "environment" - ? connectedEnvironments.map((environment) => ({ - environmentId: environment.environmentId, - label: environment.label, - patch: - environment.serverConfig?.settings.worktreeCleanup != null && - serverPatch.storageCleanup && - Object.keys(serverPatch.storageCleanup).some((key) => key.startsWith("worktree")) - ? { - ...serverPatch, - worktreeCleanup: { - mode: "custom", - rules: { - ...resolveWorktreeCleanup(environment.serverConfig.settings, null), - ...Object.fromEntries( - Object.entries(serverPatch.storageCleanup).filter(([key]) => - key.startsWith("worktree"), - ), - ), - }, - }, - } - : serverPatch, - })) + ? connectedEnvironments.flatMap((environment) => { + const previous = environment.serverConfig?.settings; + if (!previous) return []; + return [ + { + environmentId: environment.environmentId, + label: environment.label, + previous, + patch: + previous.worktreeCleanup != null && + serverPatch.storageCleanup && + Object.keys(serverPatch.storageCleanup).some((key) => + key.startsWith("worktree"), + ) + ? { + ...serverPatch, + worktreeCleanup: { + mode: "custom", + rules: { + ...resolveWorktreeCleanup(previous, null), + ...Object.fromEntries( + Object.entries(serverPatch.storageCleanup).filter(([key]) => + key.startsWith("worktree"), + ), + ), + }, + }, + } + : serverPatch, + }, + ]; + }) : []; const hasClientWrite = Object.keys(clientPatch).length > 0; const hasWrite = hasClientWrite || serverWrites.length > 0; @@ -371,6 +383,7 @@ export function planProjectOverridesClear( writes.set(environmentId, { environmentId, label: environment.label, + previous: settings, patch: { projectSettingsOverrides: { ...existing?.patch.projectSettingsOverrides, @@ -389,13 +402,56 @@ export function planProjectOverridesClear( }; } +/** + * Whether a server kept `before` where `value` asked for a change. Objects are + * compared field by field, so a nested key an older server dropped is caught + * even when a sibling field saved. A null clears the field; strings may come + * back trimmed. + */ +function writeWasDropped(value: unknown, before: unknown, saved: unknown): boolean { + if (value === null) return before != null && Equal.equals(saved, before); + if (isPlainObject(value) && !Array.isArray(before) && !Array.isArray(saved)) { + const field = (source: unknown, key: string) => + isPlainObject(source) ? source[key] : undefined; + return Object.entries(value).some(([key, next]) => + writeWasDropped(next, field(before, key), field(saved, key)), + ); + } + const unchanged = + typeof value === "string" && typeof before === "string" + ? value.trim() === before + : Equal.equals(value, before); + return !unchanged && Equal.equals(saved, before); +} + +/** + * An older server drops keys it does not know, top-level or nested, and still + * succeeds. Left unreported, the row stays mixed forever. + */ +function droppedSettingKeys(write: ScopedServerWrite, saved: ServerSettings | undefined) { + if (saved === undefined) return []; + return Object.entries(write.patch).flatMap(([key, value]) => + writeWasDropped( + value, + write.previous[key as keyof ServerSettings], + saved[key as keyof ServerSettings], + ) + ? [key] + : [], + ); +} + /** Wait for every target so a failed environment does not hide successful or later writes. */ export async function persistScopedSettingsPatch( plan: ReturnType, persistServer: (input: { environmentId: EnvironmentId; input: { patch: ServerSettingsPatch }; - }) => Promise<{ readonly _tag: "Success" | "Failure"; readonly cause?: Cause.Cause }>, + }) => Promise<{ + readonly _tag: "Success" | "Failure"; + readonly cause?: Cause.Cause; + readonly value?: ServerSettings; + }>, persistClient: (patch: ClientSettingsPatch) => void, ) { if (plan.hasClientWrite) persistClient(plan.clientPatch); @@ -404,9 +460,25 @@ export async function persistScopedSettingsPatch( persistServer({ environmentId, input: { patch } }), ), ); + const saved = (index: number) => { + const result = results[index]; + return ( + result?.status === "fulfilled" && + result.value._tag === "Success" && + droppedSettingKeys(plan.serverWrites[index]!, result.value.value).length === 0 + ); + }; const failedEnvironments = plan.serverWrites.flatMap((environment, index) => { + if (saved(index)) return []; const result = results[index]; - if (result?.status === "fulfilled" && result.value._tag === "Success") return []; + if (result?.status === "fulfilled" && result.value._tag === "Success") { + return [ + { + ...environment, + message: "This environment does not support this setting. Update T3 Code on it.", + }, + ]; + } const error: unknown = result?.status === "rejected" ? result.reason @@ -423,10 +495,7 @@ export async function persistScopedSettingsPatch( }, ]; }); - const savedEnvironments = plan.serverWrites.filter((_, index) => { - const result = results[index]; - return result?.status === "fulfilled" && result.value._tag === "Success"; - }); + const savedEnvironments = plan.serverWrites.filter((_, index) => saved(index)); return { failedEnvironments, savedEnvironments, diff --git a/packages/provider-pi/src/server/adapter.test.ts b/packages/provider-pi/src/server/adapter.test.ts index cab0043ab185..98a956cc5a08 100644 --- a/packages/provider-pi/src/server/adapter.test.ts +++ b/packages/provider-pi/src/server/adapter.test.ts @@ -90,6 +90,8 @@ interface FakePi { readonly vetoNextSwitch: () => void; /** Fields overriding the recorded idle state in the next `get_state` acks, in order. */ readonly queueState: (data: Record) => void; + /** Change Pi's streaming state before its agent_start event reaches T3. */ + readonly setStreaming: (streaming: boolean) => void; /** Hold the next `get_state` response until the test resolves it. */ readonly deferNextState: () => void; /** Resolve the held `get_state` request. */ @@ -154,6 +156,7 @@ const makeFakePi: Effect.Effect = Effect.gen(function* () { let sessionFile = FAKE_SESSION_FILE; let sessionGeneration = 0; let models: ReadonlyArray = []; + let isStreaming: boolean | null = null; let stdinBuffer = ""; const emit = (record: PiRpcRecord, chunkBytes?: number) => @@ -166,7 +169,8 @@ const makeFakePi: Effect.Effect = Effect.gen(function* () { }); const respondTo = (record: PiRpcRecord): PiRpcRecord | null => { - if (typeof record["id"] !== "string") return null; + if (typeof record["id"] !== "string" && !(record["type"] === "prompt" && isStreaming !== null)) + return null; const base = { type: "response", id: record["id"], @@ -181,7 +185,24 @@ const makeFakePi: Effect.Effect = Effect.gen(function* () { } // Queued data overrides fields of the recorded idle state, so a test // that only cares about the session file still gets a real shape. - return { ...base, data: { ...recordedIdleState(sessionFile), ...stateQueue.shift() } }; + return { + ...base, + data: { + ...recordedIdleState(sessionFile), + isStreaming: isStreaming ?? false, + ...stateQueue.shift(), + }, + }; + case "prompt": + if (isStreaming && record["streamingBehavior"] === undefined) { + return { + ...base, + success: false, + error: + "Agent is already processing. Specify streamingBehavior ('steer' or 'followUp') to queue the message.", + }; + } + return base; case "get_available_models": return { ...base, data: { models } }; case "new_session": { @@ -312,6 +333,9 @@ const makeFakePi: Effect.Effect = Effect.gen(function* () { vetoSwitch = true; }, queueState: (data) => stateQueue.push(data), + setStreaming: (streaming) => { + isStreaming = streaming; + }, queueStats: (data) => statsQueue.push(data), closeStdout: Queue.end(stdout), lastSpawn: () => lastSpawn, @@ -488,6 +512,37 @@ const expectModelFailure = (errorMessage: string) => }).pipe(Effect.scoped, Effect.provide(layerTest)); describe("PiAdapterV2", () => { + it.effect.each([false, true])( + "completes a resumed user turn before native wake events arrive, Pi streaming=%s", + (streaming) => + Effect.gen(function* () { + const fake = yield* makeFakePi; + const { runtime, takeEvent } = yield* openRuntime(fake); + const providerThread = yield* runtime.ensureThread({ + threadId: THREAD_ID, + modelSelection: modelSelection("default"), + runtimePolicy, + }); + fake.setStreaming(streaming); + yield* runtime.resumeThread({ providerThread }); + yield* startTurn(runtime, providerThread); + yield* fake.takeRequest("prompt"); + yield* fake.emit({ type: "agent_start" }); + yield* fake.emit({ + type: "message_end", + message: { + role: "assistant", + content: [{ type: "text", text: "Resumed prompt completed" }], + stopReason: "stop", + }, + }); + fake.queueState({ isStreaming: false }); + yield* fake.emit({ type: "agent_settled" }); + const terminal = yield* takeEvent((event) => event.type === "turn.terminal"); + assert.isTrue(terminal.type === "turn.terminal" && terminal.status === "completed"); + }).pipe(Effect.scoped, Effect.provide(layerTest)), + ); + it.effect.each([false, true])( "serializes rollback fork hooks with native wake ownership, wake=%s", (nativeWake) => diff --git a/packages/provider-pi/src/server/adapter.ts b/packages/provider-pi/src/server/adapter.ts index 96d2783ecf3f..722ac13b3989 100644 --- a/packages/provider-pi/src/server/adapter.ts +++ b/packages/provider-pi/src/server/adapter.ts @@ -2697,10 +2697,12 @@ export const makePiAdapterV2 = Effect.fn("makePiAdapterV2")(function* ( kind: "turn_start", }); } else if (payload !== null) { + // Pi may already be streaming before its wake reaches T3. + // Let Pi atomically queue the prompt or start an idle run. yield* connection.send({ type: "prompt", message: payload.message, - ...(wake === null ? {} : { streamingBehavior: "steer" }), + streamingBehavior: "steer", ...(payload.images.length === 0 ? {} : { images: payload.images }), }); pendingPromptResponses.push({ diff --git a/packages/provider-pi/src/server/mcpBridge.testkit.ts b/packages/provider-pi/src/server/mcpBridge.testkit.ts index afaa4888b320..2b9b4701e367 100644 --- a/packages/provider-pi/src/server/mcpBridge.testkit.ts +++ b/packages/provider-pi/src/server/mcpBridge.testkit.ts @@ -30,7 +30,10 @@ interface RegisteredTool { type AgentStartHook = ( event: { systemPrompt: string }, - ctx: { ui: { notify: (message: string, severity: string) => void } }, + ctx: { + model?: { api: string }; + ui: { notify: (message: string, severity: string) => void }; + }, ) => Promise<{ systemPrompt: string }>; export async function loadMcpBridge( diff --git a/packages/provider-pi/src/server/mcpExtensionSource.test.ts b/packages/provider-pi/src/server/mcpExtensionSource.test.ts index b8857d3c629c..8151878afafb 100644 --- a/packages/provider-pi/src/server/mcpExtensionSource.test.ts +++ b/packages/provider-pi/src/server/mcpExtensionSource.test.ts @@ -111,18 +111,40 @@ describe("Pi MCP tool results", () => { }); describe("Pi MCP tool exposure", () => { + const discoveryContext = { + model: { api: "anthropic-messages" }, + ui: { notify: () => undefined }, + }; + + it.each(["session_start", "session_tree", "model_select", "before_agent_start"])( + "declares optional T3 tools for OpenAI-compatible gateways at %s", + async (hookName) => { + const bridge = await loadMcpBridge({ modern: true, toolSearchAvailable: true }); + const hook = bridge.handlers.get(hookName); + assert.isDefined(hook); + await hook!( + { systemPrompt: "Pi system prompt" }, + { model: { api: "openai-completions" }, ui: { notify: assert.fail } }, + ); + assert.isTrue( + bridge.tools + .filter((tool) => tool.name.startsWith("mcp__t3-code__")) + .every((tool) => bridge.getActiveTools().includes(tool.name)), + ); + const result = await bridge.tools + .find((tool) => tool.name === "mcp__t3-code__preview_snapshot")! + .execute("gateway-call", {}); + assert.equal(result.content[0]?.text, "browser snapshot"); + assert.equal(bridge.tools.filter((tool) => tool.exposure === "hidden").length, 4); + }, + ); + it("keeps orchestration direct and optional bridge tools discoverable on modern Pi", async () => { const bridge = await loadMcpBridge({ modern: true, toolSearchAvailable: true }); - await bridge.handlers.get("session_start")!( - { systemPrompt: "" }, - { ui: { notify: () => undefined } }, - ); + await bridge.handlers.get("session_start")!({ systemPrompt: "" }, discoveryContext); const start = bridge.handlers.get("before_agent_start"); assert.isDefined(start); - const prompt = await start!( - { systemPrompt: "Pi system prompt" }, - { ui: { notify: () => undefined } }, - ); + const prompt = await start!({ systemPrompt: "Pi system prompt" }, discoveryContext); assert.include(prompt.systemPrompt, "orchestrator_capabilities"); assert.equal(bridge.servers.length, 0); assert.equal(bridge.tools.length, 8); @@ -154,6 +176,33 @@ describe("Pi MCP tool exposure", () => { assert.equal(bridge.requests.at(-1)?.method, "tools/call"); }); + it("restores gateway declarations after model switches and older tree loadouts without excluded tools", async () => { + const bridge = await loadMcpBridge({ + modern: true, + toolSearchAvailable: true, + allowsTool: (name) => name !== "mcp__t3-code__delegate_task", + }); + const event = { systemPrompt: "" }; + await bridge.handlers.get("session_start")!(event, discoveryContext); + const gatewayContext = { model: { api: "openai-completions" }, ui: { notify: assert.fail } }; + await bridge.handlers.get("model_select")!(event, gatewayContext); + bridge.restoreActiveTools(["read", "mcp__t3-code__task_status"]); + await bridge.handlers.get("session_tree")!(event, gatewayContext); + assert.deepEqual(bridge.getActiveTools(), [ + "read", + "mcp__t3-code__task_status", + "tool_search", + "mcp__t3-code__orchestrator_capabilities", + "mcp__t3-code__preview_snapshot", + ]); + await bridge.handlers.get("model_select")!(event, discoveryContext); + assert.equal( + bridge.tools.find((tool) => tool.name === "mcp__t3-code__preview_snapshot")?.exposure, + "deferred", + ); + assert.include(bridge.getActiveTools(), "mcp__t3-code__preview_snapshot"); + }); + it.each(["legacy Pi", "disabled tool search"])( "keeps tool execution available with %s", async (mode) => { @@ -164,7 +213,7 @@ describe("Pi MCP tool exposure", () => { }); if (mode !== "legacy Pi") { const start = bridge.handlers.get("session_start"); - await start!({ systemPrompt: "Pi system prompt" }, { ui: { notify: () => undefined } }); + await start!({ systemPrompt: "Pi system prompt" }, discoveryContext); } assert.equal(bridge.tools.filter((tool) => tool.exposure !== "hidden").length, 4); assert.isTrue( @@ -198,10 +247,7 @@ describe("Pi MCP tool exposure", () => { allowsTool: (name) => name === "read" || name === "tool_search" || name.startsWith("mcp__t3-code__"), }); - await bridge.handlers.get("session_start")!( - { systemPrompt: "" }, - { ui: { notify: () => undefined } }, - ); + await bridge.handlers.get("session_start")!({ systemPrompt: "" }, discoveryContext); assert.equal(bridge.tools.length, 4); assert.equal(bridge.tools.filter((tool) => tool.exposure === "direct").length, 3); const tool = bridge.tools.find((tool) => tool.name === "mcp__t3-code__preview_snapshot"); @@ -218,10 +264,7 @@ describe("Pi MCP tool exposure", () => { allowsTool: (name) => name !== "mcp__t3-code__delegate_task" && (!excludeSearch || name !== "tool_search"), }); - await bridge.handlers.get("session_start")!( - { systemPrompt: "" }, - { ui: { notify: () => undefined } }, - ); + await bridge.handlers.get("session_start")!({ systemPrompt: "" }, discoveryContext); bridge.restoreActiveTools([ "read", "mcp__t3-code__task_status", @@ -229,7 +272,7 @@ describe("Pi MCP tool exposure", () => { ]); const tree = bridge.handlers.get("session_tree"); assert.isDefined(tree); - await tree!({ systemPrompt: "" }, { ui: { notify: () => undefined } }); + await tree!({ systemPrompt: "" }, discoveryContext); assert.deepEqual(bridge.getActiveTools(), [ "read", "mcp__t3-code__task_status", diff --git a/packages/provider-pi/src/server/mcpExtensionSource.ts b/packages/provider-pi/src/server/mcpExtensionSource.ts index 7071b1ae6209..5ee4342e5bb9 100644 --- a/packages/provider-pi/src/server/mcpExtensionSource.ts +++ b/packages/provider-pi/src/server/mcpExtensionSource.ts @@ -1,8 +1,9 @@ /** * Source for the T3-owned Pi extension that consumes T3's HTTP MCP server. * - * Pi 0.99+ discovers optional HTTP bridge tools on demand. Older versions - * keep them directly available. Pi loads this TypeScript via `--extension`; the + * Pi 0.99+ discovers optional HTTP bridge tools on demand. OpenAI completion + * gateways need their schemas loaded up front. Older versions keep tools + * directly available. Pi loads this TypeScript via `--extension`; the * server writes it to a cache so packaged builds need no sibling .ts file. * * Do not import t3code modules from the string body. The Pi process resolves @@ -24,7 +25,7 @@ export const T3_PI_MCP_EXTENSION_PATH_ENV = "T3_PI_MCP_EXTENSION_PATH"; export const PI_FILE_CHANGE_TOOLS = ["edit", "write"] as const; export const PI_T3_MCP_EXTENSION_SOURCE = `\ -import { stripFrontmatter, type ExtensionAPI } from "@earendil-works/pi-coding-agent"; +import { stripFrontmatter, type ExtensionAPI, type ExtensionContext } from "@earendil-works/pi-coding-agent"; import * as NodeFSP from "node:fs/promises"; import * as NodePath from "node:path"; import { Type } from "typebox"; @@ -421,23 +422,31 @@ export default async function t3McpExtension(pi: ExtensionAPI) { // mcp.json configures it; retry a failed connection at session_start. await ensureStarted().catch(() => undefined); - const reconcileDiscovery = () => { + const reconcileDiscovery = (ctx: Pick) => { if (!supportsExposure) return; // A disabled or replaced search builtin cannot discover deferred tools. const hasToolSearch = hasBuiltinToolSearch(); const exposureChanged = deferOptionalTools !== hasToolSearch; deferOptionalTools = hasToolSearch; if (exposureChanged) registerTools(); - if (hasToolSearch) { - const active = pi.getActiveTools(); - if (!active.includes("tool_search")) pi.setActiveTools([...active, "tool_search"]); - } + const active = pi.getActiveTools(); + // Completion gateways can silently drop calls to undeclared tools named + // in history or guidance. Eagerly load T3 schemas, including after tree + // restores, while retaining deferred exposure for Pi's allowlist rules. + // Pi still applies --tools/--exclude-tools to this active loadout. + const required = [ + ...(hasToolSearch ? ["tool_search"] : []), + ...(ctx.model?.api === "openai-completions" ? catalog.map((tool) => "mcp__t3-code__" + tool.name) : []), + ]; + const missing = required.filter((name) => !active.includes(name)); + if (missing.length > 0) pi.setActiveTools([...active, ...missing]); }; // Tree navigation restores its saved loadout after session_start. - pi.on("session_tree", reconcileDiscovery); + pi.on("session_tree", (_event, ctx) => reconcileDiscovery(ctx)); + pi.on("model_select", (_event, ctx) => reconcileDiscovery(ctx)); pi.on("session_start", async (_event, ctx) => { - reconcileDiscovery(); + reconcileDiscovery(ctx); try { await ensureStarted(); } catch (error) { @@ -449,8 +458,9 @@ export default async function t3McpExtension(pi: ExtensionAPI) { // Deliver orchestration guidance through pi's real system-prompt channel. // Wrapping the first user message instead would stop it from starting // with "/" and silently break slash-command expansion. - pi.on("before_agent_start", (event) => ({ - systemPrompt: event.systemPrompt + "\\n\\n" + ORCHESTRATION_INSTRUCTIONS, - })); + pi.on("before_agent_start", (event, ctx) => { + reconcileDiscovery(ctx); + return { systemPrompt: event.systemPrompt + "\\n\\n" + ORCHESTRATION_INSTRUCTIONS }; + }); } `;