From 5603db9bf246b24be5b28f16af038fe001ca5b86 Mon Sep 17 00:00:00 2001 From: dan Date: Mon, 2 Nov 2026 22:21:24 +0000 Subject: [PATCH 1/4] feat(web): optionally show running ports beside sidebar threads --- .../components/settings/SettingsPanels.tsx | 25 +++++++++++++++++++ .../src/components/settings/settingsSearch.ts | 6 +++++ .../components/sidebar/SidebarThreadRow.tsx | 25 +++++++++++++++++++ packages/contracts/src/settings.ts | 2 ++ 4 files changed, 58 insertions(+) diff --git a/apps/web/src/components/settings/SettingsPanels.tsx b/apps/web/src/components/settings/SettingsPanels.tsx index 362210f96eee1..b1e5a7f3b7e59 100644 --- a/apps/web/src/components/settings/SettingsPanels.tsx +++ b/apps/web/src/components/settings/SettingsPanels.tsx @@ -480,6 +480,9 @@ export function useSettingsRestore(onRestored?: () => void) { ...(settings.sidebarProjectSortOrder !== DEFAULT_UNIFIED_SETTINGS.sidebarProjectSortOrder ? ["Project order"] : []), + ...(settings.sidebarShowPorts !== DEFAULT_UNIFIED_SETTINGS.sidebarShowPorts + ? ["Show thread ports"] + : []), ...(settings.sidebarWorkingShelfEnabled !== DEFAULT_UNIFIED_SETTINGS.sidebarWorkingShelfEnabled ? ["Working section"] : []), @@ -648,6 +651,7 @@ export function useSettingsRestore(onRestored?: () => void) { panelAnimationDurationMs: DEFAULT_UNIFIED_SETTINGS.panelAnimationDurationMs, sidebarProjectGroupingMode: DEFAULT_UNIFIED_SETTINGS.sidebarProjectGroupingMode, sidebarProjectSortOrder: DEFAULT_UNIFIED_SETTINGS.sidebarProjectSortOrder, + sidebarShowPorts: DEFAULT_UNIFIED_SETTINGS.sidebarShowPorts, sidebarWorkingShelfEnabled: DEFAULT_UNIFIED_SETTINGS.sidebarWorkingShelfEnabled, sidebarAutoSettleAfterDays: DEFAULT_UNIFIED_SETTINGS.sidebarAutoSettleAfterDays, sidebarAutoSettleKeepPullRequests: DEFAULT_UNIFIED_SETTINGS.sidebarAutoSettleKeepPullRequests, @@ -878,6 +882,27 @@ export function GeneralSettingsPanel() { } /> + + updateSettings({ sidebarShowPorts: DEFAULT_UNIFIED_SETTINGS.sidebarShowPorts }) + } + /> + ) : null + } + control={ + updateSettings({ sidebarShowPorts: Boolean(checked) })} + aria-label="Show thread ports" + /> + } + /> ; +function ThreadPorts({ thread }: { thread: SidebarThreadSummary }) { + const enabled = useClientSettings((settings) => settings.sidebarShowPorts); + const servers = useThreadDiscoveredPorts({ + environmentId: enabled ? thread.environmentId : null, + threadId: thread.id, + }); + const ports = [...new Set(servers.map((server) => server.port))].sort((a, b) => a - b); + if (ports.length === 0) return null; + return ( + + + ); +} + function compactSidebarTimeLabel(label: string): string { if (label === "just now") return "now"; return label.endsWith(" ago") ? label.slice(0, -4) : label; @@ -1249,6 +1271,7 @@ const SidebarThreadRow = memo(function SidebarThreadRow(props: SidebarThreadRowP )} {draftIndicator} {title} + {multiline ? titleLineIcons : null} {pinIndicator} {multiline ? null : terminalStatusIcon} @@ -1574,6 +1597,7 @@ const SidebarThreadRow = memo(function SidebarThreadRow(props: SidebarThreadRowP
{title} + {isRegeneratingTitle ? ( Regenerating title @@ -1738,6 +1762,7 @@ export const SidebarSearchResultRow = memo(function SidebarSearchResultRow(props > {thread.title} + Date: Tue, 3 Nov 2026 17:33:00 +0000 Subject: [PATCH 2/4] fix(web): constrain and announce sidebar thread ports --- .../sidebar/SidebarThreadRow.browser.test.tsx | 41 ++++++++++++++++ .../components/sidebar/SidebarThreadRow.tsx | 48 ++++++++++++++----- 2 files changed, 77 insertions(+), 12 deletions(-) create mode 100644 apps/web/src/components/sidebar/SidebarThreadRow.browser.test.tsx diff --git a/apps/web/src/components/sidebar/SidebarThreadRow.browser.test.tsx b/apps/web/src/components/sidebar/SidebarThreadRow.browser.test.tsx new file mode 100644 index 0000000000000..2712a875b3661 --- /dev/null +++ b/apps/web/src/components/sidebar/SidebarThreadRow.browser.test.tsx @@ -0,0 +1,41 @@ +import { expect, it, vi } from "vite-plus/test"; + +import { render } from "../../../test/render"; +import { __setClientSettingsForTests, getClientSettings } from "../../hooks/useSettings"; +import { makeThreadFixture } from "../../test-fixtures"; +import { SidebarSearchResultRow } from "./SidebarThreadRow"; + +vi.mock("../../portDiscoveryState", () => ({ + useThreadDiscoveredPorts: () => [{ port: 3000 }, { port: 4000 }, { port: 5000 }], +})); + +it("announces running ports when navigating a search result with an explicit name", async () => { + const initial = getClientSettings(); + using _settings = { [Symbol.dispose]: () => __setClientSettingsForTests(initial) }; + __setClientSettingsForTests({ ...initial, sidebarShowPorts: true }); + await using screen = await render( +
    + +
, + ); + + await expect + .element(screen.getByRole("option", { name: "Dev server" })) + .toHaveAccessibleDescription("Running ports: 3000, 4000, 5000"); +}); diff --git a/apps/web/src/components/sidebar/SidebarThreadRow.tsx b/apps/web/src/components/sidebar/SidebarThreadRow.tsx index 1ab5478587ee5..3ac6922d4c7da 100644 --- a/apps/web/src/components/sidebar/SidebarThreadRow.tsx +++ b/apps/web/src/components/sidebar/SidebarThreadRow.tsx @@ -54,6 +54,7 @@ import { } from "lucide-react"; import { memo, + useId, useRef, useState, type KeyboardEvent as ReactKeyboardEvent, @@ -132,7 +133,13 @@ export type SidebarProviderEntry = Pick< | "acpRegistryIconUrl" | "models" >; -function ThreadPorts({ thread }: { thread: SidebarThreadSummary }) { +function ThreadPorts({ + thread, + descriptionId, +}: { + thread: SidebarThreadSummary; + descriptionId: string; +}) { const enabled = useClientSettings((settings) => settings.sidebarShowPorts); const servers = useThreadDiscoveredPorts({ environmentId: enabled ? thread.environmentId : null, @@ -140,14 +147,27 @@ function ThreadPorts({ thread }: { thread: SidebarThreadSummary }) { }); const ports = [...new Set(servers.map((server) => server.port))].sort((a, b) => a - b); if (ports.length === 0) return null; + const description = `Running ports: ${ports.join(", ")}`; return ( - - + <> + + {description} + + + + {description} + + ); } @@ -675,6 +695,7 @@ const SidebarThreadRow = memo(function SidebarThreadRow(props: SidebarThreadRowP variant, variantAction, } = props; + const portsDescriptionId = useId(); const threadRef = scopeThreadRef(thread.environmentId, thread.id); const threadKey = scopedThreadKey(threadRef); const canOperateThread = useEnvironmentScope(thread.environmentId, AuthOrchestrationOperateScope); @@ -1235,6 +1256,7 @@ const SidebarThreadRow = memo(function SidebarThreadRow(props: SidebarThreadRowP data-sidebar-row={threadKey} data-sidebar-parent={props.navigationParent} aria-label={accessibility.label} + aria-describedby={portsDescriptionId} aria-current={accessibility.current} data-testid={inFolder ? "sidebar-row-folder" : "sidebar-row-slim"} aria-busy={isRegeneratingTitle || undefined} @@ -1271,7 +1293,7 @@ const SidebarThreadRow = memo(function SidebarThreadRow(props: SidebarThreadRowP )} {draftIndicator} {title} - + {multiline ? titleLineIcons : null} {pinIndicator} {multiline ? null : terminalStatusIcon} @@ -1432,6 +1454,7 @@ const SidebarThreadRow = memo(function SidebarThreadRow(props: SidebarThreadRowP data-sidebar-row={threadKey} data-sidebar-parent={props.navigationParent} aria-label={accessibility.label} + aria-describedby={portsDescriptionId} aria-current={accessibility.current} data-testid="sidebar-row-card" aria-busy={isRegeneratingTitle || undefined} @@ -1597,7 +1620,7 @@ const SidebarThreadRow = memo(function SidebarThreadRow(props: SidebarThreadRowP
{title} - + {isRegeneratingTitle ? ( Regenerating title @@ -1655,6 +1678,7 @@ export const SidebarSearchResultRow = memo(function SidebarSearchResultRow(props projectDisplayName: props.projectDisplayName, isActive: props.isRouteActive, }); + const portsDescriptionId = useId(); const threadRef = scopeThreadRef(thread.environmentId, thread.id); const { leaseLiveStatus, rowRef } = useSidebarRowSubscriptionLease( props.isHighlighted || props.isRouteActive, @@ -1721,7 +1745,7 @@ export const SidebarSearchResultRow = memo(function SidebarSearchResultRow(props tabIndex={-1} aria-selected={props.isHighlighted} aria-current={accessibility.current} - aria-describedby={`${props.resultId}-pr`} + aria-describedby={`${props.resultId}-pr ${portsDescriptionId}`} aria-label={accessibility.label} onMouseMove={props.onHighlight} onClick={props.onSelect} @@ -1762,7 +1786,7 @@ export const SidebarSearchResultRow = memo(function SidebarSearchResultRow(props > {thread.title} - + Date: Tue, 3 Nov 2026 20:35:44 +0000 Subject: [PATCH 3/4] fix(server): discover servers started by active provider turns --- .../DelegatedCompletionDelivery.test.ts | 3 ++- .../ProjectSettingsUpgrade.integration.test.ts | 3 ++- .../ProviderSessionManager.test.ts | 17 +++++++++++++++++ .../orchestration-v2/ProviderSessionManager.ts | 9 +++++++++ .../src/orchestration-v2/runtimeLayer.test.ts | 3 ++- .../testkit/ProviderReplayHarness.ts | 2 ++ 6 files changed, 34 insertions(+), 3 deletions(-) diff --git a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts index cabc322d31331..ae41ea66ddcc3 100644 --- a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts +++ b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts @@ -45,13 +45,14 @@ import * as EventSink from "./EventSink.ts"; import * as Orchestrator from "./Orchestrator.ts"; import { continueRestartedRun } from "./RestartContinuation.ts"; import * as RuntimeLayer from "./runtimeLayer.ts"; +import * as PortScanner from "../preview/PortScanner.ts"; import * as ThreadCommandExecutor from "./ThreadCommandExecutor.ts"; import * as ProviderTurnStartServiceTestkit from "./ProviderTurnStartService.testkit.ts"; import type * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter"; import * as ProviderContinuationRequests from "@t3tools/provider-core/server/ProviderContinuationRequests"; const layerPlatformTest = Layer.merge( - Layer.mergeAll(NodeServices.layer, NodeCrypto.layer), + Layer.mergeAll(NodeServices.layer, NodeCrypto.layer, PortScanner.layerTest), Layer.mock(SourceControlProviderRegistry.SourceControlProviderRegistry)({ resolveLink: () => Effect.die("unused title link"), }), diff --git a/apps/server/src/orchestration-v2/ProjectSettingsUpgrade.integration.test.ts b/apps/server/src/orchestration-v2/ProjectSettingsUpgrade.integration.test.ts index 03d4dcff87ef0..a51fa89accbb3 100644 --- a/apps/server/src/orchestration-v2/ProjectSettingsUpgrade.integration.test.ts +++ b/apps/server/src/orchestration-v2/ProjectSettingsUpgrade.integration.test.ts @@ -23,6 +23,7 @@ import * as ProjectEnrichmentService from "../project/ProjectEnrichmentService.t import * as ProjectService from "../project/ProjectService.ts"; import * as ProviderInstanceRegistry from "../provider/ProviderInstanceRegistry.ts"; import * as ServerSettings from "../serverSettings.ts"; +import * as PortScanner from "../preview/PortScanner.ts"; import * as SourceControlProviderRegistry from "../sourceControl/SourceControlProviderRegistry.ts"; import * as VcsDriverRegistry from "../vcs/VcsDriverRegistry.ts"; import * as VcsProcess from "../vcs/VcsProcess.ts"; @@ -117,7 +118,7 @@ const unusedEnrichment = { const layerRuntime = (dbPath: string) => { const layerPlatform = Layer.merge( - Layer.mergeAll(NodeServices.layer, NodeCrypto.layer), + Layer.mergeAll(NodeServices.layer, NodeCrypto.layer, PortScanner.layerTest), Layer.mock(SourceControlProviderRegistry.SourceControlProviderRegistry)({ resolveLink: () => Effect.die("unused"), }), diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts index d8c046c018651..811c9680488e7 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts @@ -41,6 +41,7 @@ import * as McpProviderSessions from "@t3tools/provider-core/server/McpProviderS import * as McpSessionRegistry from "../mcp/McpSessionRegistry.ts"; import * as SqlitePersistence from "../persistence/Sqlite.ts"; import * as ServerSettings from "../serverSettings.ts"; +import * as PortScanner from "../preview/PortScanner.ts"; import { CodexProviderCapabilitiesV2 } from "./Adapters/CodexAdapterV2.ts"; import * as EventSink from "./EventSink.ts"; import * as EventStore from "./EventStore.ts"; @@ -589,6 +590,7 @@ function layerTest(input: { readonly scopeCloseReached?: Deferred.Deferred; readonly serverSettingsLayer?: ReturnType; readonly projectServiceLayer?: Layer.Layer; + readonly portDiscoveryLayer?: Layer.Layer; }) { const layerConfiguredEventSink = input.flakyReleaseWrites !== undefined @@ -669,6 +671,7 @@ function layerTest(input: { layerProviderEventIngestorTest, layerConfiguredMcpRegistry, layerTestStores, + input.portDiscoveryLayer ?? PortScanner.layerTest, ...(input.serverSettingsLayer === undefined ? [] : [input.serverSettingsLayer]), ...(input.projectServiceLayer === undefined ? [] : [input.projectServiceLayer]), ), @@ -1361,6 +1364,14 @@ it.effect( // Like OpenCode2, a stopped start can still emit turn.terminal later. const holdStart = yield* Ref.make(false); const startHeld = yield* Deferred.make(); + const discoveryLeases = new Set(); + const portDiscoveryLayer = Layer.mock(PortScanner.PortDiscovery)({ + trackAgent: () => { + const lease = Symbol(); + discoveryLeases.add(lease); + return () => void discoveryLeases.delete(lease); + }, + }); yield* Effect.gen(function* () { const eventSink = yield* EventSink.EventSinkV2; const idAllocator = yield* IdAllocator.IdAllocatorV2; @@ -1423,11 +1434,13 @@ it.effect( }); yield* startTurn(threadB); + assert.equal(discoveryLeases.size, 1); yield* Ref.set(holdStart, true); const startingA = yield* startTurn(threadA).pipe(Effect.forkChild); yield* Deferred.await(startHeld); yield* Fiber.interrupt(startingA); + assert.equal(discoveryLeases.size, 1); const queue = (yield* Ref.get(state)).eventQueues.get(String(providerSessionId)); assert.isDefined(queue); @@ -1449,11 +1462,15 @@ it.effect( yield* Effect.yieldNow; assert.isTrue(Option.isSome(yield* manager.get(providerSessionId))); assert.equal((yield* Ref.get(state)).closeCount, 0); + assert.equal(discoveryLeases.size, 1); + yield* manager.close(providerSessionId); + assert.equal(discoveryLeases.size, 0); }).pipe( Effect.provide( layerTest({ state, idleTimeoutMs: 1000, + portDiscoveryLayer, startTurn: Ref.get(holdStart).pipe( Effect.flatMap((hold) => hold diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.ts index f06011676c667..0bf168a201740 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.ts @@ -42,6 +42,7 @@ import { } from "../observability/Metrics.ts"; import { ProviderWorkspaceMissingError } from "../provider/Errors.ts"; import * as ProjectService from "../project/ProjectService.ts"; +import * as PortScanner from "../preview/PortScanner.ts"; import * as McpProviderSessions from "@t3tools/provider-core/server/McpProviderSessions"; import * as ServerSettings from "../serverSettings.ts"; import * as McpSessionRegistry from "../mcp/McpSessionRegistry.ts"; @@ -230,6 +231,7 @@ interface LiveSessionEntry { readonly scope: Scope.Closeable; readonly idleGeneration: number; readonly busyTurns: ReadonlySet; + readonly releasePortDiscovery: (() => void) | null; readonly idleFiber: Fiber.Fiber | null; readonly pinnedSinceMs: number | null; readonly idleThreadUnloads: ReadonlyMap; @@ -329,11 +331,13 @@ export const layerWithOptions = ( | ProjectionStore.ProjectionStoreV2 | ProviderEventIngestor.ProviderEventIngestorV2 | ProviderAdapterRegistry.ProviderAdapterRegistryV2 + | PortScanner.PortDiscovery > => Layer.effect( ProviderSessionManagerV2, Effect.gen(function* () { const registry = yield* ProviderAdapterRegistry.ProviderAdapterRegistryV2; + const portDiscovery = yield* PortScanner.PortDiscovery; const fileSystem = yield* FileSystem.FileSystem; const mcpSessionRegistry = yield* McpSessionRegistry.McpSessionRegistry; const mcpSessions = yield* McpProviderSessions.McpProviderSessions; @@ -953,6 +957,7 @@ export const layerWithOptions = ( } const updated = new Map(current); updated.delete(key); + existing.releasePortDiscovery?.(); return ["removed", updated] as const; }), DateTime.now, @@ -1438,6 +1443,7 @@ export const layerWithOptions = ( updated.set(key, { ...entry, busyTurns: new Set(entry.busyTurns).add(turnKey), + releasePortDiscovery: entry.releasePortDiscovery ?? portDiscovery.trackAgent(), idleFiber: null, pinnedSinceMs: null, }); @@ -1465,10 +1471,12 @@ export const layerWithOptions = ( } const busyTurns = new Set(entry.busyTurns); busyTurns.delete(busyTurnKey(providerThreadId, runOrdinal)); + if (busyTurns.size === 0) entry.releasePortDiscovery?.(); const updated = new Map(current); updated.set(key, { ...entry, busyTurns, + releasePortDiscovery: busyTurns.size === 0 ? null : entry.releasePortDiscovery, }); return updated; }); @@ -2250,6 +2258,7 @@ export const layerWithOptions = ( scope: sessionScope, idleGeneration: 0, busyTurns: new Set(), + releasePortDiscovery: null, idleFiber: null, pinnedSinceMs: null, idleThreadUnloads: new Map(), diff --git a/apps/server/src/orchestration-v2/runtimeLayer.test.ts b/apps/server/src/orchestration-v2/runtimeLayer.test.ts index b38fb737c91db..7d5c8638aebcd 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.test.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.test.ts @@ -86,6 +86,7 @@ import * as ManagedPolicy from "../policy/ManagedPolicy.ts"; import * as Scheduler from "../scheduling/Scheduler.ts"; import * as SecretRequests from "../secrets/SecretRequests.ts"; import * as PreviewManager from "../preview/Manager.ts"; +import * as PortScanner from "../preview/PortScanner.ts"; import * as TerminalManager from "../terminal/Manager.ts"; import * as ResourceCleanupService from "./ResourceCleanupService.ts"; import * as ThreadLaunchService from "./ThreadLaunchService.ts"; @@ -93,7 +94,7 @@ import * as ThreadCommandExecutor from "./ThreadCommandExecutor.ts"; import type * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter"; const layerPlatformTest = Layer.merge( - Layer.mergeAll(NodeServices.layer, NodeCrypto.layer), + Layer.mergeAll(NodeServices.layer, NodeCrypto.layer, PortScanner.layerTest), Layer.mock(SourceControlProviderRegistry.SourceControlProviderRegistry)({ resolveLink: () => Effect.die("unused title link"), }), diff --git a/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts b/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts index 746ff0f844fe2..dfe32404bdfe4 100644 --- a/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts +++ b/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts @@ -20,6 +20,7 @@ import * as CheckpointStore from "../../checkpointing/CheckpointStore.ts"; import * as ServerConfig from "../../config.ts"; import * as SqlitePersistence from "../../persistence/Sqlite.ts"; import * as ServerSettings from "../../serverSettings.ts"; +import * as PortScanner from "../../preview/PortScanner.ts"; import * as ManagedPolicy from "../../policy/ManagedPolicy.ts"; import * as ThreadManagementService from "../ThreadManagementService.ts"; import * as McpProviderSessions from "@t3tools/provider-core/server/McpProviderSessions"; @@ -542,6 +543,7 @@ export function layerWithRegistry( layerCommandOutputHubProvided, ).pipe( Layer.provide(ProviderTurnStartServiceTestkit.layer), + Layer.provide(PortScanner.layerTest), Layer.provide(Layer.mergeAll(NodeServices.layer, NodeCrypto.layer)), ); From d85e140794b96ddde03cc72c5b7bc122e1530176 Mon Sep 17 00:00:00 2001 From: dan Date: Tue, 3 Nov 2026 20:58:20 +0000 Subject: [PATCH 4/4] fix(server): retain port discovery for background work --- .../ProviderSessionManager.test.ts | 201 +++++++++++++++++- .../ProviderSessionManager.ts | 53 ++++- 2 files changed, 245 insertions(+), 9 deletions(-) diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts index 811c9680488e7..d55b7662a49c1 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts @@ -3,6 +3,7 @@ import * as NodeCrypto from "@effect/platform-node/NodeCrypto"; import * as NodeServices from "@effect/platform-node/NodeServices"; import { assert, it } from "@effect/vitest"; import { AgentScope } from "@t3tools/shared/AgentScope"; +import * as HostProcess from "@t3tools/shared/HostProcess"; import { EnvironmentId, type ModelSelection, @@ -31,7 +32,7 @@ import * as Queue from "effect/Queue"; import * as Ref from "effect/Ref"; import * as Stream from "effect/Stream"; import { TestClock } from "effect/testing"; -import { HttpServer } from "effect/http"; +import { FetchHttpClient, HttpServer } from "effect/http"; import { ProviderWorkspaceMissingError } from "../provider/Errors.ts"; import * as ServerEnvironment from "../environment/ServerEnvironment.ts"; @@ -42,6 +43,7 @@ import * as McpSessionRegistry from "../mcp/McpSessionRegistry.ts"; import * as SqlitePersistence from "../persistence/Sqlite.ts"; import * as ServerSettings from "../serverSettings.ts"; import * as PortScanner from "../preview/PortScanner.ts"; +import * as ProcessRunner from "../processRunner.ts"; import { CodexProviderCapabilitiesV2 } from "./Adapters/CodexAdapterV2.ts"; import * as EventSink from "./EventSink.ts"; import * as EventStore from "./EventStore.ts"; @@ -1484,6 +1486,203 @@ it.effect( }), ); +const openDiscoveryTurn = Effect.fnUntraced(function* (threadId: ThreadId) { + const eventSink = yield* EventSink.EventSinkV2; + const idAllocator = yield* IdAllocator.IdAllocatorV2; + const manager = yield* ProviderSessionManager.ProviderSessionManagerV2; + const projectionStore = yield* ProjectionStore.ProjectionStoreV2; + const now = yield* DateTime.now; + const providerSessionId = idAllocator.derive.providerSession({ + providerInstanceId: modelSelection.instanceId, + }); + yield* eventSink.write({ + events: [yield* makeThreadCreatedEvent({ idAllocator, threadId, now })], + }); + const open = manager.open({ threadId, providerSessionId, modelSelection, runtimePolicy }); + const runtime = yield* open; + const providerThread = makeProviderThread({ idAllocator, threadId, providerSessionId, now }); + const runId = idAllocator.derive.run({ threadId, ordinal: 1 }); + const input = { + appThread: (yield* projectionStore.getThreadProjection(threadId)).thread, + threadId, + runId, + runOrdinal: 1, + providerTurnOrdinal: 1, + attemptId: idAllocator.derive.runAttempt({ runId, attemptOrdinal: 1 }), + rootNodeId: idAllocator.derive.rootNode({ runId }), + providerThread, + message: { + createdBy: "user", + creationSource: "web", + messageId: yield* idAllocator.allocate.message({ threadId, ordinal: 1 }), + text: "start a server", + attachments: [], + }, + modelSelection, + runtimePolicy, + } satisfies Parameters[0]; + const terminal = { + type: "turn.terminal", + driver: CODEX_DRIVER, + providerThreadId: providerThread.id, + providerTurnId: idAllocator.derive.providerTurn({ + driver: CODEX_DRIVER, + nativeTurnId: "discovery-turn", + }), + runOrdinal: 1, + status: "completed", + failure: null, + threadDisposition: "reusable", + } satisfies ProviderAdapter.ProviderAdapterV2Event; + return { manager, providerSessionId, runtime, open, input, terminal }; +}); + +it.effect("ProviderSessionManagerV2 ignores idle signals from a replaced runtime", () => + Effect.gen(function* () { + const state = yield* Ref.make(emptyState); + const startHeld = yield* Deferred.make(); + const finishStart = yield* Deferred.make(); + const holdStart = yield* Ref.make(true); + const discoveryLeases = new Set(); + const portDiscoveryLayer = Layer.mock(PortScanner.PortDiscovery)({ + trackAgent: () => { + const lease = Symbol(); + discoveryLeases.add(lease); + return () => void discoveryLeases.delete(lease); + }, + }); + yield* Effect.gen(function* () { + const { manager, providerSessionId, runtime, open, input, terminal } = + yield* openDiscoveryTurn(ThreadId.make("discovery-replaced-runtime")); + const starting = yield* runtime.startTurn(input).pipe(Effect.forkChild); + yield* Deferred.await(startHeld); + assert.equal(discoveryLeases.size, 1); + const oldQueue = (yield* Ref.get(state)).eventQueues.get(String(providerSessionId))!; + yield* manager.close(providerSessionId); + assert.equal(discoveryLeases.size, 0); + const replacement = yield* open; + yield* replacement.startTurn(input); + assert.equal(discoveryLeases.size, 1); + + yield* Deferred.succeed(finishStart, undefined); + assert.isTrue(Exit.isFailure(yield* Fiber.await(starting))); + assert.equal(discoveryLeases.size, 1); + + const staleSubscription = yield* runtime.subscribeEvents!; + const lateEvent = yield* staleSubscription.events.pipe(Stream.runHead, Effect.forkChild); + yield* Queue.offer(oldQueue, terminal); + assert.deepStrictEqual(yield* Fiber.join(lateEvent), Option.some(terminal)); + assert.equal(discoveryLeases.size, 1); + yield* manager.close(providerSessionId); + assert.equal(discoveryLeases.size, 0); + }).pipe( + Effect.provide( + layerTest({ + state, + idleTimeoutMs: 60_000, + portDiscoveryLayer, + startTurn: Effect.gen(function* () { + if (!(yield* Ref.getAndSet(holdStart, false))) return; + yield* Deferred.succeed(startHeld, undefined); + yield* Deferred.await(finishStart); + return yield* Effect.die("The replaced runtime failed its start."); + }), + }), + ), + ); + }), +); + +it.effect("ProviderSessionManagerV2 discovers listeners started after a turn ends", () => + Effect.gen(function* () { + const state = yield* Ref.make(emptyState); + const pending = yield* Ref.make(true); + const threadId = ThreadId.make("discovery-background-work"); + let listeners = ""; + let scans = 0; + const portDiscoveryLayer = PortScanner.layer.pipe( + Layer.provide( + Layer.mergeAll( + Layer.succeed(HostProcess.Platform, "linux"), + FileSystem.layerNoop({ + readFileString: () => Effect.succeed(`${PortScanner.AGENT_THREAD_ENV}=${threadId}\0`), + }), + Layer.succeed(ProcessRunner.ProcessRunner, { + run: () => + Effect.sync(() => { + scans += 1; + return { + stdout: listeners, + stderr: "", + code: null, + timedOut: false, + stdoutTruncated: false, + stderrTruncated: false, + stdoutInvalidUtf8: false, + stderrInvalidUtf8: false, + }; + }), + }), + FetchHttpClient.layer, + ), + ), + ); + yield* Effect.gen(function* () { + const scanner = yield* PortScanner.PortDiscovery; + const { manager, providerSessionId, runtime, input, terminal } = + yield* openDiscoveryTurn(threadId); + yield* runtime.startTurn(input); + const subscription = yield* runtime.subscribeEvents!; + const observed = yield* Queue.unbounded(); + yield* subscription.events.pipe( + Stream.runForEach((event) => Queue.offer(observed, event)), + Effect.forkChild, + ); + const events = (yield* Ref.get(state)).eventQueues.get(String(providerSessionId))!; + yield* Queue.offer(events, terminal); + assert.deepStrictEqual(yield* Queue.take(observed), terminal); + assert.deepStrictEqual(yield* scanner.scan(), []); + + const backgroundEvent = { + type: "background_work.updated", + driver: CODEX_DRIVER, + providerTurnId: terminal.providerTurnId, + pending: true, + } satisfies ProviderAdapter.ProviderAdapterV2Event; + yield* Queue.offer(events, backgroundEvent); + yield* Queue.take(observed); + listeners = "p4242\ncnode\nn127.0.0.1:43171\n"; + assert.deepStrictEqual( + (yield* scanner.scan()).map((server) => ({ port: server.port, agent: server.agent })), + [{ port: 43171, agent: { threadId } }], + ); + + yield* Ref.set(pending, false); + yield* Queue.offer(events, { ...backgroundEvent, pending: false }); + yield* Queue.take(observed); + listeners = ""; + assert.deepStrictEqual(yield* scanner.scan(), []); + const scansAfterWork = scans; + assert.deepStrictEqual(yield* scanner.scan(), []); + assert.equal(scans, scansAfterWork); + yield* manager.close(providerSessionId); + }).pipe( + Effect.provide( + Layer.mergeAll( + portDiscoveryLayer, + layerTest({ + state, + idleTimeoutMs: 60_000, + portDiscoveryLayer, + hasPendingBackgroundWork: Ref.get(pending), + hasPendingBackgroundWorkForThread: Ref.get(pending), + }), + ), + ), + ); + }), +); + it.effect("ProviderSessionManagerV2 gives up on a provider that never confirms a Stop", () => Effect.gen(function* () { const state = yield* Ref.make(emptyState); diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.ts index 0bf168a201740..1efb99381bee1 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.ts @@ -1455,31 +1455,60 @@ export const layerWithOptions = ( }), ); + // Rechecked on provider events, including streaming updates; avoid a span per event. + const releaseIdlePortDiscovery = Effect.fnUntraced(function* ( + runtime: ProviderAdapter.ProviderAdapterV2SessionRuntime, + ) { + const key = sessionKey(runtime.providerSessionId); + const entry = (yield* Ref.get(sessions)).get(key); + if ( + entry?.runtime !== runtime || + entry.busyTurns.size > 0 || + entry.releasePortDiscovery === null + ) { + return; + } + const pending = + runtime.hasPendingBackgroundWork === undefined + ? false + : yield* runtime.hasPendingBackgroundWork.pipe( + Effect.catchCause(() => Effect.succeed(true)), + ); + if (pending) return; + yield* Ref.update(sessions, (current) => { + const latest = current.get(key); + if (latest?.runtime !== runtime || latest.busyTurns !== entry.busyTurns) return current; + latest.releasePortDiscovery?.(); + return new Map(current).set(key, { ...latest, releasePortDiscovery: null }); + }); + }); + const markIdle = ( providerSessionId: ProviderSessionId, providerThreadId: ProviderThreadId, runOrdinal: number, + runtime: ProviderAdapter.ProviderAdapterV2SessionRuntime, ) => withActivityError( providerSessionId, Effect.gen(function* () { const key = sessionKey(providerSessionId); - yield* Ref.update(sessions, (current) => { + const marked = yield* Ref.modify(sessions, (current) => { const entry = current.get(key); - if (entry === undefined) { - return current; + if (entry === undefined || entry.runtime !== runtime) { + return [false, current] as const; } const busyTurns = new Set(entry.busyTurns); busyTurns.delete(busyTurnKey(providerThreadId, runOrdinal)); - if (busyTurns.size === 0) entry.releasePortDiscovery?.(); const updated = new Map(current); updated.set(key, { ...entry, busyTurns, - releasePortDiscovery: busyTurns.size === 0 ? null : entry.releasePortDiscovery, }); - return updated; + return [true, updated] as const; }); + if (!marked) return; + yield* releaseIdlePortDiscovery(runtime); yield* scheduleIdleReleaseInternal(providerSessionId); yield* scheduleThreadUnload(providerSessionId, providerThreadId); }), @@ -1872,7 +1901,12 @@ export const layerWithOptions = ( Exit.isFailure(exit) ? observeActivity( providerSessionId, - markIdle(providerSessionId, input.providerThread.id, input.runOrdinal), + markIdle( + providerSessionId, + input.providerThread.id, + input.runOrdinal, + runtime, + ), ) : Effect.void, ), @@ -1979,8 +2013,11 @@ export const layerWithOptions = ( entry.runtime.providerSessionId, event.providerThreadId, event.runOrdinal, + entry.runtime, ) - : touchActivity(entry.runtime.providerSessionId, { providerActivity: true }), + : touchActivity(entry.runtime.providerSessionId, { providerActivity: true }).pipe( + Effect.andThen(releaseIdlePortDiscovery(entry.runtime)), + ), ).pipe( Effect.andThen( event.type === "provider_session.updated"