diff --git a/.github/scripts/ci-changes.cjs b/.github/scripts/ci-changes.cjs index 77a526076de4..0f817be60c35 100644 --- a/.github/scripts/ci-changes.cjs +++ b/.github/scripts/ci-changes.cjs @@ -27,6 +27,8 @@ const EXTRA_OWNERS = [ [/^native\/browser-secret\//, ["@t3tools/desktop"]], [/^native\/kde-snap-shot\//, ["@t3tools/desktop"]], [/^native\/libghostty-vt\//, ["@t3tools/web"]], + // apps/server/src/cloud/ConnectReachability.test.ts checks the guide's headings. + [/^docs\/user\/t3-connect-troubleshooting\.md$/, ["t3"]], ]; const RUST = /^native\/(resource-monitor|kde-snap-shot|hyprland-snap-shot)\//; @@ -92,11 +94,11 @@ function readWorkspace() { // [] means inert; null means unknown and requires every job. function ownersOf(file, packages) { - if (INERT.some((pattern) => pattern.test(file))) return []; + const extra = EXTRA_OWNERS.flatMap(([pattern, names]) => (pattern.test(file) ? names : [])); + if (INERT.some((pattern) => pattern.test(file))) return extra; // A manifest can change dependencies or versions other packages read. if (path.posix.basename(file) === "package.json") return null; if (RUNS_EVERYTHING.some((pattern) => pattern.test(file))) return null; - const extra = EXTRA_OWNERS.flatMap(([pattern, names]) => (pattern.test(file) ? names : [])); // scripts/build-desktop-artifact.test.ts reads the crates' sources. if (RUST.test(file)) return ["@t3tools/scripts", ...extra]; diff --git a/.macroscope/check-run-agents/ui-consistency.md b/.macroscope/check-run-agents/ui-consistency.md index 047dafbdd4be..905823d1d1a9 100644 --- a/.macroscope/check-run-agents/ui-consistency.md +++ b/.macroscope/check-run-agents/ui-consistency.md @@ -44,7 +44,7 @@ The reference for how this codebase wants UI built is the composer banner system - `ComposerBanner.tsx` exports one object of small slot components (`Root`, `Row`, `Icon`, `Content`, `Actions`, `Dismiss`, `Children`, and so on). Each slot owns its own Tailwind classes, exposes a `data-slot` attribute, takes `className` and spreads the rest of its props, and uses `cn` so callers can adjust without overriding. Variants and density are props on `Root`, not ad hoc class strings at call sites. Where a slot needs a real control it renders `Button` from `ui/` rather than a styled `"); const locator = buttonLocator((await takeSnapshot()).accessibilityTree, "same label"); diff --git a/apps/server/src/preview/ServerBrowserPage.ts b/apps/server/src/preview/ServerBrowserPage.ts index d803805abc68..f023b5f3553d 100644 --- a/apps/server/src/preview/ServerBrowserPage.ts +++ b/apps/server/src/preview/ServerBrowserPage.ts @@ -701,6 +701,7 @@ const capture = async ( }); return { ...page, + url: withoutDataPayload(page.url), // Mask text before truncating so a value crossing the cut cannot leak its prefix. visibleText: mask(page.visibleText).slice(0, MAX_VISIBLE_TEXT_LENGTH), accessibilityTree, diff --git a/apps/server/src/process/externalLauncher.test.ts b/apps/server/src/process/externalLauncher.test.ts index 42acda726926..29928b2146eb 100644 --- a/apps/server/src/process/externalLauncher.test.ts +++ b/apps/server/src/process/externalLauncher.test.ts @@ -6,6 +6,7 @@ import * as ConfigProvider from "effect/ConfigProvider"; import * as Deferred from "effect/Deferred"; import * as Cause from "effect/Cause"; import * as Effect from "effect/Effect"; +import * as TestClock from "effect/testing/TestClock"; import * as Exit from "effect/Exit"; import * as Fiber from "effect/Fiber"; import * as FileSystem from "effect/FileSystem"; @@ -735,6 +736,79 @@ it.effect.skipIf(windowsHost)("ignores unusable app bundles and keeps PATH launc }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)), ); +it.effect( + "memoizes editor discovery and refreshes in the background after the cache window", + () => { + let statCalls = 0; + const rescanned = Deferred.makeUnsafe(); + let initialScanDone = false; + const fileInfo = { type: "File" } as FileSystem.File.Info; + const layerLauncher = ExternalLauncher.layer.pipe( + Layer.provide( + Layer.mergeAll( + FileSystem.layerNoop({ + stat: () => + Effect.sync(() => { + statCalls += 1; + return fileInfo; + }).pipe( + Effect.tap(() => + initialScanDone ? Deferred.succeed(rescanned, undefined) : Effect.void, + ), + ), + }), + Path.layer, + Layer.succeed( + ChildProcessSpawner.ChildProcessSpawner, + ChildProcessSpawner.make(() => Effect.sync(() => makeMockDetachedHandle())), + ), + ), + ), + ); + + return Effect.gen(function* () { + const launcher = yield* ExternalLauncher.ExternalLauncher; + + const first = yield* launcher.resolveAvailableEditors(); + assert.equal(first.includes("vscode"), true); + const statCallsAfterFirstScan = statCalls; + initialScanDone = true; + assert.isAbove(statCallsAfterFirstScan, 0); + + // Past the shared command-resolution cache TTL (30s) but within the + // discovery cache window: the memoized set is reused without any scan. + yield* TestClock.adjust("31 seconds"); + const second = yield* launcher.resolveAvailableEditors(); + assert.deepEqual([...second], [...first]); + assert.equal(statCalls, statCallsAfterFirstScan); + + // Past the discovery cache window the next call still answers from the + // memoized set and rescans in the background. + yield* TestClock.adjust("30 seconds"); + const third = yield* launcher.resolveAvailableEditors(); + assert.deepEqual([...third], [...first]); + yield* Deferred.await(rescanned); + assert.isAbove(statCalls, statCallsAfterFirstScan); + }).pipe( + Effect.provide( + Layer.mergeAll( + layerLauncher, + Layer.succeed(HostProcess.Platform, "win32"), + ConfigProvider.layer( + ConfigProvider.fromEnv({ + env: { + PATH: "C:\\t3-editor-discovery-cache-test", + PATHEXT: ".COM;.EXE;.BAT;.CMD", + }, + }), + ), + TestClock.layer(), + ), + ), + ); + }, +); + it.effect("keeps scanning after the caller is interrupted and shares that scan", () => { const fileInfo = { type: "File" } as FileSystem.File.Info; const release = Deferred.makeUnsafe(); diff --git a/apps/server/src/process/externalLauncher.ts b/apps/server/src/process/externalLauncher.ts index 1c0b2c4693c2..4ecc108b9933 100644 --- a/apps/server/src/process/externalLauncher.ts +++ b/apps/server/src/process/externalLauncher.ts @@ -20,24 +20,20 @@ import { resolveSpawnCommand, withPathDirectoryListings, } from "@t3tools/shared/shell"; -import * as Clock from "effect/Clock"; import * as Config from "effect/Config"; import * as Context from "effect/Context"; -import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; import * as Base64 from "effect/encoding/Base64"; -import * as Exit from "effect/Exit"; 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 Ref from "effect/Ref"; import * as Schema from "effect/Schema"; -import * as Scope from "effect/Scope"; import * as Stream from "effect/Stream"; import * as ChildProcess from "effect/process/ChildProcess"; import * as ChildProcessSpawner from "effect/process/ChildProcessSpawner"; +import { makeStaleWhileRevalidate } from "../environment/staleWhileRevalidate.ts"; export { ExternalLauncherError, ExternalLauncherBrowserSpawnError, @@ -464,13 +460,7 @@ const resolveFileManagerRevealKind = Effect.fn("externalLauncher.resolveFileMana }, ); -// Shared scans must survive caller timeouts/disconnects; monotonic expiry prevents clock rollback extending stale entries. -const EDITOR_DISCOVERY_CACHE_TTL_NANOS = 60_000_000_000n; - -interface EditorDiscoveryCacheEntry { - readonly scan: Deferred.Deferred>; - readonly expiresAtNanos: bigint | undefined; -} +const EDITOR_DISCOVERY_CACHE_TTL = "60 seconds"; export class ExternalLauncher extends Context.Service< ExternalLauncher, @@ -870,69 +860,27 @@ export const make = Effect.gen(function* () { Effect.provideService(Path.Path, path), ); - const scope = yield* Scope.Scope; - const editorDiscoveryCache = yield* Ref.make>( - Option.none(), - ); - const runEditorDiscovery = (scan: Deferred.Deferred>) => + const cachedAvailableEditors = yield* makeStaleWhileRevalidate( provideCommandResolutionServices(resolveAvailableEditors()).pipe( Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, spawner), - Effect.onExit((exit) => - Effect.gen(function* () { - const expiresAtNanos = - (yield* Clock.monotonicTimeNanos) + EDITOR_DISCOVERY_CACHE_TTL_NANOS; - yield* Ref.update(editorDiscoveryCache, (current) => - Option.isNone(current) || current.value.scan !== scan - ? current - : Exit.isSuccess(exit) - ? Option.some({ scan, expiresAtNanos }) - : Option.none(), - ); - yield* Deferred.done(scan, exit); - }), - ), - Effect.interruptible, - Effect.forkIn(scope), - ); - // Claim and start atomically; interruption between them would leave an entry waiting on no scan. - const acquireEditorDiscovery = Effect.gen(function* () { - const nowNanos = yield* Clock.monotonicTimeNanos; - const [scan, isNewScan] = yield* Ref.modify( - editorDiscoveryCache, - ( - current, - ): [ - [EditorDiscoveryCacheEntry["scan"], boolean], - Option.Option, - ] => { - if ( - Option.isSome(current) && - (current.value.expiresAtNanos === undefined || current.value.expiresAtNanos > nowNanos) - ) { - return [[current.value.scan, false], current]; - } - const scan = Deferred.makeUnsafe>(); - return [[scan, true], Option.some({ scan, expiresAtNanos: undefined })]; - }, - ); - if (isNewScan) { - yield* runEditorDiscovery(scan); - } - return scan; - }).pipe(Effect.uninterruptible); - const cachedAvailableEditors = Effect.flatMap(acquireEditorDiscovery, Deferred.await); + ), + EDITOR_DISCOVERY_CACHE_TTL, + ); + const cachedFileManagerRevealKind = yield* makeStaleWhileRevalidate( + provideCommandResolutionServices(resolveFileManagerRevealKind()).pipe( + Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, spawner), + ), + EDITOR_DISCOVERY_CACHE_TTL, + ); // Forwarders retain the underlying launcher and discovery spans. return ExternalLauncher.of({ resolveAvailableEditors: Effect.fnUntraced(function* () { return yield* cachedAvailableEditors; }), - resolveFileManagerRevealKind: Effect.fnUntraced( - function* () { - return yield* provideCommandResolutionServices(resolveFileManagerRevealKind()); - }, - Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, spawner), - ), + resolveFileManagerRevealKind: Effect.fnUntraced(function* () { + return yield* cachedFileManagerRevealKind; + }), launchBrowser: Effect.fnUntraced( function* (target) { return yield* launchBrowser(target); diff --git a/apps/server/src/project/AgentSessionImporter.test.ts b/apps/server/src/project/AgentSessionImporter.test.ts index 0f08429956c4..c2352f82d740 100644 --- a/apps/server/src/project/AgentSessionImporter.test.ts +++ b/apps/server/src/project/AgentSessionImporter.test.ts @@ -77,6 +77,7 @@ function candidate( model: modelSelection.model, createdAt: timestamp, updatedAt: timestamp, + omittedMessageCount: 0, messages: [ { role: "user" as const, text: "Fix it", createdAt: timestamp }, { role: "assistant" as const, text: "Fixed", createdAt: timestamp }, @@ -284,6 +285,7 @@ function recentSession( { role: "user", text: `\n${prompt}\nwith details`, createdAt: "2026-09-01T10:00:00.000Z" }, { role: "assistant", text: "Done", createdAt: "2026-09-01T10:01:00.000Z" }, ], + omittedMessageCount: 0, messageCount: 12, }, }; @@ -516,6 +518,7 @@ it.effect( "claudeAgent", ); const unowned = candidate(CODEX_MODEL_SELECTION, "unowned"); + unowned.thread.omittedMessageCount = 1; const unownedPi = candidate(PI_MODEL_SELECTION, "unowned-pi", "pi"); const importer = yield* makeImporter([ active, @@ -550,9 +553,23 @@ it.effect( const threadId = importId(source); const records = yield* projections.getThreadRecords(threadId, [ "messages", + "turnItems", "providerThreads", ]); expect(records.messages.map((message) => message.text)).toEqual(["Fix it", "Fixed"]); + const notices = records.turnItems.filter((item) => item.type === "system_notice"); + if (source === unowned) { + expect(records.turnItems[0]).toBe(notices[0]); + expect(notices).toMatchObject([ + { + ordinal: 1, + runId: null, + message: expect.stringContaining("1 earlier message was not imported"), + }, + ]); + } else { + expect(notices).toEqual([]); + } expect(records.thread.historyOrigin).toBe("v1_import"); expect(records.thread.activeProviderThreadId).toBe(records.providerThreads[0]?.id); expect(records.providerThreads[0]?.nativeThreadRef?.nativeId).toBe(nativeId(source)); diff --git a/apps/server/src/project/AgentSessionImporter.ts b/apps/server/src/project/AgentSessionImporter.ts index 927e7766daea..f4c575fb689c 100644 --- a/apps/server/src/project/AgentSessionImporter.ts +++ b/apps/server/src/project/AgentSessionImporter.ts @@ -412,6 +412,34 @@ const make = Effect.gen(function* () { occurredAt: createdAt, payload: appThread, }, + ...(thread.omittedMessageCount > 0 + ? [ + { + id: EventId.make(`${IMPORT_EVENT_PREFIX}:history-truncated:${threadId}`), + type: "turn-item.updated" as const, + threadId, + occurredAt: createdAt, + payload: { + id: TurnItemId.make(`${IMPORT_EVENT_PREFIX}:history-truncated:${threadId}`), + threadId, + runId: null, + nodeId: null, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal: 0, + status: "completed" as const, + title: null, + startedAt: createdAt, + completedAt: createdAt, + updatedAt: createdAt, + type: "system_notice" as const, + message: `${thread.omittedMessageCount} earlier ${thread.omittedMessageCount === 1 ? "message was" : "messages were"} not imported. This thread keeps the first user message and the most recent history. The original session transcript is unchanged.`, + }, + }, + ] + : []), ...thread.messages.flatMap((message, index) => messageEvents({ threadId, index, message }), ), diff --git a/apps/server/src/project/AgentSessionScanner.test.ts b/apps/server/src/project/AgentSessionScanner.test.ts index 0bbdb8752d0f..10f647aab866 100644 --- a/apps/server/src/project/AgentSessionScanner.test.ts +++ b/apps/server/src/project/AgentSessionScanner.test.ts @@ -1960,6 +1960,48 @@ describe("parseAgentSessionTranscript", () => { }, ); + it.each([199, 200, 201, 205])( + "counts only omitted conversation messages for a %s-message Claude transcript", + (count) => { + const thread = AgentSessionScanner.parseAgentSessionTranscript({ + contents: [ + ...Array.from({ length: count }, (_, index) => + encodeTranscriptRecord({ + type: index === 0 ? "user" : "assistant", + sessionId: "claude-session", + message: { content: `Message ${index}` }, + }), + ), + encodeTranscriptRecord({ type: "user", isMeta: true, message: { content: "Setup" } }), + encodeTranscriptRecord({ + type: "user", + isCompactSummary: true, + message: { content: "Summary" }, + }), + encodeTranscriptRecord({ + type: "assistant", + isSidechain: true, + message: { content: "Sidechain" }, + }), + encodeTranscriptRecord({ + type: "assistant", + message: { content: [{ type: "tool_use" }] }, + }), + "malformed", + ].join("\n"), + source: "claudeAgent", + providerInstanceId: ProviderInstanceId.make("claudeAgent"), + fallbackSessionId: "fallback", + lastActiveAtMs: Date.parse("2026-08-24T12:00:00.000Z"), + }); + + expect(thread?.omittedMessageCount).toBe(Math.max(0, count - 200)); + expect(thread?.messages).toHaveLength(Math.min(200, count)); + expect(thread?.messages[0]?.text).toBe("Message 0"); + expect(thread?.messages.at(-1)?.text).toBe(`Message ${count - 1}`); + }, + ); + it("keeps Claude text and titles while dropping malformed and tool records", () => { const thread = AgentSessionScanner.parseAgentSessionTranscript({ contents: [ @@ -2111,10 +2153,10 @@ describe("parseAgentSessionTranscript", () => { ]); }); - it("keeps the canonical first prompt after long Codex transcripts are capped", () => { + it.each([199, 200])("keeps the canonical first prompt with %s later Codex messages", (count) => { const canonicalPrompt = "\n Keep the canonical prompt \n"; const canonicalTimestamp = "2026-08-24T10:01:00.000Z"; - const laterAssistantMessages = Array.from({ length: 200 }, (_, index) => + const laterAssistantMessages = Array.from({ length: count }, (_, index) => encodeTranscriptRecord({ type: "response_item", timestamp: `2026-08-24T11:${String(index % 60).padStart(2, "0")}:00.000Z`, @@ -2150,6 +2192,7 @@ describe("parseAgentSessionTranscript", () => { lastActiveAtMs: Date.parse("2026-08-24T12:00:00.000Z"), }); + expect(thread?.omittedMessageCount).toBe(count - 199); expect(thread?.messages).toHaveLength(200); expect(thread?.messages[0]).toMatchObject({ role: "user", @@ -2212,6 +2255,7 @@ describe("parseAgentSessionTranscript", () => { lastActiveAtMs: Date.parse("2026-08-24T12:00:00.000Z"), }); + expect(thread?.omittedMessageCount).toBe(1); expect(thread?.messages).toHaveLength(200); expect(thread?.messages[0]).toMatchObject({ role: "user", diff --git a/apps/server/src/project/AgentSessionScanner.ts b/apps/server/src/project/AgentSessionScanner.ts index 7220733bbe88..88c7ef8867a7 100644 --- a/apps/server/src/project/AgentSessionScanner.ts +++ b/apps/server/src/project/AgentSessionScanner.ts @@ -156,6 +156,7 @@ export interface AgentSessionThread { readonly createdAt: string; readonly updatedAt: string; readonly messages: ReadonlyArray; + readonly omittedMessageCount: number; /** Visible messages in the whole transcript; `messages` keeps only the most recent. */ readonly messageCount: number; } @@ -408,7 +409,6 @@ function parseAgentSessionRecords( let firstUserMessage: | (AgentSessionThreadMessage & { readonly codexResponseUser: boolean }) | undefined; - let messageCount = 0; const canonicalCodexResponseUserIndices = new Set(); let canonicalUserTextsInTurn = new Set(); let responseUsersInTurn: Array<{ @@ -462,16 +462,17 @@ function parseAgentSessionRecords( finishCodexTurn(); } + let conversationMessageCount = 0; const retainMessage = ( message: AgentSessionThreadMessage & { readonly codexResponseUser: boolean }, ) => { if (firstUserMessage === undefined && message.role === "user") { firstUserMessage = message; } - messageCount += 1; if (derivedTitle === undefined && message.role === "user") { derivedTitle = deriveTitle(input.source, message.text); } + conversationMessageCount += 1; messages.push(message); if (messages.length > MAX_IMPORTED_MESSAGES) messages.shift(); }; @@ -543,7 +544,7 @@ function parseAgentSessionRecords( if (message?.codexResponseUser === true && message.text.trim() === text.trim()) { if (firstUserMessage === message) firstUserMessage = undefined; messages.splice(index, 1); - messageCount -= 1; + conversationMessageCount -= 1; break; } } @@ -598,7 +599,8 @@ function parseAgentSessionRecords( createdAt: retainedMessages[0]?.createdAt ?? fallbackTimestamp, updatedAt: fallbackTimestamp, messages: retainedMessages, - messageCount: Math.max(messageCount, retainedMessages.length), + messageCount: Math.max(conversationMessageCount, retainedMessages.length), + omittedMessageCount: conversationMessageCount - retainedMessages.length, }; } diff --git a/apps/server/src/project/ProjectMutation.ts b/apps/server/src/project/ProjectMutation.ts index 8cf70f302b8f..550738b24baa 100644 --- a/apps/server/src/project/ProjectMutation.ts +++ b/apps/server/src/project/ProjectMutation.ts @@ -42,6 +42,8 @@ export const projectMutationOperation = Effect.fn("projectMutationOperation")(fu ? {} : { defaultThreadEnvMode: mutation.defaultThreadEnvMode }), ...(mutation.scripts === undefined ? {} : { scripts: mutation.scripts }), + ...(mutation.pinned === undefined ? {} : { pinned: mutation.pinned }), + ...(mutation.archived === undefined ? {} : { archived: mutation.archived }), }); case "project.delete": diff --git a/apps/server/src/project/ProjectService.ts b/apps/server/src/project/ProjectService.ts index 3ba57db6d7e5..ee7318fb5d5b 100644 --- a/apps/server/src/project/ProjectService.ts +++ b/apps/server/src/project/ProjectService.ts @@ -179,6 +179,8 @@ export const make = Effect.gen(function* () { autoPull: row.autoPull, projectIcon: row.projectIcon, scripts: row.scripts, + pinnedAt: row.pinnedAt, + archivedAt: row.archivedAt, createdAt: row.createdAt, updatedAt: row.updatedAt, deletedAt: row.deletedAt, @@ -386,6 +388,8 @@ export const make = Effect.gen(function* () { ? {} : { defaultThreadEnvMode: input.defaultThreadEnvMode }), ...(input.scripts === undefined ? {} : { scripts: input.scripts }), + ...(input.pinned === undefined ? {} : { pinned: input.pinned }), + ...(input.archived === undefined ? {} : { archived: input.archived }), }); if (workspaceRoot !== previousRoot) { yield* projectEnrichment.invalidate([previousRoot, workspaceRoot]); diff --git a/apps/server/src/project/RepositoryIdentityResolver.test.ts b/apps/server/src/project/RepositoryIdentityResolver.test.ts index 7cd1cc82d6fe..0cef7d96bbdf 100644 --- a/apps/server/src/project/RepositoryIdentityResolver.test.ts +++ b/apps/server/src/project/RepositoryIdentityResolver.test.ts @@ -156,25 +156,43 @@ it.layer(NodeServices.layer)("RepositoryIdentityResolverLive", (it) => { }).pipe(Effect.provide(RepositoryIdentityResolver.layer)), ); - it.effect("returns null for non-git folders and repos without remotes", () => - Effect.gen(function* () { - const fileSystem = yield* FileSystem.FileSystem; - const nonGitDir = yield* fileSystem.makeTempDirectoryScoped({ - prefix: "t3-repository-identity-non-git-", - }); - const gitDir = yield* fileSystem.makeTempDirectoryScoped({ - prefix: "t3-repository-identity-no-remote-", - }); + it.effect( + "returns null for non-git, missing and no-remote folders, probing only what can exist", + () => + Effect.gen(function* () { + const fileSystem = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const nonGitDir = yield* fileSystem.makeTempDirectoryScoped({ + prefix: "t3-repository-identity-non-git-", + }); + const gitDir = yield* fileSystem.makeTempDirectoryScoped({ + prefix: "t3-repository-identity-no-remote-", + }); + + yield* git(gitDir, ["init"]); + + const runner = yield* ProcessRunner.ProcessRunner; + const runs: Array> = []; + const resolver = yield* RepositoryIdentityResolver.make().pipe( + Effect.provideService(ProcessRunner.ProcessRunner, { + run: (input) => + Effect.suspend(() => { + runs.push(input.args); + return runner.run(input); + }), + }), + ); + const bareProbes = () => + runs.filter((args) => args.some((arg) => arg.startsWith("--git-dir="))); - yield* git(gitDir, ["init"]); + expect(yield* resolver.resolve(path.join(nonGitDir, "deleted"))).toBeNull(); + expect(runs).toEqual([]); - const resolver = yield* RepositoryIdentityResolver.RepositoryIdentityResolver; - const nonGitIdentity = yield* resolver.resolve(nonGitDir); - const noRemoteIdentity = yield* resolver.resolve(gitDir); + expect(yield* resolver.resolve(nonGitDir)).toBeNull(); + expect(bareProbes()).toEqual([]); - expect(nonGitIdentity).toBeNull(); - expect(noRemoteIdentity).toBeNull(); - }).pipe(Effect.provide(RepositoryIdentityResolver.layer)), + expect(yield* resolver.resolve(gitDir)).toBeNull(); + }).pipe(Effect.provide(ProcessRunner.layer)), ); it.effect("ignores local-path remotes such as GitButler's gb-local", () => diff --git a/apps/server/src/project/RepositoryIdentityResolver.ts b/apps/server/src/project/RepositoryIdentityResolver.ts index 659ace3fd2b4..cdf9dc5e1b30 100644 --- a/apps/server/src/project/RepositoryIdentityResolver.ts +++ b/apps/server/src/project/RepositoryIdentityResolver.ts @@ -6,7 +6,9 @@ import * as Data from "effect/Data"; import * as Duration from "effect/Duration"; import * as Effect from "effect/Effect"; import * as Exit from "effect/Exit"; +import * as FileSystem from "effect/FileSystem"; import * as Layer from "effect/Layer"; +import * as Path from "effect/Path"; import * as ProcessRunner from "../processRunner.ts"; import { sourceControlDefinitions } from "../sourceControl/sourceControlDefinitions.ts"; @@ -138,8 +140,18 @@ class RepositoryIdentityCacheKey extends Data.Class<{ const resolveRepositoryIdentityCacheKey = Effect.fn("RepositoryIdentityResolver.resolveCacheKey")( function* (cwd: string) { const processRunner = yield* ProcessRunner.ProcessRunner; + const fileSystem = yield* FileSystem.FileSystem; + const path = yield* Path.Path; const env = explicitBareRepositoryEnv(process.env, yield* HostProcess.Platform); + // `git -C` cannot enter a missing folder or a file, so every probe below would fail. + const isFolder = yield* fileSystem.stat(cwd).pipe( + Effect.map((info) => info.type === "Directory"), + Effect.catchReason("PlatformError", "NotFound", () => Effect.succeed(false)), + Effect.orElseSucceed(() => true), + ); + if (!isFolder) return null; + // git is always a real executable; cmd shell mode would split spaced paths on re-tokenization. const topLevelResult = yield* processRunner .run({ @@ -160,7 +172,25 @@ const resolveRepositoryIdentityCacheKey = Effect.fn("RepositoryIdentityResolver. } // Explicit Git directories preserve safe.bareRepository=explicit for both layouts. - for (const gitDir of [".", ".git"]) { + for (const [gitDir, entry] of [ + [".", "HEAD"], + [".git", ".git"], + ] as const) { + // An explicit --git-dir fails unless the folder has this entry; a symlink counts even + // when its target is missing, as git reads a symlinked HEAD's target text. + const target = path.join(cwd, entry); + const hasEntry = yield* fileSystem.exists(target).pipe( + Effect.flatMap((exists) => + exists + ? Effect.succeed(true) + : fileSystem.readLink(target).pipe( + Effect.as(true), + Effect.orElseSucceed(() => false), + ), + ), + Effect.orElseSucceed(() => true), + ); + if (!hasEntry) continue; const bareResult = yield* processRunner .run({ command: "git", @@ -224,6 +254,8 @@ export const make = Effect.fn("RepositoryIdentityResolver.make")(function* ( options: RepositoryIdentityResolverOptions = {}, ) { const processRunner = yield* ProcessRunner.ProcessRunner; + const fileSystem = yield* FileSystem.FileSystem; + const path = yield* Path.Path; const cacheCapacity = options.cacheCapacity ?? DEFAULT_REPOSITORY_IDENTITY_CACHE_CAPACITY; const refine = options.refine ?? Effect.succeed; const timeToLive = (exit: Exit.Exit) => @@ -237,6 +269,8 @@ export const make = Effect.fn("RepositoryIdentityResolver.make")(function* ( (cwd) => resolveRepositoryIdentityCacheKey(cwd).pipe( Effect.provideService(ProcessRunner.ProcessRunner, processRunner), + Effect.provideService(FileSystem.FileSystem, fileSystem), + Effect.provideService(Path.Path, path), ), { capacity: cacheCapacity, timeToLive }, ); diff --git a/apps/server/src/provider/AcpRegistryProbe.test.ts b/apps/server/src/provider/AcpRegistryProbe.test.ts index bc7c14d3b3c8..e6835159f9e7 100644 --- a/apps/server/src/provider/AcpRegistryProbe.test.ts +++ b/apps/server/src/provider/AcpRegistryProbe.test.ts @@ -229,6 +229,7 @@ describe("ACP Registry probe", () => { { id: "medium", label: "Medium" }, { id: "high", label: "High" }, ], + role: "reasoning", }, ]); }); diff --git a/apps/server/src/provider/GrokProvider.test.ts b/apps/server/src/provider/GrokProvider.test.ts index 4ecc0ad28971..8139df621f98 100644 --- a/apps/server/src/provider/GrokProvider.test.ts +++ b/apps/server/src/provider/GrokProvider.test.ts @@ -129,9 +129,9 @@ describe("buildGrokModelCapabilities", () => { type: "select", currentValue: "medium", options: [ - { id: "xhigh", label: "Extra High Effort", isDefault: true }, - { id: "high", label: "High Effort" }, { id: "medium", label: "Medium Effort" }, + { id: "high", label: "High Effort" }, + { id: "xhigh", label: "Extra High Effort", isDefault: true }, ], }, ]); diff --git a/apps/server/src/provider/agentContextBudget.json b/apps/server/src/provider/agentContextBudget.json index 0596eaf82be9..0a4be7ef896c 100644 --- a/apps/server/src/provider/agentContextBudget.json +++ b/apps/server/src/provider/agentContextBudget.json @@ -1,5 +1,5 @@ { - "claudeSystemAppendBytes": 6034, - "codexAdditionalContextBytes": 7764, - "mcpToolListJsonBytes": 76432 + "claudeSystemAppendBytes": 6058, + "codexAdditionalContextBytes": 7788, + "mcpToolListJsonBytes": 77964 } diff --git a/apps/server/src/provider/antigravityUsageLimits.test.ts b/apps/server/src/provider/antigravityUsageLimits.test.ts index 16e3c87ce1ec..3eb366f0dba3 100644 --- a/apps/server/src/provider/antigravityUsageLimits.test.ts +++ b/apps/server/src/provider/antigravityUsageLimits.test.ts @@ -110,6 +110,7 @@ const fixture = Effect.fn("AntigravityUsageTest.fixture")(function* ( ); } expect(JSON.parse(body)).toEqual({ project: "project-id" }); + expect(request.headers["user-agent"]).toMatch(/^antigravity\/hub\/2\.24\.0 /); return HttpClientResponse.fromWeb( request, request.url.endsWith(":retrieveUserQuotaSummary") @@ -475,6 +476,28 @@ describe("Antigravity subscription limits", () => { }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)), ); + it.effect("labels Argon's weekly quota apart from the default Gemini weekly quota", () => + Effect.gen(function* () { + const { probe } = yield* fixture({ + summary: { + groups: [ + ...summary.groups, + { + displayName: "Gemini 4 Argon", + buckets: [ + { bucketId: "gemini-argon-wtu-weekly", window: "weekly", remainingFraction: 0.6 }, + ], + }, + ], + }, + }); + const limits = yield* probe; + const byId = new Map(limits.windows.map((window) => [window.id, window])); + expect(byId.get("gemini-weekly")?.label).toBe("Gemini · Weekly"); + expect(byId.get("gemini-argon-wtu-weekly")?.label).toBe("Gemini 4 Argon · Weekly"); + }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)), + ); + it.effect("keeps one endpoint's quotas when the other fails", () => Effect.gen(function* () { const { probe } = yield* fixture({ diff --git a/apps/server/src/provider/antigravityUsageLimits.ts b/apps/server/src/provider/antigravityUsageLimits.ts index beecf7b53db1..f73fddb30596 100644 --- a/apps/server/src/provider/antigravityUsageLimits.ts +++ b/apps/server/src/provider/antigravityUsageLimits.ts @@ -124,11 +124,14 @@ function summaryWindows(summary: typeof Summary.Type, requestedAt: number) { for (const group of groups ?? []) { for (const bucket of group.buckets ?? []) { if (bucket.disabled) continue; - const family = bucket.bucketId.startsWith("gemini-") - ? "Gemini" - : bucket.bucketId.startsWith("3p-") - ? "Claude & GPT (shared)" - : group.displayName?.trim() || bucket.bucketId; + // Argon is metered apart from the default Gemini buckets; keep its group name. + const family = bucket.bucketId.startsWith("gemini-argon-") + ? group.displayName?.trim() || "Gemini Argon" + : bucket.bucketId.startsWith("gemini-") + ? "Gemini" + : bucket.bucketId.startsWith("3p-") + ? "Claude & GPT (shared)" + : group.displayName?.trim() || bucket.bucketId; const weekly = bucket.window === "weekly"; const session = bucket.window === "5h"; const windowDurationMins = SUMMARY_WINDOW_MINS.get(bucket.window ?? ""); @@ -284,7 +287,7 @@ export const makeAntigravityUsageProbe = Effect.fn("makeAntigravityUsageProbe")( HttpClientRequest.post(`${endpoint}/v1internal:${operation}`).pipe( HttpClientRequest.setHeaders({ Authorization: `Bearer ${accessToken}`, - "User-Agent": `antigravity/hub/1.0.0 (aidev_client; os_type=${platform}; arch=${architecture}; cl=963137146)`, + "User-Agent": `antigravity/hub/2.24.0 (aidev_client; os_type=${platform}; arch=${architecture}; cl=963137146)`, }), HttpClientRequest.bodyJsonUnsafe( token.project_id ? { project: token.project_id } : {}, diff --git a/apps/server/src/provider/providerMaintenanceRunner.test.ts b/apps/server/src/provider/providerMaintenanceRunner.test.ts index 9b7d1ffb35d4..6f3e49e4d4c4 100644 --- a/apps/server/src/provider/providerMaintenanceRunner.test.ts +++ b/apps/server/src/provider/providerMaintenanceRunner.test.ts @@ -7,6 +7,7 @@ import { } from "@t3tools/contracts"; import { ServerProviderUpdateError } from "@t3tools/contracts"; import * as Cause from "effect/Cause"; +import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; import * as Exit from "effect/Exit"; import * as Fiber from "effect/Fiber"; @@ -15,6 +16,7 @@ import * as Ref from "effect/Ref"; import * as Schema from "effect/Schema"; import * as Sink from "effect/Sink"; import * as Stream from "effect/Stream"; +import * as TestClock from "effect/testing/TestClock"; import { HttpClient, HttpClientResponse } from "effect/http"; import { ChildProcessSpawner } from "effect/process"; import * as HostProcess from "@t3tools/shared/HostProcess"; @@ -755,6 +757,41 @@ describe("providerMaintenanceRunner", () => { ), ); + // A native updater such as `claude update` downloads a ~300 MB binary and + // prints nothing while it does. Killed mid-download, it leaves a truncated + // binary behind that later updates mistake for an install. + it.effect.each([ + { runFor: "6 minutes", status: "succeeded", message: "Provider updated." }, + { runFor: "31 minutes", status: "failed", message: "Update timed out after 30 minutes." }, + ] as const)("lets a slow installer run for $runFor", ({ runFor, status, message }) => { + const started = Deferred.makeUnsafe(); + return Effect.gen(function* () { + const { registry } = yield* makeRegistry(baseNativeCliProvider); + const updater = yield* makeTestRunner(registry); + + const update = yield* updater.updateProvider(NATIVE_CLI_DRIVER).pipe(Effect.forkScoped); + yield* Deferred.await(started); + yield* TestClock.adjust(runFor); + const result = yield* Fiber.join(update); + + assert.strictEqual(result.providers[0]?.updateState?.status, status); + assert.strictEqual(result.providers[0]?.updateState?.message, message); + }).pipe( + Effect.provide( + Layer.mergeAll( + layerNonWindowsPlatform, + layerLatestVersionHttpClient("1.0.0"), + layerMockSpawner(() => ({ + exitCode: Deferred.succeed(started, undefined).pipe( + Effect.andThen(Effect.sleep(runFor)), + Effect.as(ChildProcessSpawner.ExitCode(0)), + ), + })), + ), + ), + ); + }); + it.effect("prevents concurrent updates for the same provider", () => { const startedLatch: { resolve: () => void } = { resolve: () => {} }; const releaseLatch: { resolve: () => void } = { resolve: () => {} }; @@ -856,17 +893,31 @@ describe("providerMaintenanceRunner", () => { ), ); - it.effect("resolves npm to a .cmd shim and routes through the shell on win32", () => { + it.effect("resolves npm to a .cmd shim from the provider's PATH on win32", () => { const captured: Array<{ readonly command: string; readonly args: ReadonlyArray; readonly shell: boolean | string | undefined; + readonly env: NodeJS.ProcessEnv | undefined; }> = []; return Effect.gen(function* () { const { registry } = yield* makeRegistry(baseProvider, { installedVersionAfterUpdate: "1.0.0", }); - const runner = yield* makeTestRunner(registry); + const runner = yield* makeTestRunner({ + ...registry, + getProviderMaintenanceCapabilitiesForInstance: () => + Effect.succeed( + makeProviderMaintenanceCapabilities({ + provider: CODEX_DRIVER, + packageName: "@openai/codex", + updateExecutable: "npm", + updateArgs: ["install", "-g", "@openai/codex@latest"], + updateLockKey: "npm-global", + env: { PATH: "C:\\fake\\npm" }, + }), + ), + }); const result = yield* runner.updateProvider(CODEX_DRIVER); @@ -879,17 +930,21 @@ describe("providerMaintenanceRunner", () => { ); assert.deepEqual(call.args, []); assert.strictEqual(call.shell, true); + assert.deepStrictEqual(call.env, { PATH: "C:\\fake\\npm" }); assert.strictEqual(result.providers[0]?.updateState?.status, "succeeded"); }).pipe( Effect.provide( Layer.mergeAll( Layer.succeed(HostProcess.Platform, "win32"), + // Only the provider's PATH has npm. Layer.succeed(HostProcess.Environment, { - PATH: "C:\\fake\\npm", + PATH: "C:\\Windows\\System32", PATHEXT: ".COM;.EXE;.BAT;.CMD", }), - Layer.succeed(SpawnExecutableResolution, (command) => - command === "npm" ? "C:\\fake\\npm\\npm.cmd" : undefined, + Layer.succeed(SpawnExecutableResolution, (command, _platform, env) => + command === "npm" && env.PATH === "C:\\fake\\npm" + ? "C:\\fake\\npm\\npm.cmd" + : undefined, ), layerLatestVersionHttpClient("1.0.0"), Layer.succeed( @@ -898,12 +953,16 @@ describe("providerMaintenanceRunner", () => { const childProcess = command as unknown as { readonly command: string; readonly args: ReadonlyArray; - readonly options: { readonly shell?: boolean | string | undefined }; + readonly options: { + readonly shell?: boolean | string | undefined; + readonly env?: NodeJS.ProcessEnv | undefined; + }; }; captured.push({ command: childProcess.command, args: childProcess.args, shell: childProcess.options.shell, + env: childProcess.options.env, }); return Effect.succeed(mockHandle({ stdout: "updated" })); }), diff --git a/apps/server/src/provider/providerMaintenanceRunner.ts b/apps/server/src/provider/providerMaintenanceRunner.ts index c2347da00330..ee7709e710ed 100644 --- a/apps/server/src/provider/providerMaintenanceRunner.ts +++ b/apps/server/src/provider/providerMaintenanceRunner.ts @@ -41,7 +41,11 @@ import type { ProviderMaintenanceCapabilities } from "@t3tools/provider-core/ser import { collectUint8StreamText } from "@t3tools/provider-core/server/collectStreamText"; const isServerProviderUpdateError = Schema.is(ServerProviderUpdateError); -const UPDATE_TIMEOUT_MS = 5 * 60_000; +// Native updaters (`claude update`) download a ~300 MB binary without printing +// progress, and killing one mid-download leaves a truncated binary that the +// next run treats as installed. Allow a slow connection to finish. +const UPDATE_TIMEOUT_MINUTES = 30; +const UPDATE_TIMEOUT_MS = UPDATE_TIMEOUT_MINUTES * 60_000; const UPDATE_OUTPUT_MAX_BYTES = 10_000; // Sample installer output because every progress update resends the provider list to all clients. const UPDATE_PROGRESS_INTERVAL = Duration.seconds(1); @@ -98,13 +102,14 @@ const runProviderMaintenanceCommandWithSpawner = Effect.fn("ProviderMaintenanceR }) { const collectCommandResult = Effect.fn("ProviderMaintenanceRunner.collectCommandResult")( function* () { + const env = input.env ? { env: input.env, extendEnv: true } : {}; // Windows batch shims require shell dispatch through resolveSpawnCommand. - const resolved = yield* resolveSpawnCommand(input.command, input.args); + const resolved = yield* resolveSpawnCommand(input.command, input.args, env); const child = yield* input.spawner .spawn( ChildProcess.make(resolved.command, resolved.args, { shell: resolved.shell, - ...(input.env ? { env: input.env, extendEnv: true } : {}), + ...env, }), ) .pipe( @@ -259,7 +264,7 @@ function lastOutputLine(result: ProviderMaintenanceCommandResult): string | null function failureMessage(result: ProviderMaintenanceCommandResult): string { if (result.timedOut) { - return "Update timed out."; + return `Update timed out after ${UPDATE_TIMEOUT_MINUTES} minutes.`; } const summary = result.exitCode !== null && result.exitCode !== 0 diff --git a/apps/server/src/pullRequest/pullRequestReviewGuideNormalize.test.ts b/apps/server/src/pullRequest/pullRequestReviewGuideNormalize.test.ts index a47743b34263..92704c51102f 100644 --- a/apps/server/src/pullRequest/pullRequestReviewGuideNormalize.test.ts +++ b/apps/server/src/pullRequest/pullRequestReviewGuideNormalize.test.ts @@ -152,6 +152,48 @@ describe("normalizeReviewGuide", () => { expect(result.stops).toHaveLength(1); }); + it("merges excerpts that a later excerpt bridges into one", () => { + const result = normalizeReviewGuide( + guide({ + stops: [ + stop({ + excerpts: [ + excerpt("src/a.ts", 40, 41), + { ...excerpt("src/a.ts", 50, 51), caption: "Second" }, + excerpt("src/a.ts", 41, 50), + ], + }), + ], + }), + change, + ); + expect(result.stops[0]?.excerpts).toEqual([ + { ...excerpt("src/a.ts", 40, 51), caption: "Second" }, + ]); + + const tailNote = { path: "src/a.ts", line: 40, title: "Tail", body: "Keep this note." }; + const wide = normalizeReviewGuide( + guide({ + stops: [ + stop({ + excerpts: [ + excerpt("src/a.ts", 1, 25), + { ...excerpt("src/a.ts", 30, 45), caption: "Second" }, + excerpt("src/a.ts", 24, 32), + ], + notes: [tailNote], + }), + ], + }), + parseReviewGuideChange( + "src/a.ts\0", + "diff --git a/src/a.ts b/src/a.ts\n--- a/src/a.ts\n+++ b/src/a.ts\n@@ -0,0 +1,60 @@\n+added", + ), + ); + expect(wide.stops[0]?.excerpts).toEqual([{ ...excerpt("src/a.ts", 1, 25), caption: "Second" }]); + expect(wide.stops[0]?.notes).toEqual([]); + }); + it("drops notes outside the stop's remaining excerpts", () => { const result = normalizeReviewGuide( guide({ diff --git a/apps/server/src/pullRequest/pullRequestReviewGuideNormalize.ts b/apps/server/src/pullRequest/pullRequestReviewGuideNormalize.ts index 184908e4bd7f..82edfc5fb815 100644 --- a/apps/server/src/pullRequest/pullRequestReviewGuideNormalize.ts +++ b/apps/server/src/pullRequest/pullRequestReviewGuideNormalize.ts @@ -87,26 +87,29 @@ function clampExcerpt( function mergeExcerpts( excerpts: ReadonlyArray, ): Array { - const merged: Array = []; + let merged: Array = []; for (const excerpt of excerpts) { - const index = merged.findIndex( - (other) => - other.path === excerpt.path && - other.side === excerpt.side && - other.start <= excerpt.end + 1 && - excerpt.start <= other.end + 1, - ); - const other = merged[index]; - if (other === undefined) { + const touches = (other: PullRequestReviewGuideExcerpt) => + other.path === excerpt.path && + other.side === excerpt.side && + other.start <= excerpt.end + 1 && + excerpt.start <= other.end + 1; + // An excerpt can bridge several earlier ones; fold them all into the first. + const group = merged.filter(touches); + const first = group[0]; + if (first === undefined) { merged.push(excerpt); continue; } - merged[index] = { - ...other, - start: Math.min(other.start, excerpt.start), - end: Math.max(other.end, excerpt.end), - caption: other.caption || excerpt.caption, + const combined = { + ...first, + start: Math.min(excerpt.start, ...group.map((other) => other.start)), + end: Math.max(excerpt.end, ...group.map((other) => other.end)), + caption: group.find((other) => other.caption)?.caption ?? excerpt.caption, }; + merged = merged.flatMap((other) => + other === first ? [combined] : touches(other) ? [] : [other], + ); } return merged.map((excerpt) => ({ ...excerpt, diff --git a/apps/server/src/scheduledTasks/Schedule.test.ts b/apps/server/src/scheduledTasks/Schedule.test.ts index a83de2d78f61..65604b912e1f 100644 --- a/apps/server/src/scheduledTasks/Schedule.test.ts +++ b/apps/server/src/scheduledTasks/Schedule.test.ts @@ -108,6 +108,12 @@ describe("scheduled task schedule calculation", () => { }); it("compares schedules structurally", () => { + expect( + isSameSchedule( + { type: "once", runAt: "2026-11-01T03:00:00-05:00" }, + { type: "once", runAt: "2026-11-01T08:00:00.000Z" }, + ), + ).toBe(true); expect( isSameSchedule({ type: "interval", everyMs: 60_000 }, { type: "interval", everyMs: 60_000 }), ).toBe(true); diff --git a/apps/server/src/scheduledTasks/Schedule.ts b/apps/server/src/scheduledTasks/Schedule.ts index e8d8939adb98..fbb956b94a14 100644 --- a/apps/server/src/scheduledTasks/Schedule.ts +++ b/apps/server/src/scheduledTasks/Schedule.ts @@ -1,5 +1,6 @@ import { MIN_SCHEDULED_TASK_INTERVAL_MS, type ScheduledTaskSchedule } from "@t3tools/contracts"; import * as DateTime from "effect/DateTime"; +import * as Option from "effect/Option"; const MINUTE_MS = 60_000; @@ -14,6 +15,13 @@ export function nextScheduledRunAt( from: DateTime.DateTime, ): DateTime.DateTime | null { if (schedule.type === "webhook") return null; + if (schedule.type === "once") { + const runAt = DateTime.make(schedule.runAt); + return Option.isSome(runAt) && + DateTime.toEpochMillis(runAt.value) > DateTime.toEpochMillis(from) + ? runAt.value + : null; + } if (schedule.type === "interval") { // Intervals advance elapsed time, including across daylight-saving changes. return DateTime.addDuration(from, Math.max(schedule.everyMs, MIN_SCHEDULED_TASK_INTERVAL_MS)); @@ -48,6 +56,7 @@ export function isSameSchedule(a: ScheduledTaskSchedule, b: ScheduledTaskSchedul return b.type === "interval" && a.everyMs === b.everyMs; } if (a.type === "webhook") return b.type === "webhook"; + if (a.type === "once") return b.type === "once" && Date.parse(a.runAt) === Date.parse(b.runAt); if (b.type !== "fixed_time") return false; // Compare parsed times so padded-hour edits do not reschedule pending runs. const aTime = parseTimeOfDay(a.timeOfDay); diff --git a/apps/server/src/scheduledTasks/ScheduledTaskService.interval.test.ts b/apps/server/src/scheduledTasks/ScheduledTaskService.interval.test.ts index 4a1a58505c5c..c2cc1f382b68 100644 --- a/apps/server/src/scheduledTasks/ScheduledTaskService.interval.test.ts +++ b/apps/server/src/scheduledTasks/ScheduledTaskService.interval.test.ts @@ -4,6 +4,7 @@ import { ScheduledTaskUpsertInput } from "@t3tools/contracts"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; import * as Schema from "effect/Schema"; +import * as SqlClient from "effect/sql/SqlClient"; import * as TestClock from "effect/testing/TestClock"; import { afterEach, vi } from "vite-plus/test"; @@ -129,3 +130,185 @@ it.effect.each([ Effect.provide(SqlitePersistence.layerMemory), ), ); + +// Each launch takes two minutes, so a manual run can start before the run time and end after it. +const slowLaunchLayer = ScheduledTaskService.layer.pipe( + Layer.provide( + Layer.merge( + dependencies, + Layer.mock(ThreadLaunchService.ThreadLaunchService)({ + launch: () => TestClock.adjust("2 minutes").pipe(Effect.andThen(Effect.die("synthetic"))), + }), + ), + ), +); + +it.effect("runs a one-time schedule once and rejects past times", () => + Effect.gen(function* () { + yield* TestClock.setTime(Date.parse("2026-11-01T07:59:00.000Z")); + yield* Effect.gen(function* () { + const service = yield* ScheduledTaskService.ScheduledTaskService; + const decoded = yield* decodeInput({ + ...input, + schedule: { type: "once", runAt: "2026-11-01T03:00:00-05:00" }, + }); + const created = yield* service.upsert(decoded); + expect(created.task.schedule).toEqual({ type: "once", runAt: "2026-11-01T08:00:00.000Z" }); + expect(created.task.nextRunAt).toBe("2026-11-01T08:00:00.000Z"); + + // A manual run that starts before the time keeps the scheduled run, even if it ends after. + const early = yield* service.runNow({ id: created.task.id }); + expect(early.task.nextRunAt).toBe("2026-11-01T08:00:00.000Z"); + + const due = yield* service.runNow({ id: created.task.id }); + expect(due.task.nextRunAt).toBeNull(); + expect((yield* service.list()).tasks[0]?.nextRunAt).toBeNull(); + + yield* service.setEnabled({ id: created.task.id, enabled: false }); + const resumed = yield* service + .setEnabled({ id: created.task.id, enabled: true }) + .pipe(Effect.flip); + expect(resumed.message).toBe("The run time has passed. Pick a time in the future."); + + const past = yield* service + .upsert({ ...decoded, commandId: undefined, title: "Late" }) + .pipe(Effect.flip); + expect(past.message).toBe("The run time has passed. Pick a time in the future."); + }).pipe(Effect.provide(slowLaunchLayer)); + }).pipe( + Effect.provideService(ServerActivation, Effect.never), + Effect.provide(SqlitePersistence.layerMemory), + ), +); + +const ONCE = { type: "once", runAt: "2026-11-01T08:00:00Z" } as const; +/** A next run that only the simulated concurrent write produces. */ +const MOVED = "2030-01-01T00:00:00.000Z"; + +const onceInput = { + ...input, + commandId: "once-race", + schedule: { type: "once", runAt: "2026-11-01T08:00:00Z" }, +}; + +it.effect.each([ + { change: "a run completes", saved: ONCE, edited: ONCE, pause: false, expected: MOVED }, + { + change: "a run completes during a padded-time edit", + saved: { type: "fixed_time", timeOfDay: "9:00" }, + edited: { type: "fixed_time", timeOfDay: "09:00" }, + pause: false, + expected: MOVED, + }, + { + change: "another client pauses the task", + saved: ONCE, + edited: ONCE, + pause: true, + expected: "2026-11-01T08:00:00.000Z", + }, +] as const)( + "a title save keeps the stored next run only while nothing else changed: $change", + ({ saved, edited, pause, expected }) => + Effect.gen(function* () { + yield* TestClock.setTime(Date.parse("2026-11-01T07:00:00.000Z")); + // The project check runs between the save's read and write, so it stands in for + // a concurrent write in that gap. + const writeMidSave = Layer.unwrap( + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + return Layer.mock(ManagedProjectFolders.ManagedProjectFolders)({ + namedProjectsRoot: "/projects", + isScratchProject: () => + sql`UPDATE scheduled_tasks SET next_run_at = ${MOVED}, enabled = ${pause ? 0 : 1}`.pipe( + Effect.as(false), + Effect.orDie, + ), + }); + }), + ); + const layer = ScheduledTaskService.layer.pipe( + Layer.provide(Layer.merge(dependencies, writeMidSave)), + ); + yield* Effect.gen(function* () { + const service = yield* ScheduledTaskService.ScheduledTaskService; + const worktree = { type: "worktree", baseRef: "main" }; + yield* service.upsert( + yield* decodeInput({ + ...onceInput, + schedule: saved, + workspaceStrategy: { type: "root" }, + }), + ); + const renamed = yield* service.upsert( + yield* decodeInput({ + ...onceInput, + schedule: edited, + workspaceStrategy: worktree, + title: "Renamed", + }), + ); + expect(renamed.task.enabled).toBe(true); + expect(renamed.task.nextRunAt).toBe(expected); + }).pipe(Effect.provide(layer)); + }).pipe( + Effect.provideService(ServerActivation, Effect.never), + Effect.provide(SqlitePersistence.layerMemory), + ), +); + +it.effect("keeps a one-time run after a restart interrupts an earlier manual run", () => + Effect.gen(function* () { + yield* TestClock.setTime(Date.parse("2026-11-01T07:59:00.000Z")); + const sql = yield* SqlClient.SqlClient; + const created = yield* Effect.gen(function* () { + const service = yield* ScheduledTaskService.ScheduledTaskService; + return yield* service.upsert(yield* decodeInput(onceInput)); + }).pipe(Effect.provide(serviceLayer)); + yield* sql`UPDATE scheduled_tasks SET last_run_status = 'running', last_run_at = '2026-11-01T07:59:00.000Z'`; + + yield* TestClock.setTime(Date.parse("2026-11-01T08:05:00.000Z")); + yield* Effect.gen(function* () { + const service = yield* ScheduledTaskService.ScheduledTaskService; + const [task] = (yield* service.list()).tasks; + expect(task?.id).toBe(created.task.id); + expect(task?.lastRunStatus).toBe("failed"); + expect(task?.nextRunAt).toBe("2026-11-01T08:00:00.000Z"); + }).pipe(Effect.provide(serviceLayer)); + }).pipe( + Effect.provideService(ServerActivation, Effect.never), + Effect.provide(SqlitePersistence.layerMemory), + ), +); + +it.effect("reports the stored task when the row was replaced during the run", () => + Effect.gen(function* () { + yield* TestClock.setTime(Date.parse("2026-11-01T07:00:00.000Z")); + // The launch stands in for a delete and same-id recreate while the run is out. + const replaceRowDuringLaunch = Layer.unwrap( + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + return Layer.mock(ThreadLaunchService.ThreadLaunchService)({ + launch: () => + sql`UPDATE scheduled_tasks SET last_run_status = 'never', last_run_at = NULL, run_count = 0`.pipe( + Effect.orDie, + Effect.andThen(Effect.die("synthetic dispatch failure")), + ), + }); + }), + ); + const layer = ScheduledTaskService.layer.pipe( + Layer.provide(Layer.merge(dependencies, replaceRowDuringLaunch)), + ); + yield* Effect.gen(function* () { + const service = yield* ScheduledTaskService.ScheduledTaskService; + const created = yield* service.upsert(yield* decodeInput(onceInput)); + const ran = yield* service.runNow({ id: created.task.id }); + expect(ran.task.runCount).toBe(0); + expect(ran.task.lastRunStatus).toBe("never"); + }).pipe(Effect.provide(layer)); + }).pipe( + Effect.provideService(ServerActivation, Effect.never), + Effect.provide(SqlitePersistence.layerMemory), + ), +); diff --git a/apps/server/src/scheduledTasks/ScheduledTaskService.test.ts b/apps/server/src/scheduledTasks/ScheduledTaskService.test.ts index cd10e2084ac1..b0a217dfd090 100644 --- a/apps/server/src/scheduledTasks/ScheduledTaskService.test.ts +++ b/apps/server/src/scheduledTasks/ScheduledTaskService.test.ts @@ -2,7 +2,7 @@ import * as Scheduler from "../scheduling/Scheduler.ts"; import * as NodeUtil from "node:util"; import * as NodeCrypto from "@effect/platform-node/NodeCrypto"; -import { assert, it } from "@effect/vitest"; +import { assert, it, vi } from "@effect/vitest"; import { ProjectId, ProviderInstanceId, @@ -12,6 +12,7 @@ import { } from "@t3tools/contracts"; import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; +import * as Exit from "effect/Exit"; import * as Layer from "effect/Layer"; import * as Logger from "effect/Logger"; import * as SqlClient from "effect/sql/SqlClient"; @@ -551,3 +552,70 @@ it.effect("a bound run follows its thread when promotion moves it mid-send", () }).pipe(Effect.provide(layer)); }).pipe(Effect.provide(SqlitePersistence.layerMemory)), ); + +it.effect("a run interrupted right after it reserves the task does not block the next run", () => + Effect.gen(function* () { + const launches = yield* Ref.make(0); + const layer = ScheduledTaskService.layer.pipe( + Layer.provide( + Layer.mergeAll( + Layer.mock(ThreadLaunchService.ThreadLaunchService)({ + launch: () => + Ref.update(launches, (n) => n + 1).pipe( + Effect.as({ threadId: ThreadId.make("thread:launched"), resumed: false } as never), + ), + }), + Layer.mock(ThreadManagementService.ThreadManagementService)({}), + Layer.mock(SecretRequests.SecretRequests)({}), + Layer.mock(ManagedProjectFolders.ManagedProjectFolders)({ + namedProjectsRoot: "/projects", + isScratchProject: () => Effect.succeed(false), + }), + NodeCrypto.layer, + Layer.succeed(Scheduler.Scheduler, { register: () => Effect.void }), + ), + ), + ); + yield* Effect.gen(function* () { + const service = yield* ScheduledTaskService.ScheduledTaskService; + const { task } = yield* service.upsert({ + id: ScheduledTaskId.make("task:interrupted-reservation"), + title: "Check in", + prompt: "Check progress", + enabled: true, + schedule: { type: "interval", everyMs: 60_000 }, + projectId: ProjectId.make("project:test"), + workspaceStrategy: { type: "root" }, + modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5.4" }, + runtimeMode: "full-access", + interactionMode: "default", + }); + + // The reservation is the first Set.add of the task id; interrupt the run right there. + const add = Set.prototype.add; + let interrupted = false; + const first = yield* Effect.gen(function* () { + using _reserve = vi.spyOn(Set.prototype, "add").mockImplementation(function ( + this: Set, + value: unknown, + ) { + const result = add.call(this, value); + if (value === task.id && !interrupted) { + interrupted = true; + Fiber.getCurrent()?.interruptUnsafe(); + } + return result; + }); + return yield* service + .runNow({ id: task.id }) + .pipe(Effect.forkChild, Effect.flatMap(Fiber.await)); + }); + assert.isTrue(interrupted); + assert.isTrue(Exit.hasInterrupts(first)); + + const second = yield* service.runNow({ id: task.id }); + assert.equal(second.threadId, ThreadId.make("thread:launched")); + assert.equal(yield* Ref.get(launches), 1); + }).pipe(Effect.provide(layer)); + }).pipe(Effect.provide(SqlitePersistence.layerMemory)), +); diff --git a/apps/server/src/scheduledTasks/ScheduledTaskService.ts b/apps/server/src/scheduledTasks/ScheduledTaskService.ts index 7d58e4537816..5ed161f01563 100644 --- a/apps/server/src/scheduledTasks/ScheduledTaskService.ts +++ b/apps/server/src/scheduledTasks/ScheduledTaskService.ts @@ -303,6 +303,35 @@ function nextRunAt( } } +/** + * Next run after an interrupted run. A one-time task counts from the run's start + * (`lastRunAt`), so a manual run that began before its time does not use it up. + */ +function nextRunAfterRun( + task: Pick, + now: DateTime.DateTime, +): string | null { + const startedAt = + task.schedule.type === "once" && task.lastRunAt !== null + ? DateTime.make(task.lastRunAt) + : Option.none(); + return nextRunAt( + task, + Option.getOrElse(startedAt, () => now), + ); +} + +/** An enabled one-time task with a past time would never run, so saving it is an error. */ +function requireFutureOnce( + task: Pick, + now: DateTime.DateTime, + taskId: ScheduledTaskId, +) { + return task.enabled && task.schedule.type === "once" && nextRunAt(task, now) === null + ? taskError("The run time has passed. Pick a time in the future.", { taskId }) + : Effect.void; +} + function errorMessage(error: unknown): string { if (Cause.isCause(error)) return Cause.pretty(error); if (error instanceof Error) return error.message; @@ -547,6 +576,8 @@ export const layer = Layer.effect( const saveTask = ( task: ScheduledTask, requireExisting: boolean, + /** The schedule this save read, when it left the schedule unchanged. */ + keptSchedule: ScheduledTask["schedule"] | null, webhook: { readonly token: string | null; readonly secret: string | null; @@ -617,7 +648,18 @@ export const layer = Layer.effect( interaction_mode = excluded.interaction_mode, creation_source = excluded.creation_source, updated_at = excluded.updated_at, - next_run_at = excluded.next_run_at, + -- Keep a next_run_at that a finishing run moved during this save, unless + -- another edit changed the schedule or enabled state meanwhile. + next_run_at = CASE + WHEN ${keptSchedule === null ? 0 : 1} = 1 + AND scheduled_tasks.schedule_json IN ( + excluded.schedule_json, + ${keptSchedule === null ? null : JSON.stringify(keptSchedule)} + ) + AND scheduled_tasks.enabled = excluded.enabled + THEN scheduled_tasks.next_run_at + ELSE excluded.next_run_at + END, -- Only rotate changes a live token, so a save racing a rotation -- cannot bring the old URL back. webhook_token = CASE @@ -690,7 +732,9 @@ export const layer = Layer.effect( WHERE task_id = ${input.id} AND last_run_status = 'running' AND last_run_at = ${input.startedAtIso} + RETURNING task_id `.pipe( + Effect.map((rows) => rows.length > 0), Effect.mapError((cause) => taskError("Could not record schedule task run.", { taskId: input.id, cause }), ), @@ -700,7 +744,12 @@ export const layer = Layer.effect( // runDueTasks skips 'running' rows forever, and the dispatch may already have gone out, // so next_run_at must advance. Pending webhook attempts count only with their completion. const releaseStuckRun = Effect.fn("ScheduledTaskService.releaseStuckRun")( - function* (task: ScheduledTask, message: string, pendingDeliveryId?: string) { + function* ( + task: ScheduledTask, + startedAtIso: string, + message: string, + pendingDeliveryId?: string, + ) { const now = yield* localNow; const reread = yield* Effect.result(findTask(task.id)); if (Result.isSuccess(reread) && reread.success === null) return; @@ -709,7 +758,7 @@ export const layer = Layer.effect( UPDATE scheduled_tasks SET last_run_status = 'failed', last_run_error = ${message}, - next_run_at = ${nextRunAt(source, now)}, + next_run_at = ${nextRunAfterRun({ ...source, lastRunAt: startedAtIso }, now)}, updated_at = ${iso(now)}, run_count = run_count + CASE WHEN EXISTS ( SELECT 1 FROM scheduled_task_webhook_pending @@ -719,7 +768,7 @@ export const layer = Layer.effect( `; yield* notifyChanged; }, - (effect, task, _message, _pendingDeliveryId) => + (effect, task, _startedAtIso, _message, _pendingDeliveryId) => effect.pipe( Effect.catch((cause) => Effect.logWarning("Could not release stuck schedule task run", { @@ -766,23 +815,10 @@ export const layer = Layer.effect( }, ) { const timerRun = trigger === "scheduled" || trigger === "catch_up"; - const reserved = yield* Ref.modify(activeRuns, (active) => { - if (active.has(task.id)) return [false, active] as const; - const next = new Set(active); - next.add(task.id); - return [true, next] as const; - }); - if (!reserved) { - if (!timerRun) { - return yield* taskError("Schedule task is already running.", { taskId: task.id }); - } - return { task, threadId: null }; - } - - return yield* Effect.gen(function* () { - const startedAt = yield* localNow; - const startedAtIso = iso(startedAt); - + // Read before reserving: nothing between the reservation and its release may fail. + const startedAt = yield* localNow; + const startedAtIso = iso(startedAt); + const run = Effect.gen(function* () { // Re-read stale poll snapshots so deleted, paused or postponed tasks cannot fire. const active = yield* findTask(task.id); if (active === null) { @@ -794,7 +830,7 @@ export const layer = Layer.effect( if (!timerRun) { return yield* taskError("Schedule task not found.", { taskId: task.id }); } - return { task, threadId: null }; + return { task, threadId: null, dispatchFailed: false }; } if ( webhook !== undefined && @@ -802,7 +838,11 @@ export const layer = Layer.effect( NodeSqliteClient.readOnly, )).length > 0 ) - return { task: { ...active, lastRunStatus: "succeeded" as const }, threadId: null }; + return { + task: { ...active, lastRunStatus: "succeeded" as const }, + threadId: null, + dispatchFailed: false, + }; const parsedNextRunAt = active.nextRunAt === null ? Option.none() : DateTime.make(active.nextRunAt); if ( @@ -811,7 +851,7 @@ export const layer = Layer.effect( Option.isNone(parsedNextRunAt) || DateTime.toEpochMillis(parsedNextRunAt.value) > DateTime.toEpochMillis(startedAt)) ) { - return { task: active, threadId: null }; + return { task: active, threadId: null, dispatchFailed: false }; } // Reject queued deliveries after pause, same-id recreation or trigger changes. if (webhook !== undefined) { @@ -928,18 +968,28 @@ export const layer = Layer.effect( const lastRunStatus = runSucceeded ? ("succeeded" as const) : ("failed" as const); const lastRunError = runSucceeded ? null : errorMessage(result.cause); const threadId = runSucceeded ? result.value : null; - const current = yield* findTask(task.id); - const scheduleSource = current ?? task; - const completed: ScheduledTask = { - ...scheduleSource, - updatedAt: iso(completedAt), - lastRunAt: startedAtIso, - nextRunAt: nextRunAt(scheduleSource, completedAt), - lastRunStatus, - lastRunError, - runCount: scheduleSource.runCount + 1, - }; - if (current !== null) { + // One transaction, so an edit cannot land between reading the schedule and + // writing the next run computed from it. The origin is read first: it reads + // files, and the transaction holds SQLite's writer lock. + const origin = yield* readWebhookOrigin; + const record = Effect.gen(function* () { + const [row] = yield* getRows(task.id); + const current = row === undefined ? null : yield* decodeRow(row, origin); + const scheduleSource = current ?? task; + const completed: ScheduledTask = { + ...scheduleSource, + updatedAt: iso(completedAt), + lastRunAt: startedAtIso, + // A manual run that starts before a one-time run must not use it up. + nextRunAt: nextRunAt( + scheduleSource, + scheduleSource.schedule.type === "once" ? startedAt : completedAt, + ), + lastRunStatus, + lastRunError, + runCount: scheduleSource.runCount + 1, + }; + if (current === null) return { recorded: completed, written: false }; // startedAtIso in the guard ensures this writes only to the row this // run marked as running — a task deleted mid-run and recreated with // the same id (idempotent commandId replay) must not be stamped. @@ -951,40 +1001,68 @@ export const layer = Layer.effect( error: lastRunError, startedAtIso, }); - if (webhook === undefined) yield* completion; - else - yield* sql - .withTransaction( - Effect.gen(function* () { - const pending = - yield* sql`SELECT 1 FROM scheduled_task_webhook_pending WHERE delivery_id = ${webhook.deliveryId}`; - if (pending.length === 0) yield* completion; - else { - const counted = - yield* sql`UPDATE scheduled_task_webhook_pending SET state = ${runSucceeded ? "completed" : "failed"} - WHERE delivery_id = ${webhook.deliveryId} AND state NOT IN ('completed', 'failed') RETURNING delivery_id`; - if (counted.length > 0) yield* completion; - } - }), - ) - .pipe( - Effect.mapError((cause) => - taskError("Could not record webhook completion.", { taskId: task.id, cause }), - ), - ); - yield* notifyChanged; - } - return { task: completed, threadId }; + let written = false; + if (webhook === undefined) written = yield* completion; + else { + const pending = + yield* sql`SELECT 1 FROM scheduled_task_webhook_pending WHERE delivery_id = ${webhook.deliveryId}`; + if (pending.length === 0) written = yield* completion; + else { + const counted = + yield* sql`UPDATE scheduled_task_webhook_pending SET state = ${runSucceeded ? "completed" : "failed"} + WHERE delivery_id = ${webhook.deliveryId} AND state NOT IN ('completed', 'failed') RETURNING delivery_id`; + if (counted.length > 0) written = yield* completion; + } + } + // A row recreated mid-run is not this run's, so report what is stored. + return { recorded: written ? completed : current, written }; + }); + const { recorded, written } = yield* sql + .withTransaction(record) + .pipe( + Effect.mapError((cause) => + isScheduledTaskError(cause) + ? cause + : taskError( + webhook === undefined + ? "Could not record schedule task run." + : "Could not record webhook completion.", + { taskId: task.id, cause }, + ), + ), + ); + if (written) yield* notifyChanged; + // The stored row can belong to a replacement, so report this dispatch separately. + return { task: recorded, threadId, dispatchFailed: !runSucceeded }; }).pipe( - Effect.onError((cause) => releaseStuckRun(task, errorMessage(cause), webhook?.deliveryId)), - Effect.ensuring( - Ref.update(activeRuns, (active) => { - const next = new Set(active); - next.delete(task.id); - return next; - }), + Effect.onError((cause) => + releaseStuckRun(task, startedAtIso, errorMessage(cause), webhook?.deliveryId), ), ); + + // Reserve and release as one bracket so an interrupt cannot leave the task marked busy. + return yield* Effect.acquireUseRelease( + Ref.modify(activeRuns, (active) => { + if (active.has(task.id)) return [false, active] as const; + const next = new Set(active); + next.add(task.id); + return [true, next] as const; + }), + (reserved) => + reserved + ? run + : trigger !== "scheduled" && trigger !== "catch_up" + ? Effect.fail(taskError("Schedule task is already running.", { taskId: task.id })) + : Effect.succeed({ task, threadId: null, dispatchFailed: false }), + (reserved) => + reserved + ? Ref.update(activeRuns, (active) => { + const next = new Set(active); + next.delete(task.id); + return next; + }) + : Effect.void, + ); }); const rescheduleMissedRun = Effect.fn("ScheduledTaskService.rescheduleMissedRun")(function* ( @@ -1086,7 +1164,7 @@ export const layer = Layer.effect( UPDATE scheduled_tasks SET last_run_status = 'failed', last_run_error = 'Run was interrupted by a server restart.', - next_run_at = ${nextRunAt(decoded.success, now)}, + next_run_at = ${nextRunAfterRun(decoded.success, now)}, updated_at = ${iso(now)}, run_count = run_count + CASE WHEN EXISTS ( SELECT 1 FROM scheduled_task_webhook_pending @@ -1197,7 +1275,9 @@ export const layer = Layer.effect( : undefined), ), } - : input.schedule; + : input.schedule.type === "once" + ? { type: "once", runAt: iso(DateTime.makeUnsafe(input.schedule.runAt)) } + : input.schedule; const webhook = input.schedule.type === "webhook" ? yield* Effect.gen(function* () { @@ -1243,6 +1323,8 @@ export const layer = Layer.effect( existingTask !== null && existingTask.enabled === input.enabled && isSameSchedule(existingTask.schedule, schedule); + if (!scheduleUnchanged) + yield* requireFutureOnce({ enabled: input.enabled, schedule }, now, id); const task: ScheduledTask = { id, title: input.title, @@ -1267,7 +1349,12 @@ export const layer = Layer.effect( lastRunError: existingTask?.lastRunError ?? null, runCount: existingTask?.runCount ?? 0, }; - yield* saveTask(task, input.requireExisting === true, webhook); + yield* saveTask( + task, + input.requireExisting === true, + scheduleUnchanged ? existingTask.schedule : null, + webhook, + ); yield* notifyChanged; const previousThreadId = existingTask?.threadId ?? null; if (previousThreadId !== null && previousThreadId !== task.threadId) { @@ -1283,6 +1370,11 @@ export const layer = Layer.effect( const existing = yield* loadTask(input.id); if (existing.enabled === input.enabled) return { task: existing }; const now = yield* localNow; + yield* requireFutureOnce( + { enabled: input.enabled, schedule: existing.schedule }, + now, + input.id, + ); const next = nextRunAt({ enabled: input.enabled, schedule: existing.schedule }, now); // RETURNING so a task deleted between the load and this UPDATE is a // visible not-found error, not a false success. @@ -1732,8 +1824,8 @@ export const layer = Layer.effect( ]); const dispatched = yield* Ref.make(false); return yield* runTask(task, "webhook", { deliveryId, prompt, dispatched }).pipe( - Effect.flatMap(({ task: completed }) => - completed.lastRunStatus === "failed" + Effect.flatMap(({ dispatchFailed }) => + dispatchFailed ? runOutcome("failed").pipe( Effect.andThen(markDeliveryFailed(deliveryId, "The run failed to start.")), Effect.andThen(releaseFailedDelivery(deliveryId)), @@ -2219,83 +2311,87 @@ export const layer = Layer.effect( } yield* Effect.forEach( tasks, - ({ task_id: taskId }) => - Effect.gen(function* () { - const reserved = yield* Ref.modify(pendingInFlight, (active) => - active.has(taskId) - ? ([false, active] as const) - : ([true, new Set(active).add(taskId)] as const), - ); - if (!reserved) return; - const permit = yield* webhookPermit(taskId); - // One worker per task waits on its permit, then drains persisted rows in order. - // Other tasks remain independent and ticks never reload an active task's payloads. - const dispatch = Effect.gen(function* () { - while (true) { - const [entry] = yield* sql<{ - delivery_id: string; - task_json: string; - request_json: string; - state: string; - }>` + ({ task_id: taskId }) => { + // One worker per task waits on its permit, then drains persisted rows in order. + // Other tasks remain independent and ticks never reload an active task's payloads. + const dispatch = Effect.gen(function* () { + while (true) { + const [entry] = yield* sql<{ + delivery_id: string; + task_json: string; + request_json: string; + state: string; + }>` SELECT delivery_id, task_json, request_json, state FROM scheduled_task_webhook_pending WHERE task_id = ${taskId} AND state IN ('dispatching', 'running', 'completed', 'failed') ORDER BY due_at, rowid LIMIT 1 `.pipe(NodeSqliteClient.readOnly); - if (entry === undefined) return; - const remove = sql`DELETE FROM scheduled_task_webhook_pending WHERE delivery_id = ${entry.delivery_id}`; - if (entry.state === "completed" || entry.state === "failed") { - if (entry.state === "failed") { - yield* markDeliveryFailed(entry.delivery_id, "The run failed to start."); - yield* releaseFailedDelivery(entry.delivery_id); - } - yield* remove; - continue; - } - const prepared = yield* Effect.gen(function* () { - const task = yield* decodePendingTask(entry.task_json); - const request = yield* decodePendingRequest(entry.request_json); - const frameId = Buffer.from(yield* crypto.randomBytes(6)).toString("hex"); - return { task, rendered: renderWebhookPrompt(task.prompt, request, frameId) }; - }).pipe(Effect.exit); - if (Exit.isFailure(prepared)) { - if (Cause.hasInterruptsOnly(prepared.cause)) - return yield* Effect.failCause(prepared.cause); - yield* markDeliveryFailed( - entry.delivery_id, - "The pending webhook payload could not be prepared.", - ); + if (entry === undefined) return; + const remove = sql`DELETE FROM scheduled_task_webhook_pending WHERE delivery_id = ${entry.delivery_id}`; + if (entry.state === "completed" || entry.state === "failed") { + if (entry.state === "failed") { + yield* markDeliveryFailed(entry.delivery_id, "The run failed to start."); yield* releaseFailedDelivery(entry.delivery_id); - yield* remove; - continue; - } - const { task, rendered } = prepared.value; - if (rendered.prompt.trim().length > PROVIDER_SEND_TURN_MAX_INPUT_CHARS) { - yield* markDeliveryFailed(entry.delivery_id, PROMPT_TOO_LONG); - yield* releaseFailedDelivery(entry.delivery_id); - } else { - yield* sql`UPDATE scheduled_task_webhook_deliveries SET outcome = 'accepted', error = NULL, rendered_prompt = ${rendered.prompt.slice(0, WEBHOOK_DELIVERY_LOG_PROMPT_LIMIT)}, missing_fields_json = ${encodeMissingFieldsJson(rendered.missing)} WHERE delivery_id = ${entry.delivery_id}`; - if (!(yield* runWebhookDelivery(task, entry.delivery_id, rendered.prompt))) - return; } yield* remove; + continue; } - }); - yield* dispatch.pipe( - permit.withPermits(1), - Effect.catch((cause) => - Effect.logWarning("Pending webhook dispatch failed", { cause }), - ), - Effect.ensuring( - Ref.update(pendingInFlight, (active) => { - const next = new Set(active); - next.delete(taskId); - return next; - }), - ), - Effect.withSpan("ScheduledTaskService.dispatchPendingWebhook", { root: true }), - Effect.forkIn(serviceScope), - ); - }), + const prepared = yield* Effect.gen(function* () { + const task = yield* decodePendingTask(entry.task_json); + const request = yield* decodePendingRequest(entry.request_json); + const frameId = Buffer.from(yield* crypto.randomBytes(6)).toString("hex"); + return { task, rendered: renderWebhookPrompt(task.prompt, request, frameId) }; + }).pipe(Effect.exit); + if (Exit.isFailure(prepared)) { + if (Cause.hasInterruptsOnly(prepared.cause)) + return yield* Effect.failCause(prepared.cause); + yield* markDeliveryFailed( + entry.delivery_id, + "The pending webhook payload could not be prepared.", + ); + yield* releaseFailedDelivery(entry.delivery_id); + yield* remove; + continue; + } + const { task, rendered } = prepared.value; + if (rendered.prompt.trim().length > PROVIDER_SEND_TURN_MAX_INPUT_CHARS) { + yield* markDeliveryFailed(entry.delivery_id, PROMPT_TOO_LONG); + yield* releaseFailedDelivery(entry.delivery_id); + } else { + yield* sql`UPDATE scheduled_task_webhook_deliveries SET outcome = 'accepted', error = NULL, rendered_prompt = ${rendered.prompt.slice(0, WEBHOOK_DELIVERY_LOG_PROMPT_LIMIT)}, missing_fields_json = ${encodeMissingFieldsJson(rendered.missing)} WHERE delivery_id = ${entry.delivery_id}`; + if (!(yield* runWebhookDelivery(task, entry.delivery_id, rendered.prompt))) return; + } + yield* remove; + } + }); + // The worker owns its reservation, so an interrupt anywhere still releases it. + return Effect.acquireUseRelease( + Ref.modify(pendingInFlight, (active) => + active.has(taskId) + ? ([false, active] as const) + : ([true, new Set(active).add(taskId)] as const), + ), + (reserved) => + reserved + ? webhookPermit(taskId).pipe( + Effect.flatMap((permit) => dispatch.pipe(permit.withPermits(1))), + Effect.catch((cause) => + Effect.logWarning("Pending webhook dispatch failed", { cause }), + ), + Effect.withSpan("ScheduledTaskService.dispatchPendingWebhook", { + root: true, + }), + ) + : Effect.void, + (reserved) => + reserved + ? Ref.update(pendingInFlight, (active) => { + const next = new Set(active); + next.delete(taskId); + return next; + }) + : Effect.void, + ).pipe(Effect.forkIn(serviceScope)); + }, { discard: true }, ); }); diff --git a/apps/server/src/scheduledTasks/ScheduledTaskService.webhook.test.ts b/apps/server/src/scheduledTasks/ScheduledTaskService.webhook.test.ts index cdd4e60e62d7..aaeb374bd0dd 100644 --- a/apps/server/src/scheduledTasks/ScheduledTaskService.webhook.test.ts +++ b/apps/server/src/scheduledTasks/ScheduledTaskService.webhook.test.ts @@ -12,6 +12,7 @@ import * as DateTime from "effect/DateTime"; import * as Crypto from "effect/Crypto"; import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; +import * as Exit from "effect/Exit"; import * as Fiber from "effect/Fiber"; import * as Layer from "effect/Layer"; import * as Queue from "effect/Queue"; @@ -1005,6 +1006,57 @@ it.effect("recovers a persisted coalescing window after restarting the service", ); }); +it.effect("a pending dispatch interrupted at its permit does not block later polls", () => { + const workers = new Map>(); + // One shared dispatcher lets the test run the retry's forked worker to completion. + const mixed = new EffectScheduler.MixedScheduler("sync"); + const dispatcher = mixed.makeDispatcher(); + return withService( + ({ service, sql, launches }) => + Effect.gen(function* () { + const { task } = yield* service.upsert( + yield* webhookTaskInput({ + schedule: { + type: "webhook", + dedupe: { coalesceKey: "{{headers.content-type}}", coalesceWindowSeconds: 10 }, + }, + }), + ); + yield* service.triggerWebhook(requestFor(task)); + yield* TestClock.adjust("10 seconds"); + const poll = workers.get("webhook-coalescing")!; + + const entered = yield* Deferred.make>(); + const interruptAtPermit = Tracer.make({ + span: (options) => { + if (options.name === "ScheduledTaskService.webhookPermit") { + const fiber = Fiber.getCurrent()!; + Deferred.doneUnsafe(entered, Effect.succeed(fiber)); + fiber.interruptUnsafe(); + } + return new Tracer.NativeSpan(options); + }, + }); + const firstPoll = yield* poll.pipe(Effect.withTracer(interruptAtPermit), Effect.forkChild); + assert.isTrue(Exit.hasInterrupts(yield* Fiber.await(yield* Deferred.await(entered)))); + yield* Fiber.await(firstPoll); + + yield* poll; + dispatcher.flush(); + + assert.equal(yield* Queue.size(launches), 1); + assert.lengthOf(yield* sql`SELECT 1 FROM scheduled_task_webhook_pending`, 0); + }), + { workers }, + ).pipe( + Effect.provideService(EffectScheduler.Scheduler, { + executionMode: mixed.executionMode, + makeDispatcher: () => dispatcher, + shouldYield: (fiber) => mixed.shouldYield(fiber), + }), + ); +}); + it.effect("rolls back replay and pending state when recording a delivery fails", () => { const workers = new Map>(); return withService( @@ -1199,6 +1251,32 @@ it.effect("start failures release both replay keys and relay claims for redelive ), ); +it.effect("a start failure releases the relay claim when the task row is replaced mid-run", () => + Effect.gen(function* () { + const gate = yield* Deferred.make(); + yield* withService( + ({ service, sql, launches }) => + Effect.gen(function* () { + const { task } = yield* service.upsert(yield* webhookTaskInput()); + const request = requestFor(task, { relayDeliveryId: "replaced-mid-run" }); + const first = yield* waitForSpan( + service.triggerWebhook(request), + "ScheduledTaskService.runWebhookDelivery", + ).pipe(Effect.forkChild); + yield* Queue.take(launches); + // A same-id recreate while the launch is out: the stored row is no longer this run's. + yield* sql`UPDATE scheduled_tasks SET last_run_status = 'never', last_run_at = NULL`; + yield* Deferred.succeed(gate, undefined); + yield* Fiber.join(first); + const retry = yield* service.triggerWebhook(request); + assert.equal(retry._tag === "accepted" && retry.outcome, "accepted"); + yield* Queue.take(launches); + }), + { gate, failLaunches: { remaining: 1 } }, + ); + }), +); + it.effect("main retention keeps fifty runs and preserves filtered and dedupe buckets", () => withService(({ service, sql, launches }) => Effect.gen(function* () { diff --git a/apps/server/src/scheduledTasks/webhookFilter.test.ts b/apps/server/src/scheduledTasks/webhookFilter.test.ts index 1194a1aed517..2c7f2cc4062e 100644 --- a/apps/server/src/scheduledTasks/webhookFilter.test.ts +++ b/apps/server/src/scheduledTasks/webhookFilter.test.ts @@ -16,6 +16,28 @@ const issueComment: WebhookRequest = { }; describe("findFailedWebhookFilter", () => { + it("matches exists on wildcard object values", () => { + assert.isNull( + findFailedWebhookFilter([{ path: "body.issue.labels[]", op: "exists" }], issueComment), + ); + assert.isNull( + findFailedWebhookFilter([{ path: "body.issues[].labels", op: "exists" }], { + ...issueComment, + bodyText: JSON.stringify({ issues: [{ labels: [] }] }), + }), + ); + }); + + it("rejects exists when wildcard values are all null or missing", () => { + assert.deepEqual( + findFailedWebhookFilter([{ path: "body.issues[].labels", op: "exists" }], { + ...issueComment, + bodyText: JSON.stringify({ issues: [{ labels: null }, {}] }), + }), + { index: 0, path: "body.issues[].labels" }, + ); + }); + it("matches a wildcard path when any element matches", () => { assert.isNull( findFailedWebhookFilter( diff --git a/apps/server/src/scheduledTasks/webhookFilter.ts b/apps/server/src/scheduledTasks/webhookFilter.ts index 882f30036428..2ad0e0237e07 100644 --- a/apps/server/src/scheduledTasks/webhookFilter.ts +++ b/apps/server/src/scheduledTasks/webhookFilter.ts @@ -36,6 +36,8 @@ function matchesScalar(filter: ScheduledTaskWebhookFilter, text: string): boolea */ function matchesWebhookFilter(filter: ScheduledTaskWebhookFilter, value: unknown): boolean { if (Array.isArray(value) && /\[\]/.test(filter.path)) { + if (filter.op === "exists") + return value.some((element) => element !== null && element !== undefined); const texts = value.map(scalarText).filter((text): text is string => text !== undefined); if (filter.op === "notEquals") return texts.every((text) => text !== filter.value); return texts.some((text) => matchesScalar(filter, text)); diff --git a/apps/server/src/terminal/Manager.test.ts b/apps/server/src/terminal/Manager.test.ts index f014ce413549..b79292fdfae3 100644 --- a/apps/server/src/terminal/Manager.test.ts +++ b/apps/server/src/terminal/Manager.test.ts @@ -772,6 +772,30 @@ it.layer( }), ); + it.effect("labels a running command by its whole process name, spaces included", () => + Effect.gen(function* () { + const { manager } = yield* createManager(5, { + subprocessPollIntervalMs: 1_000, + processKillGraceMs: 0, + processTable: Effect.succeed([ + { pid: 9000, ppid: 1, name: "zsh" }, + { pid: 200, ppid: 9000, name: "Google Chrome" }, + ]), + }).pipe(Effect.provide(layerWithHostPlatform("darwin"))); + const label = yield* Deferred.make(); + const unsubscribe = yield* manager.subscribe((event) => + event.type === "activity" + ? Deferred.succeed(label, event.label).pipe(Effect.asVoid) + : Effect.void, + ); + yield* Effect.addFinalizer(() => Effect.sync(unsubscribe)); + yield* manager.open(openInput()); + yield* TestClock.adjust("1 second"); + + expect(yield* Deferred.await(label)).toBe("Google Chrome"); + }).pipe(Effect.provide(TestClock.layer())), + ); + // Answers process-table queries after `queryDuration`, honoring the caller's // timeout the way ProcessRunner's `timedOutResult` mode does. const slowProcessTable = ( diff --git a/apps/server/src/terminal/Manager.ts b/apps/server/src/terminal/Manager.ts index 284beac37b8a..4bc86f5b1296 100644 --- a/apps/server/src/terminal/Manager.ts +++ b/apps/server/src/terminal/Manager.ts @@ -377,10 +377,9 @@ function normalizeChildCommandName(raw: string, platform: NodeJS.Platform): stri ) { trimmed = trimmed.slice(1, -1).trim(); } - const firstToken = (trimmed.split(/\s+/)[0] ?? trimmed).trim(); - if (firstToken.length === 0) return null; + // Process names and macOS `comm` paths can contain spaces ("Google Chrome"). const separators = platform === "win32" ? /[\\/]/ : /\//; - const base = firstToken.split(separators).at(-1) ?? firstToken; + const base = (trimmed.split(separators).at(-1) ?? trimmed).trim(); const withoutExe = platform === "win32" && base.toLowerCase().endsWith(".exe") ? base.slice(0, -4) : base; return withoutExe.length > 0 ? withoutExe : null; diff --git a/apps/server/src/textGeneration/AntigravityTextGeneration.test.ts b/apps/server/src/textGeneration/AntigravityTextGeneration.test.ts index 3fcafcc15178..7e1f1446d1a5 100644 --- a/apps/server/src/textGeneration/AntigravityTextGeneration.test.ts +++ b/apps/server/src/textGeneration/AntigravityTextGeneration.test.ts @@ -5,6 +5,7 @@ import { ProviderInstanceId, ProviderSetupError, } from "@t3tools/contracts"; +import { symlinksSupported } from "@t3tools/shared/testing/symlinks"; import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; import * as Exit from "effect/Exit"; @@ -60,6 +61,7 @@ const makeFixture = Effect.fn("makeAntigravityTextGenerationFixture")(function* readonly startError?: AcpError; readonly rejectAdmission?: boolean; readonly sessionId?: string; + readonly startupSkills?: boolean; } = {}, ) { const fs = yield* FileSystem.FileSystem; @@ -135,6 +137,13 @@ const makeFixture = Effect.fn("makeAntigravityTextGenerationFixture")(function* start: () => Effect.gen(function* () { if (options.startError) return yield* options.startError; + if (options.startupSkills) { + for (const directory of [".agents", ".gemini"]) { + yield* fs + .makeDirectory(path.join(cwd, directory, "skills"), { recursive: true }) + .pipe(Effect.orDie); + } + } yield* fs.makeDirectory(brainDirectory, { recursive: true }).pipe(Effect.orDie); yield* fs.writeFileString(`${sessionBase}.db`, "helper session").pipe(Effect.orDie); yield* fs.writeFileString(`${sessionBase}.db-wal`, "helper journal").pipe(Effect.orDie); @@ -279,11 +288,12 @@ const makeFixture = Effect.fn("makeAntigravityTextGenerationFixture")(function* }); it.layer(NodeServices.layer)("AntigravityTextGeneration", (it) => { - it.effect( - "generates all helper types in empty workspaces and removes only owned session files", - () => + it.effect.each([false, true])( + "generates all helper types and cleans up with startupSkills=%s", + (startupSkills) => Effect.gen(function* () { const fixture = yield* makeFixture({ + startupSkills, outputs: [ '{"subject":" Repair Google login.\\nExtra line","body":" Keep the remote callback. ","branch":"Repair Login"}', '```json\n{"title":" Repair Google login\\nExtra line","body":" ## Summary\\nSupport remote callbacks. "}\n```', @@ -442,29 +452,51 @@ it.layer(NodeServices.layer)("AntigravityTextGeneration", (it) => { }).pipe(Effect.scoped), ); - it.effect("rejects a helper that writes files in its temporary workspace", () => - Effect.gen(function* () { - const fs = yield* FileSystem.FileSystem; - const path = yield* Path.Path; - const fixture = yield* makeFixture({ - prompt: ({ cwd, emit }) => - Effect.gen(function* () { - yield* fs - .writeFileString(path.join(cwd, "unexpected.txt"), "unexpected") - .pipe(Effect.orDie); - yield* emit({ - sessionUpdate: "agent_message_chunk", - content: { type: "text", text: '{"title":"Ignored output"}' }, - }); - return { stopReason: "end_turn" }; - }), - }); - const error = yield* fixture.textGeneration - .generateThreadTitle(fixture.titleInput) - .pipe(Effect.flip); - expect(error.detail).toContain("wrote files"); - yield* fixture.assertCleaned; - }).pipe(Effect.scoped), + it.effect.each([ + { relativePath: "unexpected.txt", kind: "file" }, + { relativePath: ".agents/skills/unexpected.txt", kind: "file" }, + { relativePath: ".gemini/unexpected", kind: "directory" }, + { relativePath: ".agents/skills", kind: "file" }, + ...(symlinksSupported ? [{ relativePath: ".gemini/skills", kind: "symlink" }] : []), + ])( + "rejects an unexpected $kind at $relativePath in its temporary workspace", + ({ relativePath, kind }) => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const target = + kind === "symlink" + ? yield* fs.makeTempDirectoryScoped({ prefix: "t3-antigravity-link-test-" }) + : undefined; + const fixture = yield* makeFixture({ + startupSkills: true, + prompt: ({ cwd, emit }) => + Effect.gen(function* () { + const entry = path.join(cwd, relativePath); + if (yield* fs.exists(entry).pipe(Effect.orDie)) { + yield* fs.remove(entry, { recursive: true }).pipe(Effect.orDie); + } + yield* ( + target !== undefined + ? fs.symlink(target, entry) + : kind === "directory" + ? fs.makeDirectory(entry) + : fs.writeFileString(entry, "unexpected") + ).pipe(Effect.orDie); + yield* emit({ + sessionUpdate: "agent_message_chunk", + content: { type: "text", text: '{"title":"Ignored output"}' }, + }); + return { stopReason: "end_turn" }; + }), + }); + const error = yield* fixture.textGeneration + .generateThreadTitle(fixture.titleInput) + .pipe(Effect.flip); + expect(error.detail).toContain("wrote files"); + if (target !== undefined) expect(yield* fs.exists(target)).toBe(true); + yield* fixture.assertCleaned; + }).pipe(Effect.scoped), ); it.effect("uses the native default without sending T3's default selection as a model ID", () => @@ -576,3 +608,15 @@ it.layer(NodeServices.layer)("AntigravityTextGeneration", (it) => { }).pipe(Effect.scoped), ); }); + +it.effect("retains filesystem services supplied only during construction", () => + Effect.gen(function* () { + const fixture = yield* makeFixture({ startupSkills: true }).pipe( + Effect.provide(NodeServices.layer), + ); + expect(yield* fixture.textGeneration.generateThreadTitle(fixture.titleInput)).toEqual({ + title: "Repair login", + }); + yield* fixture.assertCleaned; + }).pipe(Effect.scoped), +); diff --git a/apps/server/src/textGeneration/AntigravityTextGeneration.ts b/apps/server/src/textGeneration/AntigravityTextGeneration.ts index 921a2f118e34..22fce172b183 100644 --- a/apps/server/src/textGeneration/AntigravityTextGeneration.ts +++ b/apps/server/src/textGeneration/AntigravityTextGeneration.ts @@ -94,6 +94,34 @@ export const makeAntigravityTextGeneration = Effect.fn("makeAntigravityTextGener ) { const fs = yield* FileSystem.FileSystem; const path = yield* Path.Path; + const hasOnlyRuntimeDirectories = Effect.fn( + "AntigravityTextGeneration.hasOnlyRuntimeDirectories", + )(function* (cwd: string) { + const root = yield* fs.realPath(cwd); + // ACP creates these empty skill directories before the first prompt. + const allowed = new Set([ + path.join(root, ".agents"), + path.join(root, ".agents", "skills"), + path.join(root, ".gemini"), + path.join(root, ".gemini", "skills"), + ]); + const directories = [root]; + for (const directory of directories) { + for (const name of yield* fs.readDirectory(directory)) { + const entry = path.join(directory, name); + if ( + !allowed.has(entry) || + (yield* fs.realPath(entry)) !== entry || + (yield* fs.stat(entry)).type !== "Directory" + ) { + return false; + } + directories.push(entry); + } + } + return true; + }); + const available = isAntigravityTextGenerationAvailable(options.profileDirectory).pipe( Effect.provideService(FileSystem.FileSystem, fs), Effect.provideService(Path.Path, path), @@ -253,7 +281,7 @@ export const makeAntigravityTextGeneration = Effect.fn("makeAntigravityTextGener ); }).pipe(Effect.scoped); - if ((yield* fs.readDirectory(cwd)).length > 0) { + if (!(yield* hasOnlyRuntimeDirectories(cwd))) { return yield* new TextGenerationError({ operation, detail: "Antigravity wrote files during text generation.", diff --git a/apps/server/src/usage/UsageService.test.ts b/apps/server/src/usage/UsageService.test.ts index 1a064a389449..b8668a4172bd 100644 --- a/apps/server/src/usage/UsageService.test.ts +++ b/apps/server/src/usage/UsageService.test.ts @@ -1516,6 +1516,37 @@ describe("UsageService", () => { }).pipe(Effect.scoped), ); + it.effect("ignores a stale macOS Keychain login when Cursor is not installed", () => + Effect.gen(function* () { + const { settings, home } = yield* setup; + const cursor = makeFakeCursor(); + cursor.state.error = "Sign in to Cursor again to read account usage."; + const readCursorSource = Effect.gen(function* () { + const service = yield* makeWithCursor(cursor.read); + const summary = yield* service.readSummary({ ...WINDOW, awaitRefresh: true }); + return cursorSource(summary); + }).pipe( + Effect.provide( + layerService({ + prefix: "usage-service-cursor-keychain-uninstalled", + home, + settings: { ...settings, cursorKeychainUsageEnabled: true }, + platform: "darwin", + environment: {}, + }), + ), + ); + + assert.isUndefined(yield* readCursorSource); + assert.lengthOf(cursor.state.calls, 0); + + yield* Effect.promise(() => NodeFSP.mkdir(NodePath.join(home, ".cursor"))); + const installed = yield* readCursorSource; + assert.strictEqual(installed?.message, "Sign in to Cursor again to read account usage."); + assert.lengthOf(cursor.state.calls, 1); + }).pipe(Effect.scoped), + ); + it.live("ignores stale Cursor file logins when the active credential store differs", () => Effect.gen(function* () { const { settings, home } = yield* setup; diff --git a/apps/server/src/usage/cliproxyApi.test.ts b/apps/server/src/usage/cliproxyApi.test.ts index df8181fe4c5c..3129b68ec9d3 100644 --- a/apps/server/src/usage/cliproxyApi.test.ts +++ b/apps/server/src/usage/cliproxyApi.test.ts @@ -352,6 +352,70 @@ describe("CLIProxyAPI built-in management API", () => { }), ); + it.effect.each([ + { + name: "spend budget", + budget: { + spend: { + enabled: true, + used: { amount_minor: 4_631, currency: "USD", exponent: 2 }, + limit: { amount_minor: 50_000, currency: "USD", exponent: 2 }, + }, + }, + }, + { + name: "extra usage", + budget: { + extra_usage: { + is_enabled: true, + used_credits: 4_631, + monthly_limit: 50_000, + utilization: 9, + }, + }, + }, + ])("keeps Claude monthly spend from $name beside rolling windows", ({ budget }) => + Effect.gen(function* () { + const test = fixture({ + accounts: [{ ...accounts[0]!, provider: "claude" }], + upstream: () => ({ + status: 200, + body: { five_hour: { utilization: 10, resets_at: null }, ...budget }, + }), + }); + const result = yield* (yield* test.api).readAccounts(config); + expect( + result[0]?.usageLimits.windows.map((window) => [window.id, window.usedPercent]), + ).toEqual([ + ["five_hour", 10], + ["monthly_spend", expect.closeTo(9.262, 10)], + ]); + }), + ); + + it.effect("omits disabled Claude budgets without losing rolling windows", () => + Effect.gen(function* () { + const test = fixture({ + accounts: [{ ...accounts[0]!, provider: "claude" }], + upstream: () => ({ + status: 200, + body: { + five_hour: { utilization: 10, resets_at: null }, + spend: { enabled: false, used: null, limit: null }, + extra_usage: { + is_enabled: false, + used_credits: null, + monthly_limit: null, + utilization: null, + }, + }, + }), + }); + const result = yield* (yield* test.api).readAccounts(config); + expect(result[0]?.usageLimits.windows.map((window) => window.id)).toEqual(["five_hour"]); + }), + ); + it.effect("pins redemption to the displayed credit and clears only that account's cooldown", () => Effect.gen(function* () { const test = fixture(); diff --git a/apps/server/src/usage/cliproxyApi.ts b/apps/server/src/usage/cliproxyApi.ts index 79884da26db7..12aa039d9f74 100644 --- a/apps/server/src/usage/cliproxyApi.ts +++ b/apps/server/src/usage/cliproxyApi.ts @@ -70,6 +70,17 @@ const ClaudeWindow = Schema.Struct({ const ClaudeUsage = Schema.Struct({ five_hour: Schema.optional(Schema.NullOr(ClaudeWindow)), seven_day: Schema.optional(Schema.NullOr(ClaudeWindow)), + spend: Schema.optional(Schema.Unknown), + extra_usage: Schema.optional( + Schema.NullOr( + Schema.Struct({ + is_enabled: Schema.Boolean, + monthly_limit: Schema.NullOr(Schema.Number), + used_credits: Schema.NullOr(Schema.Number), + utilization: Schema.NullOr(Schema.Number), + }), + ), + ), limits: Schema.optional( Schema.Array( Schema.Struct({ @@ -736,6 +747,13 @@ export const makeCliproxyApi = Effect.gen(function* () { ] : [], ); + const rateLimits = { + five_hour: usage.five_hour ?? null, + seven_day: usage.seven_day ?? null, + model_scoped, + spend: usage.spend, + extra_usage: usage.extra_usage ?? null, + }; return { ...base, plan: "Claude Subscription", @@ -743,11 +761,7 @@ export const makeCliproxyApi = Effect.gen(function* () { checkedAt, response: { rate_limits_available: true, - rate_limits: { - five_hour: usage.five_hour ?? null, - seven_day: usage.seven_day ?? null, - model_scoped, - }, + rate_limits: rateLimits, }, }).limits, }; diff --git a/apps/server/src/vcs/GitVcsDriverCore.test.ts b/apps/server/src/vcs/GitVcsDriverCore.test.ts index 8cef878fa322..5779e2b01958 100644 --- a/apps/server/src/vcs/GitVcsDriverCore.test.ts +++ b/apps/server/src/vcs/GitVcsDriverCore.test.ts @@ -2581,6 +2581,39 @@ it.layer(layerTest)("GitVcsDriver core integration", (it) => { }), ); + it.effect("reads range patches from source without running textconv", () => + Effect.gen(function* () { + const cwd = yield* makeTmpDir(); + const { initialBranch } = yield* initRepoWithCommit(cwd); + yield* git(cwd, ["config", "diff.probe.textconv", "touch textconv-ran; echo converted"]); + yield* git(cwd, ["checkout", "-b", "feature/range"]); + yield* writeTextFile(cwd, ".gitattributes", "*.doc diff=probe\n"); + yield* writeTextFile(cwd, "notes.doc", "source line\n"); + yield* writeTextFile(cwd, "plain.txt", "plain line\n"); + yield* writeTextFile(cwd, "binary.dat", "binary\0data"); + yield* git(cwd, ["add", "."]); + yield* git(cwd, ["commit", "-m", "range files"]); + const driver = yield* GitVcsDriver.GitVcsDriver; + const fileSystem = yield* FileSystem.FileSystem; + + const context = yield* driver.readRangeContext(cwd, initialBranch); + + assert.include(context.diffPatch, "+source line"); + assert.notInclude(context.diffPatch, "converted"); + assert.include(context.diffPatch, "+plain line"); + assert.include(context.diffPatch, "Binary files /dev/null and b/binary.dat differ"); + assert.include(context.diffSummary, "notes.doc"); + assert.isFalse(yield* fileSystem.exists(`${cwd}/textconv-ran`)); + + yield* writeTextFile(cwd, "notes.doc", "updated source line\n"); + const staged = yield* driver.prepareCommitContext(cwd, ["notes.doc"]); + assert.include(staged?.stagedPatch ?? "", "+updated source line"); + assert.notInclude(staged?.stagedPatch ?? "", "converted"); + assert.include(staged?.stagedSummary ?? "", "notes.doc"); + assert.isFalse(yield* fileSystem.exists(`${cwd}/textconv-ran`)); + }), + ); + it.effect("loads new and deleted files without reading their missing side", () => Effect.gen(function* () { const cwd = yield* makeTmpDir(); @@ -3666,6 +3699,47 @@ it.layer(layerTest)("GitVcsDriver core integration", (it) => { }), ); + it.effect("flattens a generated name when a branch at its parent path blocks it", () => + Effect.gen(function* () { + const cwd = yield* makeTmpDir(); + yield* initRepoWithCommit(cwd); + const driver = yield* GitVcsDriver.GitVcsDriver; + // A local `t3` branch is why launches start on the flat `t3-` temporary branch. + yield* git(cwd, ["branch", "t3"]); + yield* git(cwd, ["checkout", "-b", "t3-abcd1234"]); + yield* git(cwd, ["branch", "t3-fix-login"]); + + const renamed = yield* driver.renameBranch({ + cwd, + oldBranch: "t3-abcd1234", + newBranch: "t3/fix-login", + }); + + assert.equal(renamed.branch, "t3-fix-login-1"); + assert.equal(yield* git(cwd, ["branch", "--show-current"]), "t3-fix-login-1"); + }), + ); + + it.effect("skips a flattened name that a nested branch blocks", () => + Effect.gen(function* () { + const cwd = yield* makeTmpDir(); + yield* initRepoWithCommit(cwd); + const driver = yield* GitVcsDriver.GitVcsDriver; + yield* git(cwd, ["branch", "t3"]); + yield* git(cwd, ["checkout", "-b", "t3-abcd1234"]); + yield* git(cwd, ["branch", "t3-fix-login/child"]); + + const renamed = yield* driver.renameBranch({ + cwd, + oldBranch: "t3-abcd1234", + newBranch: "t3/fix-login", + }); + + assert.equal(renamed.branch, "t3-fix-login-1"); + assert.equal(yield* git(cwd, ["branch", "--show-current"]), "t3-fix-login-1"); + }), + ); + it.effect("rejects a missing branch without restoring a matching dirty file", () => Effect.gen(function* () { const cwd = yield* makeTmpDir(); @@ -4781,6 +4855,29 @@ it.layer(layerTest)("GitVcsDriver core integration", (it) => { }); describe("commit context", () => { + it.effect("prepares source patches without running textconv", () => + Effect.gen(function* () { + const cwd = yield* makeTmpDir(); + yield* initRepoWithCommit(cwd); + yield* git(cwd, ["config", "diff.probe.textconv", "touch textconv-ran; echo converted"]); + yield* writeTextFile(cwd, ".gitattributes", "*.doc diff=probe\n"); + yield* writeTextFile(cwd, "notes.doc", "original source\n"); + yield* git(cwd, ["add", "."]); + yield* git(cwd, ["commit", "-m", "source document"]); + yield* writeTextFile(cwd, "notes.doc", "updated source\n"); + const driver = yield* GitVcsDriver.GitVcsDriver; + const fileSystem = yield* FileSystem.FileSystem; + + const context = yield* driver.prepareCommitContext(cwd, ["notes.doc"]); + + assert.include(context?.stagedPatch ?? "", "-original source"); + assert.include(context?.stagedPatch ?? "", "+updated source"); + assert.notInclude(context?.stagedPatch ?? "", "converted"); + assert.include(context?.stagedSummary ?? "", "notes.doc"); + assert.isFalse(yield* fileSystem.exists(`${cwd}/textconv-ran`)); + }), + ); + it.effect("can omit the patch while preserving change detection and the real index", () => Effect.gen(function* () { const cwd = yield* makeTmpDir(); diff --git a/apps/server/src/vcs/GitVcsDriverCore.ts b/apps/server/src/vcs/GitVcsDriverCore.ts index 71945b0b6f71..11580f4c26e1 100644 --- a/apps/server/src/vcs/GitVcsDriverCore.ts +++ b/apps/server/src/vcs/GitVcsDriverCore.ts @@ -1247,16 +1247,37 @@ export const makeGitVcsDriverCore = Effect.fn("makeGitVcsDriverCore")(function* const resolveAvailableBranchName = Effect.fn("resolveAvailableBranchName")(function* ( cwd: string, - desiredBranch: string, + requestedBranch: string, ) { - const isDesiredTaken = yield* branchExists(cwd, desiredBranch); + // A branch at a parent path (such as `t3`) blocks every ref nested under it, so no suffix + // would help; flatten to a sibling the way temporary worktree branches already do. + const segments = requestedBranch.split("/"); + let blockedByParent = false; + for (let index = 1; index < segments.length && !blockedByParent; index += 1) { + blockedByParent = yield* branchExists(cwd, segments.slice(0, index).join("/")); + } + const desiredBranch = blockedByParent ? segments.join("-") : requestedBranch; + // A branch nested under the name (`t3-fix/child`) blocks it as surely as one with the name. + const isBranchNameTaken = (name: string) => + Effect.zipWith( + branchExists(cwd, name), + runGitStdout( + "GitVcsDriver.branchExists", + cwd, + ["for-each-ref", "--count=1", "--format=%(refname)", `refs/heads/${name}/`], + true, + ), + (exists, nested) => exists || nested.trim().length > 0, + ); + + const isDesiredTaken = yield* isBranchNameTaken(desiredBranch); if (!isDesiredTaken) { return desiredBranch; } for (let suffix = 1; suffix <= 100; suffix += 1) { const candidate = `${desiredBranch}-${suffix}`; - const isCandidateTaken = yield* branchExists(cwd, candidate); + const isCandidateTaken = yield* isBranchNameTaken(candidate); if (!isCandidateTaken) { return candidate; } @@ -2502,7 +2523,7 @@ export const makeGitVcsDriverCore = Effect.fn("makeGitVcsDriverCore")(function* const stagedPatch = yield* runGitStdoutWithOptions( "GitVcsDriver.prepareCommitContext.stagedPatch", cwd, - ["diff", "--no-ext-diff", "--cached", "--patch", "--minimal"], + ["diff", "--no-ext-diff", "--no-textconv", "--cached", "--patch", "--minimal"], { env, maxOutputBytes: PREPARED_COMMIT_PATCH_MAX_OUTPUT_BYTES, @@ -2913,7 +2934,15 @@ export const makeGitVcsDriverCore = Effect.fn("makeGitVcsDriverCore")(function* runGitStdoutWithOptions( "GitVcsDriver.readRangeContext.diffPatch", cwd, - ["diff", "--no-ext-diff", "--patch", "--minimal", "--end-of-options", diffRange], + [ + "diff", + "--no-ext-diff", + "--no-textconv", + "--patch", + "--minimal", + "--end-of-options", + diffRange, + ], { maxOutputBytes: RANGE_DIFF_PATCH_MAX_OUTPUT_BYTES, appendTruncationMarker: true, diff --git a/apps/server/src/workspace/WorkspaceEntries.test.ts b/apps/server/src/workspace/WorkspaceEntries.test.ts index 6bef5f04263c..d861935fe17c 100644 --- a/apps/server/src/workspace/WorkspaceEntries.test.ts +++ b/apps/server/src/workspace/WorkspaceEntries.test.ts @@ -1,6 +1,7 @@ // @effect-diagnostics nodeBuiltinImport:off import * as NodeFSP from "node:fs/promises"; import * as NodeServices from "@effect/platform-node/NodeServices"; +import { FileFinder } from "@ff-labs/fff-node"; import { it, afterEach, describe, expect } from "@effect/vitest"; import * as Effect from "effect/Effect"; import * as FileSystem from "effect/FileSystem"; @@ -673,6 +674,53 @@ it.layer(layerTest, { excludeTestServices: true })("WorkspaceEntries", (it) => { expect(match.lineContent.slice(range.start, range.end)).toBe("wörld"); }), ); + + it.effect("keeps answering from the live index when a refresh outlasts the scan wait", () => + Effect.gen(function* () { + const cwd = yield* makeTempDir({ prefix: "t3code-workspace-content-refresh-" }); + yield* writeTextFile(cwd, "src/notes.ts", "const needle = 1;\n"); + let scansOutlastWait = false; + let finders = 0; + let rescans = 0; + const createFinder = FileFinder.create.bind(FileFinder); + vi.spyOn(FileFinder, "create").mockImplementation((options) => { + const created = createFinder(options); + if (created.ok) { + const finder = created.value; + finders += 1; + const scanFiles = finder.scanFiles.bind(finder); + finder.scanFiles = () => { + rescans += 1; + return scanFiles(); + }; + const waitForIndexReady = finder.waitForIndexReady.bind(finder); + finder.waitForIndexReady = (timeoutMs) => + scansOutlastWait + ? Promise.resolve({ ok: true, value: false }) + : waitForIndexReady(timeoutMs); + } + return created; + }); + const workspaceEntries = yield* WorkspaceEntries.WorkspaceEntries; + const searchNeedle = workspaceEntries.searchContents({ + cwd, + query: "needle", + limit: 100, + caseSensitive: true, + wholeWord: false, + useRegex: false, + }); + yield* searchNeedle; + + scansOutlastWait = true; + yield* workspaceEntries.refresh(cwd); + const result = yield* searchNeedle; + + expect(result.matches.map((match) => match.path)).toEqual(["src/notes.ts"]); + // The second search ran on the rescanned live finder, not a rebuild. + expect({ finders, rescans }).toEqual({ finders: 1, rescans: 1 }); + }), + ); }); describe("browse", () => { diff --git a/apps/server/src/workspace/WorkspaceSearchIndex.ts b/apps/server/src/workspace/WorkspaceSearchIndex.ts index 8f2b64cf26a3..d5539e2f4e71 100644 --- a/apps/server/src/workspace/WorkspaceSearchIndex.ts +++ b/apps/server/src/workspace/WorkspaceSearchIndex.ts @@ -117,10 +117,7 @@ export class WorkspaceSearchIndex extends Context.Service< readonly searchContents: ( input: Omit, ) => Effect.Effect; - readonly refresh: () => Effect.Effect< - void, - WorkspaceSearchIndexRefreshFailed | WorkspaceSearchIndexScanTimedOut - >; + readonly refresh: () => Effect.Effect; } >()("t3/workspace/WorkspaceSearchIndex") {} @@ -419,6 +416,8 @@ export const make = Effect.fn("WorkspaceSearchIndex.make")(function* ( reason: result.error, }); } + // A rescan that outlasts the wait keeps running, and the finder answers + // from its previous index until it finishes, so the index stays usable. yield* waitForIndexReady( cwd, finder, @@ -428,7 +427,7 @@ export const make = Effect.fn("WorkspaceSearchIndex.make")(function* ( reason, cause, }), - ); + ).pipe(Effect.catchTags({ WorkspaceSearchIndexScanTimedOut: () => Effect.void })); }); const list: WorkspaceSearchIndex["Service"]["list"] = Effect.fn("WorkspaceSearchIndex.list")( diff --git a/apps/server/src/ws.test.ts b/apps/server/src/ws.test.ts index f2c3a80e561e..eecc99a1e51d 100644 --- a/apps/server/src/ws.test.ts +++ b/apps/server/src/ws.test.ts @@ -152,6 +152,7 @@ import * as ProjectEnrichment from "./project/ProjectEnrichmentService.ts"; import * as ProjectService from "./project/ProjectService.ts"; import { layerWsRpc, + resolveOpenDiscoveryForConfig, subscribeOrchestrationV2Shell, subscribeOrchestrationV2Thread, WS_RPC_SERVER_OPTIONS, @@ -217,6 +218,38 @@ it.effect("resends a file manager reveal kind that missed the snapshot", () => }), ); +it.effect("runs editor, remote open target and direct endpoint discovery side by side", () => + Effect.gen(function* () { + // Each discovery finishes only once the other two have started, so a + // config that ran them one after another would never resolve. + const editorsStarted = yield* Deferred.make(); + const targetsStarted = yield* Deferred.make(); + const endpointsStarted = yield* Deferred.make(); + const startThenAwait = (started: Deferred.Deferred, value: A) => + Deferred.succeed(started, undefined).pipe( + Effect.andThen( + Effect.forEach([editorsStarted, targetsStarted, endpointsStarted], Deferred.await), + ), + Effect.as(value), + ); + + const discovery = yield* resolveOpenDiscoveryForConfig({ + editors: startThenAwait(editorsStarted, ["file-manager" as const]), + fileManagerRevealKind: Effect.succeed("finder" as const), + remoteOpenTargets: startThenAwait(targetsStarted, []), + directEndpoints: startThenAwait(endpointsStarted, []), + }); + + assert.deepEqual(discovery, { + availableEditors: ["file-manager"], + shellRevealInFileManager: true, + shellRevealInFileManagerKind: "finder", + remoteOpenTargets: [], + directEndpoints: [], + }); + }), +); + const ReplayTestLayer = OrchestrationEventStore.layer.pipe( Layer.provideMerge(SqlitePersistence.layerMemory), ); diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 9687793e6309..059abe0e9fa8 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -337,7 +337,8 @@ export const withLateEditorConfig = ( editors.join() !== config.availableEditors.join() || (editors.includes("file-manager") && config.shellRevealInFileManagerKind === undefined), ), - // Leave the unshared reveal probe unbounded so timeouts cannot cancel every attempt on slow hosts. + // Unbounded, unlike the snapshot: this exists to deliver a probe that + // outlasts the snapshot's timeout. Stream.mapEffect((editors) => resolveEditorConfig(editors, launcher.resolveFileManagerRevealKind()), ), @@ -379,6 +380,31 @@ export const withLateEditorConfig = ( ); }; +// Discoveries run side by side so the slowest one, not their sum, bounds the +// server config snapshot. Each degrades on timeout rather than failing it. +export const resolveOpenDiscoveryForConfig = (discovery: { + readonly editors: Effect.Effect, E1, R1>; + readonly fileManagerRevealKind: Effect.Effect; + readonly remoteOpenTargets: Effect.Effect, E3, R3>; + readonly directEndpoints: Effect.Effect, E4, R4>; +}) => + Effect.all( + { + editorConfig: resolveAvailableEditorsForConfig(discovery.editors).pipe( + Effect.flatMap((availableEditors) => + resolveEditorConfig( + availableEditors, + resolveFileManagerRevealKindForConfig(discovery.fileManagerRevealKind), + ), + ), + ), + // Same discovery-with-timeout treatment as editors: a slow probe must + // not stall server.getConfig, so it degrades to an empty list. + remoteOpenTargets: resolveAvailableEditorsForConfig(discovery.remoteOpenTargets), + directEndpoints: resolveAvailableEditorsForConfig(discovery.directEndpoints), + }, + { concurrency: "unbounded" }, + ).pipe(Effect.map(({ editorConfig, ...endpoints }) => ({ ...editorConfig, ...endpoints }))); function withoutWorkspaceSnapshots(providers: ReadonlyArray): ServerProvider[] { return providers.map(({ workspaceSnapshots: _workspaceSnapshots, ...provider }) => provider); } @@ -1766,10 +1792,12 @@ export const layerWsRpc = ( const auth = yield* serverAuth.getDescriptor(); const scratchWorkspaceRoot = yield* managedFolders.scratchRoot; const lastEventLoopStall = yield* eventLoopMonitor.latestNoticeableStall; - const editorConfig = yield* resolveEditorConfig( - yield* resolveAvailableEditorsForConfig(externalLauncher.resolveAvailableEditors()), - resolveFileManagerRevealKindForConfig(externalLauncher.resolveFileManagerRevealKind()), - ); + const discovery = yield* resolveOpenDiscoveryForConfig({ + editors: externalLauncher.resolveAvailableEditors(), + fileManagerRevealKind: externalLauncher.resolveFileManagerRevealKind(), + remoteOpenTargets: remoteOpenTargets.resolveTargets(), + directEndpoints: directEndpoints.resolve(), + }); return { environment, @@ -1779,11 +1807,7 @@ export const layerWsRpc = ( keybindings: keybindingsConfig.keybindings, issues: keybindingsConfig.issues, providers, - ...editorConfig, - remoteOpenTargets: yield* resolveAvailableEditorsForConfig( - remoteOpenTargets.resolveTargets(), - ), - directEndpoints: yield* resolveAvailableEditorsForConfig(directEndpoints.resolve()), + ...discovery, observability: { logsDirectoryPath: config.logsDir, localTracingEnabled: true, diff --git a/apps/web/package.json b/apps/web/package.json index 5649bd0ffdee..29424571fcb7 100644 --- a/apps/web/package.json +++ b/apps/web/package.json @@ -1,6 +1,6 @@ { "name": "@t3tools/web", - "version": "0.1.5", + "version": "0.1.6", "private": true, "type": "module", "scripts": { diff --git a/apps/web/src/components/BranchToolbar.tsx b/apps/web/src/components/BranchToolbar.tsx index 27639044e758..93249bcf0c9f 100644 --- a/apps/web/src/components/BranchToolbar.tsx +++ b/apps/web/src/components/BranchToolbar.tsx @@ -214,9 +214,7 @@ const RunContextSelector = memo(function RunContextSelector({ const workspaceIcon = ( }> - + {isPanel @@ -232,7 +230,7 @@ const RunContextSelector = memo(function RunContextSelector({ }> {autoEnvironmentLabel ? (