diff --git a/apps/mobile/src/features/threads/PendingApprovalCard.tsx b/apps/mobile/src/features/threads/PendingApprovalCard.tsx index caf9bd4cd0cc..377ae82aba8b 100644 --- a/apps/mobile/src/features/threads/PendingApprovalCard.tsx +++ b/apps/mobile/src/features/threads/PendingApprovalCard.tsx @@ -14,8 +14,10 @@ export interface PendingApprovalCardProps { } export function PendingApprovalCard(props: PendingApprovalCardProps) { + // Opaque for the same reason as PendingUserInputCard: nothing blurs the feed + // behind this card, so a translucent surface bleeds messages through it. return ( - + Approval needed diff --git a/apps/mobile/src/features/threads/PendingUserInputCard.tsx b/apps/mobile/src/features/threads/PendingUserInputCard.tsx index 4d2c2c84159b..c3c9b4e7ce83 100644 --- a/apps/mobile/src/features/threads/PendingUserInputCard.tsx +++ b/apps/mobile/src/features/threads/PendingUserInputCard.tsx @@ -24,8 +24,11 @@ export interface PendingUserInputCardProps { } export function PendingUserInputCard(props: PendingUserInputCardProps) { + // The surface is opaque on purpose: the card floats over the thread feed + // with no blur behind it, so a translucent background renders the questions + // on top of whatever message happens to sit underneath. return ( - + User input needed diff --git a/apps/server/src/cloud/bootService.test.ts b/apps/server/src/cloud/bootService.test.ts index a9ae3f49f70b..12eebbbd223f 100644 --- a/apps/server/src/cloud/bootService.test.ts +++ b/apps/server/src/cloud/bootService.test.ts @@ -31,7 +31,7 @@ it("keeps systemd pinned to the stable launcher rather than a versioned server", }); expect(unit).toContain("ExecStart=/usr/bin/node /home/theo/.t3/runtime/service-launcher.mjs"); - expect(unit).toContain("KillMode=control-group"); + expect(unit).toContain("KillMode=mixed"); expect(unit).not.toContain("versions/1.2.3"); }); diff --git a/apps/server/src/cloud/bootService.ts b/apps/server/src/cloud/bootService.ts index ec110cb8b6a8..a57585be4b24 100644 --- a/apps/server/src/cloud/bootService.ts +++ b/apps/server/src/cloud/bootService.ts @@ -64,7 +64,9 @@ export function renderBootServiceUnit(plan: BootServicePlan): string { `Environment=T3CODE_HOME=${quoteSystemdValue(plan.baseDir)}`, `Environment=${BOOT_SERVICE_UNIT_ENV}=${BOOT_SERVICE_UNIT_FILE}`, `ExecStart=${quoteSystemdValue(plan.nodePath)} ${quoteSystemdValue(plan.launcherPath)}`, - "KillMode=control-group", + // Let the launcher mark an explicit stop before it signals the server. + // systemd still SIGKILLs the whole cgroup if graceful shutdown times out. + "KillMode=mixed", "Restart=always", "RestartSec=5", `StandardOutput=append:${escapeSystemdSpecifiers(plan.logPath)}`, diff --git a/apps/server/src/cloud/http.test.ts b/apps/server/src/cloud/http.test.ts index c779c13aaf4d..0f24e6f34176 100644 --- a/apps/server/src/cloud/http.test.ts +++ b/apps/server/src/cloud/http.test.ts @@ -1,7 +1,10 @@ import * as NodeServices from "@effect/platform-node/NodeServices"; import { describe, expect, it } from "@effect/vitest"; import * as Effect from "effect/Effect"; +import * as FileSystem from "effect/FileSystem"; +import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; +import * as Path from "effect/Path"; import * as PlatformError from "effect/PlatformError"; import * as Tracer from "effect/Tracer"; import { @@ -15,6 +18,14 @@ import { EnvironmentId } from "@t3tools/contracts"; import { RelayClientTracer } from "@t3tools/shared/relayTracing"; import * as EnvironmentAuth from "../auth/EnvironmentAuth.ts"; import * as ServerSecretStore from "../auth/ServerSecretStore.ts"; +import * as ServerConfigModule from "../config.ts"; +import { writeServiceState } from "../serviceLauncher.ts"; +import { + SERVICE_LAUNCHER_PROTOCOL, + SERVICE_STATE_FILE, + SERVICE_STOP_MARKER_FILE, + type ServiceUpdateRecord, +} from "./serviceProtocol.ts"; import * as ServerEnvironment from "../environment/ServerEnvironment.ts"; import { CLOUD_CLI_DESIRED_LINK_SECRET } from "./CliState.ts"; import * as CliTokenManager from "./CliTokenManager.ts"; @@ -24,6 +35,7 @@ import { consumeCloudReplayGuards, isSupportedLinkProviderKind, linkProofScopes, + pendingServiceUpdateExists, reconcileDesiredCloudLink, releaseManagedTunnelOnShutdown, } from "./http.ts"; @@ -256,6 +268,23 @@ describe("releaseManagedTunnelOnShutdown", () => { readonly respond?: () => Response; } + // Writes the launcher's durable state file into this test's baseDir with + // the launcher's own writer; the release reads it to detect an in-flight + // update handoff. + const writeLauncherState = (update: ServiceUpdateRecord) => + Effect.gen(function* () { + const path = yield* Path.Path; + const config = yield* ServerConfigModule.ServerConfig; + const statePath = path.join(config.baseDir, "runtime", SERVICE_STATE_FILE); + yield* Effect.promise(() => + writeServiceState(statePath, { + protocol: SERVICE_LAUNCHER_PROTOCOL, + activeVersion: "0.0.30", + update, + }), + ); + }); + const provideReleaseHarness = (harness: ReleaseHarness) => (effect: Effect.Effect) => @@ -306,6 +335,14 @@ describe("releaseManagedTunnelOnShutdown", () => { }), ), ), + // The release consults the launcher state file under the configured + // baseDir, so every harness run gets a scoped temp baseDir. + Effect.provide( + ServerConfigModule.layerTest("/", { prefix: "t3-http-release-test-" }).pipe( + Layer.provideMerge(NodeServices.layer), + ), + ), + Effect.scoped, ); // The persisted state of a CLI-managed link whose tunnel is releasable. @@ -390,6 +427,84 @@ describe("releaseManagedTunnelOnShutdown", () => { }).pipe(provideReleaseHarness({ store, applyConfigCalls, requests })); }); + it.effect("keeps the tunnel when shutdown hands off to a pending update", () => { + const { store, values } = makeMemorySecretStore(managedLinkSecrets); + const applyConfigCalls: Array = []; + const requests: Array = []; + + return Effect.gen(function* () { + yield* writeLauncherState({ + id: "update-1", + fromVersion: "0.0.30", + targetVersion: "0.0.31", + dbPath: "/tmp/state.sqlite", + status: "pending", + }); + + const released = yield* releaseManagedTunnelOnShutdown(); + + // The launcher restarts a server immediately, so the tunnel is not + // orphaned; keeping it avoids the hostname route re-propagation that + // dominates update downtime. The stored config must survive so the + // next boot respawns the connector against the same tunnel. + expect(released).toBe(false); + expect(applyConfigCalls).toEqual([]); + expect(requests).toEqual([]); + expect(values.has(CLOUD_ENDPOINT_RUNTIME_CONFIG)).toBe(true); + }).pipe(provideReleaseHarness({ store, applyConfigCalls, requests })); + }); + + it.effect("still releases a pending update when the launcher is stopping", () => { + // `t3 service uninstall` or `systemctl stop` during the pending window: + // the launcher writes its stop marker before signalling the child, so no + // replacement server is coming and the tunnel must not be kept. + const { store, values } = makeMemorySecretStore(managedLinkSecrets); + const applyConfigCalls: Array = []; + const requests: Array = []; + + return Effect.gen(function* () { + yield* writeLauncherState({ + id: "update-1", + fromVersion: "0.0.30", + targetVersion: "0.0.31", + dbPath: "/tmp/state.sqlite", + status: "pending", + }); + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const config = yield* ServerConfigModule.ServerConfig; + yield* fs.writeFileString(path.join(config.baseDir, "runtime", SERVICE_STOP_MARKER_FILE), ""); + + expect(yield* pendingServiceUpdateExists).toBe(true); + const released = yield* releaseManagedTunnelOnShutdown(); + + expect(released).toBe(true); + expect(requests).toHaveLength(1); + expect(values.has(CLOUD_ENDPOINT_RUNTIME_CONFIG)).toBe(false); + }).pipe(provideReleaseHarness({ store, applyConfigCalls, requests })); + }); + + it.effect("still releases when the recorded update already settled", () => { + const { store, values } = makeMemorySecretStore(managedLinkSecrets); + const applyConfigCalls: Array = []; + const requests: Array = []; + + return Effect.gen(function* () { + yield* writeLauncherState({ + id: "update-1", + fromVersion: "0.0.30", + targetVersion: "0.0.31", + status: "committed", + }); + + const released = yield* releaseManagedTunnelOnShutdown(); + + expect(released).toBe(true); + expect(requests).toHaveLength(1); + expect(values.has(CLOUD_ENDPOINT_RUNTIME_CONFIG)).toBe(false); + }).pipe(provideReleaseHarness({ store, applyConfigCalls, requests })); + }); + it.effect("keeps a runtime config that a fast restart replaced mid-release", () => { const { store, values } = makeMemorySecretStore(managedLinkSecrets); const applyConfigCalls: Array = []; diff --git a/apps/server/src/cloud/http.ts b/apps/server/src/cloud/http.ts index eaf31a37f277..5c744bbfb9d8 100644 --- a/apps/server/src/cloud/http.ts +++ b/apps/server/src/cloud/http.ts @@ -46,7 +46,9 @@ import * as DateTime from "effect/DateTime"; import * as Crypto from "effect/Crypto"; import * as Duration from "effect/Duration"; import * as Effect from "effect/Effect"; +import * as FileSystem from "effect/FileSystem"; import * as Option from "effect/Option"; +import * as Path from "effect/Path"; import * as Schema from "effect/Schema"; import * as HttpEffect from "effect/unstable/http/HttpEffect"; import { HttpServerRequest, HttpServerResponse } from "effect/unstable/http"; @@ -56,8 +58,14 @@ import * as HttpApiBuilder from "effect/unstable/httpapi/HttpApiBuilder"; import * as EnvironmentAuth from "../auth/EnvironmentAuth.ts"; import * as ServerSecretStore from "../auth/ServerSecretStore.ts"; import { requireEnvironmentScope } from "../auth/http.ts"; +import * as ServerConfig from "../config.ts"; import * as ServerEnvironment from "../environment/ServerEnvironment.ts"; import * as ManagedEndpointRuntime from "./ManagedEndpointRuntime.ts"; +import { + SERVICE_STATE_FILE, + SERVICE_STOP_MARKER_FILE, + serviceStateHasPendingUpdate, +} from "./serviceProtocol.ts"; import { CLOUD_ENDPOINT_RUNTIME_CONFIG, CLOUD_LINKED_USER_ID, @@ -627,6 +635,37 @@ export const reconcileDesiredCloudLink = Effect.fn("environment.cloud.reconcileD }, ); +// The launcher owns this durable state, so read it directly both when a trial +// decides whether it owns pre-activation cleanup and while a server tears down. +export const pendingServiceUpdateExists = Effect.gen(function* () { + const config = yield* ServerConfig.ServerConfig; + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const runtimeDir = path.join(config.baseDir, "runtime"); + const stateText = yield* fs + .readFileString(path.join(runtimeDir, SERVICE_STATE_FILE)) + .pipe(Effect.option); + return Option.isSome(stateText) && serviceStateHasPendingUpdate(stateText.value); +}); + +// A pending update alone is not proof a replacement server is coming: an +// explicit launcher stop (`t3 service uninstall`, `systemctl stop`) during +// the pending window also tears this server down. The launcher marks that case +// just before it signals the child, so pending + no marker is the handoff. +const pendingUpdateHandoffExists = Effect.gen(function* () { + if (!(yield* pendingServiceUpdateExists)) { + return false; + } + const config = yield* ServerConfig.ServerConfig; + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const runtimeDir = path.join(config.baseDir, "runtime"); + const stopping = yield* fs + .exists(path.join(runtimeDir, SERVICE_STOP_MARKER_FILE)) + .pipe(Effect.orElseSucceed(() => false)); + return !stopping; +}); + // Cloudflare bills per provisioned tunnel, so an environment that goes offline // must not leave its tunnel behind. Releasing deletes only the tunnel — the // relay keeps the link and its hostname reservation, and the next startup's @@ -649,6 +688,19 @@ export const releaseManagedTunnelOnShutdown = Effect.fn( if (!(yield* readCliDesiredCloudLink) || (yield* readCliDesiredLinkMode) !== "managed") { return false; } + // A shutdown that hands off to a pending remote update is not the + // environment going offline: the launcher immediately brings a server back + // (the new version, or the old one after a rollback). Deleting the tunnel + // here forces that server to provision a replacement UUID, and the public + // hostname's route to the new tunnel takes 1-2 minutes to propagate — the + // dominant cost of an update restart. Keep the tunnel instead: the next + // boot respawns the connector from the stored config and is reachable as + // soon as it connects, and the reconcile confirms the still-live tunnel + // without replacing it. + if (yield* pendingUpdateHandoffExists) { + yield* Effect.logInfo("Keeping the managed tunnel across the update restart"); + return false; + } const token = yield* dependencies.cliTokenManager.getExisting; if (Option.isNone(token)) { return false; diff --git a/apps/server/src/cloud/serviceProtocol.ts b/apps/server/src/cloud/serviceProtocol.ts index ebc5d15d54f0..0faf88948377 100644 --- a/apps/server/src/cloud/serviceProtocol.ts +++ b/apps/server/src/cloud/serviceProtocol.ts @@ -5,6 +5,10 @@ export const SERVICE_LAUNCHER_PROTOCOL = 2 as const; export const SERVICE_LAUNCHER_CONTEXT_ENV = "T3_SERVICE_LAUNCHER_CONTEXT"; export const SERVICE_LAUNCHER_FILE = "service-launcher.mjs"; export const SERVICE_STATE_FILE = "service-state.json"; +/** Written by the launcher just before an explicit stop kills its child, so + the child can tell "the service is going away" from "the launcher is about + to start my replacement" while a pending update is recorded. */ +export const SERVICE_STOP_MARKER_FILE = ".service-stopping"; export interface PendingServiceUpdate { readonly id: string; diff --git a/apps/server/src/orchestration/ActivityPayloadProjection.test.ts b/apps/server/src/orchestration/ActivityPayloadProjection.test.ts index 7ea1e3ea0ed7..fc9ea4b62268 100644 --- a/apps/server/src/orchestration/ActivityPayloadProjection.test.ts +++ b/apps/server/src/orchestration/ActivityPayloadProjection.test.ts @@ -44,6 +44,60 @@ describe("projectActivityPayload agent-field survival", () => { expect(data.somethingClientNeverReads).toBeUndefined(); }); + it("slims Codex-shaped mcp_tool_call items to rendered fields plus a result summary", () => { + const projected = projectActivityPayload( + activity({ + itemType: "mcp_tool_call", + data: { + item: { + type: "mcpToolCall", + id: "item-1", + tool: "fetch_pr", + server: "github", + status: "completed", + arguments: { pr: 42 }, + durationMs: 1200, + result: { + content: [{ type: "text", text: `PR body line one\n${"x".repeat(5000)}` }], + structuredContent: { huge: "y".repeat(5000) }, + }, + _meta: { internal: true }, + }, + }, + }), + ); + const data = (projected.payload as Record).data as Record; + const item = data.item as Record; + expect(item.tool).toBe("fetch_pr"); + expect(item.server).toBe("github"); + expect(item.arguments).toEqual({ pr: 42 }); + expect(item._meta).toBeUndefined(); + expect(item.result).toEqual({ content: "PR body line one" }); + expect(JSON.stringify(projected.payload).length).toBeLessThan(500); + }); + + it("slims Claude-shaped mcp_tool_call data (toolName/input/result block)", () => { + const projected = projectActivityPayload( + activity({ + itemType: "mcp_tool_call", + data: { + toolName: "mcp__github__fetch_pr", + input: { pr: 42 }, + result: { + type: "tool_result", + tool_use_id: "toolu_1", + content: [{ type: "text", text: `first line of output\n${"z".repeat(5000)}` }], + }, + }, + }), + ); + const data = (projected.payload as Record).data as Record; + expect(data.toolName).toBe("mcp__github__fetch_pr"); + expect(data.input).toEqual({ pr: 42 }); + expect(data.result).toEqual({ content: "first line of output" }); + expect(JSON.stringify(projected.payload).length).toBeLessThan(500); + }); + it("passes task lifecycle payloads (no data field) through untouched", () => { const source = activity({ taskId: "task-9", diff --git a/apps/server/src/orchestration/ActivityPayloadProjection.ts b/apps/server/src/orchestration/ActivityPayloadProjection.ts index 67896961b387..f68a3ee96e9b 100644 --- a/apps/server/src/orchestration/ActivityPayloadProjection.ts +++ b/apps/server/src/orchestration/ActivityPayloadProjection.ts @@ -123,6 +123,114 @@ function summarizeToolTextOutput(value: string): string | null { return null; } +/** + * Fields of an MCP tool-call item both clients render in the expanded + * work-log row. Everything else — notably `result`, which carries the full + * tool output and dominates wire size on MCP-heavy threads — is summarized + * or dropped. Full payloads remain in persistence. + */ +const MCP_ITEM_KEPT_FIELDS = [ + "type", + "id", + "tool", + "server", + "status", + "arguments", + "appContext", + "error", + "durationMs", +] as const; + +/** + * Pulls renderable text out of an MCP tool result: either a Codex-style + * `{content: [{type: "text", text}, ...]}` record or a raw Claude + * `tool_result` block whose `content` is a string or block array. + */ +function extractMcpResultText(result: unknown): string | null { + const record = asRecord(result); + if (!record) { + return typeof result === "string" ? result : null; + } + if (typeof record.content === "string") { + return record.content; + } + if (Array.isArray(record.content)) { + const texts: string[] = []; + for (const entry of record.content) { + const text = asRecord(entry)?.text; + if (typeof text === "string" && text.trim().length > 0) { + texts.push(text); + } + } + if (texts.length > 0) { + return texts.join("\n"); + } + } + return null; +} + +function summarizeMcpResult(result: unknown): Record | undefined { + if (result === undefined || result === null) { + return undefined; + } + const text = extractMcpResultText(result); + const summary = text ? summarizeToolTextOutput(text) : null; + return summary ? { content: summary } : undefined; +} + +/** + * MCP tool calls carry full tool results (`data.item.result` on Codex, + * `data.result` on Claude/OpenCode) that used to bypass slimming entirely to + * keep the expanded-row UI working. Keep the fields the UI actually renders + * and summarize the result like regular tool output. + */ +function projectMcpToolCallData(data: Record): Record { + const projectedData: Record = {}; + + const item = asRecord(data.item); + if (item) { + const projectedItem: Record = {}; + for (const key of MCP_ITEM_KEPT_FIELDS) { + if (key in item) { + projectedItem[key] = item[key]; + } + } + const result = summarizeMcpResult(item.result); + if (result) { + projectedItem.result = result; + } + projectedData.item = projectedItem; + } + + if ("toolName" in data) { + projectedData.toolName = data.toolName; + } + if ("input" in data) { + projectedData.input = data.input; + } + if (!item) { + const result = summarizeMcpResult(data.result); + if (result) { + projectedData.result = result; + } + } + + if ("toolCallId" in data) { + projectedData.toolCallId = data.toolCallId; + } + if ("kind" in data) { + projectedData.kind = data.kind; + } + + const changedFiles: string[] = []; + collectChangedFiles(data, changedFiles, new Set(), 0); + if (changedFiles.length > 0) { + projectedData.files = changedFiles.map((path) => ({ path })); + } + + return projectedData; +} + function projectRawOutput(value: unknown): Record | undefined { const rawOutput = asRecord(value); if (!rawOutput) { @@ -160,10 +268,20 @@ export function projectActivityPayload( ): OrchestrationThreadActivity { const payload = asRecord(activity.payload); const data = asRecord(payload?.data); - if (!payload || !data || payload.itemType === "mcp_tool_call") { + if (!payload || !data) { return activity; } + if (payload.itemType === "mcp_tool_call") { + return { + ...activity, + payload: { + ...payload, + data: projectMcpToolCallData(data), + }, + }; + } + const projectedData: Record = {}; const item = projectCommandData(data); if (item) { @@ -247,6 +365,107 @@ function dropStaleContextWindowActivities( ); } +/** + * Identity both clients use to fold a tool lifecycle row into the call it + * belongs to (`deriveToolLifecycleCollapseKey` in web's `session-logic` and + * mobile's `threadActivity`): an explicit `data.toolCallId` when the adapter + * emits one, otherwise the itemType/title/detail triple. Returns null for rows + * with no identity at all — those never collapse on the client either, so they + * must not be dropped here. + */ +function toolLifecycleIdentity(activity: OrchestrationThreadActivity): string | null { + const payload = asRecord(activity.payload); + if (!payload) { + return null; + } + + const toolCallId = asTrimmedString(asRecord(payload.data)?.toolCallId); + if (toolCallId) { + return `id:${toolCallId}`; + } + + const itemType = asTrimmedString(payload.itemType) ?? ""; + // Mirrors the clients' `normalizeCompactToolLabel`: a completion's title may + // gain a trailing "complete"/"completed" the in-flight updates lack. + const label = (asTrimmedString(payload.title) ?? activity.summary) + .replace(/\s+(?:complete|completed)\s*$/iu, "") + .trim(); + const detail = asTrimmedString(payload.detail) ?? ""; + if (itemType.length === 0 && label.length === 0 && detail.length === 0) { + return null; + } + return [itemType, label, detail].join(""); +} + +/** + * Drops `tool.updated` rows a `tool.completed` row already supersedes. An + * update is the in-flight snapshot of a call; once the call completes, the + * completion carries the final state and the clients fold every matching + * update into it, so shipping the updates buys nothing — 47k such rows exist + * in one real database, and a single thread carries 2,291 of them totalling + * ~1MB post-slimming. + * + * Matching is per turn for the same reason `dropStaleContextWindowActivities` + * retains per turn: a live `thread.reverted` makes the client discard whole + * turns, so a completion in a different turn could vanish and leave the + * dropped update unrepresented. The completion must also come *after* the + * update within the turn — a later update belongs to a subsequent call that + * reuses the same identity and is still in flight. Rows without a lifecycle + * identity pass through, matching the clients, which never collapse them. + * Live `thread.activity-appended` events are untouched: updates still stream + * in real time and the completion supersedes them on the client as before. + * + * Deliberate divergence from client collapse: clients fold only *adjacent* + * lifecycle rows, so a superseded update separated from its completion by an + * interleaved parallel call renders as its own row today, and this drop + * removes it. Measured against a real database, that affects 1.5% of dropped + * rows (553 of 36,581), all pure in-flight state whose final result the + * retained completion still shows. Dropping them is intentional; matching + * adjacency server-side would forfeit most of the win for parallel-heavy + * threads, which are exactly the heavy ones. Superseding completions always + * carry a payload superset of their updates (verified across all 49,515 + * update rows: zero dropped rows held a client-merged field — detail, title, + * command, item, kind, files — their completion lacked), so no expanded-row + * content is lost. + */ +function dropSupersededToolUpdatedActivities( + activities: ReadonlyArray, +): ReadonlyArray { + const completionIndicesByKey = new Map(); + for (let index = 0; index < activities.length; index += 1) { + const activity = activities[index]!; + if (activity.kind !== "tool.completed") { + continue; + } + const identity = toolLifecycleIdentity(activity); + if (!identity) { + continue; + } + const key = `${activity.turnId ?? ""}${identity}`; + const indices = completionIndicesByKey.get(key); + if (indices) { + indices.push(index); + } else { + completionIndicesByKey.set(key, [index]); + } + } + if (completionIndicesByKey.size === 0) { + return activities; + } + + return activities.filter((activity, index) => { + if (activity.kind !== "tool.updated") { + return true; + } + const identity = toolLifecycleIdentity(activity); + if (!identity) { + return true; + } + const indices = completionIndicesByKey.get(`${activity.turnId ?? ""}${identity}`); + return !indices?.some((completionIndex) => completionIndex > index); + }); +} + export function projectThreadDetailSnapshot( snapshot: OrchestrationThreadDetailSnapshot, ): OrchestrationThreadDetailSnapshot { @@ -254,9 +473,9 @@ export function projectThreadDetailSnapshot( ...snapshot, thread: { ...snapshot.thread, - activities: dropStaleContextWindowActivities(snapshot.thread.activities).map( - projectActivityPayload, - ), + activities: dropSupersededToolUpdatedActivities( + dropStaleContextWindowActivities(snapshot.thread.activities), + ).map(projectActivityPayload), }, }; } diff --git a/apps/server/src/provider/acp/AcpCoreRuntimeEvents.test.ts b/apps/server/src/provider/acp/AcpCoreRuntimeEvents.test.ts index 7fe25699bbc3..394ada83f763 100644 --- a/apps/server/src/provider/acp/AcpCoreRuntimeEvents.test.ts +++ b/apps/server/src/provider/acp/AcpCoreRuntimeEvents.test.ts @@ -68,6 +68,46 @@ describe("AcpCoreRuntimeEvents", () => { }); }); + it("maps generic ACP permission kinds to dynamic tool approvals", () => { + const stamp = { eventId: "event-1" as never, createdAt: "2026-03-27T00:00:00.000Z" }; + + for (const kind of ["search", "fetch", "other", "unknown", "future-tool-kind"]) { + const permissionRequest = { kind }; + const request = { + stamp, + provider: ProviderDriverKind.make("cursor"), + threadId: "thread-1" as never, + turnId: TurnId.make("turn-1"), + requestId: RuntimeRequestId.make(`request-${kind}`), + permissionRequest, + }; + + expect( + makeAcpRequestOpenedEvent({ + ...request, + detail: kind, + args: {}, + source: "acp.jsonrpc", + method: "session/request_permission", + rawPayload: { sessionId: "session-1" }, + }), + ).toMatchObject({ + type: "request.opened", + payload: { requestType: "dynamic_tool_call" }, + }); + + expect( + makeAcpRequestResolvedEvent({ + ...request, + decision: "accept", + }), + ).toMatchObject({ + type: "request.resolved", + payload: { requestType: "dynamic_tool_call" }, + }); + } + }); + it("maps ACP core plan, tool-call, and content updates", () => { const stamp = { eventId: "event-1" as never, createdAt: "2026-03-27T00:00:00.000Z" }; const turnId = TurnId.make("turn-1"); diff --git a/apps/server/src/provider/acp/AcpCoreRuntimeEvents.ts b/apps/server/src/provider/acp/AcpCoreRuntimeEvents.ts index c93e61dc37b6..bd25e9815aef 100644 --- a/apps/server/src/provider/acp/AcpCoreRuntimeEvents.ts +++ b/apps/server/src/provider/acp/AcpCoreRuntimeEvents.ts @@ -26,7 +26,7 @@ interface AcpEventStamp { type AcpCanonicalRequestType = Extract< CanonicalRequestType, - "exec_command_approval" | "file_read_approval" | "file_change_approval" | "unknown" + "exec_command_approval" | "file_read_approval" | "file_change_approval" | "dynamic_tool_call" >; function canonicalRequestTypeFromAcpKind(kind: string | "unknown"): AcpCanonicalRequestType { @@ -40,7 +40,7 @@ function canonicalRequestTypeFromAcpKind(kind: string | "unknown"): AcpCanonical case "move": return "file_change_approval"; default: - return "unknown"; + return "dynamic_tool_call"; } } diff --git a/apps/server/src/server.ts b/apps/server/src/server.ts index 49a3a31940f7..1d824afbd1be 100644 --- a/apps/server/src/server.ts +++ b/apps/server/src/server.ts @@ -83,6 +83,7 @@ import * as ServerSecretStore from "./auth/ServerSecretStore.ts"; import * as EnvironmentAuth from "./auth/EnvironmentAuth.ts"; import { connectHttpApiLayer, + pendingServiceUpdateExists, reconcileDesiredCloudLink, releaseManagedTunnelOnShutdown, } from "./cloud/http.ts"; @@ -559,26 +560,34 @@ export const makeServerLayer = Layer.unwrap( yield* Deferred.succeed(cloudLinkParked, undefined).pipe(Effect.orDie); return; } + const releaseManagedTunnel = releaseManagedTunnelOnShutdown().pipe( + Effect.timeout("10 seconds"), + Effect.tap((released) => + released ? Effect.logInfo("Released the managed tunnel on shutdown") : Effect.void, + ), + Effect.catchCause((cause) => + Effect.logWarning( + "Failed to release the managed tunnel on shutdown; the next link reuses it", + { cause }, + ), + ), + Effect.asVoid, + ); + // A launcher trial can be stopped before activation. The previous + // server is already gone, so the trial owns cleanup immediately; the + // pending-state check keeps the tunnel for normal commit or rollback, + // while the launcher's explicit-stop marker allows it to be released. + // Other runtimes wait for activation so a failed standby cannot tear + // down the active runtime's tunnel. + const cleanupBeforeActivation = yield* pendingServiceUpdateExists; + if (cleanupBeforeActivation) { + yield* Effect.addFinalizer(() => releaseManagedTunnel); + } yield* forkParked( Effect.gen(function* () { - // Only an activated runtime owns the tunnel cleanup finalizer. - yield* Effect.addFinalizer(() => - releaseManagedTunnelOnShutdown().pipe( - Effect.timeout("10 seconds"), - Effect.tap((released) => - released - ? Effect.logInfo("Released the managed tunnel on shutdown") - : Effect.void, - ), - Effect.catchCause((cause) => - Effect.logWarning( - "Failed to release the managed tunnel on shutdown; the next link reuses it", - { cause }, - ), - ), - Effect.asVoid, - ), - ); + if (!cleanupBeforeActivation) { + yield* Effect.addFinalizer(() => releaseManagedTunnel); + } if (!(yield* CloudCliState.readCliDesiredCloudLink)) return; const server = yield* HttpServer.HttpServer; const address = server.address; diff --git a/apps/server/src/serviceLauncher.test.ts b/apps/server/src/serviceLauncher.test.ts index 4562c6a6de8e..5278a227b4a2 100644 --- a/apps/server/src/serviceLauncher.test.ts +++ b/apps/server/src/serviceLauncher.test.ts @@ -10,6 +10,7 @@ import { decodeServiceState, isExactServiceVersion, SERVICE_LAUNCHER_PROTOCOL, + SERVICE_STOP_MARKER_FILE, } from "./cloud/serviceProtocol.ts"; it("accepts only exact semantic versions", () => { @@ -99,6 +100,7 @@ it.layer(NodeServices.layer)("service state persistence", (it) => { const statePath = path.join(root, "runtime", "service-state.json"); const versionDir = path.join(root, "runtime", "versions", "1.0.0"); const entryPath = path.join(versionDir, "node_modules", "t3", "dist", "bin.mjs"); + const markerPath = path.join(root, "runtime", SERVICE_STOP_MARKER_FILE); yield* fs.makeDirectory(path.dirname(entryPath), { recursive: true }); yield* fs.writeFileString(entryPath, "setInterval(() => {}, 1_000);\n"); yield* fs.writeFileString(path.join(versionDir, ".install-complete"), "1.0.0\n"); @@ -111,8 +113,15 @@ it.layer(NodeServices.layer)("service state persistence", (it) => { const launcher = new Launcher(root, yield* Effect.promise(() => readServiceState(statePath))); const running = launcher.run(); - yield* Effect.promise(() => launcher.stop("SIGTERM")); + const stopping = launcher.stop("SIGTERM"); + // An explicit stop leaves the marker that tells a child shutting down + // mid-update that no replacement server is coming. stop() writes it + // synchronously; recover must not clear it while #stopRequested. + assert.isTrue(yield* fs.exists(markerPath)); + yield* Effect.promise(() => stopping); yield* Effect.promise(() => running); + // Prove recover did not wipe the marker after stop() raced ahead of it. + assert.isTrue(yield* fs.exists(markerPath)); }), ); diff --git a/apps/server/src/serviceLauncher.ts b/apps/server/src/serviceLauncher.ts index 211c0138bc4b..1869fbbe74c0 100644 --- a/apps/server/src/serviceLauncher.ts +++ b/apps/server/src/serviceLauncher.ts @@ -5,6 +5,7 @@ // `t3 service update`. Keep runtime imports limited to Node built-ins. import * as NodeChildProcess from "node:child_process"; import * as NodeCrypto from "node:crypto"; +import * as NodeFS from "node:fs"; import * as NodeFSP from "node:fs/promises"; import * as NodePath from "node:path"; @@ -24,6 +25,7 @@ import { SERVICE_LAUNCHER_CONTEXT_ENV, SERVICE_LAUNCHER_PROTOCOL, SERVICE_STATE_FILE, + SERVICE_STOP_MARKER_FILE, } from "./cloud/serviceProtocol.ts"; const HANDOFF_DELAY_MS = 2_000; @@ -257,6 +259,9 @@ async function terminateChild( } } +const stopMarkerPath = (baseDir: string) => + NodePath.join(baseDir, "runtime", SERVICE_STOP_MARKER_FILE); + export class Launcher { readonly #baseDir: string; readonly #statePath: string; @@ -264,6 +269,7 @@ export class Launcher { #child: ManagedChild | null = null; #timer: NodeJS.Timeout | undefined; #transitions: Promise = Promise.resolve(); + #stopRequested = false; #stopping = false; #done = false; readonly #completion = Promise.withResolvers(); @@ -308,13 +314,26 @@ export class Launcher { } async stop(signal: NodeJS.Signals): Promise { - if (this.#stopping) { + // This must happen synchronously at signal receipt. A queued update + // transition may already be terminating the active child, and that child + // needs to see the marker in its shutdown finalizer. KillMode=mixed also + // ensures systemd signals the launcher before the rest of the cgroup. + try { + NodeFS.writeFileSync(stopMarkerPath(this.#baseDir), "", { mode: 0o600 }); + } catch { + // Err toward keeping the tunnel; the next link or unlink reconciles it. + } + if (this.#stopRequested || this.#stopping) { await this.#completion.promise.catch(() => undefined); return; } - this.#stopping = true; + this.#stopRequested = true; this.#clearTimer(); this.#enqueue(async () => { + // Let an update transition already in progress start its replacement + // before this queued stop tears it down. That replacement owns the + // pre-activation tunnel cleanup path and observes the marker above. + this.#stopping = true; const child = this.#child?.process; this.#child = null; if (child !== undefined) await terminateChild(child, signal); @@ -330,6 +349,13 @@ export class Launcher { } async #recover(): Promise { + // A fresh launcher means servers are running again: any stop marker from + // a previous explicit stop is stale and must not make a future update + // handoff release its tunnel. Keep the marker when stop() already asked + // for shutdown so a child started by this recover still observes it. + if (!this.#stopRequested) { + await NodeFSP.rm(stopMarkerPath(this.#baseDir), { force: true }).catch(() => undefined); + } const update = this.#state.update; if (update?.status !== "pending") { if (update !== undefined) { diff --git a/apps/server/test/ActivityPayloadProjection.test.ts b/apps/server/test/ActivityPayloadProjection.test.ts index d6098937e7fb..49f1b532a53a 100644 --- a/apps/server/test/ActivityPayloadProjection.test.ts +++ b/apps/server/test/ActivityPayloadProjection.test.ts @@ -117,9 +117,9 @@ const fixtures = [ server: "repository", tool: "search", arguments: { query: "activity projection" }, - aggregatedOutput: "mcp payload remains available", + aggregatedOutput: "mcp bulk is dropped", }, - ignored: "MCP data is rendered verbatim", + ignored: "top-level bulk", }), makeActivity("search", "web_search", { rawOutput: { @@ -184,13 +184,37 @@ describe("projectActivityPayload", () => { }); }); - it("passes MCP tool data through unchanged", () => { - expect(projectActivityPayload(fixtures[4]!)).toBe(fixtures[4]); + it("slims MCP tool data to the fields the expanded row renders", () => { + expect(projectActivityPayload(fixtures[4]!).payload).toEqual({ + itemType: "mcp_tool_call", + title: "mcp_tool_call", + detail: "mcp_tool_call detail", + status: "completed", + requestKind: "command", + data: { + item: { + server: "repository", + tool: "search", + arguments: { query: "activity projection" }, + }, + }, + }); }); it("keeps current web and mobile derived output identical for every tool item type", () => { for (const activity of fixtures) { const projected = projectActivityPayload(activity); + if (activity === fixtures[4]) { + // MCP is the one deliberate difference: the expanded row's toolData + // loses result bulk but keeps the rendered identity fields. + const [entry] = deriveWorkLogEntries([projected]); + expect(entry?.toolData).toEqual({ + server: "repository", + tool: "search", + arguments: { query: "activity projection" }, + }); + continue; + } expect(deriveWorkLogEntries([projected])).toEqual(deriveWorkLogEntries([activity])); expect(comparableThreadFeed([projected])).toEqual(comparableThreadFeed([activity])); } @@ -233,6 +257,179 @@ describe("projectActivityPayload", () => { }); }); +describe("superseded tool.updated snapshot dedup", () => { + function makeToolLifecycleActivity( + id: string, + kind: "tool.updated" | "tool.completed", + options: { + readonly turn?: string; + readonly title?: string; + readonly detail?: string; + readonly toolCallId?: string; + } = {}, + ): OrchestrationThreadActivity { + const { turn = "turn-a", title = "File change", detail, toolCallId } = options; + return { + id: EventId.make(id), + tone: "tool", + kind, + summary: title, + payload: { + itemType: "file_change", + title, + ...(detail ? { detail } : {}), + data: { + ...(toolCallId ? { toolCallId } : {}), + toolName: "Edit", + input: { file_path: "src/app.ts" }, + }, + }, + turnId: TurnId.make(turn), + createdAt: "2026-07-27T00:00:00.000Z", + }; + } + + function projectedIds(activities: ReadonlyArray) { + return projectThreadDetailSnapshot({ + snapshotSequence: 7, + thread: makeThread(activities), + }).thread.activities.map((activity) => activity.id); + } + + it("drops updates a later completion supersedes in the same turn", () => { + const update1 = makeToolLifecycleActivity("upd-1", "tool.updated"); + const update2 = makeToolLifecycleActivity("upd-2", "tool.updated"); + const completed = makeToolLifecycleActivity("done-1", "tool.completed"); + + expect(projectedIds([update1, update2, completed])).toEqual([completed.id]); + }); + + it("matches on toolCallId when the adapter emits one", () => { + const otherCall = makeToolLifecycleActivity("upd-other", "tool.updated", { + toolCallId: "call-b", + }); + const update = makeToolLifecycleActivity("upd-a", "tool.updated", { toolCallId: "call-a" }); + const completed = makeToolLifecycleActivity("done-a", "tool.completed", { + toolCallId: "call-a", + }); + + // Same itemType/title, different call: only call-a's update is superseded. + expect(projectedIds([otherCall, update, completed])).toEqual([otherCall.id, completed.id]); + }); + + it("keeps updates with no matching completion", () => { + const inFlight = makeToolLifecycleActivity("upd-live", "tool.updated", { title: "Running" }); + const other = makeToolLifecycleActivity("upd-other", "tool.updated", { title: "Reading" }); + const completed = makeToolLifecycleActivity("done-other", "tool.completed", { + title: "Reading", + }); + + expect(projectedIds([inFlight, other, completed])).toEqual([inFlight.id, completed.id]); + }); + + it("drops interleaved superseded updates even when a parallel call separates them", () => { + // Deliberate divergence from the clients' adjacency-based collapse: a + // superseded update separated from its completion by an interleaved + // parallel call renders as its own in-flight row on full history, and the + // snapshot omits it. Its final state still shows via the retained + // completion (1.5% of dropped rows on real data; see the projection's doc + // comment). + const updateA = makeToolLifecycleActivity("upd-a", "tool.updated", { toolCallId: "call-a" }); + const updateB = makeToolLifecycleActivity("upd-b", "tool.updated", { toolCallId: "call-b" }); + const completedA = makeToolLifecycleActivity("done-a", "tool.completed", { + toolCallId: "call-a", + }); + const completedB = makeToolLifecycleActivity("done-b", "tool.completed", { + toolCallId: "call-b", + }); + + expect(projectedIds([updateA, updateB, completedA, completedB])).toEqual([ + completedA.id, + completedB.id, + ]); + }); + + it("keeps an update whose completion lives in another turn", () => { + // A live thread.reverted can discard the completing turn while keeping + // the updating one, which would leave the call unrepresented. + const update = makeToolLifecycleActivity("upd-kept", "tool.updated", { turn: "turn-kept" }); + const completed = makeToolLifecycleActivity("done-later", "tool.completed", { + turn: "turn-reverted", + }); + + expect(projectedIds([update, completed])).toEqual([update.id, completed.id]); + }); + + it("keeps an update that follows its completion", () => { + // A later update under the same identity is the next call, still in flight. + const completed = makeToolLifecycleActivity("done-first", "tool.completed"); + const nextCall = makeToolLifecycleActivity("upd-next", "tool.updated"); + + expect(projectedIds([completed, nextCall])).toEqual([completed.id, nextCall.id]); + }); + + it("keeps identity-less rows the clients never collapse", () => { + const anonymous: OrchestrationThreadActivity = { + id: EventId.make("upd-anon"), + tone: "tool", + kind: "tool.updated", + summary: " ", + payload: { data: { toolName: "Edit" } }, + turnId: TurnId.make("turn-a"), + createdAt: "2026-07-27T00:00:00.000Z", + }; + const completed: OrchestrationThreadActivity = { + ...anonymous, + id: EventId.make("done-anon"), + kind: "tool.completed", + }; + + expect(projectedIds([anonymous, completed])).toEqual([anonymous.id, completed.id]); + }); + + it("does not filter live activity-appended events", () => { + const update = makeToolLifecycleActivity("upd-live-event", "tool.updated"); + const event = { + sequence: 11, + eventId: EventId.make("event-tool-updated"), + aggregateKind: "thread", + aggregateId: ThreadId.make("thread-projection"), + occurredAt: "2026-07-27T00:00:03.000Z", + commandId: null, + causationEventId: null, + correlationId: null, + metadata: {}, + type: "thread.activity-appended", + payload: { + threadId: ThreadId.make("thread-projection"), + activity: update, + }, + } satisfies Extract; + + const projected = projectActivityEvent(event); + expect( + projected.type === "thread.activity-appended" ? projected.payload.activity.id : undefined, + ).toEqual(update.id); + }); + + it("leaves the collapsed work log identical to the full history", () => { + const activities = [ + makeToolLifecycleActivity("upd-1", "tool.updated", { detail: "writing" }), + makeToolLifecycleActivity("upd-2", "tool.updated", { detail: "writing" }), + makeToolLifecycleActivity("done-1", "tool.completed", { detail: "writing" }), + ]; + const projected = projectThreadDetailSnapshot({ + snapshotSequence: 7, + thread: makeThread(activities), + }); + + const before = deriveWorkLogEntries(activities); + const after = deriveWorkLogEntries(projected.thread.activities); + expect(after).toHaveLength(before.length); + expect(after.map((entry) => entry.label)).toEqual(before.map((entry) => entry.label)); + }); +}); + describe("context-window snapshot dedup", () => { function makeContextWindowActivity( id: string, @@ -328,12 +525,12 @@ describe("context-window snapshot dedup", () => { ); }); - it("leaves snapshots without context-window activities untouched", () => { + it("applies only payload slimming when there are no context-window activities", () => { const projected = projectThreadDetailSnapshot({ snapshotSequence: 7, thread: makeThread([fixtures[4]!]), }); - expect(projected.thread.activities).toEqual([fixtures[4]]); + expect(projected.thread.activities).toEqual([projectActivityPayload(fixtures[4]!)]); }); it("does not filter live activity-appended events", () => { diff --git a/apps/web/src/components/ChatView.tsx b/apps/web/src/components/ChatView.tsx index 22915489dc44..8a67d2595b77 100644 --- a/apps/web/src/components/ChatView.tsx +++ b/apps/web/src/components/ChatView.tsx @@ -26,7 +26,11 @@ import { connectionStatusTitle, type EnvironmentConnectionPresentation, } from "@t3tools/client-runtime/connection"; -import { effectiveSettled, effectiveSnoozed } from "@t3tools/client-runtime/state/thread-settled"; +import { + effectiveSettled, + effectiveSnoozed, + threadWokeAt, +} from "@t3tools/client-runtime/state/thread-settled"; import { parseScopedThreadKey, scopedThreadKey, @@ -117,6 +121,11 @@ import { useTurnDiffSummaries } from "../hooks/useTurnDiffSummaries"; import { isCommandPaletteOpen } from "../commandPaletteBus"; import { buildTemporaryWorktreeBranchName } from "@t3tools/shared/git"; import { useMediaQuery } from "../hooks/useMediaQuery"; +import { + clearPlanSidebarDismissal, + dismissPlanSidebarForTurn, + isPlanSidebarDismissedForTurn, +} from "../planSidebarDismissal"; import { RIGHT_PANEL_INLINE_LAYOUT_MEDIA_QUERY } from "../rightPanelLayout"; import { selectActiveRightPanel, @@ -155,7 +164,6 @@ import { CheckCircle2Icon, ChevronDownIcon, GitBranchIcon, - TriangleAlertIcon, WifiOffIcon, } from "lucide-react"; import { cn, randomHex } from "~/lib/utils"; @@ -1228,9 +1236,6 @@ function ChatViewContent(props: ChatViewProps) { ); const activeServerThread = serverThread ?? loadingServerThread; const markThreadVisited = useUiStateStore((store) => store.markThreadVisited); - const activeThreadLastVisitedAt = useUiStateStore( - (store) => store.threadLastVisitedAtById[routeThreadKey], - ); const settings = useEnvironmentSettings(environmentId); // New-thread defaults live in the primary environment's settings.json (the // settings UI never writes to remote environments), so read them from the @@ -1311,8 +1316,6 @@ function ChatViewContent(props: ChatViewProps) { const [pendingUserInputQuestionIndexByRequestId, setPendingUserInputQuestionIndexByRequestId] = useState>({}); const shouldUsePlanSidebarSheet = useMediaQuery(RIGHT_PANEL_INLINE_LAYOUT_MEDIA_QUERY); - // Tracks whether the user explicitly dismissed the sidebar for the active turn. - const planSidebarDismissedForTurnRef = useRef(null); // When set, the thread-change reset effect will open the sidebar instead of closing it. // Used by "Implement in a new thread" to carry the sidebar-open intent across navigation. const planSidebarOpenOnNextThreadRef = useRef(false); @@ -1849,25 +1852,6 @@ function ChatViewContent(props: ChatViewProps) { [openOrReuseProjectDraftThread], ); - useEffect(() => { - if (!serverThread?.id) return; - const threadUpdatedAt = Date.parse(serverThread.updatedAt); - if (Number.isNaN(threadUpdatedAt)) return; - const lastVisitedAt = activeThreadLastVisitedAt ? Date.parse(activeThreadLastVisitedAt) : NaN; - if (!Number.isNaN(lastVisitedAt) && lastVisitedAt >= threadUpdatedAt) return; - - markThreadVisited( - scopedThreadKey(scopeThreadRef(serverThread.environmentId, serverThread.id)), - serverThread.updatedAt, - ); - }, [ - activeThreadLastVisitedAt, - markThreadVisited, - serverThread?.environmentId, - serverThread?.id, - serverThread?.updatedAt, - ]); - const selectedProviderByThreadId = composerActiveProvider ?? null; const threadProvider = activeThread?.modelSelection.instanceId ?? @@ -1984,31 +1968,41 @@ function ChatViewContent(props: ChatViewProps) { const updateFailed = serverUpdateState.status === "failed"; items.push({ id: `server-version:${serverUpdateEnvironmentId}`, - variant: updateFailed ? "error" : updateInProgress ? "default" : "warning", - icon: updateInProgress ? ( -