diff --git a/MISSION.md b/MISSION.md new file mode 100644 index 000000000000..af361778ff4b --- /dev/null +++ b/MISSION.md @@ -0,0 +1,3 @@ +# Mission + +Extend T3 Code with a fleet management pane that lets users discover agents, inspect their work live, and message them directly through their native harnesses. Keep orchestration and routing in their existing tools. diff --git a/OBJECTIVE.md b/OBJECTIVE.md new file mode 100644 index 000000000000..83f6d90e344a --- /dev/null +++ b/OBJECTIVE.md @@ -0,0 +1,5 @@ +# Objective + +Make T3 Code a working control surface for agents across connected machines, covering Codex, Claude Code, OpenCode, and Grok CLI. Users can see agent status, inspect live sessions, and message any agent, including the brain/planner, dispatcher, and implementation agents. + +Start by proving that one externally launched dispatcher appears in T3, streams its work, and receives and answers a message in the same native session without disrupting its ongoing task. Expand that proven integration across agents, harnesses, and machines. diff --git a/VISION.md b/VISION.md new file mode 100644 index 000000000000..daead55de4c6 --- /dev/null +++ b/VISION.md @@ -0,0 +1,3 @@ +# Vision + +One surface to understand and direct the entire agent fleet, across every machine and harness. diff --git a/apps/server/src/auth/RpcAuthorization.ts b/apps/server/src/auth/RpcAuthorization.ts index 8ab1520a6f34..5a19c17fe2b2 100644 --- a/apps/server/src/auth/RpcAuthorization.ts +++ b/apps/server/src/auth/RpcAuthorization.ts @@ -115,6 +115,10 @@ export const RPC_REQUIRED_SCOPES = { [WS_METHODS.filesystemBrowse]: AuthOrchestrationReadScope, [WS_METHODS.agentSessionsScan]: AuthOrchestrationReadScope, [WS_METHODS.agentSessionsImport]: AuthOrchestrationOperateScope, + [WS_METHODS.fleetListAgents]: AuthOrchestrationReadScope, + [WS_METHODS.fleetReadThread]: AuthOrchestrationReadScope, + [WS_METHODS.fleetSendMessage]: AuthOrchestrationOperateScope, + [WS_METHODS.fleetSubscribe]: AuthOrchestrationReadScope, [WS_METHODS.assetsCreateUrl]: AuthOrchestrationReadScope, [WS_METHODS.attachmentsCreateUploadUrl]: AuthOrchestrationOperateScope, [WS_METHODS.attachmentsDelete]: AuthOrchestrationOperateScope, diff --git a/apps/server/src/fleet/CodexNativeWs.test.ts b/apps/server/src/fleet/CodexNativeWs.test.ts new file mode 100644 index 000000000000..e4d976934685 --- /dev/null +++ b/apps/server/src/fleet/CodexNativeWs.test.ts @@ -0,0 +1,332 @@ +/** + * CodexNativeWs transport tests. + * + * An in-memory socket pair stands in for a real TCP WebSocket while the + * production `openFleetNativeClient` path runs unchanged: the same + * JSON-RPC framing, request/response matching, and notification fan-out + * from the vendored protocol layer. A fake native peer replays the proven + * sequence (initialize, loaded/list, metadata-only resume, steer) and + * records every frame so tests assert the exact wire params. + * + * Delivery is synchronous (no timers, no sleeps): frames move through + * queues between fibers, and tests yield control with `Effect.yieldNow` + * until the expected frame lands, failing fast on a bounded poll instead + * of hanging on a timeout. + */ +import { describe, expect, it } from "@effect/vitest"; +import * as Effect from "effect/Effect"; +import * as Exit from "effect/Exit"; +import * as Fiber from "effect/Fiber"; +import * as Ref from "effect/Ref"; +import * as Scope from "effect/Scope"; +import * as Stream from "effect/Stream"; + +import { + openFleetNativeClient, + type FleetNativeSocket, + type FleetNativeSocketFactory, +} from "./CodexNativeWs.ts"; +import type * as CodexError from "effect-codex-app-server/errors"; + +const THREAD_ID = "01a0d431-b396-7023-a2aa-bc7ed6c6bc0c"; +const TURN_ID = "01a0d431-b3c1-70e1-957e-88bc5fe51d2b"; + +interface WireFrame { + readonly id?: unknown; + readonly method?: unknown; + readonly params?: unknown; + readonly result?: unknown; +} + +interface FakePeer { + readonly clientSocket: FleetNativeSocket; + readonly received: Array; + readonly notifications: Array; + readonly notify: (method: string, params: Record) => void; + readonly closePeer: () => void; +} + +const makeSocketPair = (): { client: FleetNativeSocket; peer: FakePeer } => { + const received: Array = []; + const notifications: Array = []; + let clientHandlers = { + open: () => {}, + message: (_text: string) => {}, + error: (_cause: unknown) => {}, + close: () => {}, + }; + let peerHandlers = { + message: (_text: string) => {}, + close: () => {}, + }; + let peerClosed = false; + + // Synchronous delivery: every frame moves through Effect queues between + // fibers, so no timers or sleeps are needed for the fake to behave. + const deliverClient = (text: string) => { + clientHandlers.message(text); + }; + const deliverPeer = (text: string) => { + if (!peerClosed) peerHandlers.message(text); + }; + + const parse = (text: string): WireFrame => JSON.parse(text) as WireFrame; + + const peer = { + message: (text: string) => { + const frame = parse(text); + if (frame.method !== undefined && frame.id === undefined) { + notifications.push(frame); + return; + } + received.push(frame); + const id = frame.id; + const method = frame.method; + const params = (frame.params ?? {}) as Record; + const respond = (result: unknown) => { + deliverClient(JSON.stringify({ id, result })); + }; + if (method === "initialize") { + respond({ userAgent: "Codex Test/0.156.1", codexHome: "/tmp/codex-home" }); + } else if (method === "thread/loaded/list") { + const cursor = params["cursor"]; + if (cursor === undefined || cursor === null) { + respond({ data: [THREAD_ID], nextCursor: "cursor-1" }); + } else { + respond({ data: [], nextCursor: null }); + } + } else if (method === "thread/resume") { + respond({ + thread: { + id: params["threadId"], + status: { type: "idle" }, + model: "gpt-5.6-luna", + turns: [], + }, + initialTurnsPage: null, + }); + } else if (method === "thread/read") { + respond({ + thread: { + id: params["threadId"], + status: { type: "idle" }, + model: "gpt-5.6-luna", + turns: [], + }, + }); + } else if (method === "turn/steer") { + respond({ turnId: params["expectedTurnId"] }); + } else if (method === "turn/start") { + respond({ turn: { id: "turn-new" } }); + } else { + deliverClient( + JSON.stringify({ id, error: { code: -32601, message: `unknown ${String(method)}` } }), + ); + } + }, + }; + + const clientSocket: FleetNativeSocket = { + send: (text: string) => { + deliverPeer(text); + }, + close: () => { + peerClosed = true; + clientHandlers.close(); + peerHandlers.close(); + }, + onOpen: (callback: () => void) => { + clientHandlers.open = callback; + callback(); + }, + onMessage: (callback: (text: string) => void) => { + const previous = clientHandlers; + clientHandlers = { ...previous, message: callback }; + }, + onError: (callback: (cause: unknown) => void) => { + const previous = clientHandlers; + clientHandlers = { ...previous, error: callback }; + }, + onClose: (callback: () => void) => { + const previous = clientHandlers; + clientHandlers = { ...previous, close: callback }; + }, + }; + + peerHandlers.message = peer.message; + + const fakePeer: FakePeer = { + clientSocket: { + send: clientSocket.send, + close: clientSocket.close, + onOpen: clientSocket.onOpen, + onMessage: clientSocket.onMessage, + onError: clientSocket.onError, + onClose: clientSocket.onClose, + }, + received, + notifications, + notify: (method: string, params: Record) => { + deliverClient(JSON.stringify({ method, params })); + }, + closePeer: () => { + peerClosed = true; + }, + }; + return { client: clientSocket, peer: fakePeer }; +}; + +type FleetRequest = ( + method: string, + params?: Record, +) => Effect.Effect; + +const withClient = ( + peer: FakePeer, + use: (request: FleetRequest) => Effect.Effect, +): Effect.Effect => + Effect.gen(function* () { + const factory: FleetNativeSocketFactory = () => peer.clientSocket; + const client = yield* openFleetNativeClient("ws://127.0.0.1:9/fleet", factory); + return yield* use((method, params) => client.request(method, params)); + }); + +/** + * Yield until `ready()` is true. The protocol drains outbound frames on a + * forked fiber, so the fake observes them a few yields after the test + * fiber moves on. Fails fast on a bounded poll instead of sleeping. + */ +const waitFor = (ready: () => Effect.Effect): Effect.Effect => + Effect.gen(function* () { + for (let step = 0; step < 1000; step += 1) { + if (yield* ready()) return; + yield* Effect.yieldNow; + } + return yield* Effect.fail("fleet-test-timeout" as const); + }); + +describe("CodexNativeWs transport", () => { + it.effect("completes the native handshake on open", () => + Effect.gen(function* () { + const { peer } = makeSocketPair(); + yield* Effect.scoped( + withClient(peer, (request) => + // Await one round trip: the protocol drains outbound frames FIFO, + // so a completed request proves the earlier `initialized` + // notification already flushed to the peer. + request("thread/loaded/list", {}).pipe(Effect.asVoid), + ), + ); + yield* waitFor(() => Effect.succeed(peer.notifications.length > 0)); + const initialize = peer.received.find((frame) => frame.method === "initialize"); + expect(initialize).toBeDefined(); + expect((initialize?.params as Record)?.["capabilities"]).toEqual({ + experimentalApi: true, + }); + expect(peer.notifications).toContainEqual(expect.objectContaining({ method: "initialized" })); + }), + ); + + it.effect("matches concurrent requests to their responses", () => + Effect.gen(function* () { + const { peer } = makeSocketPair(); + const results = yield* Effect.scoped( + withClient(peer, (request) => + Effect.all( + [ + request("thread/loaded/list", {}), + request("thread/read", { threadId: THREAD_ID, includeTurns: true }), + request("turn/steer", { + threadId: THREAD_ID, + expectedTurnId: TURN_ID, + input: [{ type: "text", text: "hello", text_elements: [] }], + }), + ], + { concurrency: "unbounded" }, + ), + ), + ); + expect((results[0] as { data: Array }).data).toEqual([THREAD_ID]); + expect((results[1] as { thread: { id: string } }).thread.id).toBe(THREAD_ID); + expect((results[2] as { turnId: string }).turnId).toBe(TURN_ID); + }), + ); + + it.effect("pages thread/loaded/list through the cursor", () => + Effect.gen(function* () { + const { peer } = makeSocketPair(); + const ids: Array = []; + yield* Effect.scoped( + withClient(peer, (request) => + Effect.gen(function* () { + let cursor: string | null = null; + for (;;) { + const page = (yield* request("thread/loaded/list", { + ...(cursor === null ? {} : { cursor }), + })) as { data: Array; nextCursor: string | null }; + ids.push(...page.data); + if (page.nextCursor === null) return; + cursor = page.nextCursor; + } + }), + ), + ); + expect(ids).toEqual([THREAD_ID]); + expect(peer.received.filter((frame) => frame.method === "thread/loaded/list")).toHaveLength( + 2, + ); + }), + ); + + it.effect("sends metadata-only resume with no settings payload", () => + Effect.gen(function* () { + const { peer } = makeSocketPair(); + yield* Effect.scoped( + withClient(peer, (request) => + request("thread/resume", { threadId: THREAD_ID, excludeTurns: true }), + ), + ); + const resume = peer.received.find((frame) => frame.method === "thread/resume"); + expect(resume?.params).toEqual({ threadId: THREAD_ID, excludeTurns: true }); + }), + ); + + it.effect("delivers native notifications to subscribers", () => + Effect.gen(function* () { + const { peer } = makeSocketPair(); + const seen = yield* Ref.make>([]); + const scope = yield* Scope.make(); + const factory: FleetNativeSocketFactory = () => peer.clientSocket; + const client = yield* Scope.provide(scope)( + openFleetNativeClient("ws://127.0.0.1:9/fleet", factory), + ); + const fiber = yield* Stream.runForEach(client.notifications, (notification) => + Ref.update(seen, (current) => [...current, notification.method]), + ).pipe(Effect.forkDetach); + peer.notify("item/completed", { + threadId: THREAD_ID, + item: { id: "exec-1", type: "commandExecution" }, + }); + peer.notify("turn/completed", { threadId: THREAD_ID, turn: { id: TURN_ID } }); + yield* waitFor(() => Ref.get(seen).pipe(Effect.map((methods) => methods.length >= 2))); + yield* Fiber.interrupt(fiber); + yield* Scope.close(scope, Exit.void); + expect(yield* Ref.get(seen)).toEqual(["item/completed", "turn/completed"]); + }), + ); + + it.effect("fails pending requests with a protocol error, not a hang", () => + Effect.gen(function* () { + const { peer } = makeSocketPair(); + const tag = yield* Effect.scoped( + withClient(peer, (request) => + request("thread/unknown", {}).pipe( + Effect.as("ok" as const), + Effect.catch((error) => Effect.succeed(error._tag)), + ), + ), + ); + expect(tag).toBe("CodexAppServerRequestError"); + }), + ); +}); diff --git a/apps/server/src/fleet/CodexNativeWs.ts b/apps/server/src/fleet/CodexNativeWs.ts new file mode 100644 index 000000000000..b7414f5323ac --- /dev/null +++ b/apps/server/src/fleet/CodexNativeWs.ts @@ -0,0 +1,193 @@ +/** + * CodexNativeWs - production WebSocket transport for an externally launched + * Codex native app-server. + * + * The fleet pane never spawns its own app-server. It dials the WebSocket URL + * configured per Codex provider instance (`CodexSettings.nativeEndpoint`), + * runs the proven native handshake (`initialize` with `experimentalApi: true` + * followed by the `initialized` notification), and speaks the same JSON-RPC + * framing the vendored `effect-codex-app-server` protocol layer uses for + * stdio: one JSON object per line over a byte stream. WebSocket text frames + * are already discrete messages, so each frame is fed to the protocol as one + * line. + * + * Closing a client only closes T3's socket. It sends no stop, close, + * archive, or permission payload to the native backend; the native session + * keeps running under its original owner. + * + * @module fleet/CodexNativeWs + */ +import * as Cause from "effect/Cause"; +import * as Deferred from "effect/Deferred"; +import * as Effect from "effect/Effect"; +import * as PlatformError from "effect/PlatformError"; +import * as Queue from "effect/Queue"; +import * as Scope from "effect/Scope"; +import * as Sink from "effect/Sink"; +import * as Stdio from "effect/Stdio"; +import * as Stream from "effect/Stream"; +import * as CodexError from "effect-codex-app-server/errors"; +import { + makeCodexAppServerPatchedProtocol, + type CodexAppServerIncomingNotification, + type CodexAppServerPatchedProtocol, +} from "effect-codex-app-server/protocol"; + +export type { CodexAppServerIncomingNotification }; + +const encoder = new TextEncoder(); +const decoder = new TextDecoder(); + +const FLEET_CLIENT_NAME = "t3-fleet"; +const FLEET_CLIENT_VERSION = "0.0.0"; + +/** Minimal socket surface. Satisfied by the global WebSocket in production. */ +export interface FleetNativeSocket { + readonly send: (text: string) => void; + readonly close: () => void; + readonly onOpen: (callback: () => void) => void; + readonly onMessage: (callback: (text: string) => void) => void; + readonly onError: (callback: (cause: unknown) => void) => void; + readonly onClose: (callback: () => void) => void; +} + +export type FleetNativeSocketFactory = (url: string) => FleetNativeSocket; + +const toTransportError = (cause: unknown): CodexError.CodexAppServerError => + new CodexError.CodexAppServerTransportError({ operation: "read-input-stream", cause }); + +/** Production factory over the global WebSocket (Node 22+, browsers, Bun). */ +export const globalWebSocketFactory: FleetNativeSocketFactory = (url: string) => { + const socket = new WebSocket(url); + return { + send: (text: string) => { + socket.send(text); + }, + close: () => { + socket.close(); + }, + onOpen: (callback: () => void) => { + socket.onopen = () => { + callback(); + }; + }, + onMessage: (callback: (text: string) => void) => { + socket.onmessage = (event: MessageEvent) => { + const data = event.data; + callback(typeof data === "string" ? data : decoder.decode(data as ArrayBuffer)); + }; + }, + onError: (callback: (cause: unknown) => void) => { + socket.onerror = (event: Event) => { + callback(event); + }; + }, + onClose: (callback: () => void) => { + socket.onclose = () => { + callback(); + }; + }, + }; +}; + +export interface FleetNativeClient { + readonly url: string; + /** Raw JSON-RPC request. Responses decode at the call site. */ + readonly request: CodexAppServerPatchedProtocol["request"]; + readonly notify: CodexAppServerPatchedProtocol["notify"]; + readonly notifications: Stream.Stream< + CodexAppServerIncomingNotification, + CodexError.CodexAppServerError + >; + /** Close T3's socket. Sends nothing to the native backend. */ + readonly close: Effect.Effect; +} + +/** + * Dial a native endpoint and complete the native handshake. Fails when the + * socket never opens or the `initialize` round trip fails; callers treat + * that as an unreachable endpoint and never retry blindly inside one call. + */ +export const openFleetNativeClient = Effect.fn("CodexNativeWs.openFleetNativeClient")(function* ( + url: string, + factory: FleetNativeSocketFactory = globalWebSocketFactory, +): Effect.fn.Return { + const socket = yield* Effect.sync(() => factory(url)); + const inbound = yield* Queue.unbounded>(); + const opened = yield* Deferred.make(); + + yield* Effect.sync(() => { + socket.onOpen(() => { + Deferred.doneUnsafe(opened, Effect.void); + }); + socket.onMessage((text: string) => { + Queue.offerUnsafe(inbound, text); + }); + socket.onError((cause: unknown) => { + Deferred.doneUnsafe(opened, Effect.fail(toTransportError(cause))); + Queue.endUnsafe(inbound); + }); + socket.onClose(() => { + Deferred.doneUnsafe( + opened, + Effect.fail(toTransportError(new Error("Native socket closed before it opened."))), + ); + Queue.endUnsafe(inbound); + }); + }); + + yield* Deferred.await(opened); + + yield* Effect.addFinalizer(() => + Effect.sync(() => { + socket.close(); + }), + ); + + const stdio = Stdio.make({ + args: Effect.succeed([]), + stdin: Stream.fromQueue(inbound).pipe( + Stream.map((text: string) => encoder.encode(`${text}\n`)), + ), + stdout: () => + Sink.forEach((chunk: string | Uint8Array) => + Effect.try({ + try: () => { + socket.send(typeof chunk === "string" ? chunk : decoder.decode(chunk)); + }, + catch: (cause: unknown) => + PlatformError.systemError({ + _tag: "Unknown", + module: "CodexNativeWs", + method: "send", + description: "Failed to send a frame to the native Codex app-server.", + cause, + }), + }), + ), + stderr: () => Sink.drain, + }); + + const protocol = yield* makeCodexAppServerPatchedProtocol({ + stdio, + terminationError: Effect.succeed(toTransportError(new Error("Native socket failed."))), + }); + + yield* protocol + .request("initialize", { + clientInfo: { name: FLEET_CLIENT_NAME, version: FLEET_CLIENT_VERSION }, + capabilities: { experimentalApi: true }, + }) + .pipe(Effect.asVoid); + yield* protocol.notify("initialized", undefined).pipe(Effect.ignore); + + return { + url, + request: protocol.request, + notify: protocol.notify, + notifications: protocol.incomingNotifications, + close: Effect.sync(() => { + socket.close(); + }), + }; +}); diff --git a/apps/server/src/fleet/FleetService.test.ts b/apps/server/src/fleet/FleetService.test.ts new file mode 100644 index 000000000000..9c399bfee102 --- /dev/null +++ b/apps/server/src/fleet/FleetService.test.ts @@ -0,0 +1,349 @@ +/** + * FleetService integration tests. + * + * The full production path short of a real Codex binary: real + * `ServerSettingsService` (in-memory layers, same as serverSettings.test.ts) + * supplies `CodexSettings.nativeEndpoint`; the real `openFleetNativeClient` + * dials it through an in-memory socket pair; a fake native peer replays the + * recorded probe shapes (bare-id loaded list, turns nested in + * `thread.turns`, metadata-only resume, `expectedTurnId` steer). Assertions + * cover discovery identity, history, delivery semantics, live subscription, + * and explicit `notLoaded` refusal. + */ +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { ProviderDriverKind, ProviderInstanceId } from "@t3tools/contracts"; +import { assert, it } from "@effect/vitest"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Stream from "effect/Stream"; + +import * as ServerSecretStore from "../auth/ServerSecretStore.ts"; +import * as ServerConfig from "../config.ts"; +import { SqlitePersistenceMemory } from "../persistence/Layers/Sqlite.ts"; +import * as ServerSettingsModule from "../serverSettings.ts"; +import type { FleetNativeSocket, FleetNativeSocketFactory } from "./CodexNativeWs.ts"; +import { FleetService, layerWithSocketFactory } from "./FleetService.ts"; + +const THREAD_ID = "01a0d431-b396-7023-a2aa-bc7ed6c6bc0c"; +const ACTIVE_TURN_ID = "01a0d431-b3c1-70e1-957e-88bc5fe51d2b"; +const ENDPOINT_URL = "ws://127.0.0.1:9/fleet"; +const INSTANCE_ID = ProviderInstanceId.make("codex"); + +type PeerMode = "active" | "idle" | "notLoaded"; + +interface WireFrame { + readonly id?: unknown; + readonly method?: unknown; + readonly params?: unknown; +} + +const makePeer = (mode: { current: PeerMode }) => { + const received: Array = []; + const notifications: Array = []; + let handlers = { + open: () => {}, + message: (_text: string) => {}, + error: (_cause: unknown) => {}, + close: () => {}, + }; + + const deliverClient = (text: string) => { + handlers.message(text); + }; + + const threadStatus = () => + mode.current === "notLoaded" ? { type: "notLoaded" } : { type: "idle" }; + + const turns = () => + mode.current === "active" + ? [ + { + id: ACTIVE_TURN_ID, + status: "inProgress", + itemsView: "full", + items: [ + { + type: "userMessage", + id: "01a0d431-baf4-7592-a25e-0fe93a4aa226", + content: [ + { type: "text", text: "Bounded native integration probe.", text_elements: [] }, + ], + }, + { + type: "agentMessage", + id: "msg_0f810051548a869a016ab54c57d5e487d299416b59c77c13b9", + text: "FLEET_MESSAGE_ACK_8426 ORIGINAL_TASK_COMPLETE", + phase: "final_answer", + }, + ], + }, + ] + : mode.current === "idle" + ? [{ id: "turn-done", status: "completed", itemsView: "full", items: [] }] + : []; + + const onMessage = (text: string) => { + const frame = JSON.parse(text) as WireFrame; + if (frame.method !== undefined && frame.id === undefined) { + notifications.push(frame); + return; + } + received.push(frame); + const id = frame.id; + const method = frame.method; + const params = (frame.params ?? {}) as Record; + const respond = (result: unknown) => { + deliverClient(JSON.stringify({ id, result })); + }; + if (method === "initialize") { + respond({ userAgent: "Codex Test/0.156.1" }); + } else if (method === "thread/loaded/list") { + respond({ data: [THREAD_ID], nextCursor: null }); + } else if (method === "thread/read") { + respond({ + thread: { + id: params["threadId"], + status: threadStatus(), + model: "gpt-5.6-luna", + agentRole: null, + name: null, + cwd: "/tmp/t3-fleet-spec/probe", + turns: turns(), + }, + }); + } else if (method === "thread/resume") { + respond({ + thread: { + id: params["threadId"], + status: threadStatus(), + model: "gpt-5.6-luna", + turns: [], + }, + }); + } else if (method === "turn/steer") { + respond({ turnId: params["expectedTurnId"] }); + } else if (method === "turn/start") { + respond({ turn: { id: "turn-new" } }); + } else { + deliverClient(JSON.stringify({ id, error: { code: -32601, message: "unknown" } })); + } + }; + + const clientSocket: FleetNativeSocket = { + send: (text: string) => { + onMessage(text); + }, + close: () => { + handlers.close(); + }, + onOpen: (callback: () => void) => { + handlers = { ...handlers, open: callback }; + callback(); + }, + onMessage: (callback: (text: string) => void) => { + handlers = { ...handlers, message: callback }; + }, + onError: (callback: (cause: unknown) => void) => { + handlers = { ...handlers, error: callback }; + }, + onClose: (callback: () => void) => { + handlers = { ...handlers, close: callback }; + }, + }; + + const factory: FleetNativeSocketFactory = () => clientSocket; + return { factory, received, notifications }; +}; + +const makeServerSettingsLayer = () => + ServerSettingsModule.layer.pipe( + Layer.provide(ServerSecretStore.layer), + Layer.provideMerge(Layer.fresh(SqlitePersistenceMemory)), + Layer.provideMerge( + Layer.fresh( + ServerConfig.layerTest(process.cwd(), { + prefix: "t3code-fleet-service-test-", + }), + ), + ), + ); + +const configureEndpoint = Effect.gen(function* () { + const serverSettings = yield* ServerSettingsModule.ServerSettingsService; + yield* serverSettings.updateSettings({ + providerInstances: { + [INSTANCE_ID]: { + driver: ProviderDriverKind.make("codex"), + enabled: true, + config: { nativeEndpoint: ENDPOINT_URL }, + }, + }, + }); +}); + +it.layer(NodeServices.layer)("FleetService over the native WS transport", (it) => { + const testLayer = (factory: FleetNativeSocketFactory) => + layerWithSocketFactory(factory).pipe(Layer.provideMerge(makeServerSettingsLayer())); + + it.effect("discovers the native session with stable identity", () => + Effect.gen(function* () { + const peer = makePeer({ current: "active" }); + const result = yield* Effect.gen(function* () { + const fleet = yield* FleetService; + yield* configureEndpoint; + return yield* fleet.listAgents; + }).pipe(Effect.provide(testLayer(peer.factory))); + assert.equal(result.agents.length, 1); + const agent = result.agents[0]!; + assert.equal(agent.nativeThreadId, THREAD_ID); + assert.equal(agent.environmentId, "env-local"); + assert.equal(agent.instanceId, INSTANCE_ID); + assert.equal(agent.provider, "codex"); + assert.equal(agent.model, "gpt-5.6-luna"); + const resume = peer.received.find((frame) => frame.method === "thread/resume"); + assert.deepEqual(resume?.params, { threadId: THREAD_ID, excludeTurns: true }); + }), + ); + + it.effect("reads transcript history with the running turn", () => + Effect.gen(function* () { + const fleet = yield* FleetService; + yield* configureEndpoint; + const history = yield* fleet.readThread({ + instanceId: INSTANCE_ID, + nativeThreadId: THREAD_ID, + }); + assert.equal(history.activeTurnId, ACTIVE_TURN_ID); + const texts = history.events.map((event) => event.text).filter((text) => text !== null); + assert.ok(texts.includes("Bounded native integration probe.")); + assert.ok(texts.includes("FLEET_MESSAGE_ACK_8426 ORIGINAL_TASK_COMPLETE")); + }).pipe(Effect.provide(testLayer(makePeer({ current: "active" }).factory))), + ); + + it.effect("steers the active turn with the expected turn id", () => + Effect.gen(function* () { + const mode = { current: "active" as PeerMode }; + const peer = makePeer(mode); + const result = yield* Effect.gen(function* () { + const fleet = yield* FleetService; + yield* configureEndpoint; + return yield* fleet.sendMessage({ + instanceId: INSTANCE_ID, + nativeThreadId: THREAD_ID, + text: "Include FLEET_MESSAGE_ACK_8426.", + expectedActiveTurnId: ACTIVE_TURN_ID, + ownershipKnown: true, + }); + }).pipe(Effect.provide(testLayer(peer.factory))); + assert.equal(result.kind, "steered-active"); + const steer = peer.received.find((frame) => frame.method === "turn/steer"); + assert.deepEqual(steer?.params, { + threadId: THREAD_ID, + expectedTurnId: ACTIVE_TURN_ID, + input: [{ type: "text", text: "Include FLEET_MESSAGE_ACK_8426.", text_elements: [] }], + }); + }), + ); + + it.effect("refuses a stale expected turn id without sending", () => + Effect.gen(function* () { + const peer = makePeer({ current: "active" }); + const result = yield* Effect.gen(function* () { + const fleet = yield* FleetService; + yield* configureEndpoint; + return yield* fleet.sendMessage({ + instanceId: INSTANCE_ID, + nativeThreadId: THREAD_ID, + text: "hello", + expectedActiveTurnId: "turn-stale", + ownershipKnown: true, + }); + }).pipe(Effect.provide(testLayer(peer.factory))); + assert.equal(result.kind, "refused"); + assert.equal( + peer.received.some( + (frame) => frame.method === "turn/steer" || frame.method === "turn/start", + ), + false, + ); + }), + ); + + it.effect("starts an idle follow-up only with confirmed ownership", () => + Effect.gen(function* () { + const peer = makePeer({ current: "idle" }); + const run = (ownershipKnown: boolean) => + Effect.gen(function* () { + const fleet = yield* FleetService; + yield* configureEndpoint; + return yield* fleet.sendMessage({ + instanceId: INSTANCE_ID, + nativeThreadId: THREAD_ID, + text: "hello", + expectedActiveTurnId: null, + ownershipKnown, + }); + }).pipe(Effect.provide(testLayer(peer.factory))); + const refused = yield* run(false); + assert.equal(refused.kind, "refused"); + const queued = yield* run(true); + assert.equal(queued.kind, "queued-followup"); + const start = peer.received.find((frame) => frame.method === "turn/start"); + assert.deepEqual(start?.params, { + threadId: THREAD_ID, + input: [{ type: "text", text: "hello", text_elements: [] }], + }); + }), + ); + + it.effect("streams agent updates to subscribers", () => + Effect.gen(function* () { + const peer = makePeer({ current: "active" }); + const events = yield* Effect.gen(function* () { + const fleet = yield* FleetService; + yield* configureEndpoint; + return yield* fleet.subscribe.pipe(Stream.take(1), Stream.runCollect); + }).pipe(Effect.provide(testLayer(peer.factory))); + assert.equal(events.length, 1); + const first = events[0]!; + assert.equal(first.kind, "agent-updated"); + if (first.kind === "agent-updated") { + assert.equal(first.agent.nativeThreadId, THREAD_ID); + } + }), + ); + + it.effect("never attaches or messages a notLoaded session", () => + Effect.gen(function* () { + const peer = makePeer({ current: "notLoaded" }); + const outcome = yield* Effect.gen(function* () { + const fleet = yield* FleetService; + yield* configureEndpoint; + const listed = yield* fleet.listAgents; + const read = yield* Effect.flip( + fleet.readThread({ instanceId: INSTANCE_ID, nativeThreadId: THREAD_ID }), + ); + const sent = yield* fleet.sendMessage({ + instanceId: INSTANCE_ID, + nativeThreadId: THREAD_ID, + text: "hello", + expectedActiveTurnId: null, + ownershipKnown: true, + }); + return { listed, read, sent }; + }).pipe(Effect.provide(testLayer(peer.factory))); + assert.equal(outcome.listed.agents.length, 0); + assert.equal(outcome.read._tag, "FleetError"); + assert.equal(outcome.sent.kind, "refused"); + assert.equal( + peer.received.some( + (frame) => + frame.method === "thread/resume" || + frame.method === "turn/steer" || + frame.method === "turn/start", + ), + false, + ); + }), + ); +}); diff --git a/apps/server/src/fleet/FleetService.ts b/apps/server/src/fleet/FleetService.ts new file mode 100644 index 000000000000..e6d02a052300 --- /dev/null +++ b/apps/server/src/fleet/FleetService.ts @@ -0,0 +1,711 @@ +/** + * FleetService - server-side fleet state for externally launched Codex + * native sessions. + * + * Each T3 client connection gets its own service instance (scoped): native + * sockets are dialed lazily per configured endpoint, live notifications are + * fanned out to that connection's subscribers, and everything is closed + * when the connection ends. Closing never touches the native backend: no + * stop, no archive, no permission change, no resume of `notLoaded` + * sessions. + * + * Endpoints come from `ServerSettings.providerInstances`: entries whose + * driver is `codex` and whose decoded `CodexSettings.nativeEndpoint` is a + * ws:// or wss:// URL. The URL itself never leaves the server: RPC results + * carry agents and events, never the endpoint. + * + * @module fleet/FleetService + */ +import { + CodexSettings, + FleetError, + fleetAgentId, + type EnvironmentId, + type FleetAgent, + type FleetAgentListResult, + type FleetMessageDelivery, + type FleetSendMessageInput, + type FleetStreamEvent, + type FleetThreadHistoryResult, + type ProviderInstanceId, +} from "@t3tools/contracts"; +import * as Context from "effect/Context"; +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 Option from "effect/Option"; +import * as PubSub from "effect/PubSub"; +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 ServerEnvironment from "../environment/ServerEnvironment.ts"; +import * as ServerSettings from "../serverSettings.ts"; +import { + globalWebSocketFactory, + openFleetNativeClient, + type FleetNativeClient, + type FleetNativeSocketFactory, +} from "./CodexNativeWs.ts"; +import { + activeTurnOf, + applyAgentMessageDelta, + buildSteerInput, + decideSend, + decodeLoadedThreadIds, + decodeReadTurns, + deltaEventOf, + discoverFleetAgents, + executeSend, + historyEventsOf, + isAttachableStatus, + isNotLoadedStatus, + nativeItemToEvent, + rawThreadStatusText, + threadLoadStateOf, + toFleetStatus, + type FleetNativeTransport, + type NativeThread, +} from "./FleetSessions.ts"; + +const decodeCodexSettings = Schema.decodeUnknownOption(CodexSettings); + +export interface FleetServiceShape { + readonly listAgents: Effect.Effect; + readonly readThread: (input: { + readonly instanceId: ProviderInstanceId; + readonly nativeThreadId: string; + }) => Effect.Effect; + readonly sendMessage: ( + input: FleetSendMessageInput, + ) => Effect.Effect; + readonly subscribe: Stream.Stream; +} + +export class FleetService extends Context.Service()( + "t3/fleet/FleetService", +) {} + +interface FleetEndpointConfig { + readonly instanceId: ProviderInstanceId; + readonly url: string; +} + +interface OpenEndpoint { + readonly config: FleetEndpointConfig; + readonly client: FleetNativeClient; + readonly scope: Scope.Scope; + /** Latest harness status per thread, from resume results + live updates. */ + readonly statusByThread: Map; + /** Running turn per thread, from turn/started + history + completion. */ + readonly activeTurnByThread: Map; + /** Accumulated agent-message deltas per item id. */ + readonly deltasByItem: Map; + /** Threads this endpoint has reported, for removal detection. */ + readonly knownThreads: Set; +} + +const failFleet = ( + operation: FleetError["operation"], + message: string, +): Effect.Effect => + Effect.fail( + new FleetError({ + operation, + message, + }), + ); + +const withNativeTimeout = ( + operation: FleetError["operation"], + effect: Effect.Effect, +): Effect.Effect => + effect.pipe( + Effect.timeoutOption("30 seconds"), + Effect.flatMap((option) => + Option.isSome(option) + ? Effect.succeed(option.value) + : failFleet(operation, "The native server did not answer in time."), + ), + Effect.mapError((cause) => + Schema.is(FleetError)(cause) + ? cause + : new FleetError({ + operation, + message: cause instanceof Error ? cause.message : String(cause), + }), + ), + ); + +const readStringField = (value: unknown, field: string): string | null => { + if (value !== null && typeof value === "object") { + const candidate = (value as Record)[field]; + return typeof candidate === "string" ? candidate : null; + } + return null; +}; + +export const makeFleetService = Effect.fn("FleetService.make")(function* ( + factory: FleetNativeSocketFactory = globalWebSocketFactory, +): Effect.fn.Return { + const serviceScope = yield* Scope.Scope; + const environmentId: EnvironmentId = yield* Effect.serviceOption( + ServerEnvironment.ServerEnvironment, + ).pipe( + Effect.flatMap( + Option.match({ + onNone: () => Effect.succeed("env-local" as EnvironmentId), + onSome: (serverEnvironment) => + serverEnvironment.getEnvironmentId.pipe( + Effect.orElseSucceed(() => "env-local" as EnvironmentId), + ), + }), + ), + ); + const endpoints = yield* Ref.make(new Map()); + const hub = yield* PubSub.unbounded(); + yield* Effect.addFinalizer(() => PubSub.shutdown(hub)); + // Captured at layer build: reading settings per call stays fresh (an + // endpoint can be configured at any time) without requiring the service + // from every RPC caller. + const serverSettings = yield* ServerSettings.ServerSettingsService; + + const readEndpointConfigs = (): Effect.Effect, FleetError> => + serverSettings.getSettings.pipe( + Effect.mapError( + (cause) => + new FleetError({ + operation: "list-agents", + message: `Failed to read server settings: ${String(cause)}`, + }), + ), + Effect.map((settings) => { + const configs: Array = []; + for (const [instanceId, entry] of Object.entries(settings.providerInstances)) { + if (entry.driver !== "codex") continue; + const decoded = decodeCodexSettings(entry.config ?? {}); + if (Option.isNone(decoded)) continue; + const endpoint = decoded.value.nativeEndpoint.trim(); + if (!/^wss?:\/\/.+/.test(endpoint)) continue; + configs.push({ instanceId: instanceId as ProviderInstanceId, url: endpoint }); + } + return configs; + }), + ); + + const publish = (event: FleetStreamEvent): Effect.Effect => + PubSub.publish(hub, event).pipe(Effect.asVoid); + + const agentIdentity = (instanceId: ProviderInstanceId, nativeThreadId: string): string => + fleetAgentId({ environmentId, provider: "codex", instanceId, nativeThreadId }); + + const dropEndpoint = (url: string): Effect.Effect => + Ref.modify(endpoints, (current) => { + const open = current.get(url); + if (open === undefined) return [Effect.void, current] as const; + const next = new Map(current); + next.delete(url); + return [Scope.close(open.scope, Exit.void), next] as const; + }).pipe(Effect.flatten); + + /** + * One native JSON-RPC round trip with a bounded wait. Any failure drops + * the cached socket so the next call redials and re-attaches; the current + * call still fails (or reports uncertain delivery for sends), never + * retries blindly. + */ + const nativeRequest = ( + endpoint: OpenEndpoint, + operation: FleetError["operation"], + method: string, + params: Record, + ): Effect.Effect => + withNativeTimeout(operation, endpoint.client.request(method, params)).pipe( + Effect.tapError(() => dropEndpoint(endpoint.config.url)), + ); + + function handleNativeNotification( + endpoint: OpenEndpoint, + method: string, + params: unknown, + at: string, + ): Effect.Effect { + const threadId = readStringField(params, "threadId"); + switch (method) { + case "item/started": + case "item/completed": { + const item = (params as Record | null)?.["item"]; + const turnId = readStringField(params, "turnId"); + if (threadId !== null && turnId !== null) { + endpoint.activeTurnByThread.set(threadId, turnId); + } + if (threadId === null || item === undefined) return Effect.void; + const event = nativeItemToEvent({ nativeThreadId: threadId, item, at }); + if (event === null) return Effect.void; + return publish({ + kind: "events-appended", + agentId: agentIdentity(endpoint.config.instanceId, threadId), + events: [event], + }); + } + case "item/agentMessage/delta": { + const itemId = readStringField(params, "itemId"); + const delta = readStringField(params, "delta"); + const turnId = readStringField(params, "turnId"); + if (threadId !== null && turnId !== null) { + endpoint.activeTurnByThread.set(threadId, turnId); + } + if (threadId === null || itemId === null || delta === null) return Effect.void; + const accumulated = applyAgentMessageDelta(endpoint.deltasByItem, { itemId, delta }); + endpoint.deltasByItem.clear(); + for (const [key, value] of accumulated) endpoint.deltasByItem.set(key, value); + return publish({ + kind: "events-appended", + agentId: agentIdentity(endpoint.config.instanceId, threadId), + events: [ + deltaEventOf({ + nativeThreadId: threadId, + itemId, + text: accumulated.get(itemId) ?? "", + at, + }), + ], + }); + } + case "turn/started": { + const turn = (params as Record | null)?.["turn"]; + const turnId = readStringField(turn, "id") ?? readStringField(params, "turnId"); + if (threadId !== null && turnId !== null) { + endpoint.activeTurnByThread.set(threadId, turnId); + } + return Effect.void; + } + case "turn/completed": + case "turn/failed": { + const turn = (params as Record | null)?.["turn"]; + const turnId = readStringField(turn, "id") ?? readStringField(params, "turnId"); + if (threadId !== null && turnId !== null) { + if (endpoint.activeTurnByThread.get(threadId) === turnId) { + endpoint.activeTurnByThread.delete(threadId); + } + } + if (threadId === null || turnId === null) return Effect.void; + return publish({ + kind: "events-appended", + agentId: agentIdentity(endpoint.config.instanceId, threadId), + events: [ + { + id: `turn:${turnId}`, + nativeThreadId: threadId, + kind: method, + at, + text: null, + }, + ], + }); + } + case "thread/status/changed": { + const status = (params as Record | null)?.["status"]; + if (threadId === null) return Effect.void; + endpoint.statusByThread.set(threadId, status ?? null); + if (isNotLoadedStatus(status)) { + endpoint.activeTurnByThread.delete(threadId); + return publish({ + kind: "agent-removed", + agentId: agentIdentity(endpoint.config.instanceId, threadId), + environmentId, + instanceId: endpoint.config.instanceId, + nativeThreadId: threadId, + }); + } + if (rawThreadStatusText(status)?.trim().toLowerCase() === "idle") { + endpoint.activeTurnByThread.delete(threadId); + } + return publish({ + kind: "agent-updated", + agent: { + environmentId, + provider: "codex", + instanceId: endpoint.config.instanceId, + nativeThreadId: threadId, + status: toFleetStatus(rawThreadStatusText(status)), + model: null, + role: null, + cwd: null, + lastSeenAt: at, + }, + }); + } + default: + return Effect.void; + } + } + + const pumpNotifications = (endpoint: OpenEndpoint): Effect.Effect => + Stream.runForEach(endpoint.client.notifications, (notification) => + DateTime.now.pipe( + Effect.map(DateTime.formatIso), + Effect.flatMap((at) => + handleNativeNotification(endpoint, notification.method, notification.params, at), + ), + Effect.catch((cause) => + Effect.logDebug("Fleet native notification failed", { + method: notification.method, + cause: String(cause), + }), + ), + ), + ).pipe( + Effect.catch((cause) => + Effect.logDebug("Fleet native notification stream ended", { + url: endpoint.config.url, + cause: String(cause), + }), + ), + Effect.forkIn(endpoint.scope), + Effect.asVoid, + ); + + const ensureEndpoint = Effect.fn("FleetService.ensureEndpoint")(function* ( + config: FleetEndpointConfig, + ): Effect.fn.Return { + const cached = yield* Ref.get(endpoints); + const existing = cached.get(config.url); + if (existing !== undefined && existing.config.instanceId === config.instanceId) { + return existing; + } + if (existing !== undefined) { + yield* dropEndpoint(config.url); + } + const scope = yield* Scope.fork(serviceScope); + const client = yield* Scope.provide(scope)(openFleetNativeClient(config.url, factory)).pipe( + Effect.mapError( + (cause) => + new FleetError({ + operation: "connect-endpoint", + message: `Native Codex endpoint ${config.url} is unreachable: ${String(cause)}`, + }), + ), + Effect.tapError(() => Scope.close(scope, Exit.void)), + ); + const endpoint: OpenEndpoint = { + config, + client, + scope, + statusByThread: new Map(), + activeTurnByThread: new Map(), + deltasByItem: new Map(), + knownThreads: new Set(), + }; + yield* Ref.update(endpoints, (current) => new Map(current).set(config.url, endpoint)); + yield* pumpNotifications(endpoint); + return endpoint; + }); + + const listLoadedIds = ( + endpoint: OpenEndpoint, + operation: FleetError["operation"], + ): Effect.Effect, FleetError> => + Effect.gen(function* () { + const ids: Array = []; + let cursor: string | null = null; + for (;;) { + const raw = yield* nativeRequest(endpoint, operation, "thread/loaded/list", { + ...(cursor === null ? {} : { cursor }), + }); + const page = decodeLoadedThreadIds(raw); + if (page === null) { + return yield* failFleet>( + operation, + "The native server returned an unreadable thread list.", + ); + } + ids.push(...page.ids); + if (page.nextCursor === null) return ids; + cursor = page.nextCursor; + } + }); + + const readStatusOf = (reads: ReadonlyArray, threadId: string): unknown => { + for (const read of reads) { + if (read !== null && typeof read === "object") { + const thread = (read as { readonly thread?: unknown }).thread; + if (thread !== null && typeof thread === "object") { + const record = thread as Record; + if (record["id"] === threadId) return record["status"] ?? null; + } + } + } + return null; + }; + + const recordResumeState = (endpoint: OpenEndpoint, threadId: string, resumed: unknown) => + Effect.sync(() => { + if (resumed !== null && typeof resumed === "object") { + const thread = (resumed as { readonly thread?: unknown }).thread as + | NativeThread + | undefined; + if (thread !== undefined && thread !== null && typeof thread === "object") { + endpoint.statusByThread.set(threadId, thread.status ?? null); + } + } + }); + + const unreachable = (config: FleetEndpointConfig, cause: unknown): Effect.Effect => + publish({ + kind: "endpoint-unreachable", + instanceId: config.instanceId, + message: Schema.is(FleetError)(cause) ? cause.message : String(cause), + }); + + const discoverEndpoint = Effect.fn("FleetService.discoverEndpoint")(function* ( + endpoint: OpenEndpoint, + at: string, + ): Effect.fn.Return, FleetError> { + const ids = yield* listLoadedIds(endpoint, "list-agents"); + // A failed per-thread read aborts discovery: the caller surfaces an + // endpoint notice and the next call redials. A silent partial listing + // would be worse than a loud one. + const reads: Array = []; + for (const id of ids) { + reads.push( + yield* nativeRequest(endpoint, "list-agents", "thread/read", { + threadId: id, + }), + ); + } + const agents = yield* discoverFleetAgents({ + environmentId, + instanceId: endpoint.config.instanceId, + threadReads: reads, + seenAt: at, + }).pipe( + Effect.mapError( + () => + new FleetError({ + operation: "list-agents", + message: "The native server returned an unreadable thread.", + }), + ), + ); + const seen = new Set(agents.map((agent) => agent.nativeThreadId)); + for (const previous of endpoint.knownThreads) { + if (!seen.has(previous)) { + endpoint.knownThreads.delete(previous); + endpoint.activeTurnByThread.delete(previous); + endpoint.statusByThread.delete(previous); + yield* publish({ + kind: "agent-removed", + agentId: agentIdentity(endpoint.config.instanceId, previous), + environmentId, + instanceId: endpoint.config.instanceId, + nativeThreadId: previous, + }); + } + } + for (const agent of agents) { + endpoint.knownThreads.add(agent.nativeThreadId); + const status = readStatusOf(reads, agent.nativeThreadId); + endpoint.statusByThread.set(agent.nativeThreadId, status); + if (isAttachableStatus(status)) { + yield* nativeRequest(endpoint, "list-agents", "thread/resume", { + threadId: agent.nativeThreadId, + excludeTurns: true, + }).pipe( + Effect.tap((resumed) => recordResumeState(endpoint, agent.nativeThreadId, resumed)), + Effect.ignore, + ); + } + yield* publish({ kind: "agent-updated", agent }); + } + return agents; + }); + + const listAgents = Effect.fn("FleetService.listAgents")(function* (): Effect.fn.Return< + FleetAgentListResult, + FleetError + > { + const at = DateTime.formatIso(yield* DateTime.now); + const configs = yield* readEndpointConfigs(); + const agents: Array = []; + for (const config of configs) { + const endpoint = yield* ensureEndpoint(config).pipe( + Effect.catch((cause) => + unreachable(config, cause).pipe(Effect.as(null as OpenEndpoint | null)), + ), + ); + if (endpoint === null) continue; + const discovered = yield* discoverEndpoint(endpoint, at).pipe( + Effect.catch((cause) => + unreachable(config, cause).pipe(Effect.as([] as ReadonlyArray)), + ), + ); + agents.push(...discovered); + } + return { agents, scannedAt: at }; + }); + + const endpointFor = Effect.fn("FleetService.endpointFor")(function* ( + instanceId: ProviderInstanceId, + operation: FleetError["operation"], + ): Effect.fn.Return { + const configs = yield* readEndpointConfigs(); + const config = configs.find((entry) => entry.instanceId === instanceId); + if (config === undefined) { + return yield* failFleet( + operation, + `Codex instance ${instanceId} has no native fleet endpoint configured.`, + ); + } + return yield* ensureEndpoint(config); + }); + + const readThread = Effect.fn("FleetService.readThread")(function* (input: { + readonly instanceId: ProviderInstanceId; + readonly nativeThreadId: string; + }): Effect.fn.Return { + const at = DateTime.formatIso(yield* DateTime.now); + const endpoint = yield* endpointFor(input.instanceId, "read-thread"); + const raw = yield* nativeRequest(endpoint, "read-thread", "thread/read", { + threadId: input.nativeThreadId, + includeTurns: true, + }); + const turns = decodeReadTurns(raw); + if (turns === null) { + return yield* failFleet( + "read-thread", + "The native server returned an unreadable thread.", + ); + } + const events = historyEventsOf({ + nativeThreadId: input.nativeThreadId, + threadRead: raw, + at, + }); + if (events === null) { + return yield* failFleet( + "read-thread", + "The native server returned an unreadable thread.", + ); + } + const status = readStatusOf([raw], input.nativeThreadId); + if (threadLoadStateOf(status) === "notLoaded") { + return yield* failFleet( + "read-thread", + "Native session is not loaded in this backend.", + ); + } + const activeTurnId = + activeTurnOf(turns) ?? endpoint.activeTurnByThread.get(input.nativeThreadId) ?? null; + if (activeTurnId !== null) endpoint.activeTurnByThread.set(input.nativeThreadId, activeTurnId); + endpoint.statusByThread.set(input.nativeThreadId, status); + return { + agent: { + environmentId, + provider: "codex", + instanceId: input.instanceId, + nativeThreadId: input.nativeThreadId, + status: toFleetStatus(rawThreadStatusText(status)), + model: null, + role: null, + cwd: null, + lastSeenAt: at, + }, + events, + activeTurnId, + fetchedAt: at, + }; + }); + + const sendMessage = Effect.fn("FleetService.sendMessage")(function* ( + input: FleetSendMessageInput, + ): Effect.fn.Return { + const at = DateTime.formatIso(yield* DateTime.now); + const text = input.text.trim(); + if (text.length === 0) { + return yield* failFleet( + "send-message", + "Message text is empty. No message was sent.", + ); + } + const endpoint = yield* endpointFor(input.instanceId, "send-message"); + const raw = yield* nativeRequest(endpoint, "send-message", "thread/read", { + threadId: input.nativeThreadId, + }).pipe( + Effect.mapError( + () => + new FleetError({ + operation: "send-message", + message: "The native session could not be read before sending.", + }), + ), + ); + const status = readStatusOf([raw], input.nativeThreadId); + const turns = decodeReadTurns(raw) ?? []; + const resolvedTurnId = + activeTurnOf(turns) ?? endpoint.activeTurnByThread.get(input.nativeThreadId) ?? null; + const expected = input.expectedActiveTurnId ?? null; + if (expected !== null && resolvedTurnId !== null && expected !== resolvedTurnId) { + return { + kind: "refused", + reason: "The running turn changed since you read it. Read the thread again before sending.", + at, + }; + } + const decision = decideSend({ + threadId: input.nativeThreadId, + // A reported running turn means the session is active even when the + // thread-level status word lags behind (it races sends). + status: resolvedTurnId !== null ? "active" : toFleetStatus(rawThreadStatusText(status)), + loadState: threadLoadStateOf(status), + activeTurnId: resolvedTurnId, + ownershipKnown: input.ownershipKnown, + text, + }); + const recordTurnStarted = (started: unknown) => + Effect.sync(() => { + const turnId = + readStringField(started, "turnId") ?? + readStringField((started as Record | null)?.["turn"], "id"); + if (turnId !== null) endpoint.activeTurnByThread.set(input.nativeThreadId, turnId); + }); + const transport: FleetNativeTransport = { + steerTurn: (params) => + nativeRequest(endpoint, "send-message", "turn/steer", { + threadId: params.threadId, + expectedTurnId: params.expectedTurnId, + input: buildSteerInput(params.text), + }), + startTurn: (params) => + nativeRequest(endpoint, "send-message", "turn/start", { + threadId: params.threadId, + input: buildSteerInput(params.text), + }).pipe(Effect.tap((started) => recordTurnStarted(started))), + }; + return yield* executeSend(transport, decision, at); + }); + + const subscribe: Stream.Stream = Stream.unwrap( + Effect.gen(function* () { + const snapshot = yield* listAgents(); + const initial = snapshot.agents.map( + (agent) => ({ kind: "agent-updated", agent }) as FleetStreamEvent, + ); + return Stream.concat(Stream.fromIterable(initial), Stream.fromPubSub(hub)); + }), + ); + + return { + listAgents: Effect.suspend(() => listAgents()), + readThread, + sendMessage, + subscribe, + }; +}); + +export const layer = Layer.effect(FleetService, makeFleetService(globalWebSocketFactory)); + +export const layerWithSocketFactory = (factory: FleetNativeSocketFactory) => + Layer.effect(FleetService, makeFleetService(factory)); diff --git a/apps/server/src/fleet/FleetSessions.test.ts b/apps/server/src/fleet/FleetSessions.test.ts new file mode 100644 index 000000000000..1ee946f69232 --- /dev/null +++ b/apps/server/src/fleet/FleetSessions.test.ts @@ -0,0 +1,604 @@ +/** + * FleetSessions behavioral tests. + * + * Shapes replay the recorded native evidence (Codex 0.156.1, + * `/tmp/t3-fleet-spec/probe`): `thread/loaded/list` returns bare ids with a + * page cursor, `thread/read` nests turns inside `thread.turns`, resume is + * metadata-only with `excludeTurns: true`, steering carries + * `expectedTurnId`, and live traffic arrives as item/turn/status + * notifications with per-item deltas. + */ +import { describe, expect, it } from "@effect/vitest"; +import { + type EnvironmentId, + type FleetNativeEvent, + FleetError, + type ProviderInstanceId, +} from "@t3tools/contracts"; +import * as Effect from "effect/Effect"; +import * as Option from "effect/Option"; + +import { + activeTurnOf, + applyAgentMessageDelta, + attachMetadataOnly, + buildResumeParams, + buildSteerInput, + decideSend, + decodeLoadedThreadIds, + decodeReadTurns, + dedupeFleetEvents, + deltaEventOf, + detachFleetAgent, + discoverFleetAgents, + executeSend, + fleetAgentId, + historyEventsOf, + isAttachableStatus, + isNotLoadedStatus, + mergeFleetEvents, + nativeItemText, + nativeItemToEvent, + rawThreadStatusText, + resolveFleetEndpoint, + threadLoadStateOf, + toFleetStatus, + type FleetNativeTransport, + type NativeTurn, +} from "./FleetSessions.ts"; + +const ENVIRONMENT_ID = "env-local" as EnvironmentId; +const INSTANCE_ID = "codex" as ProviderInstanceId; +const NATIVE_THREAD_ID = "01a0d431-b396-7023-a2aa-bc7ed6c6bc0c"; +const ACTIVE_TURN_ID = "01a0d431-b3c1-70e1-957e-88bc5fe51d2b"; +const SEEN_AT = "2026-09-24T18:13:42.000Z"; + +const loadedListPage = (overrides: Record = {}) => ({ + data: [NATIVE_THREAD_ID], + nextCursor: null, + ...overrides, +}); + +const nativeThread = (overrides: Record = {}) => ({ + id: NATIVE_THREAD_ID, + status: { type: "idle" }, + model: "gpt-5.6-luna", + agentRole: null, + name: null, + cwd: "/tmp/t3-fleet-spec/probe", + turns: [], + ...overrides, +}); + +const threadRead = (overrides: Record = {}) => ({ + thread: nativeThread(overrides), +}); + +const userMessageItem = { + type: "userMessage", + id: "01a0d431-baf4-7592-a25e-0fe93a4aa226", + content: [ + { + type: "text", + text: "Bounded native integration probe.", + text_elements: [], + }, + ], +}; + +const agentMessageItem = { + type: "agentMessage", + id: "msg_0f810051548a869a016ab54c57d5e487d299416b59c77c13b9", + text: "FLEET_MESSAGE_ACK_8426 ORIGINAL_TASK_COMPLETE", + phase: "final_answer", +}; + +const commandItem = { + type: "commandExecution", + id: "exec-6a4dcdb0-f4fd-4811-ae8f-b9cb6ad1350b", + command: "/bin/zsh -lc 'sleep 25'", + exitCode: 0, + status: "completed", +}; + +const threadReadWithTurn = () => + threadRead({ + status: { type: "idle" }, + turns: [ + { + id: ACTIVE_TURN_ID, + status: "completed", + itemsView: "full", + items: [userMessageItem, commandItem, agentMessageItem], + }, + ], + }); + +const fleetEvent = (id: string, text: string | null = null): FleetNativeEvent => ({ + id, + nativeThreadId: NATIVE_THREAD_ID, + kind: "item/completed", + at: SEEN_AT, + text, +}); + +interface RecordedCall { + readonly method: "steer" | "start"; + readonly params: unknown; +} + +/** Fake send transport: records every send, replays probe behavior. */ +const makeFakeTransport = ( + options: { readonly failSend?: boolean } = {}, +): { + readonly transport: FleetNativeTransport; + readonly calls: Array; +} => { + const calls: Array = []; + const transport: FleetNativeTransport = { + steerTurn: (params) => { + calls.push({ method: "steer", params }); + return options.failSend + ? Effect.fail( + new FleetError({ operation: "send-message", message: "native socket dropped" }), + ) + : Effect.succeed({ turnId: params.expectedTurnId }); + }, + startTurn: (params) => { + calls.push({ method: "start", params }); + return options.failSend + ? Effect.fail( + new FleetError({ operation: "send-message", message: "native socket dropped" }), + ) + : Effect.succeed({ turn: { id: "turn-new" } }); + }, + }; + return { transport, calls }; +}; + +describe("FleetSessions endpoint configuration", () => { + it.effect("resolves a configured native WebSocket endpoint", () => + Effect.gen(function* () { + const endpoint = yield* resolveFleetEndpoint("ws://127.0.0.1:4242"); + expect(Option.isSome(endpoint)).toBe(true); + }), + ); + + it.effect("leaves the instance undiscovered when no endpoint is configured", () => + Effect.gen(function* () { + for (const value of [undefined, null, "", " "]) { + const endpoint = yield* resolveFleetEndpoint(value); + expect(Option.isNone(endpoint)).toBe(true); + } + }), + ); + + it.effect("rejects non-WebSocket endpoint values", () => + Effect.gen(function* () { + const result = yield* Effect.flip(resolveFleetEndpoint("http://127.0.0.1:4242")); + expect(result._tag).toBe("FleetNativeDecodeError"); + }), + ); +}); + +describe("FleetSessions loaded-list discovery", () => { + it("decodes bare string ids with a page cursor", () => { + const page = decodeLoadedThreadIds(loadedListPage()); + expect(page?.ids).toEqual([NATIVE_THREAD_ID]); + expect(page?.nextCursor).toBeNull(); + }); + + it("carries the cursor for the next page", () => { + const page = decodeLoadedThreadIds(loadedListPage({ data: ["a", "b"], nextCursor: "opaque" })); + expect(page?.ids).toEqual(["a", "b"]); + expect(page?.nextCursor).toBe("opaque"); + }); + + it("rejects object-shaped entries instead of guessing", () => { + expect(decodeLoadedThreadIds({ data: [{ threadId: NATIVE_THREAD_ID }] })).toBeNull(); + expect(decodeLoadedThreadIds({ threads: [NATIVE_THREAD_ID] })).toBeNull(); + expect(decodeLoadedThreadIds(null)).toBeNull(); + }); + + it.effect("discovers the externally launched dispatcher with stable identity", () => + Effect.gen(function* () { + const agents = yield* discoverFleetAgents({ + environmentId: ENVIRONMENT_ID, + instanceId: INSTANCE_ID, + threadReads: [threadRead()], + seenAt: SEEN_AT, + }); + expect(agents).toHaveLength(1); + expect(agents[0]?.nativeThreadId).toBe(NATIVE_THREAD_ID); + expect(fleetAgentId(agents[0]!)).toBe(`env-local/codex/codex/${NATIVE_THREAD_ID}`); + expect(agents[0]?.model).toBe("gpt-5.6-luna"); + expect(agents[0]?.status).toBe("idle"); + }), + ); + + it.effect("keeps the same identity across rediscovery", () => + Effect.gen(function* () { + const first = yield* discoverFleetAgents({ + environmentId: ENVIRONMENT_ID, + instanceId: INSTANCE_ID, + threadReads: [threadRead()], + seenAt: SEEN_AT, + }); + const second = yield* discoverFleetAgents({ + environmentId: ENVIRONMENT_ID, + instanceId: INSTANCE_ID, + threadReads: [threadRead({ status: { type: "active", activeFlags: [] } })], + seenAt: "2026-09-24T18:14:42.000Z", + }); + expect(fleetAgentId(first[0]!)).toBe(fleetAgentId(second[0]!)); + expect(second[0]?.status).toBe("active"); + }), + ); + + it.effect("skips notLoaded threads instead of attaching them", () => + Effect.gen(function* () { + const agents = yield* discoverFleetAgents({ + environmentId: ENVIRONMENT_ID, + instanceId: INSTANCE_ID, + threadReads: [threadRead({ status: { type: "notLoaded" } })], + seenAt: SEEN_AT, + }); + expect(agents).toHaveLength(0); + }), + ); + + it.effect("fails loudly on unreadable thread reads", () => + Effect.gen(function* () { + const result = yield* Effect.flip( + discoverFleetAgents({ + environmentId: ENVIRONMENT_ID, + instanceId: INSTANCE_ID, + threadReads: [{ thread: { status: { type: "idle" } } }], + seenAt: SEEN_AT, + }), + ); + expect(result._tag).toBe("FleetNativeDecodeError"); + }), + ); + + it.effect("reports model and role as null when the harness does not provide them", () => + Effect.gen(function* () { + const agents = yield* discoverFleetAgents({ + environmentId: ENVIRONMENT_ID, + instanceId: INSTANCE_ID, + threadReads: [threadRead({ model: null, agentRole: null, name: null })], + seenAt: SEEN_AT, + }); + expect(agents[0]?.model).toBeNull(); + expect(agents[0]?.role).toBeNull(); + }), + ); +}); + +describe("FleetSessions notLoaded handling", () => { + it("reads the explicit notLoaded status in any envelope", () => { + expect(isNotLoadedStatus({ type: "notLoaded" })).toBe(true); + expect(isNotLoadedStatus("notLoaded")).toBe(true); + expect(isNotLoadedStatus({ type: "idle" })).toBe(false); + expect(isNotLoadedStatus(null)).toBe(false); + expect(threadLoadStateOf({ type: "notLoaded" })).toBe("notLoaded"); + expect(threadLoadStateOf({ type: "idle" })).toBe("loaded"); + expect(threadLoadStateOf(null)).toBe("unknown"); + expect(toFleetStatus("notLoaded")).toBe("unknown"); + }); + + it("attaches only idle or active sessions", () => { + expect(isAttachableStatus({ type: "idle" })).toBe(true); + expect(isAttachableStatus({ type: "active", activeFlags: [] })).toBe(true); + expect(isAttachableStatus({ type: "notLoaded" })).toBe(false); + expect(isAttachableStatus(null)).toBe(false); + expect(isAttachableStatus({ type: "systemError" })).toBe(false); + }); + + it("reads raw status words from strings and objects", () => { + expect(rawThreadStatusText({ type: "active" })).toBe("active"); + expect(rawThreadStatusText("idle")).toBe("idle"); + expect(rawThreadStatusText(null)).toBeNull(); + expect(rawThreadStatusText(undefined)).toBeNull(); + }); +}); + +describe("FleetSessions metadata-only attach", () => { + it.effect("attaches with excludeTurns and no settings changes", () => + Effect.gen(function* () { + const attached = yield* attachMetadataOnly({ + threadId: NATIVE_THREAD_ID, + status: { type: "idle" }, + }); + expect(attached).toEqual({ + resumed: true, + threadId: NATIVE_THREAD_ID, + params: buildResumeParams(NATIVE_THREAD_ID), + }); + expect(attached.params).toEqual({ threadId: NATIVE_THREAD_ID, excludeTurns: true }); + }), + ); + + it("builds resume params without any settings payload", () => { + expect(buildResumeParams(NATIVE_THREAD_ID)).toEqual({ + threadId: NATIVE_THREAD_ID, + excludeTurns: true, + }); + }); + + it.effect("never auto-resumes a notLoaded session into another backend", () => + Effect.gen(function* () { + const result = yield* Effect.flip( + attachMetadataOnly({ threadId: NATIVE_THREAD_ID, status: { type: "notLoaded" } }), + ); + expect(result._tag).toBe("FleetNativeDecodeError"); + }), + ); + + it.effect("refuses attach when load state is unknown", () => + Effect.gen(function* () { + const result = yield* Effect.flip( + attachMetadataOnly({ threadId: NATIVE_THREAD_ID, status: null }), + ); + expect(result._tag).toBe("FleetNativeDecodeError"); + }), + ); +}); + +describe("FleetSessions history from thread.turns", () => { + it("decodes turns nested inside thread, not top level", () => { + const turns = decodeReadTurns(threadReadWithTurn()); + expect(turns?.map((turn) => turn.id)).toEqual([ACTIVE_TURN_ID]); + expect(decodeReadTurns({ turns: [{ id: ACTIVE_TURN_ID }] })).toBeNull(); + expect(decodeReadTurns(null)).toBeNull(); + }); + + it("finds the running turn and nothing else", () => { + const inProgress: NativeTurn = { id: "turn-live", status: "inProgress" }; + const done: NativeTurn = { id: ACTIVE_TURN_ID, status: "completed" }; + expect(activeTurnOf([done, inProgress])).toBe("turn-live"); + expect(activeTurnOf([done])).toBeNull(); + expect(activeTurnOf([])).toBeNull(); + }); + + it("normalizes recorded item shapes to text", () => { + expect(nativeItemText(userMessageItem)).toBe("Bounded native integration probe."); + expect(nativeItemText(agentMessageItem)).toBe("FLEET_MESSAGE_ACK_8426 ORIGINAL_TASK_COMPLETE"); + expect(nativeItemText(commandItem)).toBe("/bin/zsh -lc 'sleep 25' (exit 0)"); + expect(nativeItemText({ type: "reasoning", id: "rs-1", summary: [], content: [] })).toBeNull(); + expect(nativeItemText({ nope: true })).toBeNull(); + }); + + it("builds history events keyed by harness item id", () => { + const events = historyEventsOf({ + nativeThreadId: NATIVE_THREAD_ID, + threadRead: threadReadWithTurn(), + at: SEEN_AT, + }); + expect(events).not.toBeNull(); + expect(events?.map((event) => event.id)).toEqual([ + `turn:${ACTIVE_TURN_ID}`, + "01a0d431-baf4-7592-a25e-0fe93a4aa226", + "exec-6a4dcdb0-f4fd-4811-ae8f-b9cb6ad1350b", + "msg_0f810051548a869a016ab54c57d5e487d299416b59c77c13b9", + ]); + expect(events?.[1]?.text).toBe("Bounded native integration probe."); + }); + + it("converts one live item into an event", () => { + const event = nativeItemToEvent({ + nativeThreadId: NATIVE_THREAD_ID, + item: agentMessageItem, + at: SEEN_AT, + }); + expect(event?.id).toBe("msg_0f810051548a869a016ab54c57d5e487d299416b59c77c13b9"); + expect(event?.kind).toBe("agentMessage"); + expect( + nativeItemToEvent({ nativeThreadId: NATIVE_THREAD_ID, item: null, at: SEEN_AT }), + ).toBeNull(); + }); + + it("skips notLoaded turn item views when reading history", () => { + const events = historyEventsOf({ + nativeThreadId: NATIVE_THREAD_ID, + threadRead: threadRead({ + turns: [{ id: "turn-1", status: "completed", itemsView: "notLoaded" }], + }), + at: SEEN_AT, + }); + expect(events).toEqual([]); + }); +}); + +describe("FleetSessions live deltas without data loss", () => { + it("accumulates per-item deltas across notifications", () => { + const empty = new Map(); + const afterFirst = applyAgentMessageDelta(empty, { itemId: "msg-1", delta: "FLEET" }); + const afterSecond = applyAgentMessageDelta(afterFirst, { + itemId: "msg-1", + delta: "_MESSAGE_ACK", + }); + expect(afterSecond.get("msg-1")).toBe("FLEET_MESSAGE_ACK"); + expect(empty.size).toBe(0); + }); + + it("keys delta events by item id so completion upserts over them", () => { + const delta = deltaEventOf({ + nativeThreadId: NATIVE_THREAD_ID, + itemId: "msg-1", + text: "FLEET", + at: SEEN_AT, + }); + expect(delta.id).toBe("msg-1"); + const completed = fleetEvent("msg-1", "FLEET_MESSAGE_ACK_8426"); + const merged = mergeFleetEvents([delta], [completed]); + expect(merged).toHaveLength(1); + expect(merged[0]?.text).toBe("FLEET_MESSAGE_ACK_8426"); + }); + + it("replaces same-id live rows in place instead of duplicating", () => { + const history = [fleetEvent("evt-1", "first"), fleetEvent("evt-2", "partial")]; + const live = [fleetEvent("evt-2", "complete"), fleetEvent("evt-3", "third")]; + const merged = mergeFleetEvents(history, live); + expect(merged.map((event) => event.id)).toEqual(["evt-1", "evt-2", "evt-3"]); + expect(merged[1]?.text).toBe("complete"); + }); + + it("dedupes replayed events with latest value winning", () => { + const events = [ + fleetEvent("evt-1", "stale"), + fleetEvent("evt-2"), + fleetEvent("evt-1", "fresh"), + ]; + const deduped = dedupeFleetEvents(events); + expect(deduped.map((event) => event.id)).toEqual(["evt-1", "evt-2"]); + expect(deduped[0]?.text).toBe("fresh"); + }); + + it("maps harness status words without inventing states", () => { + expect(toFleetStatus("running")).toBe("active"); + expect(toFleetStatus("idle")).toBe("idle"); + expect(toFleetStatus("completed")).toBe("ended"); + expect(toFleetStatus("something-new")).toBe("unknown"); + expect(toFleetStatus(null)).toBe("unknown"); + }); +}); + +describe("FleetSessions message delivery semantics", () => { + const loadedActive = { + threadId: NATIVE_THREAD_ID, + status: "active" as const, + loadState: "loaded" as const, + activeTurnId: ACTIVE_TURN_ID, + ownershipKnown: true, + text: "Include FLEET_MESSAGE_ACK_8426 in your final answer.", + }; + + it("steers an active turn with the expected turn id", () => { + const decision = decideSend(loadedActive); + expect(decision._tag).toBe("Steer"); + if (decision._tag === "Steer") { + expect(decision.params.expectedTurnId).toBe(ACTIVE_TURN_ID); + expect(decision.params.input).toEqual( + buildSteerInput("Include FLEET_MESSAGE_ACK_8426 in your final answer."), + ); + expect(decision.params.input[0]).toEqual({ + type: "text", + text: "Include FLEET_MESSAGE_ACK_8426 in your final answer.", + text_elements: [], + }); + } + }); + + it.effect("sends active steering through turn/steer", () => + Effect.gen(function* () { + const { transport, calls } = makeFakeTransport(); + const decision = decideSend({ ...loadedActive, text: "hello" }); + const delivery = yield* executeSend(transport, decision, SEEN_AT); + expect(delivery.kind).toBe("steered-active"); + expect(calls).toHaveLength(1); + expect(calls[0]?.method).toBe("steer"); + expect(calls[0]?.params).toMatchObject({ + threadId: NATIVE_THREAD_ID, + expectedTurnId: ACTIVE_TURN_ID, + text: "hello", + }); + }), + ); + + it("refuses an active turn whose id is unknown", () => { + const decision = decideSend({ ...loadedActive, activeTurnId: null }); + expect(decision._tag).toBe("Refused"); + }); + + it("always refuses notLoaded sessions, even with an active display status", () => { + const decision = decideSend({ ...loadedActive, loadState: "notLoaded" }); + expect(decision._tag).toBe("Refused"); + if (decision._tag === "Refused") { + expect(decision.reason).toContain("not loaded"); + } + }); + + it("refuses unknown load state without sending", () => { + const decision = decideSend({ ...loadedActive, loadState: "unknown" }); + expect(decision._tag).toBe("Refused"); + }); + + it("starts an idle follow-up only when ownership is known", () => { + const allowed = decideSend({ + threadId: NATIVE_THREAD_ID, + status: "idle", + loadState: "loaded", + activeTurnId: null, + ownershipKnown: true, + text: "hello", + }); + expect(allowed._tag).toBe("Followup"); + const refused = decideSend({ + threadId: NATIVE_THREAD_ID, + status: "idle", + loadState: "loaded", + activeTurnId: null, + ownershipKnown: false, + text: "hello", + }); + expect(refused._tag).toBe("Refused"); + }); + + it.effect("sends idle follow-ups through turn/start", () => + Effect.gen(function* () { + const { transport, calls } = makeFakeTransport(); + const delivery = yield* executeSend( + transport, + { + _tag: "Followup", + params: { threadId: NATIVE_THREAD_ID, text: "hello" }, + }, + SEEN_AT, + ); + expect(delivery.kind).toBe("queued-followup"); + expect(calls.map((call) => call.method)).toEqual(["start"]); + }), + ); + + it.effect("surfaces uncertain delivery without auto resend", () => + Effect.gen(function* () { + const { transport, calls } = makeFakeTransport({ failSend: true }); + const decision = decideSend({ ...loadedActive, text: "hello" }); + const delivery = yield* executeSend(transport, decision, SEEN_AT); + expect(delivery.kind).toBe("uncertain"); + expect(delivery.reason).toContain("not resent"); + expect(calls.filter((call) => call.method === "steer")).toHaveLength(1); + }), + ); + + it("refuses ended and unknown sessions without sending", () => { + for (const status of ["ended", "unknown"] as const) { + const decision = decideSend({ + threadId: NATIVE_THREAD_ID, + status, + loadState: "loaded", + activeTurnId: null, + ownershipKnown: true, + text: "hello", + }); + expect(decision._tag).toBe("Refused"); + } + }); +}); + +describe("FleetSessions non-interrupting disconnect", () => { + it.effect("detaches without stopping or mutating the native agent", () => + Effect.gen(function* () { + const agents = yield* discoverFleetAgents({ + environmentId: ENVIRONMENT_ID, + instanceId: INSTANCE_ID, + threadReads: [threadRead()], + seenAt: SEEN_AT, + }); + const detached = detachFleetAgent(agents[0]!); + expect(detached.detached).toBe(true); + expect(detached.nativeThreadId).toBe(NATIVE_THREAD_ID); + }), + ); +}); diff --git a/apps/server/src/fleet/FleetSessions.ts b/apps/server/src/fleet/FleetSessions.ts new file mode 100644 index 000000000000..7906c59132fb --- /dev/null +++ b/apps/server/src/fleet/FleetSessions.ts @@ -0,0 +1,610 @@ +/** + * FleetSessions - protocol normalization and delivery decisions for the + * fleet pane's first slice: one externally launched Codex dispatcher reached + * through its own native WebSocket app-server. + * + * Recorded native evidence (Codex 0.156.1, `/tmp/t3-fleet-spec/probe`) pins + * the shapes used here: + * + * - discovery reads `thread/loaded/list`, whose result is + * `{data: string[], nextCursor: string | null}`: bare thread ids, paged. + * Each id then needs its own `thread/read` for metadata. + * - history lives at `thread/read` result `thread.turns`, not top level. + * - a thread whose status is `notLoaded` is reported, never attached or + * messaged, even when no `loaded` boolean is present. + * - attach is metadata-only `thread/resume` with `excludeTurns: true` and no + * settings payload, for live notifications only. + * - an active turn is messaged with `turn/steer` plus `expectedTurnId`; an + * idle follow-up uses native `turn/start` only when identity and ownership + * are known. + * - viewing or detaching never stops the native agent, mutates permissions, + * or resumes a `notLoaded` session into another backend. + * + * This module is pure: decode, normalize, decide. The live socket lives in + * `CodexNativeWs.ts`; connection lifecycle and fan-out live in + * `FleetService.ts`. + * + * @module fleet/FleetSessions + */ +import { + decodeFleetEndpointConfig, + fleetAgentId, + FleetError, + type EnvironmentId, + type FleetAgent, + type FleetCodexNativeEndpoint, + type FleetMessageDelivery, + type FleetNativeEvent, + type FleetNativeThreadStatus, + type ProviderInstanceId, +} from "@t3tools/contracts"; +import * as Effect from "effect/Effect"; +import * as Option from "effect/Option"; +import * as Schema from "effect/Schema"; + +export { fleetAgentId }; + +export class FleetNativeDecodeError extends Schema.TaggedError()( + "FleetNativeDecodeError", + { + operation: Schema.Literals(["decode-loaded-list", "decode-thread-read", "decode-notification"]), + }, +) {} + +/** + * Native app-server sends the fleet slice performs. Failures of any kind + * become explicit `uncertain` delivery in `executeSend`: surfaced, never + * auto resent. + */ +export interface FleetNativeTransport { + readonly steerTurn: (params: { + readonly threadId: string; + readonly expectedTurnId: string; + readonly text: string; + }) => Effect.Effect; + readonly startTurn: (params: { + readonly threadId: string; + readonly text: string; + }) => Effect.Effect; +} + +/** Metadata-only attach params. Never carries settings changes. */ +export function buildResumeParams(threadId: string): { + readonly threadId: string; + readonly excludeTurns: true; +} { + return { threadId, excludeTurns: true }; +} + +/** Native turn/steer input for one text message. */ +export function buildSteerInput(text: string): ReadonlyArray<{ + readonly type: "text"; + readonly text: string; + readonly text_elements: ReadonlyArray; +}> { + return [{ type: "text", text, text_elements: [] }]; +} + +/** + * Resolve the configured native endpoint. Empty or missing values mean + * the instance has no fleet endpoint; anything else must be ws:// or + * wss://. + */ +export const resolveFleetEndpoint = Effect.fn("FleetSessions.resolveFleetEndpoint")(function* ( + nativeEndpoint: string | null | undefined, +): Effect.fn.Return, FleetNativeDecodeError> { + const decoded = decodeFleetEndpointConfig(nativeEndpoint ?? null); + if (decoded._tag === "Missing") return Option.none(); + if (decoded._tag === "Invalid") { + return yield* new FleetNativeDecodeError({ operation: "decode-loaded-list" }); + } + return Option.some(decoded.endpoint); +}); + +const LoadedListResponse = Schema.Struct({ + data: Schema.Array(Schema.String), + nextCursor: Schema.optional(Schema.NullOr(Schema.String)), +}); +const decodeLoadedList = Schema.decodeUnknownOption(LoadedListResponse); + +/** Decode `thread/loaded/list` result: bare ids plus an opaque page cursor. */ +export function decodeLoadedThreadIds(raw: unknown): { + readonly ids: ReadonlyArray; + readonly nextCursor: string | null; +} | null { + const decoded = decodeLoadedList(raw); + if (Option.isNone(decoded)) return null; + return { + ids: decoded.value.data.filter((id) => id.trim().length > 0), + nextCursor: decoded.value.nextCursor ?? null, + }; +} + +const NativeTurnItem = Schema.Struct({ + id: Schema.String, + type: Schema.String, +}); +const decodeTurnItem = Schema.decodeUnknownOption(NativeTurnItem); + +const NativeTurn = Schema.Struct({ + id: Schema.String, + status: Schema.optional(Schema.String), + items: Schema.optional(Schema.Array(Schema.Unknown)), + itemsView: Schema.optional(Schema.String), +}); +export type NativeTurn = typeof NativeTurn.Type; + +const NativeThread = Schema.Struct({ + id: Schema.String, + status: Schema.optional(Schema.Unknown), + model: Schema.optional(Schema.NullOr(Schema.String)), + agentRole: Schema.optional(Schema.NullOr(Schema.String)), + name: Schema.optional(Schema.NullOr(Schema.String)), + cwd: Schema.optional(Schema.NullOr(Schema.String)), + turns: Schema.optional(Schema.Array(Schema.Unknown)), +}); +export type NativeThread = typeof NativeThread.Type; + +const ThreadReadResponse = Schema.Struct({ + thread: NativeThread, +}); +const decodeThreadRead = Schema.decodeUnknownOption(ThreadReadResponse); + +/** Raw status word from a native thread, whatever envelope it arrives in. */ +export function rawThreadStatusText(status: unknown): string | null { + if (typeof status === "string") return status; + if (status !== null && typeof status === "object") { + const type = (status as { readonly type?: unknown }).type; + if (typeof type === "string") return type; + } + return null; +} + +/** True for the explicit `notLoaded` harness status. */ +export function isNotLoadedStatus(status: unknown): boolean { + const text = rawThreadStatusText(status); + return text !== null && text.trim().toLowerCase() === "notloaded"; +} + +/** Map a harness status word onto the fleet status without guessing. */ +export function toFleetStatus(status: string | null): FleetNativeThreadStatus { + if (status === null) return "unknown"; + const normalized = status.trim().toLowerCase(); + if (normalized === "notloaded") return "unknown"; + if (normalized === "running" || normalized === "active" || normalized === "working") { + return "active"; + } + if (normalized === "idle") return "idle"; + if ( + normalized === "ended" || + normalized === "completed" || + normalized === "archived" || + normalized === "closed" + ) { + return "ended"; + } + return "unknown"; +} + +/** + * Load evidence for one thread: a `thread/read` (or resume) result proves + * the session is loaded in this backend. `notLoaded` is never attachable, + * even when no `loaded` boolean is present anywhere. + */ +export type FleetThreadLoadState = "loaded" | "notLoaded" | "unknown"; + +export function threadLoadStateOf(status: unknown): FleetThreadLoadState { + if (isNotLoadedStatus(status)) return "notLoaded"; + if (rawThreadStatusText(status) !== null) return "loaded"; + return "unknown"; +} + +/** + * True when T3 may attach metadata-only for live notifications: the session + * proved loaded and the harness reports it idle or active. Anything else is + * reported, never auto-resumed into another backend. + */ +export function isAttachableStatus(status: unknown): boolean { + const text = rawThreadStatusText(status); + if (text === null) return false; + const normalized = text.trim().toLowerCase(); + return normalized === "idle" || normalized === "active"; +} + +function nonEmpty(value: string | null | undefined): string | null { + if (value === undefined || value === null) return null; + const trimmed = value.trim(); + return trimmed.length > 0 ? trimmed : null; +} + +/** + * Discover fleet agents from per-thread `thread/read` results. Ids come + * from `thread/loaded/list`; entries that decode as `notLoaded`, or that + * lack a stable native thread id, are skipped instead of attached. + */ +export const discoverFleetAgents = Effect.fn("FleetSessions.discoverFleetAgents")( + function* (input: { + readonly environmentId: EnvironmentId; + readonly instanceId: ProviderInstanceId; + readonly threadReads: ReadonlyArray; + readonly seenAt: string; + }): Effect.fn.Return, FleetNativeDecodeError> { + const agents: Array = []; + for (const raw of input.threadReads) { + const decoded = decodeThreadRead(raw); + if (Option.isNone(decoded)) { + return yield* new FleetNativeDecodeError({ operation: "decode-thread-read" }); + } + const thread = decoded.value.thread; + if (threadLoadStateOf(thread.status) === "notLoaded") continue; + const role = nonEmpty(thread.agentRole) ?? nonEmpty(thread.name); + const model = nonEmpty(thread.model); + const cwd = thread.cwd === undefined || thread.cwd === null ? null : thread.cwd; + agents.push({ + environmentId: input.environmentId, + provider: "codex", + instanceId: input.instanceId, + nativeThreadId: thread.id, + status: toFleetStatus(rawThreadStatusText(thread.status)), + model, + role, + cwd, + lastSeenAt: input.seenAt, + }); + } + return agents; + }, +); + +/** Decode one `thread/read` result turn list (turns live inside `thread`). */ +export function decodeReadTurns(raw: unknown): ReadonlyArray | null { + const decoded = decodeThreadRead(raw); + if (Option.isNone(decoded)) return null; + const turns: Array = []; + for (const rawTurn of decoded.value.thread.turns ?? []) { + const turn = Schema.decodeUnknownOption(NativeTurn)(rawTurn); + if (Option.isNone(turn)) return null; + turns.push(turn.value); + } + return turns; +} + +/** Latest running turn, if the harness reports one. */ +export function activeTurnOf(turns: ReadonlyArray): string | null { + for (let index = turns.length - 1; index >= 0; index -= 1) { + const turn = turns[index]; + if (turn === undefined) continue; + if (turn.status !== undefined && turn.status.trim().toLowerCase() === "inprogress") { + return turn.id; + } + } + return null; +} + +function textOfContent(content: unknown): string | null { + if (!Array.isArray(content)) return null; + const parts: Array = []; + for (const block of content) { + if (block !== null && typeof block === "object") { + const record = block as { readonly type?: unknown; readonly text?: unknown }; + if (record.type === "text" && typeof record.text === "string" && record.text.length > 0) { + parts.push(record.text); + } + } + } + return parts.length > 0 ? parts.join("\n") : null; +} + +/** Best-effort transcript text for one native thread item. */ +export function nativeItemText(item: unknown): string | null { + if (item === null || typeof item !== "object") return null; + const record = item as { + readonly type?: unknown; + readonly text?: unknown; + readonly content?: unknown; + readonly command?: unknown; + readonly exitCode?: unknown; + readonly summary?: unknown; + }; + switch (record.type) { + case "userMessage": + return textOfContent(record.content); + case "agentMessage": + return typeof record.text === "string" && record.text.length > 0 ? record.text : null; + case "reasoning": + return textOfContent(record.summary) ?? textOfContent(record.content); + case "commandExecution": { + if (typeof record.command !== "string" || record.command.length === 0) return null; + return typeof record.exitCode === "number" + ? `${record.command} (exit ${record.exitCode})` + : record.command; + } + default: + return textOfContent(record.content); + } +} + +function eventAt(value: unknown, fallback: string): string { + if (typeof value === "number" && Number.isFinite(value)) { + // @effect-diagnostics-next-line globalDate:off - epoch-millis to ISO for event display; no DateTime.fromEpochMillis in this Effect version. + return new Date(value).toISOString(); + } + return fallback; +} + +/** + * Normalize one native thread item (history or live) into a fleet event. + * The event id is the harness item id, so a reconnect that re-reads history + * upserts over live rows instead of duplicating them. + */ +export function nativeItemToEvent(input: { + readonly nativeThreadId: string; + readonly item: unknown; + readonly at: string; +}): FleetNativeEvent | null { + const decoded = decodeTurnItem(input.item); + if (Option.isNone(decoded)) return null; + const record = input.item as { readonly completedAtMs?: unknown; readonly startedAtMs?: unknown }; + return { + id: decoded.value.id, + nativeThreadId: input.nativeThreadId, + kind: decoded.value.type, + at: eventAt(record.completedAtMs ?? record.startedAtMs, input.at), + text: nativeItemText(input.item), + }; +} + +/** History events for every item of every turn in a `thread/read` result. */ +export function historyEventsOf(input: { + readonly nativeThreadId: string; + readonly threadRead: unknown; + readonly at: string; +}): ReadonlyArray | null { + const decoded = decodeThreadRead(input.threadRead); + if (Option.isNone(decoded)) return null; + const events: Array = []; + const seen = new Set(); + seen.add(`turn:${decoded.value.thread.id}-header`); + for (const rawTurn of decoded.value.thread.turns ?? []) { + const turn = Schema.decodeUnknownOption(NativeTurn)(rawTurn); + if (Option.isNone(turn)) return null; + const itemsView = turn.value.itemsView?.trim().toLowerCase(); + if (itemsView === "notloaded") continue; + events.push({ + id: `turn:${turn.value.id}`, + nativeThreadId: input.nativeThreadId, + kind: `turn/${turn.value.status ?? "unknown"}`, + at: input.at, + text: null, + }); + for (const rawItem of turn.value.items ?? []) { + const event = nativeItemToEvent({ + nativeThreadId: input.nativeThreadId, + item: rawItem, + at: input.at, + }); + if (event === null) return null; + if (seen.has(event.id)) continue; + seen.add(event.id); + events.push(event); + } + } + return events; +} + +/** + * Accumulate one `item/agentMessage/delta` payload. Deltas stream per item + * id; the accumulated text replaces the previous value under the same event + * id so completion updates never lose earlier deltas (no first-ID-wins). + */ +export function applyAgentMessageDelta( + accumulated: ReadonlyMap, + params: { readonly itemId: string; readonly delta: string }, +): Map { + const next = new Map(accumulated); + next.set(params.itemId, `${next.get(params.itemId) ?? ""}${params.delta}`); + return next; +} + +/** Fleet event for the accumulated text of one streaming agent message. */ +export function deltaEventOf(input: { + readonly nativeThreadId: string; + readonly itemId: string; + readonly text: string; + readonly at: string; +}): FleetNativeEvent { + return { + id: input.itemId, + nativeThreadId: input.nativeThreadId, + kind: "agentMessage/delta", + at: input.at, + text: input.text.length > 0 ? input.text : null, + }; +} + +/** + * Merge native history (from `thread/read`) with live notification events. + * Same-id rows upsert in place: live completions and accumulated deltas + * replace their earlier value at the original position, so a reconnect that + * re-reads history never duplicates rows and never drops newer text. + */ +export function mergeFleetEvents( + history: ReadonlyArray, + live: ReadonlyArray, +): ReadonlyArray { + const merged = [...history]; + const indexById = new Map(merged.map((event, index) => [event.id, index] as const)); + for (const event of live) { + const index = indexById.get(event.id); + if (index === undefined) { + indexById.set(event.id, merged.length); + merged.push(event); + } else { + merged[index] = event; + } + } + return merged; +} + +/** + * Remove duplicate events by harness-assigned id. The latest value wins at + * the first position, so replayed completions replace stale placeholders. + */ +export function dedupeFleetEvents( + events: ReadonlyArray, +): ReadonlyArray { + const latest = new Map(); + const order: Array = []; + for (const event of events) { + if (!latest.has(event.id)) order.push(event.id); + latest.set(event.id, event); + } + return order.map((id) => latest.get(id) as FleetNativeEvent); +} + +/** + * Attach metadata-only for live notifications. Returns the exact + * `thread/resume` params to send: `excludeTurns: true` and no settings + * payload. Refuses `notLoaded` (or unknown-load) threads instead of + * resuming them into another backend. + */ +export const attachMetadataOnly = Effect.fn("FleetSessions.attachMetadataOnly")(function* (input: { + readonly threadId: string; + readonly status: unknown; +}): Effect.fn.Return< + { + readonly resumed: true; + readonly threadId: string; + readonly params: ReturnType; + }, + FleetNativeDecodeError +> { + if (threadLoadStateOf(input.status) !== "loaded" || !isAttachableStatus(input.status)) { + return yield* new FleetNativeDecodeError({ operation: "decode-thread-read" }); + } + return { resumed: true, threadId: input.threadId, params: buildResumeParams(input.threadId) }; +}); + +export type FleetSendDecision = + | { + readonly _tag: "Steer"; + readonly params: { + readonly threadId: string; + readonly expectedTurnId: string; + readonly input: ReturnType; + }; + } + | { + readonly _tag: "Followup"; + readonly params: { readonly threadId: string; readonly text: string }; + } + | { readonly _tag: "Refused"; readonly reason: string }; + +/** + * Choose how a user message reaches the native session: + * an active turn is steered in place; an idle session starts a follow-up + * turn only when identity and ownership are known; anything else is an + * explicit refusal, never a blind send. `notLoaded` is always refused, + * even when the display status alone looks usable. + */ +export function decideSend(input: { + readonly threadId: string; + readonly status: FleetNativeThreadStatus; + readonly loadState: FleetThreadLoadState; + readonly activeTurnId: string | null; + readonly ownershipKnown: boolean; + readonly text: string; +}): FleetSendDecision { + if (input.loadState !== "loaded") { + return { + _tag: "Refused", + reason: + input.loadState === "notLoaded" + ? "Native session is not loaded in this backend. No message was sent." + : "Native session load state is unknown. No message was sent.", + }; + } + if (input.status === "active") { + if (input.activeTurnId === null) { + return { + _tag: "Refused", + reason: "Native turn is active but its id is unknown. Read the thread before sending.", + }; + } + return { + _tag: "Steer", + params: { + threadId: input.threadId, + expectedTurnId: input.activeTurnId, + input: buildSteerInput(input.text), + }, + }; + } + if (input.status === "idle") { + if (!input.ownershipKnown) { + return { + _tag: "Refused", + reason: + "Idle session ownership is unknown. Confirm identity before starting a follow-up turn.", + }; + } + return { _tag: "Followup", params: { threadId: input.threadId, text: input.text } }; + } + if (input.status === "ended") { + return { _tag: "Refused", reason: "Native session has ended. No message was sent." }; + } + return { + _tag: "Refused", + reason: "Native session state is unknown. No message was sent.", + }; +} + +/** + * Carry out a send decision against the native transport. Transport + * failures become explicit `uncertain` delivery: surfaced, never auto + * resent. Retrying is a separate manual decision by the user. + */ +export const executeSend = Effect.fn("FleetSessions.executeSend")(function* ( + transport: FleetNativeTransport, + decision: FleetSendDecision, + at: string, +): Effect.fn.Return { + if (decision._tag === "Refused") { + return { kind: "refused", reason: decision.reason, at }; + } + const result = yield* ( + decision._tag === "Steer" + ? transport.steerTurn({ + threadId: decision.params.threadId, + expectedTurnId: decision.params.expectedTurnId, + text: decision.params.input[0]?.text ?? "", + }) + : transport.startTurn(decision.params) + ).pipe(Effect.option); + if (Option.isNone(result)) { + return { + kind: "uncertain", + reason: + "The native server did not confirm delivery. The message may or may not have landed; it was not resent.", + at, + }; + } + return { + kind: decision._tag === "Steer" ? "steered-active" : "queued-followup", + reason: null, + at, + }; +}); + +/** + * Detach a fleet agent from T3 viewing. Detach sends nothing to the + * native backend: no stop, no close, no archive, no permission change. + * The native session keeps running under its original owner. + */ +export function detachFleetAgent(agent: FleetAgent): { + readonly detached: true; + readonly agentId: string; + readonly nativeThreadId: string; +} { + return { detached: true, agentId: fleetAgentId(agent), nativeThreadId: agent.nativeThreadId }; +} diff --git a/apps/server/src/provider/Layers/ProviderInstanceRegistryLive.test.ts b/apps/server/src/provider/Layers/ProviderInstanceRegistryLive.test.ts index 25dafa5ba040..f4726e4bad0c 100644 --- a/apps/server/src/provider/Layers/ProviderInstanceRegistryLive.test.ts +++ b/apps/server/src/provider/Layers/ProviderInstanceRegistryLive.test.ts @@ -104,6 +104,7 @@ const makeCodexConfig = (overrides: Partial): CodexSettings => ({ homePath: "", shadowHomePath: "", launchArgs: "", + nativeEndpoint: "", customModels: [], ...overrides, }); diff --git a/apps/server/src/serverSettings.test.ts b/apps/server/src/serverSettings.test.ts index b2caba4a0347..027ebbfbd0a3 100644 --- a/apps/server/src/serverSettings.test.ts +++ b/apps/server/src/serverSettings.test.ts @@ -230,6 +230,7 @@ it.layer(NodeServices.layer)("server settings", (it) => { homePath: "/Users/julius/.codex", shadowHomePath: "", launchArgs: "", + nativeEndpoint: "", customModels: [], }); assert.deepEqual(next.providers.claudeAgent, { @@ -939,6 +940,7 @@ it.layer(NodeServices.layer)("server settings", (it) => { homePath: "", shadowHomePath: "", launchArgs: "", + nativeEndpoint: "", customModels: [], }); assert.deepEqual(next.providers.claudeAgent, { diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 17500460d9ba..c3f62259faf3 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -144,6 +144,7 @@ import * as ProjectCloneTracker from "./project/ProjectCloneTracker.ts"; import * as RepositoryIdentityResolver from "./project/RepositoryIdentityResolver.ts"; import * as WorktreeSetupTracker from "./project/WorktreeSetupTracker.ts"; import * as AgentSessionScanner from "./project/AgentSessionScanner.ts"; +import * as FleetService from "./fleet/FleetService.ts"; import { importRecentAgentThreads } from "./project/AgentSessionImporter.ts"; import * as ServerEnvironment from "./environment/ServerEnvironment.ts"; import * as RemoteOpenTargets from "./environment/RemoteOpenTargets.ts"; @@ -628,6 +629,7 @@ const makeWsRpcLayer = ( | WorkspacePaths.WorkspacePaths >(); const agentSessionScanner = yield* AgentSessionScanner.AgentSessionScanner; + const fleetService = yield* FleetService.FleetService; const serverEnvironment = yield* ServerEnvironment.ServerEnvironment; const backgroundPolicy = yield* BackgroundPolicy.BackgroundPolicy; const rpcClientIds = yield* Ref.make(new Set()); @@ -3162,6 +3164,22 @@ const makeWsRpcLayer = ( observeRpcEffect(WS_METHODS.agentSessionsScan, agentSessionScanner.scan, { "rpc.aggregate": "workspace", }), + [WS_METHODS.fleetListAgents]: () => + observeRpcEffect(WS_METHODS.fleetListAgents, fleetService.listAgents, { + "rpc.aggregate": "fleet", + }), + [WS_METHODS.fleetReadThread]: (input) => + observeRpcEffect(WS_METHODS.fleetReadThread, fleetService.readThread(input), { + "rpc.aggregate": "fleet", + }), + [WS_METHODS.fleetSendMessage]: (input) => + observeRpcEffect(WS_METHODS.fleetSendMessage, fleetService.sendMessage(input), { + "rpc.aggregate": "fleet", + }), + [WS_METHODS.fleetSubscribe]: () => + observeRpcStream(WS_METHODS.fleetSubscribe, fleetService.subscribe, { + "rpc.aggregate": "fleet", + }), [WS_METHODS.agentSessionsImport]: (input) => observeRpcEffect( WS_METHODS.agentSessionsImport, @@ -3859,6 +3877,7 @@ export const websocketRpcRouteLayer = Layer.unwrap( Layer.provideMerge(RpcSerialization.layerJson), Layer.provide(Layer.succeed(SqlClient.SqlClient, sql)), Layer.provide(AgentSessionScanner.layer), + Layer.provide(FleetService.layer), Layer.provide(ProviderMaintenanceRunner.layer), Layer.provide(Layer.succeed(ServerSelfUpdate.ServerSelfUpdate, serverSelfUpdate)), // One server-lifetime service means clients share the same PR caches, and a WS diff --git a/apps/web/src/components/AgentsPanel.tsx b/apps/web/src/components/AgentsPanel.tsx index 43f1fa123f38..ad3490f7c3e2 100644 --- a/apps/web/src/components/AgentsPanel.tsx +++ b/apps/web/src/components/AgentsPanel.tsx @@ -28,6 +28,7 @@ import { cn } from "~/lib/utils"; import { orchestrationEnvironment } from "~/state/orchestration"; import { ScrollArea } from "~/components/ui/scroll-area"; import { Button } from "~/components/ui/button"; +import { FleetSection } from "~/components/FleetSection"; /** * In-flight states all present as Working (one steady state, per the @@ -498,6 +499,13 @@ function CollapsedWorkflowSection({ ); } +/** + * Fleet status visuals. Labels name the harness-reported state directly: + * a live external session is never shown as working on T3's behalf, and an + * unknown state never borrows an in-flight style. Rows, transcript, and + * composer live in FleetSection; this file only mounts it. + */ + /** A workflow's open state is presentation state, not a status derivative. */ function WorkflowSection({ group, @@ -525,12 +533,20 @@ export function AgentsPanel({ model, environmentId = null, threadId = null, + fleetEnvironmentIds = null, }: { model: AgentPanelModel; environmentId?: EnvironmentId | null; threadId?: ThreadId | null; + /** + * Connected environments whose native sessions appear in the fleet + * section. Null (or empty) hides the section; the thread-local workflows + * below are unchanged either way. + */ + fleetEnvironmentIds?: ReadonlyArray | null; }) { - if (!model.hasAgents) { + const showFleet = fleetEnvironmentIds !== null && fleetEnvironmentIds.length > 0; + if (!model.hasAgents && !showFleet) { return (
@@ -547,6 +563,9 @@ export function AgentsPanel({
+ {showFleet && fleetEnvironmentIds !== null ? ( + + ) : null} {model.workflows.map((group) => ( environment.environmentId)} /> ) : renderedRightPanelSurface?.kind === "device" ? ( diff --git a/apps/web/src/components/FleetSection.tsx b/apps/web/src/components/FleetSection.tsx new file mode 100644 index 000000000000..890997ff93ae --- /dev/null +++ b/apps/web/src/components/FleetSection.tsx @@ -0,0 +1,438 @@ +/** + * Fleet section of the agents panel: externally launched native agents + * (Codex first slice) across connected environments. + * + * Each environment's T3 server exposes its own native sessions; this + * section aggregates them under stable + * environment/provider/instance/native-thread identity. Rows are clickable: + * selecting one opens its live transcript with the composer that messages + * the native session in place. Selecting or viewing never stops the native + * session or changes its permissions. + */ +import { + appendFleetEvents, + emptyFleetState, + fleetDeliveryLabel, + fleetPanelRows, + ingestEnvironmentSnapshot, + markEnvironmentOffline, + mergeFleetEventLists, + removeFleetAgent, + setFleetEndpointNotice, + upsertFleetAgent, + type FleetPanelRow, + type FleetState, +} from "@t3tools/client-runtime/state/fleetRuntime"; +import type { + EnvironmentId, + FleetAgent, + FleetMessageDelivery, + FleetNativeEvent, + FleetStreamEvent, + ProviderInstanceId, +} from "@t3tools/contracts"; +import { useCallback, useEffect, useMemo, useState } from "react"; + +import { cn } from "~/lib/utils"; +import { Button } from "~/components/ui/button"; +import { Checkbox } from "~/components/ui/checkbox"; +import { Input } from "~/components/ui/input"; +import { ScrollArea } from "~/components/ui/scroll-area"; +import { fleetEnvironment } from "~/state/fleet"; +import { useEnvironmentQuery } from "~/state/query"; +import { useAtomCommand } from "~/state/use-atom-command"; + +export interface FleetSelection { + readonly environmentId: EnvironmentId; + readonly instanceId: ProviderInstanceId; + readonly nativeThreadId: string; +} + +function selectionKey(selection: FleetSelection): string { + return `${selection.environmentId}/${selection.instanceId}/${selection.nativeThreadId}`; +} + +const FLEET_STATUS_VISUALS: Record = { + active: { dotClass: "bg-info", label: "Active" }, + idle: { dotClass: "bg-muted-foreground/50", label: "Idle" }, + ended: { dotClass: "bg-muted-foreground/60", label: "Ended" }, + unknown: { dotClass: "bg-muted-foreground/60", label: "Unknown" }, +}; + +function FleetAgentRow({ + row, + selected, + onSelect, +}: { + row: FleetPanelRow; + selected: boolean; + onSelect: () => void; +}) { + const visuals = FLEET_STATUS_VISUALS[row.status]; + return ( + + ); +} + +function FleetEventRow({ event }: { event: FleetNativeEvent }) { + return ( +
+
+ {event.kind} +
+ {event.text !== null && event.text.length > 0 ? ( +
+ {event.text} +
+ ) : null} +
+ ); +} + +function FleetComposer({ + selection, + activeTurnId, + status, + online, + onSent, +}: { + selection: FleetSelection; + activeTurnId: string | null; + status: FleetPanelRow["status"]; + online: boolean; + onSent: (delivery: FleetMessageDelivery) => void; +}) { + const [text, setText] = useState(""); + const [ownershipConfirmed, setOwnershipConfirmed] = useState(false); + const [sending, setSending] = useState(false); + const sendMessage = useAtomCommand(fleetEnvironment.sendMessage, { + label: "fleet send-message", + reportFailure: false, + }); + + const needsOwnership = status === "idle" && activeTurnId === null; + const canSend = + online && !sending && text.trim().length > 0 && (!needsOwnership || ownershipConfirmed); + + const send = useCallback(async () => { + const trimmed = text.trim(); + if (trimmed.length === 0 || !canSend) return; + setSending(true); + try { + const result = await sendMessage({ + environmentId: selection.environmentId, + input: { + instanceId: selection.instanceId, + nativeThreadId: selection.nativeThreadId, + text: trimmed, + expectedActiveTurnId: activeTurnId, + ownershipKnown: !needsOwnership || ownershipConfirmed, + }, + }); + if (result._tag === "Success") { + onSent(result.value); + if (result.value.kind === "steered-active" || result.value.kind === "queued-followup") { + setText(""); + setOwnershipConfirmed(false); + } + } else { + onSent({ + kind: "uncertain", + reason: "The send call failed before the server answered. It was not resent.", + at: new Date().toISOString(), + }); + } + } finally { + setSending(false); + } + }, [ + activeTurnId, + canSend, + needsOwnership, + onSent, + ownershipConfirmed, + selection, + sendMessage, + text, + ]); + + return ( +
+ {needsOwnership ? ( + + ) : null} +
+ setText(event.target.value)} + onKeyDown={(event) => { + if (event.key === "Enter" && !event.shiftKey) { + event.preventDefault(); + void send(); + } + }} + placeholder={ + !online + ? "Environment offline" + : status === "active" + ? "Steer the running turn…" + : status === "idle" + ? "Start a follow-up turn…" + : "Session is not messageable" + } + disabled={!online || sending || (status !== "active" && status !== "idle")} + aria-label="Message the native agent" + /> + +
+
+ ); +} + +function FleetDetail({ + selection, + liveEvents, + agentRow, + onClose, +}: { + selection: FleetSelection; + liveEvents: ReadonlyArray; + agentRow: FleetPanelRow | undefined; + onClose: () => void; +}) { + const [delivery, setDelivery] = useState(null); + const history = useEnvironmentQuery( + fleetEnvironment.threadHistory({ + environmentId: selection.environmentId, + input: { instanceId: selection.instanceId, nativeThreadId: selection.nativeThreadId }, + }), + ); + const events = useMemo( + () => mergeFleetEventLists(history.data?.events ?? [], liveEvents), + [history.data, liveEvents], + ); + const activeTurnId = history.data?.activeTurnId ?? null; + const status = agentRow?.status ?? history.data?.agent.status ?? "unknown"; + const online = agentRow?.online ?? true; + + useEffect(() => { + setDelivery(null); + }, [selectionKey(selection)]); + + return ( +
+
+ {selection.nativeThreadId} + + {selection.environmentId} · {status} + {online ? "" : " · offline"} + + +
+ +
+ {history.isPending && events.length === 0 ? ( +

Loading native history…

+ ) : null} + {history.error !== null && events.length === 0 ? ( +

{history.error}

+ ) : null} + {events.map((event) => ( + + ))} + {events.length === 0 && !history.isPending && history.error === null ? ( +

No transcript items yet.

+ ) : null} +
+ + {delivery !== null ? ( +

+ {fleetDeliveryLabel(delivery)} + {delivery.reason !== null && + (delivery.kind === "uncertain" || + delivery.kind === "refused" || + delivery.kind === "failed") + ? `: ${delivery.reason}` + : ""} +

+ ) : null} + +
+ ); +} + +function FleetEnvSync({ + environmentId, + onSnapshot, + onStreamEvent, + onUnreachable, +}: { + environmentId: EnvironmentId; + onSnapshot: (snapshot: { environmentId: string; agents: ReadonlyArray }) => void; + onStreamEvent: (event: FleetStreamEvent) => void; + onUnreachable: (environmentId: string) => void; +}) { + const list = useEnvironmentQuery(fleetEnvironment.agentList({ environmentId, input: {} })); + const live = useEnvironmentQuery(fleetEnvironment.subscription({ environmentId, input: {} })); + const listData = list.data; + const listError = list.error; + const liveData = live.data; + + useEffect(() => { + if (listData !== null) { + onSnapshot({ environmentId, agents: listData.agents }); + } else if (listError !== null) { + onUnreachable(environmentId); + } + }, [environmentId, listData, listError, onSnapshot, onUnreachable]); + + useEffect(() => { + if (liveData !== null) onStreamEvent(liveData); + }, [liveData, onStreamEvent]); + + return null; +} + +export function FleetSection({ environmentIds }: { environmentIds: ReadonlyArray }) { + const [fleetState, setFleetState] = useState(emptyFleetState); + const [selection, setSelection] = useState(null); + + const ingestSnapshot = useCallback( + (snapshot: { environmentId: string; agents: ReadonlyArray }) => { + setFleetState((state) => ingestEnvironmentSnapshot(state, { ...snapshot, events: [] })); + }, + [], + ); + + const handleStreamEvent = useCallback((event: FleetStreamEvent) => { + setFleetState((state) => { + switch (event.kind) { + case "agent-updated": + return upsertFleetAgent(state, event.agent); + case "events-appended": + return appendFleetEvents(state, event.agentId, event.events); + case "agent-removed": + return removeFleetAgent(state, event.agentId); + case "endpoint-unreachable": + return setFleetEndpointNotice(state, event.instanceId, event.message); + } + }); + }, []); + + const handleUnreachable = useCallback((environmentId: string) => { + setFleetState((state) => markEnvironmentOffline(state, environmentId)); + }, []); + + const rows = useMemo(() => fleetPanelRows(fleetState), [fleetState]); + const selectedKey = selection === null ? null : selectionKey(selection); + const selectedRow = rows.find((row) => row.id === selectedKey); + const selectedLiveEvents = useMemo( + () => (selectedKey === null ? [] : (fleetState.eventsByAgent[selectedKey] ?? [])), + [fleetState, selectedKey], + ); + + if (environmentIds.length === 0) return null; + + return ( +
+
+ Fleet +
+ {environmentIds.map((environmentId) => ( + + ))} + {Object.entries(fleetState.endpointNotices).map(([instanceId, notice]) => ( +

+ {instanceId}: {notice} +

+ ))} + {rows.map((row) => ( + + setSelection((current) => + current !== null && selectionKey(current) === row.id + ? null + : { + environmentId: row.environmentId as EnvironmentId, + instanceId: row.instanceId as ProviderInstanceId, + nativeThreadId: row.nativeThreadId, + }, + ) + } + /> + ))} + {selection !== null ? ( +
+ setSelection(null)} + /> +
+ ) : null} +
+ ); +} diff --git a/apps/web/src/state/fleet.ts b/apps/web/src/state/fleet.ts new file mode 100644 index 000000000000..95e63ce0ff9f --- /dev/null +++ b/apps/web/src/state/fleet.ts @@ -0,0 +1,5 @@ +import { createFleetEnvironmentAtoms } from "@t3tools/client-runtime/state/fleet"; + +import { connectionAtomRuntime } from "../connection/runtime"; + +export const fleetEnvironment = createFleetEnvironmentAtoms(connectionAtomRuntime); diff --git a/packages/client-runtime/package.json b/packages/client-runtime/package.json index dd6113ad6426..0df04dcb46d8 100644 --- a/packages/client-runtime/package.json +++ b/packages/client-runtime/package.json @@ -223,6 +223,14 @@ "types": "./src/state/subagentRuntime.ts", "default": "./src/state/subagentRuntime.ts" }, + "./state/fleetRuntime": { + "types": "./src/state/fleetRuntime.ts", + "default": "./src/state/fleetRuntime.ts" + }, + "./state/fleet": { + "types": "./src/state/fleet.ts", + "default": "./src/state/fleet.ts" + }, "./state/thread-sort": { "types": "./src/state/threadSort.ts", "default": "./src/state/threadSort.ts" diff --git a/packages/client-runtime/src/rpc/client.ts b/packages/client-runtime/src/rpc/client.ts index 4f725378ffbf..7188a4823fc3 100644 --- a/packages/client-runtime/src/rpc/client.ts +++ b/packages/client-runtime/src/rpc/client.ts @@ -59,6 +59,7 @@ export type EnvironmentSubscriptionRpcTag = | typeof WS_METHODS.subscribeVcsStatus | typeof WS_METHODS.subscribeWorktreeSetup | typeof WS_METHODS.subscribeProjectClones + | typeof WS_METHODS.fleetSubscribe | typeof WS_METHODS.terminalAttach; export type EnvironmentStreamCommandRpcTag = diff --git a/packages/client-runtime/src/state/fleet.ts b/packages/client-runtime/src/state/fleet.ts new file mode 100644 index 000000000000..97c45838777f --- /dev/null +++ b/packages/client-runtime/src/state/fleet.ts @@ -0,0 +1,49 @@ +import { WS_METHODS } from "@t3tools/contracts"; +import { Atom } from "effect/unstable/reactivity"; + +import type { EnvironmentRegistry } from "../connection/registry.ts"; +import { + createEnvironmentRpcCommand, + createEnvironmentRpcQueryAtomFamily, + createEnvironmentRpcSubscriptionAtomFamily, +} from "./runtime.ts"; + +/** + * Fleet atoms for one environment's externally launched native agents. + * Each connected environment exposes its own Codex sessions through its + * own T3 server; the pane aggregates snapshots by stable + * environment/provider/instance/native-thread identity. The native endpoint + * URL itself never crosses the wire: it stays environment-local on the + * server that dials it. + */ +export function createFleetEnvironmentAtoms( + runtime: Atom.AtomRuntime, +) { + return { + /** Agents visible on one environment right now. */ + agentList: createEnvironmentRpcQueryAtomFamily(runtime, { + label: "environment-data:fleet:agent-list", + tag: WS_METHODS.fleetListAgents, + staleTimeMs: 15_000, + idleTtlMs: 60_000, + }), + /** Transcript history + running turn for one native session. */ + threadHistory: createEnvironmentRpcQueryAtomFamily(runtime, { + label: "environment-data:fleet:thread-history", + tag: WS_METHODS.fleetReadThread, + staleTimeMs: 10_000, + idleTtlMs: 60_000, + }), + /** Live agent, event, and endpoint updates for one environment. */ + subscription: createEnvironmentRpcSubscriptionAtomFamily(runtime, { + label: "environment-data:fleet:subscription", + tag: WS_METHODS.fleetSubscribe, + idleTtlMs: 60_000, + }), + /** Message one native session: active steer or guarded idle follow-up. */ + sendMessage: createEnvironmentRpcCommand(runtime, { + label: "environment-data:fleet:send-message", + tag: WS_METHODS.fleetSendMessage, + }), + }; +} diff --git a/packages/client-runtime/src/state/fleetRuntime.test.ts b/packages/client-runtime/src/state/fleetRuntime.test.ts new file mode 100644 index 000000000000..5e6da96a6add --- /dev/null +++ b/packages/client-runtime/src/state/fleetRuntime.test.ts @@ -0,0 +1,297 @@ +import { describe, expect, it } from "vite-plus/test"; +import type { + EnvironmentId, + FleetAgent, + FleetMessageDelivery, + FleetNativeEvent, + ProviderInstanceId, +} from "@t3tools/contracts"; + +import { + aggregateFleetAgents, + appendFleetEvents, + emptyFleetState, + fleetDeliveryLabel, + fleetPanelRows, + ingestEnvironmentSnapshot, + markEnvironmentOffline, + mergeFleetEventLists, + removeEnvironment, + removeFleetAgent, + setFleetEndpointNotice, + upsertFleetAgent, + type FleetEnvironmentSnapshot, +} from "./fleetRuntime.ts"; + +const SEEN_AT = "2026-09-24T18:13:42.000Z"; +const THREAD_ID = "01a0d431-b396-7023-a2aa-bc7ed6c6bc0c"; + +function agent( + environmentId: string, + nativeThreadId: string = THREAD_ID, + overrides: Partial = {}, +): FleetAgent { + return { + environmentId: environmentId as EnvironmentId, + provider: "codex", + instanceId: "codex" as ProviderInstanceId, + nativeThreadId, + status: "idle", + model: "gpt-5.6-luna", + role: null, + cwd: "/tmp/t3-fleet-spec/probe", + lastSeenAt: SEEN_AT, + ...overrides, + }; +} + +function event(id: string, nativeThreadId: string = THREAD_ID): FleetNativeEvent { + return { + id, + nativeThreadId, + kind: "item/completed", + at: SEEN_AT, + text: null, + }; +} + +function snapshot( + environmentId: string, + agents: ReadonlyArray, + events: ReadonlyArray = [], +): FleetEnvironmentSnapshot { + return { environmentId, agents, events }; +} + +describe("fleetRuntime aggregation", () => { + it("aggregates agents by environment under stable identity", () => { + const state = aggregateFleetAgents([ + snapshot("env-a", [agent("env-a")]), + snapshot("env-b", [agent("env-b")]), + ]); + const rows = fleetPanelRows(state); + expect(rows).toHaveLength(2); + expect(rows.map((row) => row.id).sort()).toEqual( + [`env-a/codex/codex/${THREAD_ID}`, `env-b/codex/codex/${THREAD_ID}`].sort(), + ); + }); + + it("keeps environments distinct for the same native thread", () => { + const state = aggregateFleetAgents([ + snapshot("env-a", [agent("env-a")]), + snapshot("env-b", [agent("env-b")]), + ]); + expect(Object.keys(state.agents)).toHaveLength(2); + }); + + it("ignores agents that do not belong to the snapshot environment", () => { + const state = ingestEnvironmentSnapshot(emptyFleetState(), snapshot("env-a", [agent("env-b")])); + expect(fleetPanelRows(state)).toHaveLength(0); + }); +}); + +describe("fleetRuntime reconnect deduplication", () => { + it("re-ingesting a snapshot does not duplicate agents or events", () => { + const first = ingestEnvironmentSnapshot( + emptyFleetState(), + snapshot("env-a", [agent("env-a")], [event("evt-1"), event("evt-2")]), + ); + const second = ingestEnvironmentSnapshot( + first, + snapshot("env-a", [agent("env-a")], [event("evt-1"), event("evt-2")]), + ); + expect(Object.keys(second.agents)).toEqual(Object.keys(first.agents)); + expect(second.eventsByAgent[`env-a/codex/codex/${THREAD_ID}`]).toHaveLength(2); + }); + + it("preserves row order across reconnects", () => { + const first = aggregateFleetAgents([ + snapshot("env-a", [agent("env-a", "thread-1"), agent("env-a", "thread-2")]), + ]); + const second = ingestEnvironmentSnapshot( + first, + snapshot("env-a", [agent("env-a", "thread-2"), agent("env-a", "thread-1")]), + ); + expect(fleetPanelRows(second).map((row) => row.nativeThreadId)).toEqual( + fleetPanelRows(first).map((row) => row.nativeThreadId), + ); + }); + + it("drops events for unknown agents instead of creating rows", () => { + const state = ingestEnvironmentSnapshot( + emptyFleetState(), + snapshot("env-a", [], [event("evt-1")]), + ); + expect(fleetPanelRows(state)).toHaveLength(0); + expect(appendFleetEvents(state, "env-a/codex/codex/missing", [event("evt-1")])).toBe(state); + }); +}); + +describe("fleetRuntime event continuity", () => { + it("merges history reads with live events without doubles", () => { + const withHistory = ingestEnvironmentSnapshot( + emptyFleetState(), + snapshot("env-a", [agent("env-a")], [event("evt-1"), event("evt-2")]), + ); + const withLive = appendFleetEvents(withHistory, `env-a/codex/codex/${THREAD_ID}`, [ + event("evt-2"), + event("evt-3"), + ]); + expect( + withLive.eventsByAgent[`env-a/codex/codex/${THREAD_ID}`]?.map((entry) => entry.id), + ).toEqual(["evt-1", "evt-2", "evt-3"]); + expect(fleetPanelRows(withLive)[0]?.eventCount).toBe(3); + }); + + it("removes a disconnected environment without marking its agents ended", () => { + const state = aggregateFleetAgents([ + snapshot("env-a", [agent("env-a")]), + snapshot("env-b", [agent("env-b")]), + ]); + const remaining = removeEnvironment(state, "env-a"); + const rows = fleetPanelRows(remaining); + expect(rows).toHaveLength(1); + expect(rows[0]?.environmentId).toBe("env-b"); + }); +}); + +describe("fleetRuntime delivery states", () => { + it("labels active steering and queued follow-ups distinctly", () => { + const steered: FleetMessageDelivery = { + kind: "steered-active", + reason: null, + at: SEEN_AT, + }; + const queued: FleetMessageDelivery = { + kind: "queued-followup", + reason: null, + at: SEEN_AT, + }; + expect(fleetDeliveryLabel(steered)).toContain("running turn"); + expect(fleetDeliveryLabel(queued)).toContain("follow-up"); + expect(fleetDeliveryLabel(steered)).not.toBe(fleetDeliveryLabel(queued)); + }); + + it("surfaces uncertain delivery instead of claiming success", () => { + const uncertain: FleetMessageDelivery = { + kind: "uncertain", + reason: "The native server did not confirm delivery.", + at: SEEN_AT, + }; + expect(fleetDeliveryLabel(uncertain)).toContain("not resent"); + }); + + it("reports refusals and failures with their reasons", () => { + const refused: FleetMessageDelivery = { + kind: "refused", + reason: "Ownership unknown.", + at: SEEN_AT, + }; + const failed: FleetMessageDelivery = { kind: "failed", reason: "Socket closed.", at: SEEN_AT }; + expect(fleetDeliveryLabel(refused)).toBe("Ownership unknown."); + expect(fleetDeliveryLabel(failed)).toBe("Socket closed."); + }); +}); + +describe("fleetRuntime panel rows", () => { + it("titles rows by native session identity with harness detail when known", () => { + const state = aggregateFleetAgents([snapshot("env-a", [agent("env-a")])]); + const rows = fleetPanelRows(state); + expect(rows[0]?.title).toBe(THREAD_ID); + expect(rows[0]?.detail).toBe("gpt-5.6-luna"); + expect(rows[0]?.environmentId).toBe("env-a"); + expect(rows[0]?.status).toBe("idle"); + }); + + it("leaves detail null when the harness reports neither role nor model", () => { + const state = aggregateFleetAgents([ + snapshot("env-a", [agent("env-a", THREAD_ID, { role: null, model: null })]), + ]); + expect(fleetPanelRows(state)[0]?.detail).toBeNull(); + }); +}); + +describe("fleetRuntime offline environments", () => { + it("keeps disconnected rows visible as offline instead of ended or idle", () => { + const online = aggregateFleetAgents([snapshot("env-a", [agent("env-a")])]); + expect(fleetPanelRows(online)[0]?.online).toBe(true); + const offline = markEnvironmentOffline(online, "env-a"); + const rows = fleetPanelRows(offline); + expect(rows).toHaveLength(1); + expect(rows[0]?.online).toBe(false); + expect(rows[0]?.status).toBe("idle"); + }); + + it("marks the environment online again on the next snapshot", () => { + const offline = markEnvironmentOffline( + aggregateFleetAgents([snapshot("env-a", [agent("env-a")])]), + "env-a", + ); + const back = ingestEnvironmentSnapshot(offline, snapshot("env-a", [agent("env-a")])); + expect(fleetPanelRows(back)[0]?.online).toBe(true); + }); + + it("removes rows only when the environment is removed, not on disconnect", () => { + const offline = markEnvironmentOffline( + aggregateFleetAgents([snapshot("env-a", [agent("env-a")])]), + "env-a", + ); + expect(fleetPanelRows(offline)).toHaveLength(1); + expect(fleetPanelRows(removeEnvironment(offline, "env-a"))).toHaveLength(0); + }); +}); + +describe("fleetRuntime live updates", () => { + it("upserts one agent without disturbing row order or events", () => { + const first = upsertFleetAgent(emptyFleetState(), agent("env-a")); + const second = upsertFleetAgent( + appendFleetEvents(first, "env-a/codex/codex/" + THREAD_ID, [event("evt-1")]), + agent("env-a", THREAD_ID, { status: "active" }), + ); + const rows = fleetPanelRows(second); + expect(rows).toHaveLength(1); + expect(rows[0]?.status).toBe("active"); + expect(rows[0]?.eventCount).toBe(1); + }); + + it("removes one agent while keeping its environment siblings", () => { + const state = aggregateFleetAgents([ + snapshot("env-a", [agent("env-a", "thread-1"), agent("env-a", "thread-2")]), + ]); + const removed = removeFleetAgent(state, "env-a/codex/codex/thread-1"); + expect(fleetPanelRows(removed).map((row) => row.nativeThreadId)).toEqual(["thread-2"]); + }); + + it("records and clears native endpoint notices per instance", () => { + const noticed = setFleetEndpointNotice(emptyFleetState(), "codex", "unreachable"); + expect(noticed.endpointNotices["codex"]).toBe("unreachable"); + expect(setFleetEndpointNotice(noticed, "codex", null).endpointNotices["codex"]).toBeUndefined(); + }); + + it("upserts same-id live rows in place so deltas never duplicate", () => { + const history = [ + event("evt-1"), + { ...event("msg-1"), kind: "agentMessage/delta", text: "FLEET" }, + ]; + const live = [ + { ...event("msg-1"), kind: "agentMessage/delta", text: "FLEET_MESSAGE_ACK_8426" }, + ]; + const merged = mergeFleetEventLists(history, live); + expect(merged.map((entry) => entry.id)).toEqual(["evt-1", "msg-1"]); + expect(merged[1]?.text).toBe("FLEET_MESSAGE_ACK_8426"); + }); + + it("re-ingesting a reconnect snapshot neither duplicates rows nor events", () => { + const first = ingestEnvironmentSnapshot( + emptyFleetState(), + snapshot("env-a", [agent("env-a")], [event("evt-1")]), + ); + const second = ingestEnvironmentSnapshot( + first, + snapshot("env-a", [agent("env-a")], [event("evt-1")]), + ); + expect(fleetPanelRows(second)).toHaveLength(1); + const key = "env-a/codex/codex/" + THREAD_ID; + expect(second.eventsByAgent[key]).toHaveLength(1); + }); +}); diff --git a/packages/client-runtime/src/state/fleetRuntime.ts b/packages/client-runtime/src/state/fleetRuntime.ts new file mode 100644 index 000000000000..ec9ec6287cb0 --- /dev/null +++ b/packages/client-runtime/src/state/fleetRuntime.ts @@ -0,0 +1,258 @@ +/** + * Fleet runtime model: aggregates native agents across connected + * environments for the fleet pane. + * + * Each environment supplies its own native sessions; this module combines + * them under the stable contract identity + * (`environmentId/codex/instanceId/nativeThreadId`) so reconnects reuse + * rows, history merges with live events without doubles, and delivery + * states stay truthful per harness semantics. + * + * Pure functions over plain data: no atoms, no sockets. The pane renders + * `fleetPanelRows`; networking stays with the existing environment + * connections. + */ +import { + fleetAgentId, + type FleetAgent, + type FleetMessageDelivery, + type FleetNativeEvent, +} from "@t3tools/contracts"; + +export interface FleetEnvironmentSnapshot { + readonly environmentId: string; + readonly agents: ReadonlyArray; + readonly events: ReadonlyArray; +} + +export interface FleetState { + /** Keyed by stable fleet agent id; insertion order is first-seen order. */ + readonly agents: Readonly>; + /** Keyed by stable fleet agent id; history first, then live, deduped. */ + readonly eventsByAgent: Readonly>>; + /** + * Last-known reachability per environment. A disconnected environment + * keeps its rows marked offline instead of vanishing or reporting its + * agents as ended or idle. + */ + readonly onlineByEnvironment: Readonly>; + /** Native endpoint problems per instance id, for truthful pane notices. */ + readonly endpointNotices: Readonly>; +} + +export interface FleetPanelRow { + readonly id: string; + readonly title: string; + readonly detail: string | null; + readonly environmentId: string; + readonly provider: FleetAgent["provider"]; + readonly instanceId: string; + readonly nativeThreadId: string; + readonly status: FleetAgent["status"]; + readonly eventCount: number; + readonly lastSeenAt: string; + /** False when the owning environment is disconnected; row stays visible. */ + readonly online: boolean; +} + +export function emptyFleetState(): FleetState { + return { agents: {}, eventsByAgent: {}, onlineByEnvironment: {}, endpointNotices: {} }; +} + +function mergeEvents( + current: ReadonlyArray | undefined, + incoming: ReadonlyArray, +): ReadonlyArray { + const merged = [...(current ?? [])]; + const indexById = new Map(merged.map((event, index) => [event.id, index] as const)); + for (const event of incoming) { + const index = indexById.get(event.id); + if (index === undefined) { + indexById.set(event.id, merged.length); + merged.push(event); + } else { + merged[index] = event; + } + } + return merged; +} + +/** + * Fold one environment snapshot into state. Re-ingesting the same + * snapshot (reconnect, rescan) updates `lastSeenAt` in place without + * duplicating agents, rows, or events, and marks the environment online. + */ +export function ingestEnvironmentSnapshot( + state: FleetState, + snapshot: FleetEnvironmentSnapshot, +): FleetState { + const agents: Record = { ...state.agents }; + const eventsByAgent: Record> = { + ...state.eventsByAgent, + }; + for (const agent of snapshot.agents) { + if (agent.environmentId !== snapshot.environmentId) continue; + agents[fleetAgentId(agent)] = agent; + } + const incomingByAgent = new Map>(); + for (const event of snapshot.events) { + const owner = snapshot.agents.find((agent) => agent.nativeThreadId === event.nativeThreadId); + if (owner === undefined) continue; + const key = fleetAgentId(owner); + const list = incomingByAgent.get(key) ?? []; + list.push(event); + incomingByAgent.set(key, list); + } + for (const [key, incoming] of incomingByAgent) { + eventsByAgent[key] = mergeEvents(eventsByAgent[key], incoming); + } + return { + agents, + eventsByAgent, + onlineByEnvironment: { ...state.onlineByEnvironment, [snapshot.environmentId]: true }, + endpointNotices: state.endpointNotices, + }; +} + +/** Combine snapshots from every connected environment. */ +export function aggregateFleetAgents( + snapshots: ReadonlyArray, +): FleetState { + return snapshots.reduce(ingestEnvironmentSnapshot, emptyFleetState()); +} + +/** + * Mark one environment offline. Its last-known agents and events stay + * visible as offline rows instead of being reported as ended or idle. + * A later snapshot marks the environment online again in place. + */ +export function markEnvironmentOffline(state: FleetState, environmentId: string): FleetState { + if (state.onlineByEnvironment[environmentId] === false) return state; + return { + ...state, + onlineByEnvironment: { ...state.onlineByEnvironment, [environmentId]: false }, + }; +} + +/** Upsert one live agent update without disturbing row order or events. */ +export function upsertFleetAgent(state: FleetState, agent: FleetAgent): FleetState { + return { + ...state, + agents: { ...state.agents, [fleetAgentId(agent)]: agent }, + onlineByEnvironment: { ...state.onlineByEnvironment, [agent.environmentId]: true }, + }; +} + +/** Remove one agent row, keeping its environment's other rows intact. */ +export function removeFleetAgent(state: FleetState, agentId: string): FleetState { + if (state.agents[agentId] === undefined) return state; + const agents: Record = { ...state.agents }; + const eventsByAgent: Record> = { + ...state.eventsByAgent, + }; + delete agents[agentId]; + delete eventsByAgent[agentId]; + return { ...state, agents, eventsByAgent }; +} + +/** Record or clear a native endpoint problem for one instance. */ +export function setFleetEndpointNotice( + state: FleetState, + instanceId: string, + notice: string | null, +): FleetState { + const endpointNotices: Record = { ...state.endpointNotices }; + if (notice === null) { + delete endpointNotices[instanceId]; + } else { + endpointNotices[instanceId] = notice; + } + return { ...state, endpointNotices }; +} + +/** + * Merge history events with live events for one agent. Same-id rows + * upsert in place so accumulated deltas and completions replace their + * earlier value instead of duplicating rows. + */ +export function mergeFleetEventLists( + history: ReadonlyArray, + live: ReadonlyArray, +): ReadonlyArray { + return mergeEvents(history, live); +} + +/** + * Drop everything one environment supplied. Used when the environment is + * removed, not when it disconnects; disconnects use + * `markEnvironmentOffline` so rows stay truthful. + */ +export function removeEnvironment(state: FleetState, environmentId: string): FleetState { + const agents: Record = {}; + const eventsByAgent: Record> = {}; + const onlineByEnvironment: Record = {}; + for (const [env, online] of Object.entries(state.onlineByEnvironment)) { + if (env !== environmentId) onlineByEnvironment[env] = online; + } + for (const [key, agent] of Object.entries(state.agents)) { + if (agent.environmentId === environmentId) continue; + agents[key] = agent; + const events = state.eventsByAgent[key]; + if (events !== undefined) eventsByAgent[key] = events; + } + return { agents, eventsByAgent, onlineByEnvironment, endpointNotices: state.endpointNotices }; +} + +/** Append live notification events to one agent, deduped by event id. */ +export function appendFleetEvents( + state: FleetState, + agentId: string, + events: ReadonlyArray, +): FleetState { + if (state.agents[agentId] === undefined) return state; + return { + ...state, + eventsByAgent: { + ...state.eventsByAgent, + [agentId]: mergeEvents(state.eventsByAgent[agentId], events), + }, + }; +} + +/** Pane rows in stable first-seen order. Titles use native identity only. */ +export function fleetPanelRows(state: FleetState): ReadonlyArray { + return Object.entries(state.agents).map(([id, agent]) => { + const detail = [agent.role, agent.model].filter( + (value): value is string => value !== null && value.trim().length > 0, + ); + return { + id, + title: agent.nativeThreadId, + detail: detail.length > 0 ? detail.join(" · ") : null, + environmentId: agent.environmentId, + provider: agent.provider, + instanceId: agent.instanceId, + nativeThreadId: agent.nativeThreadId, + status: agent.status, + eventCount: state.eventsByAgent[id]?.length ?? 0, + lastSeenAt: agent.lastSeenAt, + online: state.onlineByEnvironment[agent.environmentId] !== false, + }; + }); +} + +/** User-visible delivery wording. Uncertain stays uncertain. */ +export function fleetDeliveryLabel(delivery: FleetMessageDelivery): string { + switch (delivery.kind) { + case "steered-active": + return "Steered into the running turn"; + case "queued-followup": + return "Queued as a follow-up turn"; + case "uncertain": + return "Uncertain delivery, not resent"; + case "refused": + return delivery.reason ?? "Not sent"; + case "failed": + return delivery.reason ?? "Send failed"; + } +} diff --git a/packages/contracts/src/fleet.ts b/packages/contracts/src/fleet.ts new file mode 100644 index 000000000000..4c50d7a3895f --- /dev/null +++ b/packages/contracts/src/fleet.ts @@ -0,0 +1,226 @@ +import * as Schema from "effect/Schema"; +import { EnvironmentId, IsoDateTime, TrimmedNonEmptyString } from "./baseSchemas.ts"; +import { ProviderInstanceId } from "./providerInstance.ts"; + +/** + * Fleet pane contracts for externally launched native agents. + * + * First slice covers one Codex dispatcher reached through its own native + * WebSocket app-server endpoint. The endpoint is configured per Codex + * provider instance (see `CodexSettings.nativeEndpoint`); T3 attaches + * read-only and never takes over the native session lifecycle. + * + * Identity rule: a fleet agent is keyed by environment plus stable native + * identity (`environmentId`, driver `codex`, instance id, native thread + * id). Reconnects reuse the same key; no duplicate rows or events. + * + * @module fleet + */ + +/** WebSocket URL of an externally launched Codex native app-server. */ +export const FleetCodexNativeEndpoint = Schema.Struct({ + url: TrimmedNonEmptyString, +}); +export type FleetCodexNativeEndpoint = typeof FleetCodexNativeEndpoint.Type; + +const FLEET_NATIVE_ENDPOINT_PATTERN = /^wss?:\/\/.+/; + +/** Non-empty ws:// or wss:// URL. */ +export const FleetCodexNativeEndpointUrl = TrimmedNonEmptyString.check( + Schema.isPattern(FLEET_NATIVE_ENDPOINT_PATTERN), +); +export type FleetCodexNativeEndpointUrl = typeof FleetCodexNativeEndpointUrl.Type; + +/** True for values T3 may dial as a native Codex endpoint. */ +export function isFleetNativeEndpointUrl(value: string): boolean { + return FLEET_NATIVE_ENDPOINT_PATTERN.test(value.trim()); +} + +const decodeEndpointUrl = Schema.decodeUnknownOption(FleetCodexNativeEndpointUrl); + +/** + * Decode a configured endpoint value. Empty or missing values decode to + * null so an unconfigured instance stays disabled without an error; + * anything else must be a ws:// or wss:// URL. + */ +export function decodeFleetEndpointConfig(raw: unknown): + | { readonly _tag: "Missing" } + | { readonly _tag: "Endpoint"; readonly endpoint: FleetCodexNativeEndpoint } + | { + readonly _tag: "Invalid"; + } { + if (raw === null || raw === undefined) return { _tag: "Missing" }; + if (typeof raw !== "string" || raw.trim().length === 0) return { _tag: "Missing" }; + const decoded = decodeEndpointUrl(raw.trim()); + if (decoded._tag === "None") return { _tag: "Invalid" }; + return { _tag: "Endpoint", endpoint: { url: decoded.value } }; +} + +/** Lifecycle of a native session as the harness reports it. */ +export const FleetNativeThreadStatus = Schema.Literals(["active", "idle", "ended", "unknown"]); +export type FleetNativeThreadStatus = typeof FleetNativeThreadStatus.Type; + +/** + * One externally launched native agent visible in the fleet pane. + * Role and model stay null unless the harness reports them; T3 never + * guesses. + */ +export const FleetAgent = Schema.Struct({ + environmentId: EnvironmentId, + provider: Schema.Literal("codex"), + instanceId: ProviderInstanceId, + /** Stable native session identity (Codex thread id). */ + nativeThreadId: TrimmedNonEmptyString, + status: FleetNativeThreadStatus, + model: Schema.NullOr(TrimmedNonEmptyString), + role: Schema.NullOr(TrimmedNonEmptyString), + cwd: Schema.NullOr(Schema.String), + lastSeenAt: IsoDateTime, +}); +export type FleetAgent = typeof FleetAgent.Type; + +/** + * Stable pane identity for a fleet agent. Reconnects and rediscovery + * produce the same key, so rows, history, and events deduplicate. + */ +export function fleetAgentId( + agent: Pick, +): string { + return `${agent.environmentId}/${agent.provider}/${agent.instanceId}/${agent.nativeThreadId}`; +} + +/** How a user message reaches the native session. */ +export const FleetDeliveryKind = Schema.Literals([ + /** Active turn: steered in place with turn/steer and expectedTurnId. */ + "steered-active", + /** Idle session with known identity and ownership: native turn/start. */ + "queued-followup", + /** Send outcome unknown: surfaced, never auto resent. */ + "uncertain", + /** Rejected before send (for example unknown ownership, notLoaded session). */ + "refused", + /** The harness reported a send failure. */ + "failed", +]); +export type FleetDeliveryKind = typeof FleetDeliveryKind.Type; + +export const FleetMessageDelivery = Schema.Struct({ + kind: FleetDeliveryKind, + /** Present for refused, uncertain, and failed outcomes. */ + reason: Schema.NullOr(Schema.String), + at: IsoDateTime, +}); +export type FleetMessageDelivery = typeof FleetMessageDelivery.Type; + +/** + * One native transcript or live event, keyed for dedupe. History comes + * from native read APIs; live rows come from attached notifications. + * Both carry harness-assigned ids so reconnects merge without doubles. + */ +export const FleetNativeEvent = Schema.Struct({ + id: TrimmedNonEmptyString, + nativeThreadId: TrimmedNonEmptyString, + kind: TrimmedNonEmptyString, + at: IsoDateTime, + text: Schema.NullOr(Schema.String), +}); +export type FleetNativeEvent = typeof FleetNativeEvent.Type; + +/** Fleet RPC failure. The native session is never touched on failure. */ +export class FleetError extends Schema.TaggedError()("FleetError", { + operation: Schema.Literals([ + "list-agents", + "read-thread", + "send-message", + "subscribe", + "connect-endpoint", + ]), + message: Schema.String, + cause: Schema.optional(Schema.Defect()), +}) {} + +/** Empty for now; kept as a struct so filters can be added without a new method. */ +export const FleetListAgentsInput = Schema.Struct({}); +export type FleetListAgentsInput = typeof FleetListAgentsInput.Type; + +/** + * Agents visible on this environment right now. The endpoint URL itself is + * never included: native endpoints stay environment-local and never leak + * through remote links. + */ +export const FleetAgentListResult = Schema.Struct({ + agents: Schema.Array(FleetAgent), + scannedAt: IsoDateTime, +}); +export type FleetAgentListResult = typeof FleetAgentListResult.Type; + +export const FleetReadThreadInput = Schema.Struct({ + instanceId: ProviderInstanceId, + nativeThreadId: TrimmedNonEmptyString, +}); +export type FleetReadThreadInput = typeof FleetReadThreadInput.Type; + +/** + * Transcript history for one native session, read with native read APIs. + * `activeTurnId` is the running turn when the harness reports one; the + * composer steers against it with `expectedTurnId`. + */ +export const FleetThreadHistoryResult = Schema.Struct({ + agent: FleetAgent, + events: Schema.Array(FleetNativeEvent), + activeTurnId: Schema.NullOr(TrimmedNonEmptyString), + fetchedAt: IsoDateTime, +}); +export type FleetThreadHistoryResult = typeof FleetThreadHistoryResult.Type; + +export const FleetSendMessageInput = Schema.Struct({ + instanceId: ProviderInstanceId, + nativeThreadId: TrimmedNonEmptyString, + text: TrimmedNonEmptyString, + /** Running turn the sender saw; steer requires it to still match. */ + expectedActiveTurnId: Schema.optional(Schema.NullOr(TrimmedNonEmptyString)), + /** Sender confirms this idle session is ours to continue. */ + ownershipKnown: Schema.Boolean, +}); +export type FleetSendMessageInput = typeof FleetSendMessageInput.Type; + +export const FleetAgentUpdatedEvent = Schema.Struct({ + kind: Schema.Literal("agent-updated"), + agent: FleetAgent, +}); +export type FleetAgentUpdatedEvent = typeof FleetAgentUpdatedEvent.Type; + +export const FleetEventsAppendedEvent = Schema.Struct({ + kind: Schema.Literal("events-appended"), + agentId: TrimmedNonEmptyString, + events: Schema.Array(FleetNativeEvent), +}); +export type FleetEventsAppendedEvent = typeof FleetEventsAppendedEvent.Type; + +export const FleetAgentRemovedEvent = Schema.Struct({ + kind: Schema.Literal("agent-removed"), + agentId: TrimmedNonEmptyString, + environmentId: EnvironmentId, + instanceId: ProviderInstanceId, + nativeThreadId: TrimmedNonEmptyString, +}); +export type FleetAgentRemovedEvent = typeof FleetAgentRemovedEvent.Type; + +export const FleetEndpointUnreachableEvent = Schema.Struct({ + kind: Schema.Literal("endpoint-unreachable"), + instanceId: ProviderInstanceId, + message: TrimmedNonEmptyString, +}); +export type FleetEndpointUnreachableEvent = typeof FleetEndpointUnreachableEvent.Type; + +/** Live fleet updates for one environment. Scoped; disposed when unused. */ +export const FleetStreamEvent = Schema.Union([ + FleetAgentUpdatedEvent, + FleetEventsAppendedEvent, + FleetAgentRemovedEvent, + FleetEndpointUnreachableEvent, +]); +export type FleetStreamEvent = typeof FleetStreamEvent.Type; + +export const FleetSubscribeInput = Schema.Struct({}); +export type FleetSubscribeInput = typeof FleetSubscribeInput.Type; diff --git a/packages/contracts/src/index.ts b/packages/contracts/src/index.ts index 978a0459e69b..c99bfddad326 100644 --- a/packages/contracts/src/index.ts +++ b/packages/contracts/src/index.ts @@ -33,6 +33,7 @@ export * from "./editor.ts"; export * from "./project.ts"; export * from "./filesystem.ts"; export * from "./agentSessions.ts"; +export * from "./fleet.ts"; export * from "./assets.ts"; export * from "./review.ts"; export * from "./browserImport.ts"; diff --git a/packages/contracts/src/rpc.ts b/packages/contracts/src/rpc.ts index dbc143048a72..b4397a89c73c 100644 --- a/packages/contracts/src/rpc.ts +++ b/packages/contracts/src/rpc.ts @@ -39,6 +39,17 @@ import { AgentSessionScanResult, AgentSessionScanError, } from "./agentSessions.ts"; +import { + FleetAgentListResult, + FleetError, + FleetListAgentsInput, + FleetMessageDelivery, + FleetReadThreadInput, + FleetSendMessageInput, + FleetStreamEvent, + FleetSubscribeInput, + FleetThreadHistoryResult, +} from "./fleet.ts"; import { AssetAccessError, AssetCreateUrlInput, @@ -429,6 +440,12 @@ export const WS_METHODS = { projectCloneRetry: "projectClone.retry", subscribeProjectClones: "subscribeProjectClones", + // Fleet methods (externally launched native agents, Codex first slice) + fleetListAgents: "fleet.listAgents", + fleetReadThread: "fleet.readThread", + fleetSendMessage: "fleet.sendMessage", + fleetSubscribe: "fleet.subscribe", + // Streaming subscriptions subscribeVcsStatus: "subscribeVcsStatus", subscribeWorktreeSetup: "subscribeWorktreeSetup", @@ -1006,6 +1023,31 @@ const WsProviderUploadFeedbackRpc = Rpc.make(WS_METHODS.providerUploadFeedback, error: Schema.Union([ProviderUploadFeedbackError, EnvironmentAuthorizationError]), }); +const WsFleetListAgentsRpc = Rpc.make(WS_METHODS.fleetListAgents, { + payload: FleetListAgentsInput, + success: FleetAgentListResult, + error: Schema.Union([FleetError, EnvironmentAuthorizationError]), +}); + +const WsFleetReadThreadRpc = Rpc.make(WS_METHODS.fleetReadThread, { + payload: FleetReadThreadInput, + success: FleetThreadHistoryResult, + error: Schema.Union([FleetError, EnvironmentAuthorizationError]), +}); + +const WsFleetSendMessageRpc = Rpc.make(WS_METHODS.fleetSendMessage, { + payload: FleetSendMessageInput, + success: FleetMessageDelivery, + error: Schema.Union([FleetError, EnvironmentAuthorizationError]), +}); + +const WsFleetSubscribeRpc = Rpc.make(WS_METHODS.fleetSubscribe, { + payload: FleetSubscribeInput, + success: FleetStreamEvent, + error: Schema.Union([FleetError, EnvironmentAuthorizationError]), + stream: true, +}); + const WsSubscribeVcsStatusRpc = Rpc.make(WS_METHODS.subscribeVcsStatus, { payload: VcsStatusInput, success: VcsStatusStreamEvent, @@ -1473,6 +1515,10 @@ export const WsRpcGroup = RpcGroup.make( WsFilesystemBrowseRpc, WsAgentSessionsScanRpc, WsAgentSessionsImportRpc, + WsFleetListAgentsRpc, + WsFleetReadThreadRpc, + WsFleetSendMessageRpc, + WsFleetSubscribeRpc, WsAssetsCreateUrlRpc, WsAttachmentsCreateUploadUrlRpc, WsAttachmentsDeleteRpc, diff --git a/packages/contracts/src/settings.ts b/packages/contracts/src/settings.ts index 3e301201910c..6968df5aca57 100644 --- a/packages/contracts/src/settings.ts +++ b/packages/contracts/src/settings.ts @@ -612,13 +612,25 @@ export const CodexSettings = makeProviderSettingsSchema( description: "Additional CLI arguments passed to codex app-server on session start.", }), ), + nativeEndpoint: TrimmedString.pipe( + Schema.withDecodingDefault(Effect.succeed("")), + Schema.annotateKey({ + title: "Native fleet endpoint", + description: + "WebSocket URL of an externally launched Codex app-server to observe in the fleet pane. T3 attaches read-only and never stops that server.", + providerSettingsForm: { + placeholder: "ws://127.0.0.1:PORT", + clearWhenEmpty: "omit", + }, + }), + ), customModels: Schema.Array(CustomModelSetting).pipe( Schema.withDecodingDefault(Effect.succeed([])), Schema.annotateKey({ providerSettingsForm: { hidden: true } }), ), }, { - order: ["binaryPath", "homePath", "shadowHomePath", "launchArgs"], + order: ["binaryPath", "homePath", "shadowHomePath", "launchArgs", "nativeEndpoint"], }, ); export type CodexSettings = typeof CodexSettings.Type; @@ -1390,6 +1402,7 @@ const CodexSettingsPatch = Schema.Struct({ homePath: Schema.optionalKey(TrimmedString), shadowHomePath: Schema.optionalKey(TrimmedString), launchArgs: Schema.optionalKey(TrimmedString), + nativeEndpoint: Schema.optionalKey(TrimmedString), customModels: Schema.optionalKey(Schema.Array(CustomModelSetting)), }); diff --git a/packages/effect-codex-app-server/package.json b/packages/effect-codex-app-server/package.json index 616d71bdf4f9..d953ca3db1e0 100644 --- a/packages/effect-codex-app-server/package.json +++ b/packages/effect-codex-app-server/package.json @@ -18,6 +18,10 @@ "./errors": { "types": "./src/errors.ts", "import": "./src/errors.ts" + }, + "./protocol": { + "types": "./src/protocol.ts", + "import": "./src/protocol.ts" } }, "scripts": {