diff --git a/.github/triage/PLAYBOOK.md b/.github/triage/PLAYBOOK.md index 39bf3ea01052..32def6bc0c16 100644 --- a/.github/triage/PLAYBOOK.md +++ b/.github/triage/PLAYBOOK.md @@ -59,8 +59,9 @@ different code depending on it: Then work from evidence, not assumption. In rough order of value: -- The server log and the trace file (`server.trace.ndjson`) around the time of the - problem. Recent failures usually leave a trail here. +- The trace file (`server.trace.ndjson`) around the time of the problem, plus the + service log or desktop backend logs from the context file if they exist. Recent + failures usually leave a trail here. - The provider event log, for problems with claude/codex/cursor sessions. - The SQLite database. Read it freely, but only write when a write is necessary to fix the problem the user described, and get their explicit permission diff --git a/apps/mobile/src/features/threads/ThreadDetailScreen.tsx b/apps/mobile/src/features/threads/ThreadDetailScreen.tsx index e32240748861..14beefd6d178 100644 --- a/apps/mobile/src/features/threads/ThreadDetailScreen.tsx +++ b/apps/mobile/src/features/threads/ThreadDetailScreen.tsx @@ -679,10 +679,12 @@ export const ThreadDetailScreen = memo(function ThreadDetailScreen(props: Thread const isSplitLayout = layoutVariant === "split"; const contentMaxWidth = isSplitLayout ? CHAT_CONTENT_MAX_WIDTH : undefined; const workspaceContentWidth = useWorkspaceContentWidth(); + // Clearing animated width can retain the unfolded width after Android resumes folded. + // Assign both layouts explicitly so the dock always follows its current parent. const composerWidthStyle = useAnimatedStyle(() => isSplitLayout && workspaceContentWidth !== null - ? { width: workspaceContentWidth.value, right: undefined } - : { width: undefined, right: 0 }, + ? { width: workspaceContentWidth.value } + : { width: "100%" }, ); const selectedInstanceId = props.selectedThread.modelSelection.instanceId; useStreamingHaptics(props.selectedThread.id, props.selectedThreadFeed); @@ -964,7 +966,7 @@ export const ThreadDetailScreen = memo(function ThreadDetailScreen(props: Thread {/* No paddingTop here: the overlay's measured height becomes the list's bottom inset, so any padding above the pill/composer diff --git a/apps/mobile/src/features/threads/ThreadFeed.tsx b/apps/mobile/src/features/threads/ThreadFeed.tsx index 20848cd2ad10..83b0bef9c022 100644 --- a/apps/mobile/src/features/threads/ThreadFeed.tsx +++ b/apps/mobile/src/features/threads/ThreadFeed.tsx @@ -2477,7 +2477,10 @@ export const ThreadFeed = memo(function ThreadFeed(props: ThreadFeedProps) { // content-inset override. Seed the fresh instance synchronously with the // current overlay height before the scroll integration's next reaction; // on Android the declarative contentInset floor covers this same window. - const listMountKey = `${feedThreadKey}:${presentedFeed.length === 0 ? "empty" : "filled"}`; + // The thinking row a running thread shows while its messages load is not + // content: the list must still remount, and so open at the end, when they + // arrive. + const listMountKey = `${feedThreadKey}:${presentedFeed.some((entry) => entry.type !== "thinking") ? "filled" : "empty"}`; useLayoutEffect(() => { const bottom = props.contentInsetEndAdjustment.value; if (bottom > 0) { diff --git a/apps/mobile/src/features/usage/UsageRouteScreen.tsx b/apps/mobile/src/features/usage/UsageRouteScreen.tsx index 2d5e5398bc51..10878179d2ac 100644 --- a/apps/mobile/src/features/usage/UsageRouteScreen.tsx +++ b/apps/mobile/src/features/usage/UsageRouteScreen.tsx @@ -101,8 +101,8 @@ export function UsageRouteScreen() { ); const isFocused = useIsFocused(); const limits = useRefreshLimits(selectedEnvironmentIds, isFocused && tab === "limits"); - const cursorAccessEnvironments = selectedEnvironments.filter((environment) => - environment.summary?.sources.some((source) => source.action === "enableCursorKeychain"), + const cursorAccessEnvironments = selectedEnvironments.filter( + (environment) => environment.needsCursorKeychainAccess, ); const refreshAfterCursorEnable = () => { void refresh(); diff --git a/apps/mobile/src/state/usage.ts b/apps/mobile/src/state/usage.ts index d49c26a40a44..c5895d353fef 100644 --- a/apps/mobile/src/state/usage.ts +++ b/apps/mobile/src/state/usage.ts @@ -16,7 +16,7 @@ import { type UsageSummary, type UsageSummaryInput, } from "@t3tools/contracts"; -import { refreshUsage } from "@t3tools/client-runtime/state/usage"; +import { needsCursorKeychainAccess, refreshUsage } from "@t3tools/client-runtime/state/usage"; import { mergeUsage, type EnvironmentUsage, type MergedUsage } from "@t3tools/shared/usageMerge"; import * as Option from "effect/Option"; import { AsyncResult, Atom } from "effect/unstable/reactivity"; @@ -33,6 +33,7 @@ export interface EnvironmentUsageStatus { readonly isConnected: boolean; readonly error: string | null; readonly summary: UsageSummary | null; + readonly needsCursorKeychainAccess: boolean; } /** @@ -50,13 +51,18 @@ const usageByWindowAtom = Atom.family((windowKey: string) => const statuses: EnvironmentUsageStatus[] = []; for (const [environmentId, presentation] of presentations) { const result = get(serverEnvironment.usageSummary({ environmentId, input })); + const summary = Option.getOrNull(AsyncResult.value(result)); statuses.push({ environmentId, label: presentation.entry.target.label, isPending: result.waiting, isConnected: presentation.connection.phase === "connected", error: result._tag === "Failure" ? "This environment could not report usage." : null, - summary: Option.getOrNull(AsyncResult.value(result)), + summary, + needsCursorKeychainAccess: needsCursorKeychainAccess( + summary, + get(serverEnvironment.providersValueAtom(environmentId)), + ), }); } return statuses; diff --git a/apps/server/package.json b/apps/server/package.json index afa2ca18fdfc..105e4d0d96c3 100644 --- a/apps/server/package.json +++ b/apps/server/package.json @@ -31,7 +31,7 @@ "@opencode-ai/sdk": "^1.3.15", "diff": "8.0.3", "effect": "catalog:", - "node-pty": "^1.1.0", + "node-pty": "^1.2.0-beta.15", "stream-chain": "^4.2.5", "stream-json": "3.6.0", "yaml": "catalog:", diff --git a/apps/server/src/auth/dpop.ts b/apps/server/src/auth/dpop.ts index a841f4fd444c..5e026f584273 100644 --- a/apps/server/src/auth/dpop.ts +++ b/apps/server/src/auth/dpop.ts @@ -18,6 +18,9 @@ import { } from "./EnvironmentAuth.ts"; import * as ServerSecretStore from "./ServerSecretStore.ts"; +/** Secret store name prefix of DPoP replay markers. The server prunes expired ones. */ +export const DPOP_REPLAY_MARKER_PREFIX = "dpop-proof-"; + export const mapDpopFailureReason = (code: DpopVerificationFailureCodeType): DpopFailureReason => { switch (code) { case "time_window": @@ -96,7 +99,7 @@ export const verifyRequestDpopProof = (input: { ); yield* secretStore .create( - `dpop-proof-${replayKey}`, + `${DPOP_REPLAY_MARKER_PREFIX}${replayKey}`, new TextEncoder().encode( [ `thumbprint=${result.thumbprint}`, diff --git a/apps/server/src/auth/replayMarkers.test.ts b/apps/server/src/auth/replayMarkers.test.ts new file mode 100644 index 000000000000..76c31b0162bd --- /dev/null +++ b/apps/server/src/auth/replayMarkers.test.ts @@ -0,0 +1,92 @@ +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { assert, it } from "@effect/vitest"; +import * as DateTime from "effect/DateTime"; +import * as Duration from "effect/Duration"; +import * as Effect from "effect/Effect"; +import * as FileSystem from "effect/FileSystem"; +import * as Layer from "effect/Layer"; +import * as Path from "effect/Path"; +import * as TestClock from "effect/testing/TestClock"; + +import * as ServerConfig from "../config.ts"; +import { pruneExpiredReplayMarkers, REPLAY_MARKER_MAX_AGE } from "./replayMarkers.ts"; +import * as ServerSecretStore from "./ServerSecretStore.ts"; + +// Every secret name the server stores today. The last three stand for names +// built from an id at runtime. +const REAL_SECRET_NAMES = [ + "server-signing-key", + "asset-access-signing-key", + "cloud-cli-oauth-token", + "cloud-cli-desired-link", + "cloud-link-ed25519-key-pair", + "cloud-link-ed25519-private-key", + "cloud-link-ed25519-public-key", + "cloud-mint-ed25519-public-key", + "cloud-endpoint-runtime-config", + "cloud-endpoint-confirmed-origin", + "cloud-linked-user-id", + "cloud-relay-url", + "cloud-relay-issuer", + "cloud-relay-environment-credential", + "cloud-publish-agent-activity", + "provider-env-Y29kZXg-T1BFTkFJX0FQSV9LRVk", + "provider-auth-0f1e2d3c4b5a69788796a5b4c3d2e1f00f1e2d3c4b5a69788796a5b4c3d2e1f0", + "usage-limit-source-aHVi", +]; + +it.layer(NodeServices.layer)("replayMarkers", (it) => { + it.effect("prunes only replay markers older than the max age", () => + Effect.gen(function* () { + const secretStore = yield* ServerSecretStore.ServerSecretStore; + const { secretsDir } = yield* ServerConfig.ServerConfig; + const fileSystem = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const now = DateTime.makeUnsafe("2026-01-01T00:00:00Z"); + const setAge = (fileName: string, age: Duration.Duration) => { + const mtime = DateTime.toDateUtc(DateTime.subtractDuration(now, age)); + return fileSystem.utimes(path.join(secretsDir, fileName), mtime, mtime); + }; + const writeAged = (name: string, age: Duration.Duration) => + secretStore + .create(name, Uint8Array.from([1])) + .pipe(Effect.andThen(setAge(`${name}.bin`, age))); + + const justExpired = Duration.sum(REPLAY_MARKER_MAX_AGE, Duration.seconds(1)); + const expiredMarkers = [ + "dpop-proof-old", + "cloud-mint-jti-old", + "cloud-mint-nonce-old", + "cloud-health-jti-old", + "cloud-health-nonce-old", + ]; + for (const name of expiredMarkers) yield* writeAged(name, justExpired); + yield* writeAged("dpop-proof-at-max-age", REPLAY_MARKER_MAX_AGE); + for (const name of REAL_SECRET_NAMES) yield* writeAged(name, Duration.days(30)); + const pendingSetFile = "dpop-proof-pending.bin.0000.tmp"; + yield* fileSystem.writeFile(path.join(secretsDir, pendingSetFile), Uint8Array.from([1])); + yield* setAge(pendingSetFile, Duration.days(30)); + yield* TestClock.setTime(DateTime.toEpochMillis(now)); + + yield* pruneExpiredReplayMarkers(); + + const remaining = yield* fileSystem.readDirectory(secretsDir); + assert.deepStrictEqual( + remaining.toSorted(), + [ + ...REAL_SECRET_NAMES.map((name) => `${name}.bin`), + "dpop-proof-at-max-age.bin", + pendingSetFile, + ].toSorted(), + ); + }).pipe( + Effect.provide( + ServerSecretStore.layer.pipe( + Layer.provideMerge( + ServerConfig.layerTest(process.cwd(), { prefix: "t3-replay-markers-test-" }), + ), + ), + ), + ), + ); +}); diff --git a/apps/server/src/auth/replayMarkers.ts b/apps/server/src/auth/replayMarkers.ts new file mode 100644 index 000000000000..67b56b722334 --- /dev/null +++ b/apps/server/src/auth/replayMarkers.ts @@ -0,0 +1,77 @@ +import * as Clock from "effect/Clock"; +import * as Duration from "effect/Duration"; +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 Schedule from "effect/Schedule"; + +import { CLOUD_REPLAY_MARKER_PREFIXES } from "../cloud/http.ts"; +import * as ServerConfig from "../config.ts"; +import { forkParked } from "../serverActivation.ts"; +import { DPOP_REPLAY_MARKER_PREFIX } from "./dpop.ts"; + +const REPLAY_MARKER_PREFIXES = [DPOP_REPLAY_MARKER_PREFIX, ...CLOUD_REPLAY_MARKER_PREFIXES]; + +/** + * How long a replay marker stays on disk. A marker only matters while its proof + * can pass the time check (about 5 minutes for DPoP, 7 for cloud proofs). After + * that, the time check rejects a replay by itself. The sweep and the time check + * both use the wall clock, so a pruned marker can let a replay through only if + * the clock moves back by almost a day, or if the filesystem stamps mtimes almost + * a day behind. Markers are files, so a restart does not reset them. + */ +export const REPLAY_MARKER_MAX_AGE = Duration.days(1); + +/** + * Deletes replay markers whose mtime is older than `REPLAY_MARKER_MAX_AGE`. + * `ServerSecretStore` saves each secret as `.bin`, so only + * `*.bin` names match. Other secrets and the `*.bin..tmp` + * files that `set` writes are never touched. + */ +export const pruneExpiredReplayMarkers = Effect.fn("replayMarkers.pruneExpired")(function* () { + const fileSystem = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const { secretsDir } = yield* ServerConfig.ServerConfig; + const cutoff = (yield* Clock.currentTimeMillis) - Duration.toMillis(REPLAY_MARKER_MAX_AGE); + const markers = (yield* fileSystem.readDirectory(secretsDir)).filter( + (name) => + name.endsWith(".bin") && REPLAY_MARKER_PREFIXES.some((prefix) => name.startsWith(prefix)), + ); + // `partition` visits every marker, so one locked file does not stop the sweep. + const [failures, removed] = yield* Effect.partition(markers, (name) => { + const markerPath = path.join(secretsDir, name); + return fileSystem.stat(markerPath).pipe( + Effect.flatMap((info) => + Option.exists(info.mtime, (mtime) => mtime.getTime() < cutoff) + ? fileSystem.remove(markerPath).pipe(Effect.as(true)) + : Effect.succeed(false), + ), + Effect.catchReason("PlatformError", "NotFound", () => Effect.succeed(false)), + ); + }); + yield* Effect.annotateCurrentSpan({ + "replay_markers.matched": markers.length, + "replay_markers.removed": removed.filter(Boolean).length, + "replay_markers.failed": failures.length, + }); + if (failures.length > 0) { + yield* Effect.logWarning("Failed to prune some replay markers", { + failed: failures.length, + cause: failures[0], + }); + } +}); + +/** Prunes expired replay markers after server activation, then every hour. */ +export const layer = Layer.effectDiscard( + forkParked( + pruneExpiredReplayMarkers().pipe( + Effect.catch((cause) => + Effect.logWarning("Failed to prune expired replay markers", { cause }), + ), + Effect.repeat(Schedule.spaced(Duration.hours(1))), + ), + ), +); diff --git a/apps/server/src/bin.ts b/apps/server/src/bin.ts index 0538b94fcde9..e30870aac873 100644 --- a/apps/server/src/bin.ts +++ b/apps/server/src/bin.ts @@ -24,6 +24,7 @@ import { serviceLauncherCommand } from "./cli/serviceLauncher.ts"; import { servicePreflightCommand } from "./cli/servicePreflight.ts"; import { sshHelperCommand } from "./cli/sshHelper.ts"; import { themeCommand } from "./cli/theme.ts"; +import { traceCommand } from "./cli/trace.ts"; import { triageCommand } from "./cli/triage.ts"; const CliRuntimeLayer = Layer.mergeAll(NodeServices.layer, NetService.layer); @@ -71,6 +72,7 @@ export const makeCli = ({ cloudEnabled = hasCloudPublicConfig } = {}) => servicePreflightCommand, sshHelperCommand, themeCommand, + traceCommand, triageCommand, cloudEnabled ? connectCommand : connectUnavailableCommand, ]), diff --git a/apps/server/src/checkpointing/CheckpointDiffQuery.test.ts b/apps/server/src/checkpointing/CheckpointDiffQuery.test.ts index 7fa6065109c3..6df7d7534130 100644 --- a/apps/server/src/checkpointing/CheckpointDiffQuery.test.ts +++ b/apps/server/src/checkpointing/CheckpointDiffQuery.test.ts @@ -84,6 +84,7 @@ describe("CheckpointDiffQuery.layer", () => { getShellSnapshot: () => Effect.die("CheckpointDiffQuery should not request the orchestration shell snapshot"), getDeletedWorktreeThreads: () => Effect.die("unused"), + listThreadsWithPullRequests: () => Effect.die("unused"), getArchivedShellSnapshot: () => Effect.die("CheckpointDiffQuery should not request archived shell snapshots"), getSnapshotSequence: () => Effect.succeed({ snapshotSequence: 0 }), @@ -201,6 +202,7 @@ describe("CheckpointDiffQuery.layer", () => { getShellSnapshot: () => Effect.die("CheckpointDiffQuery should not request the orchestration shell snapshot"), getDeletedWorktreeThreads: () => Effect.die("unused"), + listThreadsWithPullRequests: () => Effect.die("unused"), getArchivedShellSnapshot: () => Effect.die("CheckpointDiffQuery should not request archived shell snapshots"), getSnapshotSequence: () => Effect.succeed({ snapshotSequence: 0 }), @@ -293,6 +295,7 @@ describe("CheckpointDiffQuery.layer", () => { getShellSnapshot: () => Effect.die("CheckpointDiffQuery should not request the orchestration shell snapshot"), getDeletedWorktreeThreads: () => Effect.die("unused"), + listThreadsWithPullRequests: () => Effect.die("unused"), getArchivedShellSnapshot: () => Effect.die("CheckpointDiffQuery should not request archived shell snapshots"), getSnapshotSequence: () => Effect.succeed({ snapshotSequence: 0 }), @@ -370,6 +373,7 @@ describe("CheckpointDiffQuery.layer", () => { getShellSnapshot: () => Effect.die("CheckpointDiffQuery should not request the orchestration shell snapshot"), getDeletedWorktreeThreads: () => Effect.die("unused"), + listThreadsWithPullRequests: () => Effect.die("unused"), getArchivedShellSnapshot: () => Effect.die("CheckpointDiffQuery should not request archived shell snapshots"), getSnapshotSequence: () => Effect.succeed({ snapshotSequence: 0 }), @@ -432,6 +436,7 @@ describe("CheckpointDiffQuery.layer", () => { getShellSnapshot: () => Effect.die("CheckpointDiffQuery should not request the orchestration shell snapshot"), getDeletedWorktreeThreads: () => Effect.die("unused"), + listThreadsWithPullRequests: () => Effect.die("unused"), getArchivedShellSnapshot: () => Effect.die("CheckpointDiffQuery should not request archived shell snapshots"), getSnapshotSequence: () => Effect.succeed({ snapshotSequence: 0 }), diff --git a/apps/server/src/cli/config.test.ts b/apps/server/src/cli/config.test.ts index ef7cdd578899..42932b927008 100644 --- a/apps/server/src/cli/config.test.ts +++ b/apps/server/src/cli/config.test.ts @@ -508,7 +508,6 @@ it.layer(NodeServices.layer)("cli config resolution", (it) => { resolved.terminalLogsDir, resolved.attachmentsDir, resolved.worktreesDir, - path.dirname(resolved.serverLogPath), path.dirname(resolved.serverTracePath), ]) { expect(yield* fs.exists(directory)).toBe(true); diff --git a/apps/server/src/cli/config.ts b/apps/server/src/cli/config.ts index d724730c953f..62461e286748 100644 --- a/apps/server/src/cli/config.ts +++ b/apps/server/src/cli/config.ts @@ -82,16 +82,22 @@ const tailscaleServePortFlag = Flag.Int("tailscale-serve-port").pipe( Flag.optional, ); +// Trace file location, shared by the server and `t3 trace summary`. +export const traceFileConfig = Config.String("T3CODE_TRACE_FILE").pipe( + Config.option, + Config.map(Option.getOrUndefined), +); +export const traceMaxFilesConfig = Config.Int("T3CODE_TRACE_MAX_FILES").pipe( + Config.withDefault(10), +); + const EnvServerConfig = Config.all({ logLevel: Config.LogLevel("T3CODE_LOG_LEVEL").pipe(Config.withDefault("Info")), traceMinLevel: Config.LogLevel("T3CODE_TRACE_MIN_LEVEL").pipe(Config.withDefault("Info")), traceTimingEnabled: Config.Boolean("T3CODE_TRACE_TIMING_ENABLED").pipe(Config.withDefault(true)), - traceFile: Config.String("T3CODE_TRACE_FILE").pipe( - Config.option, - Config.map(Option.getOrUndefined), - ), + traceFile: traceFileConfig, traceMaxBytes: Config.Int("T3CODE_TRACE_MAX_BYTES").pipe(Config.withDefault(10 * 1024 * 1024)), - traceMaxFiles: Config.Int("T3CODE_TRACE_MAX_FILES").pipe(Config.withDefault(10)), + traceMaxFiles: traceMaxFilesConfig, traceBatchWindowMs: Config.Int("T3CODE_TRACE_BATCH_WINDOW_MS").pipe(Config.withDefault(1_000)), otlpTracesUrl: Config.String("T3CODE_OTLP_TRACES_URL").pipe( Config.option, diff --git a/apps/server/src/cli/trace.test.ts b/apps/server/src/cli/trace.test.ts new file mode 100644 index 000000000000..b81a3ed53f25 --- /dev/null +++ b/apps/server/src/cli/trace.test.ts @@ -0,0 +1,100 @@ +import { assert, it } from "@effect/vitest"; + +import { makeTraceSpanSummary } from "./trace.ts"; + +const MINUTE_MS = 60_000; + +function span(name: string, durationMs: number, endMs: number, exitTag = "Success") { + return JSON.stringify({ + type: "effect-span", + name, + traceId: "trace", + spanId: "span", + durationMs, + endTimeUnixNano: String(BigInt(endMs) * 1_000_000n), + exit: { _tag: exitTag, cause: "cause" }, + }); +} + +function browserSpan(name: string, status: { code: string; message?: string }) { + return JSON.stringify({ + type: "otlp-span", + name, + durationMs: 1, + endTimeUnixNano: "1000000", + status, + }); +} + +it("reports count, rate, percentiles, and exits per span name", () => { + // Ten `refresh` spans of 1..10 ms end over minutes 0..9, and one `probe` + // span ends at minute 10, so the recorded window is 10 minutes. + const refreshes = Array.from({ length: 10 }, (_, index) => + span( + "refresh", + index + 1, + index * MINUTE_MS, + index === 0 ? "Interrupted" : index === 1 ? "Failure" : "Success", + ), + ); + const summarizer = makeTraceSpanSummary(); + [...refreshes, span("probe", 2_500, 10 * MINUTE_MS)].forEach(summarizer.addLine); + const summary = summarizer.finish(); + + assert.strictEqual(summary.spanCount, 11); + assert.strictEqual(summary.minutes, 10); + assert.deepStrictEqual(summary.spans, [ + { + name: "refresh", + count: 10, + perMinute: 1, + p50Ms: 5, + p90Ms: 9, + maxMs: 10, + interrupted: 1, + failures: 1, + }, + { + name: "probe", + count: 1, + perMinute: 0.1, + p50Ms: 2_500, + p90Ms: 2_500, + maxMs: 2_500, + interrupted: 0, + failures: 0, + }, + ]); +}); + +it("drops spans that ended before the window and counts unreadable lines", () => { + const summarizer = makeTraceSpanSummary(MINUTE_MS); + [ + span("old", 1, 0), + "", + "{not json", + JSON.stringify({ name: "no-duration" }), + // Ends past the largest Date, so the report could not print it. + JSON.stringify({ name: "far-future", durationMs: 1, endTimeUnixNano: "9".repeat(22) }), + span("recent", 4, 5 * MINUTE_MS), + ].forEach(summarizer.addLine); + const summary = summarizer.finish(); + + assert.strictEqual(summary.skippedLineCount, 3); + assert.deepStrictEqual( + summary.spans.map((entry) => [entry.name, entry.count, entry.perMinute]), + [["recent", 1, undefined]], + ); +}); + +it("reads failures and interrupts of browser spans from their OTLP status", () => { + const summarizer = makeTraceSpanSummary(); + [ + browserSpan("render", { code: "2", message: "boom" }), + browserSpan("render", { code: "1", message: "Interrupted" }), + browserSpan("render", { code: "1" }), + ].forEach(summarizer.addLine); + + const [render] = summarizer.finish().spans; + assert.deepStrictEqual([render?.count, render?.interrupted, render?.failures], [3, 1, 1]); +}); diff --git a/apps/server/src/cli/trace.ts b/apps/server/src/cli/trace.ts new file mode 100644 index 000000000000..7b8a83592ef8 --- /dev/null +++ b/apps/server/src/cli/trace.ts @@ -0,0 +1,227 @@ +/** + * `t3 trace summary` - per-span counts, rates, and latency percentiles from + * the local server trace file and its rotated backups. It reads the files + * directly, so it works while the server is stalled or stopped. + */ +import { PositiveInt } from "@t3tools/contracts"; +import * as Clock from "effect/Clock"; +import * as Config from "effect/Config"; +import * as Console from "effect/Console"; +import * as DateTime from "effect/DateTime"; +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 Schema from "effect/Schema"; +import * as Stream from "effect/Stream"; +import { Command, Flag } from "effect/unstable/cli"; + +import * as ServerConfig from "../config.ts"; +import { toRotatedTracePaths, TraceFileReadError } from "../diagnostics/TraceDiagnostics.ts"; +import { resolveBaseDir } from "../os-jank.ts"; +import { baseDirFlag, DurationFromString, traceFileConfig, traceMaxFilesConfig } from "./config.ts"; + +// Only the fields the summary needs. Other record fields are ignored. +const decodeTraceSpanLine = Schema.decodeUnknownOption( + Schema.fromJsonString( + Schema.Struct({ + name: Schema.String, + durationMs: Schema.Finite, + endTimeUnixNano: Schema.FiniteFromString, + // Server (`effect-span`) records. + exit: Schema.optional(Schema.Struct({ _tag: Schema.String })), + // Browser (`otlp-span`) records. Effect's OTLP tracer writes code "2" for + // errors and code "1" with message "Interrupted" for interrupts. + status: Schema.optional( + Schema.Struct({ + code: Schema.optional(Schema.String), + message: Schema.optional(Schema.String), + }), + ), + }), + ), +); + +/** + * Groups trace NDJSON by span name. Call `addLine` once per line as the files + * stream in, then `finish` for the summary. Spans that ended before `sinceMs` + * are left out. Rates are per minute between the first and last span end, + * since spans are written when they end. + */ +export function makeTraceSpanSummary(sinceMs = -Infinity) { + // Keep each span's duration (8 bytes, a few MB for the default 110 MB of + // rotated traces) for exact percentiles. A bounded sketch would save little + // and make p50 and p90 approximate. + const byName = new Map(); + let spanCount = 0; + let skippedLineCount = 0; + let firstEndMs = Infinity; + let lastEndMs = -Infinity; + + const addLine = (line: string) => { + if (line.trim().length === 0) return; + const span = Option.getOrUndefined(decodeTraceSpanLine(line)); + if (span === undefined) { + skippedLineCount += 1; + return; + } + const endMs = span.endTimeUnixNano / 1_000_000; + // The report prints end times as dates, so skip ones outside the Date range. + if (Option.isNone(DateTime.make(endMs))) { + skippedLineCount += 1; + return; + } + if (endMs < sinceMs) return; + + spanCount += 1; + firstEndMs = Math.min(firstEndMs, endMs); + lastEndMs = Math.max(lastEndMs, endMs); + const stats = byName.get(span.name) ?? { durations: [], interrupted: 0, failures: 0 }; + stats.durations.push(span.durationMs); + if (span.exit?._tag === "Interrupted" || span.status?.message === "Interrupted") { + stats.interrupted += 1; + } + if (span.exit?._tag === "Failure" || span.status?.code === "2") stats.failures += 1; + byName.set(span.name, stats); + }; + + const finish = () => { + const minutes = (lastEndMs - firstEndMs) / 60_000; + const spans = [...byName] + .map(([name, { durations, interrupted, failures }]) => { + const sorted = durations.toSorted((left, right) => left - right); + // Nearest-rank percentile. + const percentile = (p: number) => sorted[Math.ceil(p * sorted.length) - 1]!; + return { + name, + count: sorted.length, + perMinute: minutes > 0 ? sorted.length / minutes : undefined, + p50Ms: percentile(0.5), + p90Ms: percentile(0.9), + maxMs: sorted[sorted.length - 1]!, + interrupted, + failures, + }; + }) + .toSorted((left, right) => right.count - left.count || left.name.localeCompare(right.name)); + + return { spanCount, skippedLineCount, firstEndMs, lastEndMs, minutes, spans }; + }; + + return { addLine, finish }; +} + +const formatMs = (ms: number) => + ms < 1_000 ? `${Math.round(ms)}ms` : `${(ms / 1_000).toFixed(1)}s`; + +function formatTraceSummary( + summary: ReturnType["finish"]>, + limit: number, +) { + const header = ["span", "count", "/min", "p50", "p90", "max", "interrupted", "failed"]; + const rows = summary.spans + .slice(0, limit) + .map((span) => [ + span.name, + String(span.count), + span.perMinute === undefined + ? "-" + : span.perMinute < 0.1 + ? "<0.1" + : span.perMinute.toFixed(1), + formatMs(span.p50Ms), + formatMs(span.p90Ms), + formatMs(span.maxMs), + String(span.interrupted), + String(span.failures), + ]); + const table = [header, ...rows]; + const widths = header.map((_, column) => Math.max(...table.map((row) => row[column]!.length))); + const formatIso = (ms: number) => DateTime.formatIso(DateTime.makeUnsafe(ms)); + return [ + `${summary.spanCount} spans ended from ${formatIso(summary.firstEndMs)} to ${formatIso(summary.lastEndMs)} (${summary.minutes.toFixed(1)} min).`, + ...(summary.skippedLineCount > 0 + ? [`Skipped ${summary.skippedLineCount} lines that are not spans.`] + : []), + "", + ...table.map((row) => + row + .map((cell, column) => + column === 0 ? cell.padEnd(widths[column]!) : cell.padStart(widths[column]!), + ) + .join(" "), + ), + ...(summary.spans.length > limit + ? ["", `${summary.spans.length - limit} more span names. Use --limit to show more.`] + : []), + ].join("\n"); +} + +const traceSummaryCommand = Command.make("summary", { + baseDir: baseDirFlag, + since: Flag.String("since").pipe( + Flag.withSchema(DurationFromString), + Flag.withDescription("Only count spans that ended in this window, for example 30m or 2h."), + Flag.optional, + ), + limit: Flag.Int("limit").pipe( + Flag.withSchema(PositiveInt), + Flag.withDescription("Number of span names to show, busiest first."), + Flag.withDefault(25), + ), +}).pipe( + Command.withDescription("Summarize the local server trace file: counts, rates, and latency."), + Command.withHandler( + Effect.fn("cli.trace.summary")(function* (flags) { + const fs = yield* FileSystem.FileSystem; + // T3CODE_TRACE_FILE, else the userdata trace file for --base-dir or + // T3CODE_HOME. Implicit dev runs write elsewhere; set T3CODE_TRACE_FILE. + const envHome = yield* Config.String("T3CODE_HOME").pipe(Config.option); + const baseDir = yield* resolveBaseDir( + Option.getOrUndefined(Option.orElse(flags.baseDir, () => envHome)), + ); + const traceFilePath = + (yield* traceFileConfig) ?? + (yield* ServerConfig.deriveServerPaths(baseDir, undefined)).serverTracePath; + const sinceMs = Option.isSome(flags.since) + ? (yield* Clock.currentTimeMillis) - Duration.toMillis(flags.since.value) + : undefined; + const summarizer = makeTraceSpanSummary(sinceMs); + // Stream each file so only one chunk of text is in memory at a time. + yield* Effect.forEach( + toRotatedTracePaths(traceFilePath, yield* traceMaxFilesConfig), + (path) => + fs.stream(path).pipe( + Stream.decodeText, + Stream.splitLines, + Stream.runForEachArray((lines) => Effect.sync(() => lines.forEach(summarizer.addLine))), + Effect.catchTags({ + PlatformError: (cause) => + cause.reason._tag === "NotFound" + ? Effect.void + : Effect.fail( + new TraceFileReadError({ + traceFilePath: path, + causeTag: cause.reason._tag, + cause, + }), + ), + }), + ), + { discard: true }, + ); + const summary = summarizer.finish(); + + yield* Console.log( + summary.spanCount === 0 + ? `No spans found in ${traceFilePath} or its rotated files${sinceMs === undefined ? "" : " in that window"}.${summary.skippedLineCount > 0 ? ` Skipped ${summary.skippedLineCount} lines that are not spans.` : ""}` + : formatTraceSummary(summary, flags.limit), + ); + }), + ), +); + +export const traceCommand = Command.make("trace").pipe( + Command.withDescription("Inspect the local server trace file."), + Command.withSubcommands([traceSummaryCommand]), +); diff --git a/apps/server/src/cli/triage.ts b/apps/server/src/cli/triage.ts index c621d331c404..b408e3550112 100644 --- a/apps/server/src/cli/triage.ts +++ b/apps/server/src/cli/triage.ts @@ -27,6 +27,7 @@ import * as Schema from "effect/Schema"; import { Command, Flag } from "effect/unstable/cli"; import packageJson from "../../package.json" with { type: "json" }; +import * as BootService from "../cloud/bootService.ts"; import * as ServerConfig from "../config.ts"; import { resolveBaseDir } from "../os-jank.ts"; import { isProcessAlive, readPersistedServerRuntimeState } from "../serverRuntimeState.ts"; @@ -201,7 +202,11 @@ export const triageCommand = Command.make("triage", { dbPath: paths.dbPath, settingsPath: paths.settingsPath, logsDir: paths.logsDir, - serverLogPath: paths.serverLogPath, + // The server writes no log file of its own. Service installs and the + // desktop app capture its output. The glob covers every desktop backend + // (such as WSL) and rotated copies; names come from DesktopObservability.ts. + serviceLogPath: path.join(paths.logsDir, BootService.BOOT_SERVICE_LOG_FILE), + desktopBackendLogGlob: path.join(paths.logsDir, "server-child*.log*"), serverTracePath: paths.serverTracePath, providerEventLogPath: paths.providerEventLogPath, terminalLogsDir: paths.terminalLogsDir, diff --git a/apps/server/src/cli/triagePrompt.test.ts b/apps/server/src/cli/triagePrompt.test.ts index bf1ac5dbbe5e..fe65bf0dc444 100644 --- a/apps/server/src/cli/triagePrompt.test.ts +++ b/apps/server/src/cli/triagePrompt.test.ts @@ -52,7 +52,8 @@ it("context file carries every path the playbook depends on", () => { dbPath: "/home/u/.t3/userdata/state.sqlite", settingsPath: "/home/u/.t3/userdata/settings.json", logsDir: "/home/u/.t3/userdata/logs", - serverLogPath: "/home/u/.t3/userdata/logs/server.log", + serviceLogPath: "/home/u/.t3/userdata/logs/boot-service.log", + desktopBackendLogGlob: "/home/u/.t3/userdata/logs/server-child*.log*", serverTracePath: "/home/u/.t3/userdata/logs/server.trace.ndjson", providerEventLogPath: "/home/u/.t3/userdata/logs/provider/events.log", terminalLogsDir: "/home/u/.t3/userdata/logs/terminals", @@ -63,6 +64,8 @@ it("context file carries every path the playbook depends on", () => { }); assert.include(context, "/home/u/.t3/userdata/state.sqlite"); assert.include(context, "/home/u/.t3/userdata/logs/server.trace.ndjson"); + assert.include(context, "/home/u/.t3/userdata/logs/boot-service.log"); + assert.include(context, "/home/u/.t3/userdata/logs/server-child*.log*"); assert.include(context, "/home/u/.t3/userdata/logs/provider/events.log"); assert.include(context, "/home/u/.t3/userdata/secrets"); assert.include(context, "/home/u/.t3/source"); diff --git a/apps/server/src/cli/triagePrompt.ts b/apps/server/src/cli/triagePrompt.ts index c2b93a1840a1..1df712854263 100644 --- a/apps/server/src/cli/triagePrompt.ts +++ b/apps/server/src/cli/triagePrompt.ts @@ -71,8 +71,9 @@ different code depending on it: Then work from evidence, not assumption. In rough order of value: -- The server log and the trace file (\`server.trace.ndjson\`) around the time of the - problem. Recent failures usually leave a trail here. +- The trace file (\`server.trace.ndjson\`) around the time of the problem, plus the + service log or desktop backend logs from the context file if they exist. Recent + failures usually leave a trail here. - The provider event log, for problems with claude/codex/cursor sessions. - The SQLite database. Read it freely, but only write when a write is necessary to fix the problem the user described, and get their explicit permission @@ -176,7 +177,8 @@ export interface TriageContextInput { readonly dbPath: string; readonly settingsPath: string; readonly logsDir: string; - readonly serverLogPath: string; + readonly serviceLogPath: string; + readonly desktopBackendLogGlob: string; readonly serverTracePath: string; readonly providerEventLogPath: string; readonly terminalLogsDir: string; @@ -205,7 +207,8 @@ Generated by \`t3 triage\` at ${input.generatedAt}. - Database (SQLite; write only with the user's explicit permission): ${input.paths.dbPath} - Settings: ${input.paths.settingsPath} - Logs dir: ${input.paths.logsDir} -- Server log: ${input.paths.serverLogPath} +- Service log (systemd/launchd service installs only): ${input.paths.serviceLogPath} +- Desktop backend logs (glob; one file per backend plus rotated copies; written only when a backend crashes or fails to start): ${input.paths.desktopBackendLogGlob} - Server trace (ndjson): ${input.paths.serverTracePath} - Provider event log: ${input.paths.providerEventLogPath} - Terminal logs: ${input.paths.terminalLogsDir} diff --git a/apps/server/src/cloud/bootService.ts b/apps/server/src/cloud/bootService.ts index 0a57ed822af4..fc8fc1549a6b 100644 --- a/apps/server/src/cloud/bootService.ts +++ b/apps/server/src/cloud/bootService.ts @@ -43,6 +43,8 @@ const BOOT_SERVICE_UNIT_FILE = `${BOOT_SERVICE_NAME}.service`; const BOOT_SERVICE_LAUNCHD_LABEL = "com.t3tools.t3code.service"; const BOOT_SERVICE_PLIST_FILE = `${BOOT_SERVICE_LAUNCHD_LABEL}.plist`; const BOOT_SERVICE_UNIT_ENV = "T3_BOOT_SERVICE_UNIT"; +/** File in the logs dir that receives the service's stdout and stderr. `t3 triage` points agents at it. */ +export const BOOT_SERVICE_LOG_FILE = "boot-service.log"; /** systemd expands `%` specifiers, including in unquoted append-log paths. */ function escapeSystemdSpecifiers(value: string): string { @@ -599,7 +601,7 @@ export const make = Effect.fn("cloud.boot_service.make")(function* (input: { environmentPath, }); const unitPath = detectedManager?.unitPath ?? ""; - const logPath = path.join(input.logsDir, "boot-service.log"); + const logPath = path.join(input.logsDir, BOOT_SERVICE_LOG_FILE); const statePath = path.join(input.baseDir, "runtime", SERVICE_STATE_FILE); const restartPendingPath = path.join(input.baseDir, "runtime", SERVICE_RESTART_PENDING_FILE); const runtimePaths = pinnedRuntimePaths(path, input.baseDir, input.cliVersion, platform); diff --git a/apps/server/src/cloud/http.ts b/apps/server/src/cloud/http.ts index b4c366a500c1..8e2d0e361458 100644 --- a/apps/server/src/cloud/http.ts +++ b/apps/server/src/cloud/http.ts @@ -104,6 +104,13 @@ const CLOUD_MINT_NONCE_PREFIX = "cloud-mint-nonce-"; const CLOUD_MINT_JTI_PREFIX = "cloud-mint-jti-"; const CLOUD_HEALTH_NONCE_PREFIX = "cloud-health-nonce-"; const CLOUD_HEALTH_JTI_PREFIX = "cloud-health-jti-"; +/** Secret store name prefixes of cloud replay markers. The server prunes expired ones. */ +export const CLOUD_REPLAY_MARKER_PREFIXES = [ + CLOUD_MINT_NONCE_PREFIX, + CLOUD_MINT_JTI_PREFIX, + CLOUD_HEALTH_NONCE_PREFIX, + CLOUD_HEALTH_JTI_PREFIX, +] as const; const CLOUD_PROOF_MAX_LIFETIME_SECONDS = 5 * 60; const CLOUD_PROOF_CLOCK_SKEW_SECONDS = 60; // The desktop app stops its backends within seconds of writing the marker. diff --git a/apps/server/src/config.ts b/apps/server/src/config.ts index 74ffde8efef0..d8dad5ae4d24 100644 --- a/apps/server/src/config.ts +++ b/apps/server/src/config.ts @@ -44,7 +44,6 @@ export interface ServerDerivedPaths { /** Screenshots the agent asks the collaborative browser to keep for the user. */ readonly browserArtifactsDir: string; readonly logsDir: string; - readonly serverLogPath: string; readonly serverTracePath: string; readonly providerLogsDir: string; readonly providerEventLogPath: string; @@ -154,7 +153,6 @@ export const deriveServerPaths = Effect.fn(function* ( attachmentsDir, browserArtifactsDir: join(stateDir, "browser-artifacts"), logsDir, - serverLogPath: join(logsDir, "server.log"), serverTracePath: join(logsDir, "server.trace.ndjson"), providerLogsDir, providerEventLogPath: join(providerLogsDir, "events.log"), diff --git a/apps/server/src/diagnostics/TraceDiagnostics.ts b/apps/server/src/diagnostics/TraceDiagnostics.ts index ca5552058f3f..afb7b4e923bb 100644 --- a/apps/server/src/diagnostics/TraceDiagnostics.ts +++ b/apps/server/src/diagnostics/TraceDiagnostics.ts @@ -81,7 +81,12 @@ interface TraceDiagnosticsErrorSummary { const DEFAULT_SLOW_SPAN_THRESHOLD_MS = 1_000; const TOP_LIMIT = 10; const RECENT_LIMIT = 20; -function toRotatedTracePaths(traceFilePath: string, maxFiles: number): ReadonlyArray { + +/** The trace file and its rotated backups, oldest first. */ +export function toRotatedTracePaths( + traceFilePath: string, + maxFiles: number, +): ReadonlyArray { const backupCount = Math.max(0, Math.floor(maxFiles)); const backups = Array.from( { length: backupCount }, diff --git a/apps/server/src/observability/EventLoopMonitor.test.ts b/apps/server/src/observability/EventLoopMonitor.test.ts new file mode 100644 index 000000000000..fbe30157aec2 --- /dev/null +++ b/apps/server/src/observability/EventLoopMonitor.test.ts @@ -0,0 +1,77 @@ +import { assert, describe, it } from "@effect/vitest"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Tracer from "effect/Tracer"; +import * as TestClock from "effect/testing/TestClock"; + +import { type EventLoopReadings, layerWith, stallMs } from "./EventLoopMonitor.ts"; + +const ms = (value: number) => value * 1e6; + +// Node reports a stall of S as a gap of up to S + 1 s, the histogram resolution. +const stalled: EventLoopReadings = { + delayMaxNs: ms(5_950), + activeMs: 6_200, + utilization: 0.176, + usage: { + userCPUTime: 310_400, + systemCPUTime: 95_600, + majorPageFault: 8_412, + minorPageFault: 20_031, + involuntaryContextSwitches: 57, + }, + rssBytes: 1536 * 1024 * 1024, +}; +// Over the threshold as read, but not once the resolution is subtracted. +const quiet: EventLoopReadings = { ...stalled, delayMaxNs: ms(2_950) }; + +describe("EventLoopMonitor", () => { + it.effect("records a warning span only for samples that saw a stall", () => + Effect.gen(function* () { + const spans: Array = []; + const tracer = Tracer.make({ + span: (options) => { + const span = new Tracer.NativeSpan(options); + spans.push(span); + return span; + }, + }); + // The first sample covers startup, so the monitor discards it. + const samples = [stalled, quiet, stalled]; + + yield* Effect.gen(function* () { + yield* Layer.build(layerWith(Effect.succeed(Effect.sync(() => samples.shift() ?? quiet)))); + yield* TestClock.adjust("60 seconds"); + assert.lengthOf(spans, 0); + yield* TestClock.adjust("30 seconds"); + }).pipe(Effect.scoped, Effect.withTracer(tracer)); + + assert.deepStrictEqual( + spans.map((span) => span.name), + ["server.eventLoop.stall"], + ); + const [span] = spans; + assert.deepStrictEqual(Object.fromEntries(span!.attributes), { + delayMaxMs: 4_950, + utilization: 0.18, + cpuUserMs: 310, + cpuSystemMs: 96, + majorPageFaults: 8_412, + minorPageFaults: 20_031, + involuntaryContextSwitches: 57, + rssMb: 1536, + }); + assert.deepStrictEqual( + span!.events.map(([name, , attributes]) => [name, attributes["effect.logLevel"]]), + [["event loop stalled for 4950 ms", "WARN"]], + ); + }), + ); + + it("ignores delay the loop spent idle, such as a system sleep", () => { + // Waking from sleep reads as a long gap, but the loop was idle in poll for it. + const asleep: EventLoopReadings = { ...stalled, delayMaxNs: ms(600_000), activeMs: 900 }; + assert.isUndefined(stallMs(asleep)); + assert.strictEqual(stallMs({ ...asleep, activeMs: 600_000 }), 599_000); + }); +}); diff --git a/apps/server/src/observability/EventLoopMonitor.ts b/apps/server/src/observability/EventLoopMonitor.ts new file mode 100644 index 000000000000..13b7b48c4cf7 --- /dev/null +++ b/apps/server/src/observability/EventLoopMonitor.ts @@ -0,0 +1,135 @@ +// @effect-diagnostics nodeBuiltinImport:off - only node:perf_hooks exposes the event loop delay histogram. +import * as NodePerfHooks from "node:perf_hooks"; + +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import type * as Scope from "effect/Scope"; + +// Node's delay histogram wakes a native timer every RESOLUTION_MS and records the +// gap between wakeups, so an idle loop reads about RESOLUTION_MS and a stall of S +// reads between S and S + RESOLUTION_MS. We subtract the resolution, so a delay can +// undercount a stall by up to RESOLUTION_MS. With these values every stall over 3 s +// is caught, at 1 wakeup per second that never enters JS. +const RESOLUTION_MS = 1000; +const STALL_THRESHOLD_MS = 2000; +const SAMPLE_INTERVAL = "30 seconds"; + +/** One sample interval as Node reports it. Delay in ns, active time in ms, CPU in µs. */ +export interface EventLoopReadings { + readonly delayMaxNs: number; + readonly activeMs: number; + readonly utilization: number; + readonly usage: Pick< + NodeJS.ResourceUsage, + | "userCPUTime" + | "systemCPUTime" + | "majorPageFault" + | "minorPageFault" + | "involuntaryContextSwitches" + >; + readonly rssBytes: number; +} + +// Enables the delay histogram for the layer's lifetime. Each read returns the +// readings since the previous read and resets the histogram. Node skips the first +// gap after a reset, so a stall right at a sample boundary can be missed. +const makeNodeSampler = Effect.gen(function* () { + const histogram = yield* Effect.acquireRelease( + Effect.sync(() => { + const histogram = NodePerfHooks.monitorEventLoopDelay({ resolution: RESOLUTION_MS }); + histogram.enable(); + return histogram; + }), + (histogram) => Effect.sync(() => histogram.disable()), + ); + let elu = NodePerfHooks.performance.eventLoopUtilization(); + let usage = process.resourceUsage(); + + // @effect-diagnostics-next-line returnEffectInGen:off - the read effect is the result. + return Effect.sync(() => { + const nextElu = NodePerfHooks.performance.eventLoopUtilization(); + const nextUsage = process.resourceUsage(); + const loop = NodePerfHooks.performance.eventLoopUtilization(nextElu, elu); + const readings: EventLoopReadings = { + delayMaxNs: histogram.max, + activeMs: loop.active, + utilization: loop.utilization, + usage: { + userCPUTime: nextUsage.userCPUTime - usage.userCPUTime, + systemCPUTime: nextUsage.systemCPUTime - usage.systemCPUTime, + majorPageFault: nextUsage.majorPageFault - usage.majorPageFault, + minorPageFault: nextUsage.minorPageFault - usage.minorPageFault, + involuntaryContextSwitches: + nextUsage.involuntaryContextSwitches - usage.involuntaryContextSwitches, + }, + rssBytes: process.memoryUsage.rss(), + }; + histogram.reset(); + elu = nextElu; + usage = nextUsage; + return readings; + }); +}); + +/** + * Returns the stall to report for one sample in ms, or undefined when there was none. + */ +export const stallMs = ({ delayMaxNs, activeMs }: EventLoopReadings) => { + const delayMs = Math.round(delayMaxNs / 1e6) - RESOLUTION_MS; + // A stall is time the loop spent running code, so it counts as active time. libuv's + // clock keeps running while the system sleeps on macOS and Windows, so a sleep also + // reads as delay, but the loop spent it idle in poll. + if (delayMs <= STALL_THRESHOLD_MS || activeMs < delayMs) return undefined; + return delayMs; +}; + +/** + * Samples event loop health every 30 s and records a `server.eventLoop.stall` span + * with a warning when the loop stalled for more than 2 s, so stalls land in + * the local trace file and Settings > Diagnostics without OTLP. Takes the sampler + * so tests can inject readings. + */ +export const layerWith = ( + makeSampler: Effect.Effect, never, Scope.Scope>, +) => + Layer.effectDiscard( + Effect.gen(function* () { + const sample = yield* makeSampler; + const tick = Effect.gen(function* () { + const readings = yield* sample; + const delayMaxMs = stallMs(readings); + if (delayMaxMs === undefined) return; + const { utilization, usage, rssBytes } = readings; + // Root, as the stall has no caller to attach to. Warn level keeps it when + // T3CODE_TRACE_MIN_LEVEL is raised to cut trace noise. + yield* Effect.logWarning(`event loop stalled for ${delayMaxMs} ms`).pipe( + Effect.withSpan("server.eventLoop.stall", { + root: true, + level: "Warn", + attributes: { + delayMaxMs, + utilization: Math.round(utilization * 100) / 100, + cpuUserMs: Math.round(usage.userCPUTime / 1000), + cpuSystemMs: Math.round(usage.systemCPUTime / 1000), + majorPageFaults: usage.majorPageFault, + minorPageFaults: usage.minorPageFault, + involuntaryContextSwitches: usage.involuntaryContextSwitches, + rssMb: Math.round(rssBytes / 1024 / 1024), + }, + }), + ); + }); + const wait = Effect.sleep(SAMPLE_INTERVAL); + // The layer builds before the rest of the server, so the first sample covers + // startup work such as migrations and projection bootstrap. That can block the + // loop for seconds on a large database, so skip it rather than warn at every + // launch. Layers build outside any span, so this fiber retains no parent span. + yield* wait.pipe( + Effect.andThen(sample), + Effect.andThen(wait.pipe(Effect.andThen(tick), Effect.forever)), + Effect.forkScoped, + ); + }), + ); + +export const layer = layerWith(makeNodeSampler); diff --git a/apps/server/src/observability/HeapSnapshot.test.ts b/apps/server/src/observability/HeapSnapshot.test.ts new file mode 100644 index 000000000000..0beb466e5019 --- /dev/null +++ b/apps/server/src/observability/HeapSnapshot.test.ts @@ -0,0 +1,36 @@ +// @effect-diagnostics nodeBuiltinImport:off - tests fake a failed write at the native v8 boundary. +import * as NodeServices from "@effect/platform-node/NodeServices"; +import * as NodeFS from "node:fs"; +import * as NodePath from "node:path"; +import * as NodeV8 from "node:v8"; +import { assert, it } from "@effect/vitest"; +import * as Effect from "effect/Effect"; +import * as FileSystem from "effect/FileSystem"; +import { vi } from "vite-plus/test"; + +import { writeHeapSnapshot } from "./HeapSnapshot.ts"; + +vi.mock("node:v8", async (importOriginal) => { + const actual = await importOriginal(); + return { ...actual, writeHeapSnapshot: vi.fn(actual.writeHeapSnapshot) }; +}); + +it.layer(NodeServices.layer)("writeHeapSnapshot", (it) => { + it.effect("removes the partial file when the write fails", () => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const logsDir = yield* fs.makeTempDirectoryScoped({ prefix: "t3-heap-snapshot-test-" }); + let partialPath: string | undefined; + vi.mocked(NodeV8.writeHeapSnapshot).mockImplementationOnce((path) => { + partialPath = path; + if (path) NodeFS.writeFileSync(path, "partial"); + throw new Error("ENOSPC: no space left on device"); + }); + + yield* writeHeapSnapshot(logsDir); + + assert.strictEqual(NodePath.dirname(partialPath ?? ""), logsDir); + assert.deepEqual(yield* fs.readDirectory(logsDir), []); + }), + ); +}); diff --git a/apps/server/src/observability/HeapSnapshot.ts b/apps/server/src/observability/HeapSnapshot.ts new file mode 100644 index 000000000000..827dfab2c6d6 --- /dev/null +++ b/apps/server/src/observability/HeapSnapshot.ts @@ -0,0 +1,52 @@ +// @effect-diagnostics nodeBuiltinImport:off - v8.writeHeapSnapshot has no Effect equivalent. +import * as NodePath from "node:path"; +import * as NodeV8 from "node:v8"; + +import { HostProcessPlatform } from "@t3tools/shared/hostProcess"; +import * as DateTime from "effect/DateTime"; +import * as Effect from "effect/Effect"; +import * as FileSystem from "effect/FileSystem"; +import * as Layer from "effect/Layer"; + +import * as ServerConfig from "../config.ts"; + +/** + * Writes one V8 heap snapshot into `logsDir` and logs its path. A failed write + * logs a warning and removes any partial file, because that file can hold + * secrets and the failure is often a full disk. + */ +export const writeHeapSnapshot = Effect.fn("server.heapSnapshot", { root: true })( + function* (logsDir: string) { + const fs = yield* FileSystem.FileSystem; + const timestamp = DateTime.formatIso(yield* DateTime.now).replaceAll(":", "-"); + const path = NodePath.join(logsDir, `server-${process.pid}-${timestamp}.heapsnapshot`); + yield* Effect.annotateCurrentSpan({ path }); + yield* Effect.try(() => NodeV8.writeHeapSnapshot(path)).pipe( + Effect.tapError(() => fs.remove(path, { force: true }).pipe(Effect.ignore)), + ); + yield* Effect.logInfo("Wrote heap snapshot.", { path }); + }, + Effect.catch((cause) => Effect.logWarning("Failed to write heap snapshot.", { cause })), +); + +/** + * Writes a heap snapshot when the process gets SIGUSR2 (`kill -USR2 `), + * so a maintainer can see what a long-running server holds. See "Heap + * Snapshots" in docs/operations/observability.md. + * + * The write blocks the event loop, so two snapshots never overlap: a signal + * sent during a write waits until it finishes. Windows has no SIGUSR2, so the + * layer does nothing there. + */ +export const layer = Layer.effectDiscard( + Effect.gen(function* () { + if ((yield* HostProcessPlatform) === "win32") return; + const { logsDir } = yield* ServerConfig.ServerConfig; + const runFork = Effect.runForkWith(yield* Effect.context()); + const onSignal = () => void runFork(writeHeapSnapshot(logsDir)); + yield* Effect.acquireRelease( + Effect.sync(() => process.on("SIGUSR2", onSignal)), + () => Effect.sync(() => process.off("SIGUSR2", onSignal)), + ); + }), +); diff --git a/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts b/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts index f855771b8f01..9ccf78ca1744 100644 --- a/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts +++ b/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts @@ -435,6 +435,7 @@ describe("OrchestrationEngine", () => { updatedAt: projectionSnapshot.updatedAt, }), getDeletedWorktreeThreads: () => Effect.die("unused"), + listThreadsWithPullRequests: () => Effect.die("unused"), getArchivedShellSnapshot: () => Effect.succeed({ snapshotSequence: projectionSnapshot.snapshotSequence, diff --git a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts index 890c8ae55c53..b80e13726811 100644 --- a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts +++ b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts @@ -30,6 +30,7 @@ import * as ThreadPlanProgress from "../ThreadPlanProgress.ts"; import { ProjectionSnapshotQuery } from "../Services/ProjectionSnapshotQuery.ts"; import { encodeThreadDetailPageCursor } from "../threadDetailCursor.ts"; import { projectThreadDetailSnapshot } from "../ActivityPayloadProjection.ts"; +import { readSweepSnapshot } from "../ThreadPullRequestReactor.ts"; import { makeSqlStatementCounter } from "../../../integration/SqlStatementCounter.integration.ts"; const asProjectId = (value: string): ProjectId => ProjectId.make(value); @@ -3503,6 +3504,145 @@ it.effect("omits foreign-host PRs from legacy snapshots while preserving native }).pipe(Effect.provide(layer)); }); +it.effect( + "lists linked threads like the shell snapshot, in one query and without identities", + () => { + const resolved: string[] = []; + const layer = OrchestrationProjectionSnapshotQueryLive.pipe( + Layer.provide(ThreadBackgroundLiveness.layer), + Layer.provide(ThreadPlanProgress.layer), + Layer.provide( + Layer.succeed(RepositoryIdentityResolver.RepositoryIdentityResolver, { + resolve: (root) => + Effect.sync(() => { + resolved.push(root); + return null; + }), + }), + ), + Layer.provideMerge(SqlitePersistenceMemory), + ); + return Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + const query = yield* ProjectionSnapshotQuery; + yield* sql`INSERT INTO projection_projects (project_id, title, workspace_root, scripts_json, created_at, updated_at) + VALUES ('p1', 'One', '/one', '[]', '2026-09-01T00:00:00Z', '2026-09-01T00:00:00Z'), + ('p2', 'Two', '/two', '[]', '2026-09-01T00:00:00Z', '2026-09-01T00:00:00Z')`; + yield* sql`INSERT INTO projection_threads (thread_id, project_id, title, model_selection_json, runtime_mode, interaction_mode, created_at, updated_at, archived_at, deleted_at, settled_override, settled_at) + VALUES + ('t-late', 'p1', 'Late', '{"provider":"codex","model":"gpt-5"}', 'full-access', 'default', '2026-09-03T00:00:00Z', '2026-09-03T00:00:00Z', NULL, NULL, 'settled', '2026-09-04T00:00:00Z'), + ('t-early', 'p2', 'Early', '{"provider":"codex","model":"gpt-5"}', 'full-access', 'default', '2026-09-01T00:00:00Z', '2026-09-01T00:00:00Z', NULL, NULL, NULL, NULL), + ('t-first', 'p1', 'First', '{"provider":"codex","model":"gpt-5"}', 'full-access', 'default', '2026-09-02T00:00:00Z', '2026-09-02T00:00:00Z', NULL, NULL, NULL, NULL), + ('t-plain', 'p1', 'Plain', '{"provider":"codex","model":"gpt-5"}', 'full-access', 'default', '2026-09-02T00:00:00Z', '2026-09-02T00:00:00Z', NULL, NULL, NULL, NULL), + ('t-archived', 'p1', 'Archived', '{"provider":"codex","model":"gpt-5"}', 'full-access', 'default', '2026-09-02T00:00:00Z', '2026-09-02T00:00:00Z', '2026-09-05T00:00:00Z', NULL, NULL, NULL), + ('t-deleted', 'p1', 'Deleted', '{"provider":"codex","model":"gpt-5"}', 'full-access', 'default', '2026-09-02T00:00:00Z', '2026-09-02T00:00:00Z', NULL, '2026-09-05T00:00:00Z', NULL, NULL)`; + yield* sql`INSERT INTO projection_thread_pull_requests (thread_id, host, repository, number, url, source, linked_at, snapshot_json) + VALUES + ('t-late', 'github.com', 'acme/web', 3, 'https://github.com/acme/web/pull/3', 'manual', '2026-09-03T00:00:00Z', NULL), + ('t-early', 'github.com', 'acme/api', 4, 'https://github.com/acme/api/pull/4', 'agent', '2026-09-01T00:00:00Z', NULL), + ('t-first', 'github.com', 'acme/web', 2, 'https://github.com/acme/web/pull/2', 'stack-dismissed', '2026-09-02T00:00:00Z', NULL), + ('t-first', 'github.com', 'acme/web', 1, 'https://github.com/acme/web/pull/1', 'created', '2026-09-02T00:00:00Z', + '{"state":"open","title":"One","headBranch":"one","baseBranch":"main","isDraft":false,"updatedAt":null,"syncedAt":"2026-09-02T00:00:00Z"}'), + ('t-archived', 'github.com', 'acme/web', 5, 'https://github.com/acme/web/pull/5', 'manual', '2026-09-02T00:00:00Z', NULL), + ('t-deleted', 'github.com', 'acme/web', 6, 'https://github.com/acme/web/pull/6', 'manual', '2026-09-02T00:00:00Z', NULL)`; + const expected = (yield* query.getShellSnapshot()).threads + .filter((thread) => thread.pullRequests.length > 0) + .map(({ id, projectId, settledOverride, settledAt, pullRequests }) => ({ + id, + projectId, + settledOverride, + settledAt, + pullRequests, + })); + resolved.length = 0; + + const counter = makeSqlStatementCounter(); + const threads = yield* query + .listThreadsWithPullRequests() + .pipe(Effect.withTracer(counter.tracer)); + assert.deepStrictEqual( + threads.map((thread) => [thread.id, thread.pullRequests.map((link) => link.number)]), + [ + ["t-first", [1, 2]], + ["t-late", [3]], + ["t-early", [4]], + ], + ); + assert.deepStrictEqual(threads, expected); + assert.strictEqual(counter.count(), 1); + assert.deepStrictEqual(resolved, []); + }).pipe(Effect.provide(layer)); + }, +); + +it.effect("reads one sweep thread and its projects like the shell snapshot", () => { + const layer = OrchestrationProjectionSnapshotQueryLive.pipe( + Layer.provide(ThreadBackgroundLiveness.layer), + Layer.provide(ThreadPlanProgress.layer), + Layer.provide( + Layer.succeed(RepositoryIdentityResolver.RepositoryIdentityResolver, { + resolve: () => + Effect.succeed({ + canonicalKey: "github.com/acme/web", + provider: "github", + displayName: "acme/web", + locator: { + source: "git-remote" as const, + remoteName: "origin", + remoteUrl: "https://github.com/acme/web.git", + }, + }), + }), + ), + Layer.provideMerge(SqlitePersistenceMemory), + ); + return Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + const query = yield* ProjectionSnapshotQuery; + yield* sql`INSERT INTO projection_projects (project_id, title, workspace_root, scripts_json, created_at, updated_at) + VALUES ('p1', 'One', '/one', '[]', '2026-09-01T00:00:00Z', '2026-09-01T00:00:00Z'), + ('p2', 'Two', '/two', '[]', '2026-09-02T00:00:00Z', '2026-09-02T00:00:00Z'), + ('p3', 'Three', '/three', '[]', '2026-09-03T00:00:00Z', '2026-09-03T00:00:00Z')`; + yield* sql`INSERT INTO projection_threads (thread_id, project_id, title, model_selection_json, runtime_mode, interaction_mode, branch, worktree_path, branch_pull_request_json, latest_turn_id, latest_user_message_at, pending_approval_count, snoozed_until, snoozed_at, created_at, updated_at, settled_override, settled_at) + VALUES + ('t-linked', 'p1', 'Linked', '{"provider":"codex","model":"gpt-5"}', 'full-access', 'default', 'feature', '/one/wt', NULL, 'turn-1', '2026-09-02T00:00:00Z', 1, NULL, NULL, '2026-09-01T00:00:00Z', '2026-09-02T00:00:00Z', NULL, NULL), + ('t-branch', 'p1', 'Branch', '{"provider":"codex","model":"gpt-5"}', 'full-access', 'default', 'other', NULL, + '{"projectId":"p2","repository":"acme/web","number":8,"url":"https://github.com/acme/web/pull/8"}', + NULL, NULL, 0, '2026-09-10T00:00:00Z', '2026-09-02T00:00:00Z', '2026-09-01T00:00:00Z', '2026-09-02T00:00:00Z', 'settled', '2026-09-03T00:00:00Z'), + ('t-other', 'p3', 'Other', '{"provider":"codex","model":"gpt-5"}', 'full-access', 'default', NULL, NULL, NULL, NULL, NULL, 0, NULL, NULL, '2026-09-01T00:00:00Z', '2026-09-01T00:00:00Z', NULL, NULL)`; + yield* sql`INSERT INTO projection_thread_pull_requests (thread_id, host, repository, number, url, source, linked_at) + VALUES ('t-linked', 'github.com', 'acme/web', 7, 'https://github.com/acme/web/pull/7', 'agent', '2026-09-02T00:00:00Z')`; + yield* sql`INSERT INTO projection_turns (thread_id, turn_id, state, requested_at, started_at, completed_at, checkpoint_files_json) + VALUES ('t-linked', 'turn-1', 'completed', '2026-09-02T00:00:00Z', '2026-09-02T00:00:01Z', '2026-09-02T00:00:02Z', '[]')`; + yield* sql`INSERT INTO projection_thread_sessions (thread_id, status, provider_name, active_turn_id, last_error, updated_at) + VALUES ('t-linked', 'ready', 'codex', NULL, NULL, '2026-09-02T00:00:03Z')`; + for (const projector of Object.values(ORCHESTRATION_PROJECTOR_NAMES)) { + yield* sql`INSERT INTO projection_state (projector, last_applied_sequence, updated_at) + VALUES (${projector}, 9, '2026-09-02T00:00:03Z')`; + } + + const full = yield* query.getShellSnapshot(); + // The seeded fields must reach the snapshot, or the parity check is empty. + const linked = full.threads.find((thread) => thread.id === ThreadId.make("t-linked")); + assert.strictEqual(full.snapshotSequence, 9); + assert.strictEqual(linked?.linkedPullRequest?.number, 7); + assert.strictEqual(linked?.latestTurn?.turnId, asTurnId("turn-1")); + assert.strictEqual(linked?.session?.status, "ready"); + + for (const [threadId, projectIds] of [ + [ThreadId.make("t-linked"), [asProjectId("p1")]], + // Settlement also needs the project that the saved branch PR names. + [ThreadId.make("t-branch"), [asProjectId("p1"), asProjectId("p2")]], + ] as const) { + assert.deepStrictEqual(yield* readSweepSnapshot(query, threadId), { + snapshotSequence: full.snapshotSequence, + projects: full.projects.filter((project) => projectIds.includes(project.id)), + threads: full.threads.filter((thread) => thread.id === threadId), + }); + } + }).pipe(Effect.provide(layer)); +}); + projectionSnapshotLayer("ProjectionSnapshotQuery activities by kind", (it) => { it.effect("lists one kind across active threads only, without hydrating the threads", () => Effect.gen(function* () { diff --git a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts index 1b44054c7a32..71ef1ad97321 100644 --- a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts +++ b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts @@ -12,7 +12,7 @@ import { OrchestrationProposedPlanId, OrchestrationReadModel, OrchestrationThreadSearchSource, - OrchestrationShellSnapshot, + type OrchestrationShellSnapshot, OrchestrationThread, OrchestrationThreadDetailSnapshot, ProjectScript, @@ -77,11 +77,11 @@ import { type ProjectionSnapshotCounts, type ProjectionThreadCheckpointContext, type ProjectionThreadDetailQuery, + type ProjectionThreadPullRequests, type ProjectionSnapshotQueryShape, } from "../Services/ProjectionSnapshotQuery.ts"; const decodeReadModel = Schema.decodeUnknownEffect(OrchestrationReadModel); -const decodeShellSnapshot = Schema.decodeUnknownEffect(OrchestrationShellSnapshot); const decodeThread = Schema.decodeUnknownEffect(OrchestrationThread); const decodeImportedTranscriptsPayload = Schema.decodeUnknownOption( Schema.fromJsonString( @@ -802,6 +802,41 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { `, }); + // One row per link, in the shell snapshot's thread order and link order. + const listActiveThreadPullRequestSyncRows = SqlSchema.findAll({ + Request: Schema.Void, + Result: ProjectionThreadPullRequestDbRowSchema.mapFields( + Struct.assign({ + projectId: ProjectionThread.fields.projectId, + settledOverride: ProjectionThread.fields.settledOverride, + settledAt: ProjectionThread.fields.settledAt, + }), + ), + execute: () => + sql` + SELECT + links.thread_id AS "threadId", + threads.project_id AS "projectId", + threads.settled_override AS "settledOverride", + threads.settled_at AS "settledAt", + links.host, + links.repository, + links.number, + links.url, + links.source, + links.linked_at AS "linkedAt", + links.snapshot_json AS "snapshot", + links.stack_json AS "stack" + FROM projection_thread_pull_requests links + INNER JOIN projection_threads threads + ON threads.thread_id = links.thread_id + WHERE threads.deleted_at IS NULL + AND threads.archived_at IS NULL + ORDER BY threads.project_id ASC, threads.created_at ASC, threads.thread_id ASC, + links.linked_at ASC, links.number ASC + `, + }); + const listArchivedThreadPullRequestRows = SqlSchema.findAll({ Request: Schema.Void, Result: ProjectionThreadPullRequestDbRowSchema, @@ -2710,7 +2745,10 @@ pending_approval_requests AS ( ); const pullRequestsByThread = groupPullRequestRowsByThread(pullRequestRows); - const snapshot = { + // Built from schema-decoded rows, so no second decode here. The HTTP + // and RPC layers encode it against OrchestrationShellSnapshot on the + // way out, like the per-item shells from getThreadShellById. + return { snapshotSequence: computeSnapshotSequence(stateRows), projects: Arr.filterMap(projectRows, (row) => row.deletedAt === null @@ -2764,15 +2802,7 @@ pending_approval_requests AS ( : Result.failVoid, ), updatedAt: updatedAt ?? "1970-01-01T00:00:00.000Z", - }; - - return yield* decodeShellSnapshot(snapshot).pipe( - Effect.mapError( - toPersistenceDecodeError( - "ProjectionSnapshotQuery.getShellSnapshot:decodeShellSnapshot", - ), - ), - ); + } satisfies OrchestrationShellSnapshot; }), ), Effect.mapError((error) => { @@ -2783,6 +2813,35 @@ pending_approval_requests AS ( }), ); + const listThreadsWithPullRequests: ProjectionSnapshotQueryShape["listThreadsWithPullRequests"] = + () => + listActiveThreadPullRequestSyncRows(undefined).pipe( + Effect.map((rows) => { + const threads = new Map< + ThreadId, + ProjectionThreadPullRequests & { readonly pullRequests: Array } + >(); + for (const row of rows) { + const thread = threads.get(row.threadId) ?? { + id: row.threadId, + projectId: row.projectId, + settledOverride: row.settledOverride, + settledAt: row.settledAt, + pullRequests: [], + }; + thread.pullRequests.push(mapPullRequestRow(row)); + threads.set(row.threadId, thread); + } + return [...threads.values()]; + }), + Effect.mapError( + toPersistenceSqlOrDecodeError( + "ProjectionSnapshotQuery.listThreadsWithPullRequests:query", + "ProjectionSnapshotQuery.listThreadsWithPullRequests:decodeRows", + ), + ), + ); + const getArchivedShellSnapshot: ProjectionSnapshotQueryShape["getArchivedShellSnapshot"] = () => sql .withTransaction( @@ -2876,7 +2935,7 @@ pending_approval_requests AS ( sessionRows.map((row) => [row.threadId, mapSessionRow(row)] as const), ); - const snapshot = { + return { snapshotSequence: computeSnapshotSequence(stateRows), projects: Arr.filterMap(projectRows, (row) => row.deletedAt === null && activeProjectIds.has(row.projectId) @@ -2926,15 +2985,7 @@ pending_approval_requests AS ( planProgress: threadPlanProgress.getThreadPlanProgress(row.threadId), })), updatedAt: updatedAt ?? "1970-01-01T00:00:00.000Z", - }; - - return yield* decodeShellSnapshot(snapshot).pipe( - Effect.mapError( - toPersistenceDecodeError( - "ProjectionSnapshotQuery.getArchivedShellSnapshot:decodeShellSnapshot", - ), - ), - ); + } satisfies OrchestrationShellSnapshot; }), ), Effect.mapError((error) => { @@ -3778,6 +3829,7 @@ pending_approval_requests AS ( listActivitiesByKind, getSnapshot, getShellSnapshot, + listThreadsWithPullRequests, getArchivedShellSnapshot, getDeletedWorktreeThreads, searchThreads, diff --git a/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts b/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts index 254aec4bde7e..87f9bd03d46c 100644 --- a/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts +++ b/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts @@ -891,8 +891,92 @@ describe("ProviderCommandReactor", () => { expect(thread?.session?.threadId).toBe("thread-1"); expect(thread?.session?.status).toBe("starting"); expect(thread?.session?.runtimeMode).toBe("approval-required"); + expect(harness.startSession.mock.calls[0]?.[1]).not.toHaveProperty("title"); }); + effectIt.effect("forwards only a user-renamed title when starting a provider session", () => + Effect.gen(function* () { + const harness = yield* Effect.promise(() => + createHarness({ initialTitle: "Add a progressive blur as you scroll" }), + ); + const now = "2026-01-01T00:00:00.000Z"; + const modelSelection = { + instanceId: ProviderInstanceId.make("codex"), + model: "gpt-5-codex", + }; + const startTurn = (threadId: string, text: string, titleSeed: string) => + harness.engine.dispatch({ + type: "thread.turn.start", + commandId: CommandId.make(`cmd-title-${threadId}`), + threadId: ThreadId.make(threadId), + message: { + messageId: asMessageId(`message-${threadId}`), + role: "user", + text, + attachments: [], + }, + titleSeed, + interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE, + runtimeMode: "approval-required", + createdAt: now, + }); + + yield* startTurn( + "thread-1", + "Add a progressive blur as you scroll", + "Add a progressive blur as you scroll", + ); + yield* Effect.promise(() => waitFor(() => harness.startSession.mock.calls.length === 1)); + expect(harness.startSession.mock.calls[0]?.[1]).not.toHaveProperty("title"); + + yield* harness.engine.dispatch({ + type: "thread.create", + commandId: CommandId.make("cmd-thread-create-renamed"), + threadId: ThreadId.make("thread-renamed"), + projectId: asProjectId("project-1"), + title: "New thread", + modelSelection, + interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE, + runtimeMode: "approval-required", + branch: null, + worktreePath: null, + createdAt: now, + }); + yield* harness.engine.dispatch({ + type: "thread.meta.update", + commandId: CommandId.make("cmd-thread-rename"), + threadId: ThreadId.make("thread-renamed"), + title: "Keep this name", + }); + yield* startTurn("thread-renamed", "hello there", "hello there"); + yield* Effect.promise(() => waitFor(() => harness.startSession.mock.calls.length === 2)); + expect(harness.startSession.mock.calls[1]?.[1]).toMatchObject({ title: "Keep this name" }); + + yield* harness.engine.dispatch({ + type: "thread.create", + commandId: CommandId.make("cmd-thread-create-seeded"), + threadId: ThreadId.make("thread-seeded"), + projectId: asProjectId("project-1"), + title: "New thread", + modelSelection, + interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE, + runtimeMode: "approval-required", + branch: null, + worktreePath: null, + createdAt: now, + }); + yield* harness.engine.dispatch({ + type: "thread.meta.update", + commandId: CommandId.make("cmd-thread-autotitle"), + threadId: ThreadId.make("thread-seeded"), + title: "hello there", + }); + yield* startTurn("thread-seeded", "hello there", "hello there"); + yield* Effect.promise(() => waitFor(() => harness.startSession.mock.calls.length === 3)); + expect(harness.startSession.mock.calls[2]?.[1]).not.toHaveProperty("title"); + }), + ); + effectIt.effect("projects inline context before sending the provider turn", () => Effect.gen(function* () { const harness = yield* Effect.promise(() => createHarness()); diff --git a/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts b/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts index f6794135a925..77022a8518d7 100644 --- a/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts +++ b/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts @@ -575,6 +575,9 @@ const make = Effect.gen(function* () { options?: { readonly modelSelection?: ModelSelection; readonly pendingTurnStart?: boolean; + // First-turn prompt seed. A manual title that still equals this seed was + // written by the client's auto-title, not a user rename. + readonly titleSeed?: string; }, ) { const thread = yield* resolveThreadShell(threadId); @@ -712,6 +715,15 @@ const make = Effect.gen(function* () { .refreshWorkspaceSnapshot({ instanceId: desiredInstanceId, cwd: effectiveCwd }) .pipe(Effect.forkDetach) : Effect.void; + // OpenCode skips SessionPrompt.ensureTitle when session.create already has + // a title. Prompt seeds and "New thread" are not user titles, so omit them + // and let the provider generate one. A real rename is source "manual" and + // differs from the first-turn prompt seed (the web client writes that seed + // through thread.meta.update, which also marks the title manual). + const manualTitle = thread.titleState?.source === "manual" ? thread.title.trim() : ""; + const promptSeed = options?.titleSeed?.trim(); + const sessionTitle = + manualTitle.length > 0 && manualTitle !== promptSeed ? thread.title : undefined; const startProviderSession = (input?: { readonly resumeCursor?: unknown; @@ -723,7 +735,7 @@ const make = Effect.gen(function* () { ...(preferredProvider ? { provider: preferredProvider } : {}), providerInstanceId: desiredInstanceId, ...(effectiveCwd ? { cwd: effectiveCwd } : {}), - ...(thread.title ? { title: thread.title } : {}), + ...(sessionTitle ? { title: sessionTitle } : {}), modelSelection: desiredModelSelection, ...(input?.resumeCursor !== undefined ? { resumeCursor: input.resumeCursor } : {}), runtimeMode: desiredRuntimeMode, @@ -839,6 +851,7 @@ const make = Effect.gen(function* () { readonly modelSelection?: ModelSelection; readonly interactionMode?: "default" | "plan"; readonly createdAt: string; + readonly titleSeed?: string; }) { const thread = yield* resolveThreadShell(input.threadId); if (!thread) { @@ -848,6 +861,7 @@ const make = Effect.gen(function* () { } yield* ensureSessionForThread(input.threadId, input.createdAt, { ...(input.modelSelection !== undefined ? { modelSelection: input.modelSelection } : {}), + ...(input.titleSeed !== undefined ? { titleSeed: input.titleSeed } : {}), pendingTurnStart: true, }); if (input.modelSelection !== undefined) { @@ -1488,6 +1502,11 @@ const make = Effect.gen(function* () { : {}), interactionMode: event.payload.interactionMode, createdAt: event.payload.createdAt, + // Later turns must not reuse the current title as titleSeed. Only the + // first prompt seed should suppress a not-yet-renamed session title. + ...(!hasOtherUserMessages && event.payload.titleSeed !== undefined + ? { titleSeed: event.payload.titleSeed } + : {}), }).pipe( Effect.asSome, Effect.catchCause((cause) => handleTurnStartFailure(cause).pipe(Effect.as(Option.none()))), diff --git a/apps/server/src/orchestration/PullRequestSyncReactor.test.ts b/apps/server/src/orchestration/PullRequestSyncReactor.test.ts index 5502376e691e..e6a18c2856a1 100644 --- a/apps/server/src/orchestration/PullRequestSyncReactor.test.ts +++ b/apps/server/src/orchestration/PullRequestSyncReactor.test.ts @@ -172,6 +172,7 @@ const makeHarness = Effect.fn("makePullRequestSyncHarness")(function* (options: const snapshots = yield* Ref.make(options.snapshot); const events = yield* PubSub.unbounded(); const snapshotReads = yield* Queue.unbounded(); + const shellSnapshotReads = yield* Ref.make(0); const syncCommands = yield* Ref.make>([]); const linkCommands = yield* Ref.make>([]); const summaryCalls = yield* Ref.make>([]); @@ -209,8 +210,16 @@ const makeHarness = Effect.fn("makePullRequestSyncHarness")(function* (options: const dependencies = Layer.mergeAll( Layer.mock(ProjectionSnapshotQuery)({ + listThreadsWithPullRequests: () => + Queue.offer(snapshotReads, undefined).pipe( + Effect.andThen(Ref.get(snapshots)), + Effect.map((snapshot) => snapshot.threads), + ), getShellSnapshot: () => - Queue.offer(snapshotReads, undefined).pipe(Effect.andThen(Ref.get(snapshots))), + Ref.update(shellSnapshotReads, (count) => count + 1).pipe( + Effect.andThen(Queue.offer(snapshotReads, undefined)), + Effect.andThen(Ref.get(snapshots)), + ), }), Layer.mock(PullRequestService)({ summary, @@ -235,6 +244,7 @@ const makeHarness = Effect.fn("makePullRequestSyncHarness")(function* (options: activation, snapshots, snapshotReads, + shellSnapshotReads, syncCommands, linkCommands, summaryCalls, @@ -511,6 +521,8 @@ describe("PullRequestSyncReactor", () => { ], ); assert.strictEqual((yield* Ref.get(fixture.stackCalls)).length, 1); + // Reads only linked threads, never the full shell snapshot of every thread. + assert.strictEqual(yield* Ref.get(fixture.shellSnapshotReads), 0); }).pipe(Effect.provide(fixture.layer)); }), ), diff --git a/apps/server/src/orchestration/PullRequestSyncReactor.ts b/apps/server/src/orchestration/PullRequestSyncReactor.ts index 2bdf3ff9e71d..ffcc9fb3c26d 100644 --- a/apps/server/src/orchestration/PullRequestSyncReactor.ts +++ b/apps/server/src/orchestration/PullRequestSyncReactor.ts @@ -1,7 +1,6 @@ import { siblingPullRequestUrl } from "@t3tools/shared/changeRequestUrl"; import { CommandId, - type OrchestrationThreadShell, type PullRequestSummary, type ThreadPullRequestKey, type ThreadPullRequestLink, @@ -36,7 +35,7 @@ const SLOW_SYNC_INTERVAL_MS = 15 * 60 * 1_000; type SnapshotFields = Omit; interface LinkEntry { - readonly thread: OrchestrationThreadShell; + readonly thread: ProjectionSnapshotQuery.ProjectionThreadPullRequests; readonly link: ThreadPullRequestLink; } @@ -104,15 +103,15 @@ function stacksEqual( ); } -function isUnsettled(thread: OrchestrationThreadShell): boolean { +function isUnsettled(thread: ProjectionSnapshotQuery.ProjectionThreadPullRequests): boolean { return thread.settledOverride !== "settled" && thread.settledAt === null; } /** * Keeps every thread ↔ pull request link's host snapshot current. One sweep a minute reads - * the shell snapshot, groups visible links by pull request so the host is asked once per PR - * no matter how many threads share it, and writes back only what changed. Native stacks the - * host reports are auto-linked to the thread as `source: "stack"`. + * only the active threads that have links, groups visible links by pull request so the host + * is asked once per PR no matter how many threads share it, and writes back only what + * changed. Native stacks the host reports are auto-linked to the thread as `source: "stack"`. */ export class PullRequestSyncReactor extends Context.Service< PullRequestSyncReactor, @@ -153,14 +152,13 @@ export const make = Effect.gen(function* () { Cause.hasInterruptsOnly(cause) ? Effect.failCause(cause) : Effect.logWarning(message, fields); const sweep = Effect.fn("PullRequestSyncReactor.sweep")(function* (requestedKey?: string) { - const snapshot = yield* snapshots.getShellSnapshot(); + const threads = yield* snapshots.listThreadsWithPullRequests(); const now = yield* DateTime.now; const nowMs = DateTime.toEpochMillis(now); const nowIso = DateTime.formatIso(now); const groups = new Map>(); - for (const thread of snapshot.threads) { - if (thread.archivedAt !== null) continue; + for (const thread of threads) { for (const link of visibleThreadPullRequests(thread.pullRequests)) { const key = threadPullRequestKeyOf(link); const entries = groups.get(key) ?? []; diff --git a/apps/server/src/orchestration/Services/ProjectionSnapshotQuery.ts b/apps/server/src/orchestration/Services/ProjectionSnapshotQuery.ts index eac3ede9c1ee..48e63d6fc8b0 100644 --- a/apps/server/src/orchestration/Services/ProjectionSnapshotQuery.ts +++ b/apps/server/src/orchestration/Services/ProjectionSnapshotQuery.ts @@ -64,6 +64,12 @@ export interface ProjectionFullThreadDiffContext { readonly toCheckpointRef: CheckpointRef | null; } +/** The thread fields pull request sync reads, for a thread with at least one link. */ +export type ProjectionThreadPullRequests = Pick< + OrchestrationThreadShell, + "id" | "projectId" | "settledOverride" | "settledAt" | "pullRequests" +>; + export interface ProjectionThreadDetailQuery { /** * Limit activities before SQLite returns and decodes their payloads. @@ -131,6 +137,16 @@ export interface ProjectionSnapshotQueryShape { ProjectionRepositoryError >; + /** + * Read active (not deleted, not archived) threads that have at least one pull + * request link, in shell snapshot order. Skips repository identity, so no + * legacy `linkedPullRequest` is derived. + */ + readonly listThreadsWithPullRequests: () => Effect.Effect< + ReadonlyArray, + ProjectionRepositoryError + >; + /** Durable worktree ownership retained after thread deletion, including across restarts. */ readonly getDeletedWorktreeThreads: () => Effect.Effect< ReadonlyArray<{ diff --git a/apps/server/src/orchestration/ThreadPullRequestReactor.test.ts b/apps/server/src/orchestration/ThreadPullRequestReactor.test.ts index d6572083dad2..cefd5bcb42b4 100644 --- a/apps/server/src/orchestration/ThreadPullRequestReactor.test.ts +++ b/apps/server/src/orchestration/ThreadPullRequestReactor.test.ts @@ -21,6 +21,7 @@ import * as Deferred from "effect/Deferred"; 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 PubSub from "effect/PubSub"; import * as Queue from "effect/Queue"; import * as Ref from "effect/Ref"; @@ -142,7 +143,8 @@ const makeHarness = Effect.fn("makeThreadPullRequestHarness")(function* (options threads: options.threads, updatedAt: NOW, }); - const reads = yield* Queue.unbounded(); + // Each shell read: a thread id for a one-thread read, null for a full read. + const reads = yield* Queue.unbounded(); const events = yield* PubSub.unbounded(); const commands = yield* Ref.make>([]); const branchCalls = yield* Ref.make< @@ -152,8 +154,24 @@ const makeHarness = Effect.fn("makeThreadPullRequestHarness")(function* (options let uuid = 0; const dependencies = Layer.mergeAll( Layer.mock(ProjectionSnapshotQuery)({ - getShellSnapshot: () => - Ref.get(snapshots).pipe(Effect.tap(() => Queue.offer(reads, undefined))), + getShellSnapshot: () => Ref.get(snapshots).pipe(Effect.tap(() => Queue.offer(reads, null))), + getSnapshotSequence: () => + Ref.get(snapshots).pipe(Effect.map(({ snapshotSequence }) => ({ snapshotSequence }))), + getThreadShellById: (threadId) => + Ref.get(snapshots).pipe( + Effect.map(({ threads }) => + Option.fromUndefinedOr( + threads.find((thread) => thread.id === threadId && thread.archivedAt === null), + ), + ), + Effect.tap(() => Queue.offer(reads, threadId)), + ), + getProjectShells: (projectIds) => + Ref.get(snapshots).pipe( + Effect.map(({ projects }) => + projects.filter((project) => projectIds?.includes(project.id) ?? true), + ), + ), }), Layer.mock(GitManager)({ branchPullRequest: (input, readOptions) => @@ -376,7 +394,7 @@ describe("ThreadPullRequestReactor", () => { : [checkpointEvent, sessionEvent]; for (const event of events) { yield* fixture.publish(event); - yield* Queue.take(fixture.reads); + expect(yield* Queue.take(fixture.reads)).toBe(current.id); yield* reactor.drain; } expect((yield* Ref.get(fixture.commands))[0]?.branchPullRequest).toEqual(reference(42)); @@ -388,6 +406,51 @@ describe("ThreadPullRequestReactor", () => { ), ); + it.effect("refreshes the project identity when a turn adds the remote", () => + Effect.scoped( + Effect.gen(function* () { + const current = thread("new-remote"); + const fixture = yield* makeHarness({ + threads: [current], + project: { ...project, repositoryIdentity: null }, + branchPullRequest: () => Effect.succeed(branchPullRequest()), + resolveRepositoryIdentity: (_cwd, options) => + Effect.succeed(options?.refresh ? project.repositoryIdentity : null), + }); + yield* Effect.gen(function* () { + const reactor = yield* fixture.start(); + expect(yield* Ref.get(fixture.commands)).toHaveLength(0); + + yield* fixture.publish({ + type: "thread.turn-diff-completed", + sequence: 2, + eventId: EventId.make("checkpoint-finished"), + aggregateKind: "thread", + aggregateId: current.id, + occurredAt: NOW, + commandId: null, + causationEventId: null, + correlationId: null, + metadata: {}, + payload: { + threadId: current.id, + turnId: TurnId.make("turn"), + checkpointTurnCount: 1, + checkpointRef: CheckpointRef.make("checkpoint"), + status: "ready", + files: [], + assistantMessageId: null, + completedAt: NOW, + }, + }); + yield* Queue.take(fixture.reads); + yield* reactor.drain; + expect((yield* Ref.get(fixture.commands))[0]?.branchPullRequest).toEqual(reference(42)); + }).pipe(Effect.provide(fixture.layer)); + }), + ), + ); + it.effect("uses live worktrees and falls back to the project for removed worktrees", () => Effect.scoped( Effect.gen(function* () { @@ -516,6 +579,23 @@ describe("ThreadPullRequestReactor", () => { yield* Effect.gen(function* () { const reactor = yield* fixture.start(); expect(yield* Ref.get(fixture.commands)).toHaveLength(0); + // A one-thread read cannot show that other pending threads are gone. + const gone = ThreadId.make("gone"); + yield* fixture.publish({ + type: "thread.unarchived", + sequence: 2, + eventId: EventId.make("gone-unarchived"), + aggregateKind: "thread", + aggregateId: gone, + occurredAt: NOW, + commandId: null, + causationEventId: null, + correlationId: null, + metadata: {}, + payload: { threadId: gone, updatedAt: NOW }, + }); + expect(yield* Queue.take(fixture.reads)).toBe(gone); + yield* reactor.drain; yield* Ref.set(online, true); yield* TestClock.adjust("1 minute"); yield* Queue.take(fixture.reads); diff --git a/apps/server/src/orchestration/ThreadPullRequestReactor.ts b/apps/server/src/orchestration/ThreadPullRequestReactor.ts index 17efc74e4f40..2d9d92d64364 100644 --- a/apps/server/src/orchestration/ThreadPullRequestReactor.ts +++ b/apps/server/src/orchestration/ThreadPullRequestReactor.ts @@ -6,6 +6,7 @@ import { CommandId, type OrchestrationEvent, type OrchestrationProjectShell, + type OrchestrationShellSnapshot, type ThreadId, type ThreadLinkedPullRequest, } from "@t3tools/contracts"; @@ -16,11 +17,13 @@ import * as Crypto from "effect/Crypto"; 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 Schedule from "effect/Schedule"; import type * as Scope from "effect/Scope"; import * as Stream from "effect/Stream"; import * as GitManager from "../git/GitManager.ts"; +import type { ProjectionRepositoryError } from "../persistence/Errors.ts"; import * as PullRequestService from "../pullRequest/PullRequestService.ts"; import * as RepositoryIdentityResolver from "../project/RepositoryIdentityResolver.ts"; import { forkParked } from "../serverActivation.ts"; @@ -69,6 +72,35 @@ export function pullRequestMatchesProject( ); } +/** + * Read the shell state for a discovery or settlement sweep. A sweep for one + * thread reads that thread and the projects it names, not every thread. + */ +export const readSweepSnapshot = ( + snapshots: ProjectionSnapshotQuery.ProjectionSnapshotQueryShape, + threadId: ThreadId | null, +): Effect.Effect< + Pick, + ProjectionRepositoryError +> => + threadId === null + ? snapshots.getShellSnapshot() + : Effect.gen(function* () { + // Read the sequence first. The thread is then at least this new, so a + // command guarded by the sequence is rejected rather than missing a change. + const { snapshotSequence } = yield* snapshots.getSnapshotSequence(); + const thread = yield* snapshots.getThreadShellById(threadId); + if (Option.isNone(thread)) return { snapshotSequence, projects: [], threads: [] }; + // Settlement also checks the project a saved pull request names. + const reference = thread.value.linkedPullRequest ?? thread.value.branchPullRequest; + const projects = yield* snapshots.getProjectShells( + reference == null + ? [thread.value.projectId] + : [thread.value.projectId, reference.projectId], + ); + return { snapshotSequence, projects, threads: [thread.value] }; + }); + /** @public Service construction is part of the canonical Effect module API. */ export const make = Effect.gen(function* () { const engine = yield* OrchestrationEngine.OrchestrationEngineService; @@ -97,7 +129,7 @@ export const make = Effect.gen(function* () { const synchronize = Effect.fn("ThreadPullRequestReactor.synchronize")(function* ( request: RefreshRequest, ) { - const snapshot = yield* snapshots.getShellSnapshot(); + const snapshot = yield* readSweepSnapshot(snapshots, request.threadId); const projects = new Map(snapshot.projects.map((project) => [project.id, project])); if (request.backfill) { for (const thread of snapshot.threads) { @@ -109,14 +141,15 @@ export const make = Effect.gen(function* () { } } } + // A single-thread read only shows whether its own thread is gone. const threadIds = new Set(snapshot.threads.map((thread) => thread.id)); - for (const threadId of pendingBackfill.keys()) { + const checkedIds = request.threadId === null ? pendingBackfill.keys() : [request.threadId]; + for (const threadId of checkedIds) { if (!threadIds.has(threadId)) pendingBackfill.delete(threadId); } const threads = snapshot.threads.filter( (thread) => thread.archivedAt === null && - (request.threadId === null || thread.id === request.threadId) && ((thread.settledOverride !== "settled" && thread.settledAt === null) || request.threadId !== null || pendingBackfill.has(thread.id)) && @@ -131,8 +164,19 @@ export const make = Effect.gen(function* () { (group) => Effect.gen(function* () { const first = group[0]!; - const project = projects.get(first.projectId); - if (project === undefined) return finishBackfill(group); + const snapshotProject = projects.get(first.projectId); + if (snapshotProject === undefined) return finishBackfill(group); + // A finished turn may have added the remote this PR lives on. A failed + // refresh resolves to null, so keep the snapshot's identity then. + const project = request.refresh + ? { + ...snapshotProject, + repositoryIdentity: + (yield* repositoryIdentities.resolve(snapshotProject.workspaceRoot, { + refresh: true, + })) ?? snapshotProject.repositoryIdentity, + } + : snapshotProject; const repository = sourceControlRepositorySelector(project.repositoryIdentity); if (first.branch !== null && repository === null) return finishBackfill(group); const worktreeExists = diff --git a/apps/server/src/orchestration/ThreadSettlementReactor.test.ts b/apps/server/src/orchestration/ThreadSettlementReactor.test.ts index 07afbaf05e6a..bee487ceb721 100644 --- a/apps/server/src/orchestration/ThreadSettlementReactor.test.ts +++ b/apps/server/src/orchestration/ThreadSettlementReactor.test.ts @@ -24,6 +24,7 @@ import * as Effect from "effect/Effect"; import * as Fiber from "effect/Fiber"; import * as FileSystem from "effect/FileSystem"; import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; import * as PubSub from "effect/PubSub"; import * as Queue from "effect/Queue"; import * as Ref from "effect/Ref"; @@ -181,7 +182,8 @@ const makeHarness = Effect.fn("makeThreadSettlementHarness")(function* (options: const activation = yield* Deferred.make(); const snapshots = yield* Ref.make(options.snapshot); const snapshotReadCount = yield* Ref.make(0); - const snapshotReads = yield* Queue.unbounded(); + // Each shell read: a thread id for a one-thread read, null for a full read. + const snapshotReads = yield* Queue.unbounded(); const settings = yield* Ref.make(options.settings ?? DEFAULT_SERVER_SETTINGS); const settingsReads = yield* Queue.unbounded(); const settingsChanges = yield* PubSub.unbounded(); @@ -255,10 +257,27 @@ const makeHarness = Effect.fn("makeThreadSettlementHarness")(function* (options: const dependencies = Layer.mergeAll( Layer.mock(ProjectionSnapshotQuery)({ getShellSnapshot: () => - Ref.updateAndGet(snapshotReadCount, (count) => count + 1).pipe( - Effect.tap((count) => Queue.offer(snapshotReads, count)), + Ref.update(snapshotReadCount, (count) => count + 1).pipe( + Effect.andThen(Queue.offer(snapshotReads, null)), Effect.andThen(Ref.get(snapshots)), ), + getSnapshotSequence: () => + Ref.get(snapshots).pipe(Effect.map(({ snapshotSequence }) => ({ snapshotSequence }))), + getThreadShellById: (threadId) => + Ref.get(snapshots).pipe( + Effect.map(({ threads }) => + Option.fromUndefinedOr( + threads.find((thread) => thread.id === threadId && thread.archivedAt === null), + ), + ), + Effect.tap(() => Queue.offer(snapshotReads, threadId)), + ), + getProjectShells: (projectIds) => + Ref.get(snapshots).pipe( + Effect.map(({ projects }) => + projects.filter((project) => projectIds?.includes(project.id) ?? true), + ), + ), }), Layer.mock(GitManager)({ branchPullRequest, @@ -313,7 +332,7 @@ const makeHarness = Effect.fn("makeThreadSettlementHarness")(function* (options: const startHarness = Effect.fn("startThreadSettlementHarness")(function* ( reactor: ThreadSettlementReactor.ThreadSettlementReactor["Service"], activation: Deferred.Deferred, - snapshotReads: Queue.Queue, + snapshotReads: Queue.Queue, ) { yield* reactor.start(); yield* Deferred.succeed(activation, undefined); @@ -443,7 +462,7 @@ describe("ThreadSettlementReactor", () => { updatedAt: NOW, }, }); - yield* Queue.take(fixture.snapshotReads); + assert.strictEqual(yield* Queue.take(fixture.snapshotReads), thread.id); yield* reactor.drain; } assert.deepStrictEqual( @@ -463,7 +482,7 @@ describe("ThreadSettlementReactor", () => { aggregateId: readySession.threadId, payload: { threadId: readySession.threadId, session: readySession }, }); - yield* Queue.take(fixture.snapshotReads); + assert.strictEqual(yield* Queue.take(fixture.snapshotReads), readySession.threadId); yield* reactor.drain; assert.deepStrictEqual( (yield* Ref.get(fixture.commands)).map(({ threadId }) => threadId), diff --git a/apps/server/src/orchestration/ThreadSettlementReactor.ts b/apps/server/src/orchestration/ThreadSettlementReactor.ts index 4192896efed5..22d21ff27996 100644 --- a/apps/server/src/orchestration/ThreadSettlementReactor.ts +++ b/apps/server/src/orchestration/ThreadSettlementReactor.ts @@ -23,7 +23,7 @@ import * as ServerSettings from "../serverSettings.ts"; import { forkParked } from "../serverActivation.ts"; import * as OrchestrationEngine from "./Services/OrchestrationEngine.ts"; import * as ProjectionSnapshotQuery from "./Services/ProjectionSnapshotQuery.ts"; -import { pullRequestMatchesProject } from "./ThreadPullRequestReactor.ts"; +import { pullRequestMatchesProject, readSweepSnapshot } from "./ThreadPullRequestReactor.ts"; import { isAutoSettlementCandidate, resolveAutoSettlementAt, @@ -95,20 +95,16 @@ export const make = Effect.gen(function* () { if (!autoSettlementConfigured(settings)) { return; } - const snapshot = yield* snapshots.getShellSnapshot(); + const snapshot = yield* readSweepSnapshot(snapshots, threadId ?? null); const now = DateTime.formatIso(yield* DateTime.now); const projects = new Map(snapshot.projects.map((project) => [project.id, project])); // A merge rechecks all candidates, including branches that discovery has // not linked yet. Those lookups can still have cached the PR as open. - const candidates = snapshot.threads.filter( - (thread) => - (threadId === undefined || thread.id === threadId) && - isAutoSettlementCandidate(thread, now), - ); + const candidates = snapshot.threads.filter((thread) => isAutoSettlementCandidate(thread, now)); // Return the thread when it still needs a pull request decision. A rejected // dispatch skips it for this snapshot instead of retrying through a lookup. - const settleThread = Effect.fn("ThreadSettlementReactor.settleThread")( + const settleThread = Effect.fnUntraced( function* (thread: (typeof candidates)[number], pullRequest: SettlementPullRequest | null) { const settings = resolveProjectSettings( yield* settingsService.getSettings, diff --git a/apps/server/src/orchestration/projector.ts b/apps/server/src/orchestration/projector.ts index 85d9db3fdfed..dd418b95b9bc 100644 --- a/apps/server/src/orchestration/projector.ts +++ b/apps/server/src/orchestration/projector.ts @@ -117,12 +117,26 @@ function settledTurnStateForSessionStatus( } } +// Runs for every thread event (including streaming deltas) against every +// thread the server has ever seen, so copy the array rather than map it. function updateThread( threads: ReadonlyArray, threadId: ThreadId, patch: ThreadPatch, -): OrchestrationThread[] { - return threads.map((thread) => (thread.id === threadId ? { ...thread, ...patch } : thread)); +): ReadonlyArray { + const index = threads.findIndex((thread) => thread.id === threadId); + return index === -1 ? threads : patchThreadAt(threads, index, patch); +} + +/** For callers that already located the thread and must not scan again. */ +function patchThreadAt( + threads: ReadonlyArray, + index: number, + patch: ThreadPatch, +): ReadonlyArray { + const next = threads.slice(); + next[index] = { ...threads[index]!, ...patch }; + return next; } /** Patch that swaps a thread's links and re-derives the legacy single-PR field from them. */ @@ -767,7 +781,8 @@ export function projectEvent( event.type, "payload", ); - const thread = nextBase.threads.find((entry) => entry.id === payload.threadId); + const threadIndex = nextBase.threads.findIndex((entry) => entry.id === payload.threadId); + const thread = nextBase.threads[threadIndex]; if (!thread) { return nextBase; } @@ -815,7 +830,7 @@ export function projectEvent( return { ...nextBase, - threads: updateThread(nextBase.threads, payload.threadId, { + threads: patchThreadAt(nextBase.threads, threadIndex, { messages: cappedMessages, updatedAt: event.occurredAt, }), @@ -1055,7 +1070,8 @@ export function projectEvent( "payload", ).pipe( Effect.map((payload) => { - const thread = nextBase.threads.find((entry) => entry.id === payload.threadId); + const threadIndex = nextBase.threads.findIndex((entry) => entry.id === payload.threadId); + const thread = nextBase.threads[threadIndex]; if (!thread) { return nextBase; } @@ -1069,7 +1085,7 @@ export function projectEvent( return { ...nextBase, - threads: updateThread(nextBase.threads, payload.threadId, { + threads: patchThreadAt(nextBase.threads, threadIndex, { activities, updatedAt: event.occurredAt, }), diff --git a/apps/server/src/persistence/Layers/Sqlite.test.ts b/apps/server/src/persistence/Layers/Sqlite.test.ts index 0b64e4f7fdcb..5bcbd35e918c 100644 --- a/apps/server/src/persistence/Layers/Sqlite.test.ts +++ b/apps/server/src/persistence/Layers/Sqlite.test.ts @@ -10,7 +10,11 @@ import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; import * as SqlClient from "effect/unstable/sql/SqlClient"; -import { SqlitePersistenceMemory, makeSqlitePersistenceLive } from "./Sqlite.ts"; +import { + SqlitePersistenceMemory, + WAL_SIZE_LIMIT_BYTES, + makeSqlitePersistenceLive, +} from "./Sqlite.ts"; const lockHolderSource = ` const { DatabaseSync } = require("node:sqlite"); @@ -57,6 +61,32 @@ it.effect("waits out a concurrent writer instead of failing with SQLITE_BUSY", ( ); }); +it.effect("shrinks the WAL file back to the size limit after a large write", () => { + const tempDir = NodeFS.mkdtempSync(NodePath.join(NodeOS.tmpdir(), "t3-sqlite-wal-")); + const dbPath = NodePath.join(tempDir, "state.sqlite"); + const walFileSize = () => NodeFS.statSync(`${dbPath}-wal`).size; + // About 25% more 4 KB rows than the limit holds, in one transaction. + const rowCount = Math.ceil((WAL_SIZE_LIMIT_BYTES * 1.25) / 4000); + + return Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + yield* sql`CREATE TABLE wal_probe(payload BLOB)`; + yield* sql` + WITH RECURSIVE n(i) AS (SELECT 1 UNION ALL SELECT i + 1 FROM n WHERE i < ${rowCount}) + INSERT INTO wal_probe(payload) SELECT randomblob(4000) FROM n + `; + assert.isAbove(walFileSize(), WAL_SIZE_LIMIT_BYTES); + + // The auto-checkpoint after the large commit copied every frame into the + // database, so the next commit restarts the WAL and cuts the file back. + yield* sql`INSERT INTO wal_probe(payload) VALUES (x'00')`; + assert.isAtMost(walFileSize(), WAL_SIZE_LIMIT_BYTES); + }).pipe( + Effect.provide(makeSqlitePersistenceLive(dbPath).pipe(Layer.provide(NodeServices.layer))), + Effect.ensuring(Effect.sync(() => NodeFS.rmSync(tempDir, { recursive: true, force: true }))), + ); +}); + it.effect("applies busy_timeout in the shared persistence setup", () => Effect.gen(function* () { const sql = yield* SqlClient.SqlClient; diff --git a/apps/server/src/persistence/Layers/Sqlite.ts b/apps/server/src/persistence/Layers/Sqlite.ts index 032b44645089..56536087d69c 100644 --- a/apps/server/src/persistence/Layers/Sqlite.ts +++ b/apps/server/src/persistence/Layers/Sqlite.ts @@ -8,6 +8,9 @@ import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient"; import { runMigrations } from "../Migrations.ts"; import { ServerConfig } from "../../config.ts"; +// Size the -wal file is cut back to on the first commit after a WAL reset. +export const WAL_SIZE_LIMIT_BYTES = 32 * 1024 * 1024; + const setup = Layer.effectDiscard( Effect.gen(function* () { const sql = yield* SqlClient.SqlClient; @@ -15,6 +18,9 @@ const setup = Layer.effectDiscard( yield* sql`PRAGMA busy_timeout = 5000;`; yield* sql`PRAGMA foreign_keys = ON;`; yield* sql`PRAGMA journal_mode = WAL;`; + // PASSIVE checkpoints never shrink the -wal file, so it otherwise keeps its + // largest size until the last connection closes. + yield* sql.unsafe(`PRAGMA journal_size_limit = ${WAL_SIZE_LIMIT_BYTES};`); yield* runMigrations(); }), ); diff --git a/apps/server/src/process/externalLauncher.test.ts b/apps/server/src/process/externalLauncher.test.ts index f714a70f783d..aab4d78f4d71 100644 --- a/apps/server/src/process/externalLauncher.test.ts +++ b/apps/server/src/process/externalLauncher.test.ts @@ -923,6 +923,18 @@ it.effect("discovers editors through the service API", () => ); for (const { platform, installPath, editor, args } of [ + { + platform: "darwin", + installPath: "Applications/Antigravity IDE.app/Contents/Resources/app/bin/antigravity-ide", + editor: "antigravity", + args: ["--goto", "/workspace with spaces/file.ts:12:4"], + }, + { + platform: "linux", + installPath: ".local/bin/antigravity-ide", + editor: "antigravity", + args: ["--goto", "/workspace with spaces/file.ts:12:4"], + }, { platform: "darwin", installPath: "Applications/Cursor.app/Contents/Resources/app/bin/code", @@ -1028,6 +1040,45 @@ for (const { platform, installPath, editor, args } of [ ); } +// `agy` is the standalone Antigravity CLI, which installs to ~/.local/bin on +// macOS and Linux and to its own bin folder on Windows. It is not the IDE. +for (const { platform, installPath, onPath } of [ + { platform: "darwin", installPath: ".local/bin/agy", onPath: true }, + { platform: "linux", installPath: ".local/bin/agy", onPath: false }, + { platform: "win32", installPath: "agy/bin/agy.cmd", onPath: true }, +] as const) { + it.effect.skipIf(windowsHost && platform !== "win32")( + `does not report the agy CLI as the Antigravity IDE on ${platform}`, + () => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const home = yield* fs.makeTempDirectoryScoped({ prefix: "t3-agy-cli-" }); + const executable = path.join(home, installPath); + yield* fs.makeDirectory(path.dirname(executable), { recursive: true }); + yield* fs.writeFileString(executable, "#!/bin/sh\n"); + yield* fs.chmod(executable, 0o755); + const editors = yield* Effect.gen(function* () { + const launcher = yield* ExternalLauncher.ExternalLauncher; + return yield* launcher.resolveAvailableEditors(); + }).pipe( + Effect.provide( + testLayer({ + platform, + env: { + HOME: home, + LOCALAPPDATA: home, + PATH: onPath ? path.dirname(executable) : path.join(home, "empty"), + PATHEXT: ".COM;.EXE;.BAT;.CMD", + }, + }), + ), + ); + assert.notInclude(editors, "antigravity"); + }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)), + ); +} + it.effect.skipIf(windowsHost)("ignores unusable app bundles and keeps PATH launchers first", () => Effect.gen(function* () { const fs = yield* FileSystem.FileSystem; diff --git a/apps/server/src/process/externalLauncher.ts b/apps/server/src/process/externalLauncher.ts index 29c25e790c61..ab287e4d78ee 100644 --- a/apps/server/src/process/externalLauncher.ts +++ b/apps/server/src/process/externalLauncher.ts @@ -20,7 +20,11 @@ import { } from "@t3tools/contracts"; import { resolveEditorCommand } from "@t3tools/shared/editor"; import { HostProcessPlatform } from "@t3tools/shared/hostProcess"; -import { isCommandAvailable, resolveSpawnCommand } from "@t3tools/shared/shell"; +import { + isCommandAvailable, + resolveSpawnCommand, + withPathDirectoryListings, +} from "@t3tools/shared/shell"; import * as Clock from "effect/Clock"; import * as Config from "effect/Config"; import * as Context from "effect/Context"; @@ -442,7 +446,7 @@ const resolveBrowserLaunch = Effect.fn("externalLauncher.resolveBrowserLaunch")( const resolveAvailableEditors = Effect.fn("externalLauncher.resolveAvailableEditors")(function* () { const platform = yield* HostProcessPlatform; const env = { ...(yield* readBrowserLaunchEnv), ...(yield* readCommandLookupEnv) }; - return yield* buildAvailableEditors(platform, env); + return yield* buildAvailableEditors(platform, env).pipe(withPathDirectoryListings); }); const resolveFileManagerRevealKind = Effect.fn("externalLauncher.resolveFileManagerRevealKind")( diff --git a/apps/server/src/processRunner.test.ts b/apps/server/src/processRunner.test.ts index e264ba7849da..d7f3799e103b 100644 --- a/apps/server/src/processRunner.test.ts +++ b/apps/server/src/processRunner.test.ts @@ -412,3 +412,11 @@ describe("isWindowsCommandNotFound", () => { }), ); }); + +describe("commandName", () => { + it("drops the directory from POSIX and Windows paths", () => { + expect(ProcessRunner.commandName("/Users/me/.local/bin/claude")).toBe("claude"); + expect(ProcessRunner.commandName("C:\\Program Files\\nodejs\\npx.cmd")).toBe("npx.cmd"); + expect(ProcessRunner.commandName("git")).toBe("git"); + }); +}); diff --git a/apps/server/src/processRunner.ts b/apps/server/src/processRunner.ts index 0a9bb9b04a43..36bb5b649f06 100644 --- a/apps/server/src/processRunner.ts +++ b/apps/server/src/processRunner.ts @@ -285,10 +285,14 @@ function finalizeRunProcess( ); } +/** The executable name without its directory, recorded as `process.command` on process spans. */ +export const commandName = (command: string) => command.replace(/^.*[\\/]/, ""); + const runProcessCore = Effect.fn("processRunner.runProcessCore")(function* ( spawner: ChildProcessSpawner.ChildProcessSpawner["Service"], input: ProcessRunInput, ): Effect.fn.Return { + yield* Effect.annotateCurrentSpan("process.command", commandName(input.command)); const maxOutputBytes = input.maxOutputBytes ?? DEFAULT_MAX_OUTPUT_BYTES; const outputMode = input.outputMode ?? "error"; const truncatedMarker = input.truncatedMarker ?? ""; diff --git a/apps/server/src/project/AgentSessionScanner.test.ts b/apps/server/src/project/AgentSessionScanner.test.ts index 792ef92310ff..2dde0643ed60 100644 --- a/apps/server/src/project/AgentSessionScanner.test.ts +++ b/apps/server/src/project/AgentSessionScanner.test.ts @@ -48,6 +48,7 @@ const makeProjectionSnapshotQueryLayer = (importedWorkspaceRoots: ReadonlyArray< updatedAt: "2026-01-01T00:00:00.000Z", }), getDeletedWorktreeThreads: () => Effect.die("unused"), + listThreadsWithPullRequests: () => Effect.die("unused"), getArchivedShellSnapshot: () => Effect.die("unused"), getSnapshotSequence: () => Effect.die("unused"), getCounts: () => Effect.die("unused"), diff --git a/apps/server/src/project/ProjectSetupScriptRunner.test.ts b/apps/server/src/project/ProjectSetupScriptRunner.test.ts index 650f304e8481..69c553a4deaf 100644 --- a/apps/server/src/project/ProjectSetupScriptRunner.test.ts +++ b/apps/server/src/project/ProjectSetupScriptRunner.test.ts @@ -34,6 +34,7 @@ const makeProjectionSnapshotQueryLayer = (project: OrchestrationProject) => getSnapshot: () => Effect.die("unused"), getShellSnapshot: () => Effect.die("unused"), getDeletedWorktreeThreads: () => Effect.die("unused"), + listThreadsWithPullRequests: () => Effect.die("unused"), getArchivedShellSnapshot: () => Effect.die("unused"), getSnapshotSequence: () => Effect.succeed({ snapshotSequence: 1 }), getCounts: () => Effect.die("unused"), diff --git a/apps/server/src/project/RepositoryIdentityResolver.test.ts b/apps/server/src/project/RepositoryIdentityResolver.test.ts index d6ddb0b9263f..58f199b834e2 100644 --- a/apps/server/src/project/RepositoryIdentityResolver.test.ts +++ b/apps/server/src/project/RepositoryIdentityResolver.test.ts @@ -94,6 +94,8 @@ it.layer(NodeServices.layer)("RepositoryIdentityResolverLive", (it) => { const resolver = yield* RepositoryIdentityResolver.RepositoryIdentityResolver; const first = yield* resolver.resolve("/repo/packages/web"); rootPath = "/repo/packages/web"; + // Longer than the one-minute cadence of the background sweeps. + yield* TestClock.adjust(Duration.minutes(10)); const second = yield* resolver.resolve("/repo/packages/web"); expect(first?.canonicalKey).toBe("github.com/t3tools/t3code"); @@ -123,10 +125,10 @@ it.layer(NodeServices.layer)("RepositoryIdentityResolverLive", (it) => { const unavailable = yield* resolver.resolve(rootPath, { refresh: true }); expect(unavailable?.webUrl).toBeUndefined(); expect(unavailable?.canonicalKey).toBe("ssh.forge.test/team/repo"); - }).pipe(Effect.provide(resolverLayer)); + }).pipe(Effect.provide(Layer.merge(TestClock.layer(), resolverLayer))); }); - it.effect("retries Git root discovery after a failed lookup", () => { + it.effect("retries Git root discovery after the negative TTL", () => { const calls: Array> = []; let rootAttempts = 0; const processRunner = Layer.succeed(ProcessRunner.ProcessRunner, { @@ -159,7 +161,9 @@ it.layer(NodeServices.layer)("RepositoryIdentityResolverLive", (it) => { return Effect.gen(function* () { const resolver = yield* RepositoryIdentityResolver.RepositoryIdentityResolver; expect(yield* resolver.resolve("/repo/packages/web")).toBeNull(); + expect(yield* resolver.resolve("/repo/packages/web")).toBeNull(); + yield* TestClock.adjust(Duration.minutes(1)); const recovered = yield* resolver.resolve("/repo/packages/web"); expect(recovered?.rootPath).toBe("/repo"); expect(calls).toEqual([ @@ -167,7 +171,7 @@ it.layer(NodeServices.layer)("RepositoryIdentityResolverLive", (it) => { ["-C", "/repo/packages/web", "rev-parse", "--show-toplevel"], ["-C", "/repo", "remote", "-v"], ]); - }).pipe(Effect.provide(resolverLayer)); + }).pipe(Effect.provide(Layer.merge(TestClock.layer(), resolverLayer))); }); it.effect("normalizes equivalent GitHub remotes into a stable repository identity", () => diff --git a/apps/server/src/project/RepositoryIdentityResolver.ts b/apps/server/src/project/RepositoryIdentityResolver.ts index 2d7f5d02d02e..5acafa47e2e2 100644 --- a/apps/server/src/project/RepositoryIdentityResolver.ts +++ b/apps/server/src/project/RepositoryIdentityResolver.ts @@ -13,7 +13,11 @@ import * as Layer from "effect/Layer"; import * as ProcessRunner from "../processRunner.ts"; const DEFAULT_REPOSITORY_IDENTITY_CACHE_CAPACITY = 512; -const DEFAULT_POSITIVE_CACHE_TTL = Duration.minutes(1); +// Background sweeps resolve every project each minute. A long TTL keeps them +// from spawning git each time. Clone, publish, and PR discovery (after a turn +// and before it saves links) resolve with `refresh: true`. +const DEFAULT_POSITIVE_CACHE_TTL = Duration.minutes(15); +// Short, so a folder that gains a repository or a remote shows up quickly. const DEFAULT_NEGATIVE_CACHE_TTL = Duration.minutes(1); export interface RepositoryIdentityResolverOptions { @@ -142,20 +146,23 @@ export const make = Effect.fn("RepositoryIdentityResolver.make")(function* ( const processRunner = yield* ProcessRunner.ProcessRunner; const cacheCapacity = options.cacheCapacity ?? DEFAULT_REPOSITORY_IDENTITY_CACHE_CAPACITY; const refine = options.refine ?? Effect.succeed; + // Git errors and timeouts resolve to null, so they use the negative TTL like + // "no repository" or "no remote". Only interrupts and defects skip the cache. + const timeToLive = (exit: Exit.Exit) => + Exit.match(exit, { + onSuccess: (value) => + value === null + ? (options.negativeCacheTtl ?? DEFAULT_NEGATIVE_CACHE_TTL) + : (options.positiveCacheTtl ?? DEFAULT_POSITIVE_CACHE_TTL), + onFailure: () => Duration.zero, + }); const repositoryRootCache = yield* Cache.makeWith( (cwd) => resolveRepositoryIdentityCacheKey(cwd).pipe( Effect.provideService(ProcessRunner.ProcessRunner, processRunner), ), - { - capacity: cacheCapacity, - timeToLive: Exit.match({ - onSuccess: (value) => - value === null ? Duration.zero : (options.positiveCacheTtl ?? DEFAULT_POSITIVE_CACHE_TTL), - onFailure: () => Duration.zero, - }), - }, + { capacity: cacheCapacity, timeToLive }, ); const repositoryIdentityCache = yield* Cache.makeWith( @@ -167,27 +174,20 @@ export const make = Effect.fn("RepositoryIdentityResolver.make")(function* ( (identity) => refine(identity).pipe(Effect.orElseSucceed(() => identity)), ), ), - { - capacity: cacheCapacity, - timeToLive: Exit.match({ - onSuccess: (value) => - value === null - ? (options.negativeCacheTtl ?? DEFAULT_NEGATIVE_CACHE_TTL) - : (options.positiveCacheTtl ?? DEFAULT_POSITIVE_CACHE_TTL), - onFailure: () => Duration.zero, - }), - }, + { capacity: cacheCapacity, timeToLive }, ); - const resolve: RepositoryIdentityResolver["Service"]["resolve"] = Effect.fn( - "RepositoryIdentityResolver.resolve", - )(function* (cwd, options) { - if (options?.refresh) yield* Cache.invalidate(repositoryRootCache, cwd); - const cacheKey = yield* Cache.get(repositoryRootCache, cwd); - if (cacheKey === null) return null; - if (options?.refresh) yield* Cache.invalidate(repositoryIdentityCache, cacheKey); - return yield* Cache.get(repositoryIdentityCache, cacheKey); - }); + // Untraced because almost every call is a cache hit. The lookups that spawn + // git keep their own spans. + const resolve: RepositoryIdentityResolver["Service"]["resolve"] = Effect.fnUntraced( + function* (cwd, options) { + if (options?.refresh) yield* Cache.invalidate(repositoryRootCache, cwd); + const cacheKey = yield* Cache.get(repositoryRootCache, cwd); + if (cacheKey === null) return null; + if (options?.refresh) yield* Cache.invalidate(repositoryIdentityCache, cacheKey); + return yield* Cache.get(repositoryIdentityCache, cacheKey); + }, + ); return RepositoryIdentityResolver.of({ resolve }); }); diff --git a/apps/server/src/provider/Layers/ClaudeAdapter.test.ts b/apps/server/src/provider/Layers/ClaudeAdapter.test.ts index 502620359836..295926b26644 100644 --- a/apps/server/src/provider/Layers/ClaudeAdapter.test.ts +++ b/apps/server/src/provider/Layers/ClaudeAdapter.test.ts @@ -6691,6 +6691,56 @@ describe("ClaudeAdapterLive", () => { ); }); + it.effect("completed turns keep their ids but not the SDK messages", () => { + const harness = makeHarness(); + return Effect.gen(function* () { + const adapter = yield* ClaudeAdapter; + const session = yield* adapter.startSession({ + threadId: THREAD_ID, + provider: ProviderDriverKind.make("claudeAgent"), + runtimeMode: "full-access", + }); + const turn = yield* adapter.sendTurn({ + threadId: session.threadId, + input: "hello", + attachments: [], + }); + const completedFiber = yield* Stream.filter( + adapter.streamEvents, + (event) => event.type === "turn.completed", + ).pipe(Stream.runHead, Effect.forkChild); + + harness.query.emit({ + type: "assistant", + session_id: "sdk-session-1", + uuid: "assistant-1", + parent_tool_use_id: null, + message: { + id: "assistant-message-1", + content: [{ type: "text", text: "Hi" }], + }, + } as unknown as SDKMessage); + harness.query.emit({ + type: "result", + subtype: "success", + is_error: false, + errors: [], + session_id: "sdk-session-1", + uuid: "result-1", + } as unknown as SDKMessage); + yield* Fiber.join(completedFiber); + + const snapshot = yield* adapter.readThread(session.threadId); + assert.deepEqual( + snapshot.turns.map((entry) => ({ id: String(entry.id), items: entry.items })), + [{ id: String(turn.turnId), items: [] }], + ); + }).pipe( + Effect.provideService(Random.Random, makeDeterministicRandomService()), + Effect.provide(harness.layer), + ); + }); + it.effect("rewinds Claude history when the fork omits retained system messages", () => { const forkCalls: Array>> = []; let firstTurnId = ""; diff --git a/apps/server/src/provider/Layers/ClaudeAdapter.ts b/apps/server/src/provider/Layers/ClaudeAdapter.ts index 11a5322b4eee..6e30f5241780 100644 --- a/apps/server/src/provider/Layers/ClaudeAdapter.ts +++ b/apps/server/src/provider/Layers/ClaudeAdapter.ts @@ -267,7 +267,6 @@ interface ClaudeTurnState { * steered instead (the queued message continues the same turn). */ readonly synthetic?: boolean; - readonly items: Array; readonly assistantTextBlocks: Map; readonly assistantTextBlockOrder: Array; readonly capturedProposedPlanKeys: Set; @@ -422,10 +421,11 @@ interface ClaudeSessionContext { resumeSessionId: string | undefined; readonly pendingApprovals: Map; readonly pendingUserInputs: Map; - readonly turns: Array<{ - id: TurnId; - items: Array; - }>; + /** Completed turn ids, reported by readThread and trimmed on rollback. + * SDK messages are not kept: rollback reads Claude's own history through + * turnStartMessageIds, and a long-lived session would otherwise hold every + * message it ever produced. */ + readonly turns: Array<{ readonly id: TurnId }>; readonly inFlightTools: Map; readonly claudeTasks: Map; readonly taskAgents: Map; @@ -2176,10 +2176,7 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( } return { threadId, - turns: context.turns.map((turn) => ({ - id: turn.id, - items: [...turn.items], - })), + turns: context.turns.map((turn) => ({ id: turn.id, items: [] })), }; }); @@ -2808,10 +2805,7 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( }); } - context.turns.push({ - id: turnState.turnId, - items: [...turnState.items], - }); + context.turns.push({ id: turnState.turnId }); yield* emitThreadTokenUsage(context, usageSnapshot, { rawMethod: "claude/result", @@ -3175,10 +3169,6 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( return; } - if (context.turnState) { - context.turnState.items.push(message.message); - } - for (const toolResult of toolResultBlocksFromUserMessage(message)) { const toolEntry = Array.from(context.inFlightTools.entries()).find( ([, tool]) => tool.itemId === toolResult.toolUseId, @@ -3378,7 +3368,6 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( turnId, startedAt, synthetic: true, - items: [], assistantTextBlocks: new Map(), assistantTextBlockOrder: [], capturedProposedPlanKeys: new Set(), @@ -3461,7 +3450,6 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( cwd: path.resolve(context.session.cwd ?? "."), }); } - context.turnState.items.push(message.message); if ( normalizeClaudeActiveTokenUsage( message.message.usage, @@ -5205,7 +5193,6 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( const turnState: ClaudeTurnState = { turnId, startedAt: yield* nowIso, - items: [], assistantTextBlocks: new Map(), assistantTextBlockOrder: [], capturedProposedPlanKeys: new Set(), diff --git a/apps/server/src/provider/Layers/ProviderService.test.ts b/apps/server/src/provider/Layers/ProviderService.test.ts index 36d884d8ac98..242d0b55aa2f 100644 --- a/apps/server/src/provider/Layers/ProviderService.test.ts +++ b/apps/server/src/provider/Layers/ProviderService.test.ts @@ -648,6 +648,78 @@ it.effect("ProviderServiceLive catches stopAll failures during shutdown", () => }), ); +it.effect("ProviderServiceLive shutdown leaves settled session rows untouched", () => + Effect.gen(function* () { + const recordedAnalytics = makeRecordingAnalytics(); + const codex = makeFakeCodexAdapter(); + const persistence = yield* Layer.build( + ProviderSessionDirectoryLive.pipe( + Layer.provide(ProviderSessionRuntime.layer.pipe(Layer.provide(SqlitePersistenceMemory))), + ), + ); + const directory = yield* ProviderSessionDirectory.ProviderSessionDirectory.pipe( + Effect.provide(persistence), + ); + const seed = (threadId: ThreadId, status: "running" | "stopped", activeTurnId: TurnId | null) => + directory.upsert({ + threadId, + provider: CODEX_DRIVER, + providerInstanceId: codexInstanceId, + status, + runtimePayload: { cwd: "/repo", activeTurnId }, + }); + const readBindings = directory + .listBindings() + .pipe( + Effect.map((bindings) => new Map(bindings.map((binding) => [binding.threadId, binding]))), + ); + const settledId = asThreadId("shutdown-settled"); + const runningId = asThreadId("shutdown-running"); + const stoppedWithTurnId = asThreadId("shutdown-stopped-with-turn"); + yield* seed(settledId, "stopped", null); + yield* seed(runningId, "running", asTurnId("running-turn")); + yield* seed(stoppedWithTurnId, "stopped", asTurnId("stale-turn")); + const settledBefore = (yield* readBindings).get(settledId); + assert(settledBefore !== undefined); + + const scope = yield* Scope.make(); + yield* Layer.build( + makeProviderServiceLive().pipe( + Layer.provide(NodeServices.layer), + Layer.provide(Layer.succeed(ProviderSessionDirectory.ProviderSessionDirectory, directory)), + Layer.provide( + Layer.succeed( + ProviderAdapterRegistry.ProviderAdapterRegistry, + makeStaticInstanceRegistry([[codexInstanceId, codex.adapter]]), + ), + ), + Layer.provide(defaultServerSettingsLayer), + Layer.provide(serverConfigTestLayer), + Layer.provide(recordedAnalytics.layer), + Layer.provide( + Layer.succeed( + ProviderEventLoggers.ProviderEventLoggers, + ProviderEventLoggers.NoOpProviderEventLoggers, + ), + ), + ), + ).pipe(Scope.provide(scope)); + yield* TestClock.adjust("1 minute"); + yield* Scope.close(scope, Exit.void); + + const byThread = yield* readBindings; + assert.deepStrictEqual(byThread.get(settledId), settledBefore); + for (const threadId of [runningId, stoppedWithTurnId]) { + const binding = byThread.get(threadId); + assert.equal(binding?.status, "stopped"); + assert.propertyVal(binding?.runtimePayload, "activeTurnId", null); + assert.propertyVal(binding?.runtimePayload, "lastRuntimeEvent", "provider.stopAll"); + } + const [stoppedAll] = recordedAnalytics.eventsByName("provider.sessions.stopped_all"); + assert.equal(stoppedAll?.properties?.stoppedSessionCount, 2); + }).pipe(Effect.provide(NodeServices.layer)), +); + it.effect("ProviderServiceLive flushes deferred completions during shutdown", () => Effect.gen(function* () { const recordedAnalytics = makeRecordingAnalytics(); @@ -4969,6 +5041,7 @@ describe("agent browser access", () => { getSnapshot: () => Effect.die("unused"), getShellSnapshot: () => Effect.die("unused"), getDeletedWorktreeThreads: () => Effect.die("unused"), + listThreadsWithPullRequests: () => Effect.die("unused"), getArchivedShellSnapshot: () => Effect.die("unused"), getSnapshotSequence: () => Effect.die("unused"), getCounts: () => Effect.die("unused"), diff --git a/apps/server/src/provider/Layers/ProviderService.ts b/apps/server/src/provider/Layers/ProviderService.ts index 5e5052d5ef23..12f8869da486 100644 --- a/apps/server/src/provider/Layers/ProviderService.ts +++ b/apps/server/src/provider/Layers/ProviderService.ts @@ -427,6 +427,14 @@ function readPersistedCwd( return trimmed.length > 0 ? trimmed : undefined; } +/** Stopped rows with no active turn are settled; shutdown leaves them untouched. */ +function isSettledBinding(binding: ProviderSessionDirectory.ProviderRuntimeBinding): boolean { + if (binding.status !== "stopped") return false; + const payload = binding.runtimePayload; + if (!payload || typeof payload !== "object" || Array.isArray(payload)) return true; + return !("activeTurnId" in payload) || payload.activeTurnId == null; +} + const dieOnMissingBindingInstanceId = ( operation: string, payload: { @@ -2331,7 +2339,6 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( return [completed, state] as const; }); yield* recordCompletedTurnProperties(properties); - const threadIds = yield* directory.listThreadIds(); const currentAdapters = yield* getAdapterEntries; const activeSessions = yield* Effect.forEach(currentAdapters, ([instanceId, adapter]) => adapter.listSessions().pipe( @@ -2362,7 +2369,12 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( yield* Effect.forEach(currentAdapters, ([, adapter]) => adapter.stopAll()).pipe(Effect.asVoid); yield* McpSessionRegistry.revokeAllActiveMcpCredentials(); McpProviderSession.clearAllMcpProviderSessions(); - const bindings = yield* directory.listBindings().pipe(Effect.orElseSucceed(() => [])); + // Stopped rows stay for their resume cursors, so long-lived installs hold + // thousands. Only rewrite the ones this shutdown actually stops. + const bindings = yield* directory.listBindings().pipe( + Effect.map((all) => all.filter((binding) => !isSettledBinding(binding))), + Effect.orElseSucceed(() => []), + ); yield* Effect.forEach(bindings, (binding) => Effect.gen(function* () { const providerInstanceId = dieOnMissingBindingInstanceId( @@ -2382,8 +2394,10 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( }); }), ).pipe(Effect.asVoid); + // Not `sessionCount`: that older property counted every row, so a new name + // keeps the two meanings in separate series. yield* analytics.record("provider.sessions.stopped_all", { - sessionCount: threadIds.length, + stoppedSessionCount: bindings.length, }); yield* analytics.flush; }); diff --git a/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts b/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts index d8226648e9f3..7b1fec90f867 100644 --- a/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts +++ b/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts @@ -239,6 +239,7 @@ describe("ProviderSessionReaper", () => { getSnapshot: () => Effect.die("unused"), getShellSnapshot: () => Effect.die("unused"), getDeletedWorktreeThreads: () => Effect.die("unused"), + listThreadsWithPullRequests: () => Effect.die("unused"), getArchivedShellSnapshot: () => Effect.die("unused"), getSnapshotSequence: () => Effect.succeed({ snapshotSequence: input.readModel.snapshotSequence }), diff --git a/apps/server/src/provider/opencodeRuntime.ts b/apps/server/src/provider/opencodeRuntime.ts index e87f758e0e3d..d8319bc44a71 100644 --- a/apps/server/src/provider/opencodeRuntime.ts +++ b/apps/server/src/provider/opencodeRuntime.ts @@ -78,7 +78,6 @@ export function resolveOpenCodeServerPassword( : input.environment.OPENCODE_SERVER_PASSWORD; } -const OPENCODE_SERVER_READY_PREFIX = "opencode server listening"; const DEFAULT_OPENCODE_SERVER_TIMEOUT_MS = 30_000; const DEFAULT_HOSTNAME = "127.0.0.1"; const OPENCODE_SERVER_STARTUP_MAX_OUTPUT_CHARS = 64 * 1024; @@ -289,11 +288,8 @@ export interface OpenCodeRuntimeShape { function parseServerUrlFromOutput(output: string): string | null { for (const line of output.split("\n")) { - if (!line.startsWith(OPENCODE_SERVER_READY_PREFIX)) { - continue; - } - const match = line.match(/on\s+(https?:\/\/[^\s]+)/); - return match?.[1] ?? null; + const match = line.match(/server listening on\s+(https?:\/\/[^\s]+)/i); + if (match?.[1]) return match[1]; } return null; } diff --git a/apps/server/src/server.test.ts b/apps/server/src/server.test.ts index 6836d28e25c1..5dc793b5cdf1 100644 --- a/apps/server/src/server.test.ts +++ b/apps/server/src/server.test.ts @@ -176,6 +176,7 @@ import * as VcsProcess from "./vcs/VcsProcess.ts"; import * as GitWorkflowService from "./git/GitWorkflowService.ts"; import * as ReviewService from "./review/ReviewService.ts"; import * as SourceControlRepositoryService from "./sourceControl/SourceControlRepositoryService.ts"; +import { REPLAY_MARKER_MAX_AGE } from "./auth/replayMarkers.ts"; import * as ServerSecretStore from "./auth/ServerSecretStore.ts"; import * as EnvironmentAuth from "./auth/EnvironmentAuth.ts"; import * as PairingGrantStore from "./auth/PairingGrantStore.ts"; @@ -2643,6 +2644,40 @@ it.layer(NodeServices.layer)("server router seam", (it) => { }).pipe(Effect.provide(NodeHttpServer.layerTest)), ); + it.effect("rejects a DPoP replay by time alone once its marker can be pruned", () => + Effect.gen(function* () { + yield* buildAppUnderTest(); + + const ownerCookie = yield* getAuthenticatedSessionCookieHeader(); + const credentialResponse = yield* HttpClient.post("/api/auth/pairing-token", { + headers: { cookie: ownerCookie }, + body: yield* HttpBody.json({}), + }); + const credential = (yield* credentialResponse.json) as { readonly credential: string }; + const tokenUrl = yield* getHttpServerUrl("/oauth/token"); + const acceptedAt = yield* DateTime.now; + // The longest-lived proof: `iat` at the 5 s future skew the verifier allows. + const dpop = makeDpopProof({ + method: "POST", + url: tokenUrl, + iat: Math.floor(acceptedAt.epochMilliseconds / 1_000) + 5, + }); + const exchange = exchangeAccessToken(credential.credential, { + headers: { dpop: dpop.proof }, + scope: "orchestration:read orchestration:operate terminal:operate review:write", + }); + + assert.equal((yield* exchange).response.status, 200); + // While the proof is fresh, only the replay marker rejects it. + assert.equal((yield* exchange).body.dpopFailureReason, "replay"); + // Once the marker can be pruned, the time check rejects the proof by itself. + yield* TestClock.setTime( + acceptedAt.epochMilliseconds + Duration.toMillis(REPLAY_MARKER_MAX_AGE), + ); + assert.equal((yield* exchange).body.dpopFailureReason, "time_window"); + }).pipe(Effect.provide(NodeHttpServer.layerTest)), + ); + it.effect("ignores forwarded host headers when validating token exchange DPoP URLs", () => Effect.gen(function* () { yield* buildAppUnderTest(); @@ -3710,6 +3745,82 @@ it.layer(NodeServices.layer)("server router seam", (it) => { }).pipe(Effect.provide(NodeHttpServer.layerTest)), ); + it.effect("rejects cloud replays by time alone once their markers can be pruned", () => + Effect.gen(function* () { + yield* buildAppUnderTest(); + + const cloudKeyPair = NodeCrypto.generateKeyPairSync("ed25519", { + privateKeyEncoding: { format: "pem", type: "pkcs8" }, + publicKeyEncoding: { format: "pem", type: "spki" }, + }); + const ownerCookie = yield* getAuthenticatedSessionCookieHeader(); + const relayConfigResponse = yield* fetchEffect( + yield* getHttpServerUrl("/api/connect/relay-config"), + { + method: "POST", + headers: { cookie: ownerCookie, "content-type": "application/json" }, + body: jsonRequestBody({ + relayUrl: "https://relay.example.test", + cloudUserId: "user_123", + environmentCredential: "t3env_test_credential", + cloudMintPublicKey: cloudKeyPair.publicKey, + endpointRuntime: null, + }), + }, + ); + assert.equal(relayConfigResponse.status, 200); + + const acceptedAt = yield* DateTime.now; + // The longest-lived proofs: `iat` at the 60 s future skew the handlers + // allow, and the 5 minute maximum lifetime. + const issuedAt = DateTime.add(acceptedAt, { minutes: 1 }); + const proofTimes = { + issuedAt: DateTime.formatIso(issuedAt), + expiresAt: DateTime.formatIso(DateTime.add(issuedAt, { minutes: 5 })), + }; + const requests = [ + [ + "/api/t3-connect/health", + makeCloudEnvironmentHealthRequest({ + privateKey: cloudKeyPair.privateKey, + environmentId: testEnvironmentDescriptor.environmentId, + nonce: "cloud-health-nonce-pruned", + ...proofTimes, + }), + ], + [ + "/api/t3-connect/mint-credential", + makeCloudMintCredentialRequest({ + privateKey: cloudKeyPair.privateKey, + environmentId: testEnvironmentDescriptor.environmentId, + clientProofKeyThumbprint: "client-proof-key-thumbprint", + nonce: "cloud-mint-nonce-pruned", + ...proofTimes, + }), + ], + ] as const; + const postAll = Effect.forEach(requests, ([pathname, request]) => + Effect.gen(function* () { + const response = yield* fetchEffect(yield* getHttpServerUrl(pathname), { + method: "POST", + headers: { "content-type": "application/json" }, + body: jsonRequestBody(request), + }); + return response.status; + }), + ); + + assert.deepStrictEqual(yield* postAll, [200, 200]); + // While the proofs are fresh, only the replay markers reject them (409). + assert.deepStrictEqual(yield* postAll, [409, 409]); + // Once the markers can be pruned, the time checks reject the proofs by themselves (401). + yield* TestClock.setTime( + acceptedAt.epochMilliseconds + Duration.toMillis(REPLAY_MARKER_MAX_AGE), + ); + assert.deepStrictEqual(yield* postAll, [401, 401]); + }).pipe(Effect.provide(NodeHttpServer.layerTest)), + ); + it.effect( "validates cloud proofs against the configured relay issuer, not the transport URL", () => diff --git a/apps/server/src/server.ts b/apps/server/src/server.ts index 4f264ae1cb0d..b7020458160d 100644 --- a/apps/server/src/server.ts +++ b/apps/server/src/server.ts @@ -119,9 +119,12 @@ import * as SourceControlRepositoryService from "./sourceControl/SourceControlRe import * as ProjectSetupScriptRunner from "./project/ProjectSetupScriptRunner.ts"; import * as WorktreeSetupTracker from "./project/WorktreeSetupTracker.ts"; import { ObservabilityLive } from "./observability/Layers/Observability.ts"; +import * as HeapSnapshot from "./observability/HeapSnapshot.ts"; +import * as EventLoopMonitor from "./observability/EventLoopMonitor.ts"; import * as ServerEnvironment from "./environment/ServerEnvironment.ts"; import * as RemoteOpenTargets from "./environment/RemoteOpenTargets.ts"; import { authHttpApiLayer, environmentAuthenticatedAuthLayer } from "./auth/http.ts"; +import * as ReplayMarkers from "./auth/replayMarkers.ts"; import * as ServerSecretStore from "./auth/ServerSecretStore.ts"; import * as EnvironmentAuth from "./auth/EnvironmentAuth.ts"; import { @@ -182,7 +185,8 @@ export const HTTP_ROUTER_CONFIG = { // those finalizers get a chance to run. const HTTP_PREEMPTIVE_SHUTDOWN_GRACE_MS = 0; const ResourceAttributionLayerLive = ResourceAttribution.layer; -const ApplicationObservabilityLive = ObservabilityLive.pipe( +const ApplicationObservabilityLive = EventLoopMonitor.layer.pipe( + Layer.provideMerge(ObservabilityLive), Layer.provideMerge(ResourceAttributionLayerLive), ); @@ -499,6 +503,7 @@ const AntigravityInstallationRefreshLive = Layer.effectDiscard( const RuntimeCoreDependenciesLive = ReactorLayerLive.pipe( Layer.provideMerge(AntigravityInstallationRefreshLive), + Layer.provideMerge(ReplayMarkers.layer), Layer.provideMerge(ProviderAuthServiceLive), // Core Services Layer.provideMerge(ServerSettingsLayerLive), @@ -957,6 +962,7 @@ const makeServerLayer = Layer.unwrap( runtimeStateLayer.pipe(Layer.provide(launcherLayer)), tailscaleServeLayer, cloudDesiredLinkReconcileLayer, + HeapSnapshot.layer, ); return serverApplicationLayer.pipe( diff --git a/apps/server/src/serverRuntimeStartup.test.ts b/apps/server/src/serverRuntimeStartup.test.ts index df879a2cf307..4bbdb2e9c8b5 100644 --- a/apps/server/src/serverRuntimeStartup.test.ts +++ b/apps/server/src/serverRuntimeStartup.test.ts @@ -169,6 +169,7 @@ it.effect("resolveAutoBootstrapWelcomeTargets returns existing project and threa getSnapshot: () => Effect.die("unused"), getShellSnapshot: () => Effect.die("unused"), getDeletedWorktreeThreads: () => Effect.die("unused"), + listThreadsWithPullRequests: () => Effect.die("unused"), getArchivedShellSnapshot: () => Effect.die("unused"), getSnapshotSequence: () => Effect.die("unused"), getCounts: () => Effect.die("unused"), @@ -298,6 +299,7 @@ it.effect.each([ getSnapshot: () => Effect.die("unused"), getShellSnapshot: () => Effect.die("unused"), getDeletedWorktreeThreads: () => Effect.die("unused"), + listThreadsWithPullRequests: () => Effect.die("unused"), getArchivedShellSnapshot: () => Effect.die("unused"), getSnapshotSequence: () => Effect.die("unused"), getCounts: () => Effect.die("unused"), @@ -385,6 +387,7 @@ it.effect( getSnapshot: () => Effect.die("unused"), getShellSnapshot: () => Effect.die("unused"), getDeletedWorktreeThreads: () => Effect.die("unused"), + listThreadsWithPullRequests: () => Effect.die("unused"), getArchivedShellSnapshot: () => Effect.die("unused"), getSnapshotSequence: () => Effect.die("unused"), getCounts: () => Effect.die("unused"), @@ -450,6 +453,7 @@ it.effect("resolveAutoBootstrapWelcomeTargets preserves typed UUID generation fa getSnapshot: () => Effect.die("unused"), getShellSnapshot: () => Effect.die("unused"), getDeletedWorktreeThreads: () => Effect.die("unused"), + listThreadsWithPullRequests: () => Effect.die("unused"), getArchivedShellSnapshot: () => Effect.die("unused"), getSnapshotSequence: () => Effect.die("unused"), getCounts: () => Effect.die("unused"), diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 830bb35b86db..077087a7d84e 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -3038,9 +3038,13 @@ const makeWsRpcLayer = ( [WS_METHODS.sourceControlPublishRepository]: (input) => observeRpcEffect( WS_METHODS.sourceControlPublishRepository, - sourceControlRepositories - .publishRepository(input) - .pipe(Effect.tap(() => refreshGitStatus(input.cwd))), + sourceControlRepositories.publishRepository(input).pipe( + // A new remote can change the cached identity. Only the `cwd` entry + // refreshes, so after a publish from a linked worktree the project + // root entry waits for its TTL. + Effect.tap(() => repositoryIdentityResolver.resolve(input.cwd, { refresh: true })), + Effect.tap(() => refreshGitStatus(input.cwd)), + ), { "rpc.aggregate": "source-control", }, diff --git a/apps/web/src/components/ChatView.tsx b/apps/web/src/components/ChatView.tsx index f5c3825cee33..b08bf12176a5 100644 --- a/apps/web/src/components/ChatView.tsx +++ b/apps/web/src/components/ChatView.tsx @@ -917,7 +917,14 @@ const PersistentThreadTerminalDrawer = memo(function PersistentThreadTerminalDra const writeTerminal = useAtomCommand(terminalEnvironment.write, "terminal write"); const closeTerminalMutation = useAtomCommand(terminalEnvironment.close, "terminal close"); const draftThread = useComposerDraftStore((store) => store.getDraftThreadByRef(threadRef)); - const serverThread = useThread(threadRef, { waitForShell: draftThread !== null }); + // Hidden drawers stay mounted (see MAX_HIDDEN_MOUNTED_TERMINAL_THREADS), so they read only + // the shell: a detail subscription would keep each hidden thread's history in memory. The + // active drawer shares ChatView's detail, which also covers archived threads (no shell). + const activeServerThread = useThread(active ? threadRef : null, { + waitForShell: draftThread !== null, + }); + const serverThreadShell = useThreadShell(threadRef); + const serverThread = activeServerThread ?? serverThreadShell; const projectRef = serverThread ? scopeProjectRef(serverThread.environmentId, serverThread.projectId) : draftThread @@ -10053,7 +10060,7 @@ export default function ChatView(props: ChatViewProps) { >
{isDraftHeroState ? (
diff --git a/apps/web/src/components/NoProjectsHero.tsx b/apps/web/src/components/NoProjectsHero.tsx index de125d384eb1..2da1c0aadb8c 100644 --- a/apps/web/src/components/NoProjectsHero.tsx +++ b/apps/web/src/components/NoProjectsHero.tsx @@ -2,9 +2,11 @@ import { PlusIcon } from "lucide-react"; import { useCallback } from "react"; import { openCommandPalette } from "../commandPaletteBus"; +import { isElectron } from "../env"; import { Button } from "./ui/button"; import { Empty, EmptyDescription, EmptyHeader, EmptyTitle } from "./ui/empty"; import { SidebarInset } from "./ui/sidebar"; +import { WorkspacePageHeader } from "./WorkspacePageHeader"; export function NoProjectsHero() { const openAddProject = useCallback(() => openCommandPalette({ open: "add-project" }), []); @@ -12,6 +14,8 @@ export function NoProjectsHero() { return (
+ {/* The desktop window only moves where CSS opts in, so keep a titlebar strip. */} + {isElectron ? : null}
diff --git a/apps/web/src/components/Sidebar.tsx b/apps/web/src/components/Sidebar.tsx index 4b8027c14108..968fd0a5f310 100644 --- a/apps/web/src/components/Sidebar.tsx +++ b/apps/web/src/components/Sidebar.tsx @@ -755,7 +755,7 @@ const SidebarDraftRow = memo(function SidebarDraftRow(props: { tabIndex={0} data-testid="sidebar-draft-row" className={cn( - "group/sidebar-row relative w-full cursor-pointer overflow-hidden rounded-md text-left text-sidebar-foreground outline-none select-none", + "group/sidebar-row relative w-full cursor-pointer overflow-hidden rounded-md text-left text-sidebar-foreground outline-none select-none focus-visible:ring-2 focus-visible:ring-inset focus-visible:ring-ring", props.isActive ? "bg-sidebar-row-active" : draftSurfaceClassName, )} onClick={handleActivate} @@ -1390,7 +1390,7 @@ const SidebarThreadRow = memo(function SidebarThreadRow(props: { // a useful hierarchy nor a reliable hover cue. Status now lives in the row // content; surface is reserved for interaction (hover, multi-select, route). const rowSurfaceClassName = cn( - "group/sidebar-row relative w-full cursor-pointer overflow-hidden rounded-md text-left outline-none select-none", + "group/sidebar-row relative w-full cursor-pointer overflow-hidden rounded-md text-left outline-none select-none focus-visible:ring-2 focus-visible:ring-inset focus-visible:ring-ring", variantAction === "unsettle" && "[&:not(:hover):not(:focus-within)_*]:text-secondary-label/70", props.isActive ? "bg-sidebar-row-active text-sidebar-foreground" diff --git a/apps/web/src/components/chat/ChatComposer.tsx b/apps/web/src/components/chat/ChatComposer.tsx index fafe24e43973..d168c6db4e5f 100644 --- a/apps/web/src/components/chat/ChatComposer.tsx +++ b/apps/web/src/components/chat/ChatComposer.tsx @@ -6165,7 +6165,7 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps) if (isInsideRestingComposerControlScope(event.target)) return; composerMentionDragHandlers.onDrop(event); }} - className="mx-auto w-full min-w-0 max-w-3xl" + className="mx-auto w-full min-w-0 max-w-(--chat-max-width)" data-chat-composer-form="true" > {composerControlsInStrip && restingControlsHost diff --git a/apps/web/src/components/chat/ComposerSurface.tsx b/apps/web/src/components/chat/ComposerSurface.tsx index bcb994085eb0..83e94895eddc 100644 --- a/apps/web/src/components/chat/ComposerSurface.tsx +++ b/apps/web/src/components/chat/ComposerSurface.tsx @@ -13,7 +13,7 @@ function Shell({ data-slot="composer-shell" data-with-context={contextStrip || undefined} className={cn( - "@container/composer-surface group/composer-surface relative isolate mx-auto w-full max-w-3xl", + "@container/composer-surface group/composer-surface relative isolate mx-auto w-full max-w-(--chat-max-width)", "[--chat-composer-drawer-inset:1.375rem] [--chat-composer-glass-surface:var(--card)] [--chat-composer-outline:rgb(0_0_0/8%)]", "dark:[--chat-composer-glass-surface:var(--surface-raised)] dark:[--chat-composer-highlight:rgb(255_255_255/3%)] dark:[--chat-composer-outline:color-mix(in_srgb,var(--color-white)_5%,transparent)]", "[html[data-theme-id]_&]:[--chat-composer-glass-surface:var(--app-theme-surface-raised)] [html[data-theme-id]_&]:[--chat-composer-outline:var(--app-theme-toolbar-border)]", diff --git a/apps/web/src/components/chat/MessagesTimeline.logic.ts b/apps/web/src/components/chat/MessagesTimeline.logic.ts index 57ed45a89d1d..21508a9663ae 100644 --- a/apps/web/src/components/chat/MessagesTimeline.logic.ts +++ b/apps/web/src/components/chat/MessagesTimeline.logic.ts @@ -42,7 +42,6 @@ import { formatWorkspaceRelativePath } from "../../filePathDisplay"; const TIMELINE_MINIMAP_ITEM_SPACING = 8; export const TIMELINE_MINIMAP_MIN_ITEMS = 2; const TIMELINE_MINIMAP_MAX_HEIGHT_CSS = "calc(100vh - 18rem)"; -const TIMELINE_CONTENT_MAX_WIDTH = 768; const TIMELINE_MINIMAP_PERSISTENT_GUTTER = 48; function singleToolCallLabel(entry: WorkLogEntry): string { @@ -233,14 +232,25 @@ export function resolveTimelineMinimapCurrentIndex(input: { return precedingIndex; } -export function resolveTimelineMinimapHasPersistentGutter(viewportWidth: number): boolean { - if (!Number.isFinite(viewportWidth) || viewportWidth <= 0) { - return false; +/** + * Side gutter between the viewport edge and the centered content column. + * `contentWidth` is the rendered column width, which follows the Chat width + * setting, so callers measure it rather than assume a fixed maximum. + */ +function resolveTimelineSideGutter(viewportWidth: number, contentWidth: number): number { + if (!Number.isFinite(viewportWidth) || viewportWidth <= 0 || !Number.isFinite(contentWidth)) { + return 0; } + return Math.max(0, (viewportWidth - Math.min(viewportWidth, contentWidth)) / 2); +} - const contentWidth = Math.min(viewportWidth, TIMELINE_CONTENT_MAX_WIDTH); - const sideGutter = Math.max(0, (viewportWidth - contentWidth) / 2); - return sideGutter >= TIMELINE_MINIMAP_PERSISTENT_GUTTER; +export function resolveTimelineMinimapHasPersistentGutter( + viewportWidth: number, + contentWidth: number, +): boolean { + return ( + resolveTimelineSideGutter(viewportWidth, contentWidth) >= TIMELINE_MINIMAP_PERSISTENT_GUTTER + ); } const TIMELINE_MINIMAP_HIT_STRIP_LEFT = 12; @@ -249,18 +259,16 @@ const TIMELINE_MINIMAP_EXPANDED_HIT_STRIP_WIDTH = "22rem"; /** * The minimap overlays the viewport's left edge while the content column is - * centered, so the side gutter between them shrinks under browser zoom or a - * narrow pane. A fixed-width hover strip would then sit on top of the message + * centered, so the side gutter between them shrinks under browser zoom, a + * narrow pane, or a wider Chat width setting. A fixed-width hover strip would then sit on top of the message * text and swallow its pointer events. Cap the strip's width so it never * extends past the gutter into the content column; 0 disables the strip. */ -export function resolveTimelineMinimapHitStripWidth(viewportWidth: number): number { - if (!Number.isFinite(viewportWidth) || viewportWidth <= 0) { - return 0; - } - - const contentWidth = Math.min(viewportWidth, TIMELINE_CONTENT_MAX_WIDTH); - const sideGutter = Math.max(0, (viewportWidth - contentWidth) / 2); +export function resolveTimelineMinimapHitStripWidth( + viewportWidth: number, + contentWidth: number, +): number { + const sideGutter = resolveTimelineSideGutter(viewportWidth, contentWidth); return Math.max( 0, Math.min( @@ -270,6 +278,19 @@ export function resolveTimelineMinimapHitStripWidth(viewportWidth: number): numb ); } +// The prev/next buttons are centered 4px into the strip and 20px wide, so +// their hitbox reaches 14px past the strip's left edge. +const TIMELINE_MINIMAP_NAVIGATION_REACH = 14; + +/** + * The prev/next buttons hang outside the strip's height, so the strip's own + * width cap does not cover them. Keep them inert to the pointer unless the + * gutter can hold them; keyboard focus still reaches them. + */ +export function resolveTimelineMinimapNavigationInteractive(collapsedWidth: number): boolean { + return collapsedWidth >= TIMELINE_MINIMAP_NAVIGATION_REACH; +} + /** * Once the preview is open, keep the full preview and the space leading to it * interactive. The collapsed strip remains gutter-capped so it cannot block diff --git a/apps/web/src/components/chat/MessagesTimeline.test.tsx b/apps/web/src/components/chat/MessagesTimeline.test.tsx index 016e621be1dd..a87dc2c2a444 100644 --- a/apps/web/src/components/chat/MessagesTimeline.test.tsx +++ b/apps/web/src/components/chat/MessagesTimeline.test.tsx @@ -571,6 +571,7 @@ describe("MessagesTimeline", () => { resolveTimelineMinimapHitStripWidth, resolveTimelineMinimapIndexFromPointer, resolveTimelineMinimapInteractiveWidth, + resolveTimelineMinimapNavigationInteractive, resolveTimelineMinimapTopPercent, } = await import("./MessagesTimeline.logic"); @@ -655,22 +656,39 @@ describe("MessagesTimeline", () => { itemBounds: [{ top: 80, height: 20 }], }), ).toBeNull(); - expect(resolveTimelineMinimapHasPersistentGutter(832)).toBe(false); - expect(resolveTimelineMinimapHasPersistentGutter(863)).toBe(false); - expect(resolveTimelineMinimapHasPersistentGutter(864)).toBe(true); + // Comfortable width: the column is capped at 768px. + expect(resolveTimelineMinimapHasPersistentGutter(832, 768)).toBe(false); + expect(resolveTimelineMinimapHasPersistentGutter(863, 768)).toBe(false); + expect(resolveTimelineMinimapHasPersistentGutter(864, 768)).toBe(true); + // Wider Chat width settings consume the gutter the minimap relies on. + expect(resolveTimelineMinimapHasPersistentGutter(1400, 1152)).toBe(true); + expect(resolveTimelineMinimapHasPersistentGutter(1200, 1152)).toBe(false); + expect(resolveTimelineMinimapHasPersistentGutter(2560, 2560)).toBe(false); // No usable gutter (zoomed in / narrow pane): the strip must go inert // instead of overlaying the centered content column. - expect(resolveTimelineMinimapHitStripWidth(768)).toBe(0); - expect(resolveTimelineMinimapHitStripWidth(792)).toBe(0); + expect(resolveTimelineMinimapHitStripWidth(768, 768)).toBe(0); + expect(resolveTimelineMinimapHitStripWidth(792, 768)).toBe(0); // Partial gutter: strip shrinks to what fits between the viewport edge // and the content column. - expect(resolveTimelineMinimapHitStripWidth(820)).toBe(14); + expect(resolveTimelineMinimapHitStripWidth(820, 768)).toBe(14); // Full gutter: unchanged 40px-wide strip. - expect(resolveTimelineMinimapHitStripWidth(872)).toBe(40); - expect(resolveTimelineMinimapHitStripWidth(1400)).toBe(40); - expect(resolveTimelineMinimapHitStripWidth(0)).toBe(0); - expect(resolveTimelineMinimapHitStripWidth(Number.NaN)).toBe(0); + expect(resolveTimelineMinimapHitStripWidth(872, 768)).toBe(40); + expect(resolveTimelineMinimapHitStripWidth(1400, 768)).toBe(40); + // Full Chat width: the column spans the viewport, so the strip is inert + // however wide the window gets. + expect(resolveTimelineMinimapHitStripWidth(2560, 2560)).toBe(0); + // Wide Chat width on a window just wider than the column: partial strip. + expect(resolveTimelineMinimapHitStripWidth(1204, 1152)).toBe(14); + expect(resolveTimelineMinimapHitStripWidth(0, 0)).toBe(0); + expect(resolveTimelineMinimapHitStripWidth(Number.NaN, 768)).toBe(0); + + // Prev/next buttons reach 14px past the strip's left edge; a narrower + // strip means they would sit on the content column. + expect(resolveTimelineMinimapNavigationInteractive(40)).toBe(true); + expect(resolveTimelineMinimapNavigationInteractive(14)).toBe(true); + expect(resolveTimelineMinimapNavigationInteractive(8)).toBe(false); + expect(resolveTimelineMinimapNavigationInteractive(0)).toBe(false); // The collapsed target stays narrow, but an open preview keeps its full // 20rem width plus the 2rem offset from the minimap rail interactive. diff --git a/apps/web/src/components/chat/MessagesTimeline.tsx b/apps/web/src/components/chat/MessagesTimeline.tsx index 755657e794ab..f812f67f2bd4 100644 --- a/apps/web/src/components/chat/MessagesTimeline.tsx +++ b/apps/web/src/components/chat/MessagesTimeline.tsx @@ -189,6 +189,7 @@ import { resolveTimelineMinimapHitStripWidth, resolveTimelineMinimapIndexFromPointer, resolveTimelineMinimapInteractiveWidth, + resolveTimelineMinimapNavigationInteractive, resolveTimelineMinimapTopPercent, resolveWorkGroupScrollIndex, shouldFollowWorkGroupAppend, @@ -239,6 +240,7 @@ import { chatMarkdownClipboardPayload } from "../../markdown-clipboard"; import { ContextChip, ContextChipLabel, type ContextChipKind } from "../ContextChip"; import { createContextPresentationRegistry } from "../contextPresentationRegistry"; import { useOpenPrLink } from "~/lib/openPullRequestLink"; +import { useClientSettings } from "~/hooks/useSettings"; import type { ChatMarkdownContextReference } from "../ChatMarkdown"; import { useMediaQuery } from "~/hooks/useMediaQuery"; import { cn } from "~/lib/utils"; @@ -346,7 +348,7 @@ function TimelineLoadEarlierHeader({ }) { return (
-
+