diff --git a/apps/server/src/auth/RpcAuthorization.test.ts b/apps/server/src/auth/RpcAuthorization.test.ts index 2c747fc71310..f489d917999b 100644 --- a/apps/server/src/auth/RpcAuthorization.test.ts +++ b/apps/server/src/auth/RpcAuthorization.test.ts @@ -15,6 +15,12 @@ import { } from "./RpcAuthorization.ts"; describe("RPC authorization scopes", () => { + it("requires read scope for the only Bridget RPC", () => { + expect(requiredScopeForRpcMethod("bridget.read")).toBe(AuthOrchestrationReadScope); + expect( + [...WsRpcGroup.requests.keys()].filter((method) => method.startsWith("bridget.")), + ).toEqual(["bridget.read"]); + }); it("declares exactly one scope for every RPC in the server group", () => { expect(new Set(Object.keys(RPC_REQUIRED_SCOPES))).toEqual(new Set(WsRpcGroup.requests.keys())); }); diff --git a/apps/server/src/auth/RpcAuthorization.ts b/apps/server/src/auth/RpcAuthorization.ts index 27fde35b9e90..61f96fca9670 100644 --- a/apps/server/src/auth/RpcAuthorization.ts +++ b/apps/server/src/auth/RpcAuthorization.ts @@ -22,6 +22,7 @@ type WsRpcMethod = RpcGroup.Rpcs["_tag"]; * runtime failure. */ export const RPC_REQUIRED_SCOPES = { + [WS_METHODS.bridgetRead]: AuthOrchestrationReadScope, [ORCHESTRATION_WS_METHODS.dispatchCommand]: AuthOrchestrationOperateScope, [ORCHESTRATION_WS_METHODS.getWorkflowScript]: AuthOrchestrationReadScope, [ORCHESTRATION_WS_METHODS.getTurnDiff]: AuthOrchestrationReadScope, diff --git a/apps/server/src/bridget/BridgetReader.test.ts b/apps/server/src/bridget/BridgetReader.test.ts new file mode 100644 index 000000000000..2b1c4be17fc8 --- /dev/null +++ b/apps/server/src/bridget/BridgetReader.test.ts @@ -0,0 +1,460 @@ +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { describe, expect, it } from "@effect/vitest"; +import { + ProjectId, + ThreadId, + type OrchestrationProjectShell, + type OrchestrationThreadShell, +} from "@t3tools/contracts"; +import { HostProcessEnvironment, HostProcessPlatform } from "@t3tools/shared/hostProcess"; +import { resolveCommandPath } from "@t3tools/shared/shell"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as FileSystem from "effect/FileSystem"; +import * as Path from "effect/Path"; +import * as Schema from "effect/Schema"; +import * as ChildProcessSpawner from "effect/unstable/process/ChildProcessSpawner"; +import * as ProcessRunner from "../processRunner.ts"; +import { ProjectionSnapshotQuery } from "../orchestration/Services/ProjectionSnapshotQuery.ts"; +import * as BridgetReader from "./BridgetReader.ts"; + +const threadId = ThreadId.make("8b09a229-dc14-4b38-91b5-2bf0e2a294ac"); +const projectId = ProjectId.make("project-1"); +const sharedThreadId = "3503a3ce-cb97-46eb-b17b-339377b0c1ce"; +const input = { threadId, projectId, action: "list" } as const; +const envelope = { + version: 1, + subject: { agent_id: threadId, name: null }, + result: { status: "listed", threads: [], next_after: null }, +}; +const encodeJson = Schema.encodeSync(Schema.fromJsonString(Schema.Unknown)); +const output = (stdout = encodeJson(envelope), code = 0): ProcessRunner.ProcessRunOutput => ({ + stdout, + stderr: "private diagnostic secret", + code: ChildProcessSpawner.ExitCode(code), + timedOut: false, + stdoutTruncated: false, + stderrTruncated: false, + stdoutInvalidUtf8: false, + stderrInvalidUtf8: false, +}); + +const harness = Effect.fn("BridgetReaderTest.harness")(function* ( + options: { + missingThread?: boolean; + missingProject?: boolean; + differentProject?: boolean; + environment?: NodeJS.ProcessEnv; + result?: ProcessRunner.ProcessRunOutput; + error?: ProcessRunner.ProcessRunError; + } = {}, +) { + const calls: ProcessRunner.ProcessRunInput[] = []; + const snapshots = ProjectionSnapshotQuery.of({ + getThreadShellById: () => + Effect.succeed( + options.missingThread + ? Option.none() + : Option.some({ + id: threadId, + projectId: options.differentProject ? ProjectId.make("another-project") : projectId, + } as OrchestrationThreadShell), + ), + getProjectShellById: () => + Effect.succeed( + options.missingProject + ? Option.none() + : Option.some({ + id: projectId, + workspaceRoot: "/project with spaces/$(forged); é", + } as OrchestrationProjectShell), + ), + } as unknown as ProjectionSnapshotQuery["Service"]); + const runner = ProcessRunner.ProcessRunner.of({ + run: (request) => { + calls.push(request); + return options.error + ? Effect.fail(options.error) + : Effect.succeed(options.result ?? output()); + }, + }); + const reader = yield* BridgetReader.BridgetReader.pipe( + Effect.provide( + BridgetReader.layer.pipe( + Layer.provide(Layer.succeed(ProjectionSnapshotQuery, snapshots)), + Layer.provide(Layer.succeed(ProcessRunner.ProcessRunner, runner)), + Layer.provide(NodeServices.layer), + Layer.provide( + Layer.succeed( + HostProcessEnvironment, + options.environment ?? { + T3CODE_BRIDGET_EXECUTABLE: "/usr/bin/true", + PATH: "/usr/bin:/bin", + }, + ), + ), + Layer.provide(Layer.succeed(HostProcessPlatform, "darwin")), + ), + ), + ); + return { reader, calls }; +}); + +describe("BridgetReader", () => { + it.effect("rejects recent history bounds unrelated to the requested immutable window", () => + Effect.gen(function* () { + for (const result of [ + { snapshot_seq: 8, through_seq: 5 }, + { snapshot_seq: 9, through_seq: 2 }, + { snapshot_seq: 7, through_seq: 2 }, + ]) { + const { reader } = yield* harness({ + result: output( + encodeJson({ + ...envelope, + result: { + status: "history_recent", + thread_id: sharedThreadId, + entries: [], + has_more: false, + next_before_seq: null, + ...result, + }, + }), + ), + }); + expect( + (yield* reader + .read({ ...input, action: "history_recent", sharedThreadId, beforeSeq: 2, toSeq: 8 }) + .pipe(Effect.flip)).code, + ).toBe("invalid_output"); + } + }), + ); + it.effect("rejects recent list results that repeat the requested cursor position", () => + Effect.gen(function* () { + const after = `1791331200:${sharedThreadId}`; + const { reader } = yield* harness({ + result: output( + encodeJson({ + ...envelope, + result: { + status: "listed_recent", + threads: [ + { + thread_id: sharedThreadId, + title: "Recent", + creator_id: threadId, + state: "open", + last_seq: 0, + members: [], + last_activity_at: 1791331200, + }, + ], + next_after: null, + }, + }), + ), + }); + expect( + (yield* reader.read({ ...input, action: "list_recent", after }).pipe(Effect.flip)).code, + ).toBe("invalid_output"); + }), + ); + it.effect("passes recent list cursor as one fixed argument and refuses old listed output", () => + Effect.gen(function* () { + const after = `1791331200:${sharedThreadId}`; + const recentEnvelope = { + ...envelope, + result: { status: "listed_recent", threads: [], next_after: null }, + }; + const { reader, calls } = yield* harness({ result: output(encodeJson(recentEnvelope)) }); + expect(yield* reader.read({ ...input, action: "list_recent", after, limit: 20 })).toEqual( + recentEnvelope, + ); + expect(calls[0]?.args.slice(-4)).toEqual(["--limit", "20", "--after", after]); + expect(calls[0]?.args).toContain("list_recent"); + const legacy = yield* harness(); + expect( + (yield* legacy.reader.read({ ...input, action: "list_recent" }).pipe(Effect.flip)).code, + ).toBe("invalid_output"); + }), + ); + it.effect( + "passes descending bounds without ASC flags and refuses a mismatched shared thread", + () => + Effect.gen(function* () { + const recentEnvelope = { + ...envelope, + result: { + status: "history_recent", + thread_id: sharedThreadId, + entries: [], + snapshot_seq: 8, + through_seq: 5, + has_more: false, + next_before_seq: null, + }, + }; + const { reader, calls } = yield* harness({ result: output(encodeJson(recentEnvelope)) }); + expect( + yield* reader.read({ + ...input, + action: "history_recent", + sharedThreadId, + limit: 20, + beforeSeq: 5, + toSeq: 8, + }), + ).toEqual(recentEnvelope); + expect(calls[0]?.args.slice(-6)).toEqual([ + "--limit", + "20", + "--before-seq", + "5", + "--to-seq", + "8", + ]); + expect(calls[0]?.args).not.toContain("--from-seq"); + const forged = yield* harness({ + result: output( + encodeJson({ + ...recentEnvelope, + result: { ...recentEnvelope.result, thread_id: threadId }, + }), + ), + }); + expect( + (yield* forged.reader + .read({ ...input, action: "history_recent", sharedThreadId }) + .pipe(Effect.flip)).code, + ).toBe("invalid_output"); + }), + ); + it.effect("uses authoritative project root and fixed argv with bounded execution", () => + Effect.gen(function* () { + const { reader, calls } = yield* harness(); + expect(yield* reader.read(input)).toEqual(envelope); + expect(calls).toHaveLength(1); + expect(calls[0]).toMatchObject({ + command: "/usr/bin/true", + args: [ + "thread", + "inspect", + "--t3-thread", + threadId, + "--project-root", + "/project with spaces/$(forged); é", + "--action", + "list", + "--json", + ], + timeout: "6 seconds", + maxOutputBytes: 256 * 1024, + outputMode: "error", + }); + }), + ); + it.effect("passes only action-specific paging options", () => + Effect.gen(function* () { + const { reader, calls } = yield* harness({ + result: output( + encodeJson({ + ...envelope, + result: { + status: "history", + from_seq: 2, + thread_id: sharedThreadId, + entries: [], + snapshot_seq: 8, + through_seq: 8, + has_more: false, + next_from_seq: null, + }, + }), + ), + }); + yield* reader.read({ + threadId, + projectId, + action: "history", + sharedThreadId, + limit: 20, + fromSeq: 2, + toSeq: 8, + }); + expect(calls[0]?.args).toContain("--thread"); + expect(calls[0]?.args.slice(-6)).toEqual([ + "--limit", + "20", + "--from-seq", + "2", + "--to-seq", + "8", + ]); + }), + ); + for (const scenario of [ + { missingThread: true }, + { missingProject: true }, + { differentProject: true }, + ]) { + it.effect( + `rejects missing or mismatched environment-local context ${JSON.stringify(scenario)} before spawn`, + () => + Effect.gen(function* () { + const { reader, calls } = yield* harness(scenario); + const error = yield* reader.read(input).pipe(Effect.flip); + expect(error.code).toBe( + scenario.differentProject ? "project_mismatch" : "context_missing", + ); + expect(calls).toHaveLength(0); + }), + ); + } + it.effect("rejects injected authority even through a direct service call", () => + Effect.gen(function* () { + const { reader, calls } = yield* harness(); + const error = yield* reader + .read({ ...input, root: "/forged" } as typeof input) + .pipe(Effect.flip); + expect(error.code).toBe("invalid_request"); + expect(calls).toHaveLength(0); + }), + ); + it.effect("decodes functional refusal JSON on exit 2 without stderr", () => + Effect.gen(function* () { + const refusal = { + version: 1, + subject: null, + result: { + status: "error", + code: "binding_unavailable", + detail: "Binding unavailable", + retryable: false, + }, + }; + const { reader } = yield* harness({ result: output(encodeJson(refusal), 2) }); + const view = yield* reader.read(input); + expect(view).toMatchObject({ + version: 1, + subject: null, + result: { status: "error", code: "binding_unavailable", retryable: false }, + }); + expect(encodeJson(view)).not.toContain("private diagnostic"); + }), + ); + for (const [name, result, code] of [ + ["JSON", output("invalid json"), "invalid_output"], + ["version", output(encodeJson({ ...envelope, version: 2 })), "unsupported_version"], + ["UTF-8", { ...output(), stdoutInvalidUtf8: true }, "invalid_output"], + ["truncated", { ...output(), stdoutTruncated: true }, "output_limit"], + ["oversized projection", output(" ".repeat(128 * 1024 + 1)), "output_limit"], + ["process exit", output("", 1), "command_failed"], + ["absent daemon", output("", 3), "unavailable"], + ] as const) { + it.effect(`refuses ${name} and sanitizes diagnostics`, () => + Effect.gen(function* () { + const { reader } = yield* harness({ result }); + const error = yield* reader.read(input).pipe(Effect.flip); + expect(error.code).toBe(code); + expect(error.message).not.toContain("private diagnostic"); + expect(encodeJson(error)).not.toContain("private diagnostic"); + }), + ); + } + it.effect("maps timeout and process output limit to distinct fixed errors", () => + Effect.gen(function* () { + for (const [error, code] of [ + [ + new ProcessRunner.ProcessTimeoutError({ + command: "bridget", + argumentCount: 1, + timeoutMs: 6000, + }), + "timeout", + ], + [ + new ProcessRunner.ProcessOutputLimitError({ + command: "bridget", + argumentCount: 1, + stream: "stdout", + maxBytes: 256 * 1024, + observedBytes: 256 * 1024 + 1, + }), + "output_limit", + ], + ] as const) { + const { reader } = yield* harness({ error }); + expect((yield* reader.read(input).pipe(Effect.flip)).code).toBe(code); + } + }), + ); + it.effect("prefers the user binary then falls back to PATH", () => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const home = yield* fs.makeTempDirectoryScoped(); + yield* fs.makeDirectory(path.join(home, ".local/bin"), { recursive: true }); + yield* fs.symlink("/usr/bin/true", path.join(home, ".local/bin/bridget")); + const user = yield* harness({ environment: { HOME: home, PATH: "/usr/bin" } }); + yield* user.reader.read(input); + expect(user.calls[0]?.command).toBe(path.join(home, ".local/bin/bridget")); + const bin = yield* fs.makeTempDirectoryScoped(); + yield* fs.symlink("/usr/bin/true", path.join(bin, "bridget")); + const fallback = yield* harness({ environment: { HOME: "/absent", PATH: bin } }); + yield* fallback.reader.read(input); + expect(fallback.calls[0]?.command).toBe(path.join(bin, "bridget")); + }).pipe(Effect.provide(NodeServices.layer)), + ); + it.effect("rejects cmd and bat executables before starting a process", () => + Effect.gen(function* () { + for (const command of ["/configured/bridget.cmd", "/configured/bridget.BAT"]) { + const { reader, calls } = yield* harness({ + environment: { T3CODE_BRIDGET_EXECUTABLE: command }, + }); + expect((yield* reader.read(input).pipe(Effect.flip)).code).toBe("unavailable"); + expect(calls).toHaveLength(0); + } + }), + ); + for (const source of ["configured", "user", "PATH"] as const) { + it.effect( + `rejects Windows PATHEXT cmd resolution from ${source} before starting a process`, + () => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const home = yield* fs.makeTempDirectoryScoped(); + const directory = source === "user" ? path.join(home, ".local", "bin") : home; + yield* fs.makeDirectory(directory, { recursive: true }); + const command = path.join(directory, "bridget"); + yield* fs.symlink("/usr/bin/true", `${command}.CMD`); + const environment = { + PATHEXT: ".CMD", + ...(source === "configured" ? { T3CODE_BRIDGET_EXECUTABLE: command } : {}), + ...(source === "user" ? { HOME: home } : {}), + ...(source === "PATH" ? { PATH: home } : {}), + }; + expect( + yield* resolveCommandPath(source === "PATH" ? "bridget" : command, { + env: environment, + }), + ).toBe(`${command}.CMD`); + const { reader, calls } = yield* harness({ environment }); + const outcome = yield* reader.read(input).pipe( + Effect.match({ + onFailure: (error) => error.code, + onSuccess: () => "success", + }), + ); + expect(calls).toHaveLength(0); + expect(outcome).toBe("unavailable"); + }).pipe( + Effect.provideService(HostProcessPlatform, "win32"), + Effect.provide(NodeServices.layer), + ), + ); + } +}); diff --git a/apps/server/src/bridget/BridgetReader.ts b/apps/server/src/bridget/BridgetReader.ts new file mode 100644 index 000000000000..f6d2844b67c5 --- /dev/null +++ b/apps/server/src/bridget/BridgetReader.ts @@ -0,0 +1,200 @@ +import { BridgetHumanView, BridgetReadError, BridgetReadInput } from "@t3tools/contracts"; +import { HostProcessEnvironment } from "@t3tools/shared/hostProcess"; +import { resolveCommandPath } from "@t3tools/shared/shell"; +import * as Context from "effect/Context"; +import * as Effect from "effect/Effect"; +import * as FileSystem from "effect/FileSystem"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as Path from "effect/Path"; +import * as Schema from "effect/Schema"; +import * as ProcessRunner from "../processRunner.ts"; +import * as ProjectionSnapshotQuery from "../orchestration/Services/ProjectionSnapshotQuery.ts"; + +const decodeInput = Schema.decodeUnknownEffect(BridgetReadInput); +const decodeView = Schema.decodeUnknownEffect(Schema.fromJsonString(BridgetHumanView)); +const decodeVersion = Schema.decodeUnknownEffect( + Schema.fromJsonString(Schema.Struct({ version: Schema.Int })), +); +const refusalDetails = { + unsupported_version: "The human thread protocol is incompatible.", + binding_unavailable: "The selected conversation has no available Bridget binding.", + project_mismatch: "The selected conversation belongs to another project.", + thread_unavailable: "This shared thread is unavailable for the selected conversation.", + invalid_request: "The human thread request is invalid.", + storage_unavailable: "The shared thread storage is unavailable.", + response_too_large: "The human thread response exceeds the read limit.", +} as const; + +export class BridgetReader extends Context.Service< + BridgetReader, + { + readonly read: (input: BridgetReadInput) => Effect.Effect; + } +>()("t3/bridget/BridgetReader") {} + +const make = Effect.gen(function* () { + const snapshots = yield* ProjectionSnapshotQuery.ProjectionSnapshotQuery; + const runner = yield* ProcessRunner.ProcessRunner; + const environment = yield* HostProcessEnvironment; + const path = yield* Path.Path; + const fileSystem = yield* FileSystem.FileSystem; + const resolveExecutable = Effect.gen(function* () { + const configured = environment.T3CODE_BRIDGET_EXECUTABLE?.trim(); + if (configured) { + return yield* resolveCommandPath(configured, { env: environment }).pipe( + Effect.mapError(() => new BridgetReadError({ code: "unavailable" })), + ); + } + if (environment.HOME) { + const userBinary = yield* resolveCommandPath( + path.join(environment.HOME, ".local", "bin", "bridget"), + { env: environment }, + ).pipe(Effect.option); + if (Option.isSome(userBinary)) return userBinary.value; + } + return yield* resolveCommandPath("bridget", { env: environment }).pipe( + Effect.mapError(() => new BridgetReadError({ code: "unavailable" })), + ); + }).pipe( + Effect.flatMap((executable) => + /\.(cmd|bat)$/i.test(executable) + ? Effect.fail(new BridgetReadError({ code: "unavailable" })) + : Effect.succeed(executable), + ), + Effect.provideService(Path.Path, path), + Effect.provideService(FileSystem.FileSystem, fileSystem), + ); + + // O(n + p): bounded response decoding plus executable PATH search; no per-message I/O. + const read = Effect.fn("BridgetReader.read")(function* (raw: BridgetReadInput) { + const input = yield* decodeInput(raw).pipe( + Effect.mapError(() => new BridgetReadError({ code: "invalid_request" })), + ); + const thread = yield* snapshots + .getThreadShellById(input.threadId) + .pipe(Effect.mapError(() => new BridgetReadError({ code: "storage_unavailable" }))); + if (Option.isNone(thread)) return yield* new BridgetReadError({ code: "context_missing" }); + if (thread.value.projectId !== input.projectId) + return yield* new BridgetReadError({ code: "project_mismatch" }); + const project = yield* snapshots + .getProjectShellById(thread.value.projectId) + .pipe(Effect.mapError(() => new BridgetReadError({ code: "storage_unavailable" }))); + if (Option.isNone(project)) return yield* new BridgetReadError({ code: "context_missing" }); + const command = yield* resolveExecutable; + const args = [ + "thread", + "inspect", + "--t3-thread", + input.threadId, + "--project-root", + project.value.workspaceRoot, + "--action", + input.action, + "--json", + ]; + const isList = input.action === "list" || input.action === "list_recent"; + if (input.action === "show" || input.action === "history" || input.action === "history_recent") + args.push("--thread", input.sharedThreadId); + if (input.action !== "show" && input.limit !== undefined) + args.push("--limit", String(input.limit)); + if ((input.action === "list" || input.action === "list_recent") && input.after !== undefined) + args.push("--after", input.after); + if (input.action === "history") { + if (input.fromSeq !== undefined) args.push("--from-seq", String(input.fromSeq)); + if (input.toSeq !== undefined) args.push("--to-seq", String(input.toSeq)); + } + if (input.action === "history_recent") { + if (input.beforeSeq !== undefined) args.push("--before-seq", String(input.beforeSeq)); + if (input.toSeq !== undefined) args.push("--to-seq", String(input.toSeq)); + } + const output = yield* runner + .run({ command, args, timeout: "6 seconds", maxOutputBytes: 256 * 1024, outputMode: "error" }) + .pipe( + Effect.mapError( + (error) => + new BridgetReadError({ + code: + error._tag === "ProcessTimeoutError" + ? "timeout" + : error._tag === "ProcessOutputLimitError" + ? "output_limit" + : error._tag === "ProcessSpawnError" + ? "unavailable" + : "command_failed", + }), + ), + ); + if (output.timedOut) return yield* new BridgetReadError({ code: "timeout" }); + if ( + output.stdoutTruncated || + output.stderrTruncated || + Buffer.byteLength(output.stdout, "utf8") > 128 * 1024 + ) + return yield* new BridgetReadError({ code: "output_limit" }); + if (output.stdoutInvalidUtf8) return yield* new BridgetReadError({ code: "invalid_output" }); + if (output.code === 3) return yield* new BridgetReadError({ code: "unavailable" }); + if (output.code !== 0 && output.code !== 2) + return yield* new BridgetReadError({ code: "command_failed" }); + const header = yield* decodeVersion(output.stdout).pipe( + Effect.mapError(() => new BridgetReadError({ code: "invalid_output" })), + ); + if (header.version !== 1) return yield* new BridgetReadError({ code: "unsupported_version" }); + const view = yield* decodeView(output.stdout).pipe( + Effect.mapError(() => new BridgetReadError({ code: "invalid_output" })), + ); + if (view.result.status === "error") + return { + ...view, + subject: null, + result: { ...view.result, detail: refusalDetails[view.result.code] }, + }; + const expectedStatus = { + list: "listed", + list_recent: "listed_recent", + show: "shown", + history: "history", + history_recent: "history_recent", + }[input.action]; + if ( + output.code !== 0 || + view.subject === null || + view.result.status !== expectedStatus || + (!isList && + "sharedThreadId" in input && + view.result.status !== "listed" && + view.result.status !== "listed_recent" && + view.result.thread_id !== input.sharedThreadId) + ) { + return yield* new BridgetReadError({ code: "invalid_output" }); + } + if (input.action === "history_recent" && view.result.status === "history_recent") { + if ( + (input.toSeq !== undefined && view.result.snapshot_seq !== input.toSeq) || + view.result.through_seq !== + Math.min(input.beforeSeq ?? view.result.snapshot_seq, view.result.snapshot_seq) || + view.result.entries.length > (input.limit ?? 200) + ) + return yield* new BridgetReadError({ code: "invalid_output" }); + } + if (input.action === "list_recent" && view.result.status === "listed_recent") { + if (view.result.threads.length > (input.limit ?? 100)) + return yield* new BridgetReadError({ code: "invalid_output" }); + if (input.after !== undefined) { + const [timestamp, id] = input.after.split(":"); + if ( + view.result.threads.some( + (row) => + row.last_activity_at > Number(timestamp) || + (row.last_activity_at === Number(timestamp) && row.thread_id <= id!), + ) + ) + return yield* new BridgetReadError({ code: "invalid_output" }); + } + } + return view; + }); + return BridgetReader.of({ read }); +}); + +export const layer = Layer.effect(BridgetReader, make); diff --git a/apps/server/src/server.test.ts b/apps/server/src/server.test.ts index 221941aa7833..f7f0d0716910 100644 --- a/apps/server/src/server.test.ts +++ b/apps/server/src/server.test.ts @@ -6763,6 +6763,62 @@ it.layer(NodeServices.layer)("server router seam", (it) => { }).pipe(Effect.provide(NodeHttpServer.layerTest)), ); + it.effect("Bridget human read requires read scope and never dispatches an agent command", () => + Effect.gen(function* () { + let contextReads = 0; + let dispatched = 0; + yield* buildAppUnderTest({ + layers: { + projectionSnapshotQuery: { + getThreadShellById: () => + Effect.sync(() => { + contextReads += 1; + return Option.none(); + }), + }, + orchestrationEngine: { + dispatch: () => + Effect.sync(() => { + dispatched += 1; + return { sequence: 0 }; + }), + }, + }, + }); + for (const scope of ["access:read", "orchestration:read"]) { + const token = yield* exchangeAccessToken(defaultDesktopBootstrapToken, { scope }); + assert.equal(token.response.status, 200); + const ticketResponse = yield* HttpClient.post("/api/auth/websocket-ticket", { + headers: { authorization: `Bearer ${token.body.access_token ?? ""}` }, + }); + const { ticket } = yield* responseJsonEffect<{ readonly ticket: string }>(ticketResponse); + const wsUrl = `${yield* getWsServerUrl("/ws", { authenticated: false })}?wsTicket=${encodeURIComponent(ticket)}`; + yield* Effect.scoped( + withWsRpcClient(wsUrl, (client) => + Effect.gen(function* () { + const error = yield* client[WS_METHODS.bridgetRead]({ + threadId: ThreadId.make("8b09a229-dc14-4b38-91b5-2bf0e2a294ac"), + projectId: defaultProjectId, + action: "list", + }).pipe(Effect.flip); + if (scope === "access:read") { + assert.equal(error._tag, "EnvironmentAuthorizationError"); + if (error._tag === "EnvironmentAuthorizationError") + assert.equal(error.requiredScope, "orchestration:read"); + assert.equal(contextReads, 0); + } else { + assert.equal(error._tag, "BridgetReadError"); + if (error._tag === "BridgetReadError") assert.equal(error.code, "context_missing"); + assert.equal(contextReads, 1); + } + }), + ), + ); + } + assert.equal(dispatched, 0); + }).pipe(Effect.provide(NodeHttpServer.layerTest)), + ); + it.effect("provider setup binds private sign-in to the authenticated websocket session", () => Effect.gen(function* () { const flowId = "private-sign-in-flow"; diff --git a/apps/server/src/server.ts b/apps/server/src/server.ts index 43bba34bd8c5..65b215113adc 100644 --- a/apps/server/src/server.ts +++ b/apps/server/src/server.ts @@ -101,6 +101,7 @@ import * as ServerSettings from "./serverSettings.ts"; import * as NativeAppIconResolver from "./assets/NativeAppIconResolver.ts"; import * as ProjectFaviconResolver from "./project/ProjectFaviconResolver.ts"; import * as T3ProjectFileLoader from "./project/T3ProjectFileLoader.ts"; +import * as BridgetReader from "./bridget/BridgetReader.ts"; import * as RepositoryIdentityResolver from "./project/RepositoryIdentityResolver.ts"; import * as WorkspaceEntries from "./workspace/WorkspaceEntries.ts"; import * as WorkspaceFileSystem from "./workspace/WorkspaceFileSystem.ts"; @@ -618,6 +619,7 @@ export const makeRoutesLayer = Layer.mergeAll( // Both transports consume the same service instance, so caches single-flight across clients // and mutations observed on WebSocket invalidate patches subsequently read over HTTP. Layer.provide(PullRequestServiceLive), + Layer.provide(BridgetReader.layer.pipe(Layer.provide(ProcessRunner.layer))), Layer.provide(PreviewAutomationBroker.layer), Layer.provide(ServerSelfUpdate.layer.pipe(Layer.provide(DesktopAppUpdateLayerLive))), Layer.provide(commandReadinessLayer), diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index bcc19be00da5..fffafc6c803f 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -124,6 +124,7 @@ import * as ServerSelfUpdate from "./cloud/selfUpdate.ts"; import * as ServerLifecycleEvents from "./serverLifecycleEvents.ts"; import * as ServerRuntimeStartup from "./serverRuntimeStartup.ts"; import * as ServerSettings from "./serverSettings.ts"; +import * as BridgetReader from "./bridget/BridgetReader.ts"; import * as TerminalManager from "./terminal/Manager.ts"; import { withTerminalOutputWindow } from "./terminal/OutputProtocol.ts"; import * as PreviewAutomationBroker from "./mcp/PreviewAutomationBroker.ts"; @@ -580,6 +581,7 @@ const makeWsRpcLayer = ( const config = yield* ServerConfig.ServerConfig; const lifecycleEvents = yield* ServerLifecycleEvents.ServerLifecycleEvents; const serverSettings = yield* ServerSettings.ServerSettingsService; + const bridgetReader = yield* BridgetReader.BridgetReader; const startup = yield* ServerRuntimeStartup.ServerRuntimeStartup; const workspaceEntries = yield* WorkspaceEntries.WorkspaceEntries; const workspaceFileSystem = yield* WorkspaceFileSystem.WorkspaceFileSystem; @@ -2137,6 +2139,8 @@ const makeWsRpcLayer = ( .pipe(Effect.ignoreCause({ log: true }), Effect.forkDetach, Effect.asVoid); return WsRpcGroup.of({ + [WS_METHODS.bridgetRead]: (input) => + observeRpcEffect(WS_METHODS.bridgetRead, bridgetReader.read(input)), [ORCHESTRATION_WS_METHODS.dispatchCommand]: (command) => observeRpcEffect( ORCHESTRATION_WS_METHODS.dispatchCommand, @@ -4108,6 +4112,7 @@ export const websocketRpcRouteLayer = Layer.unwrap( ), }); const pullRequests = yield* PullRequestService.PullRequestService; + const bridgetReader = yield* BridgetReader.BridgetReader; const sql = yield* SqlClient.SqlClient; return HttpRouter.add( "GET", @@ -4156,6 +4161,7 @@ export const websocketRpcRouteLayer = Layer.unwrap( // One server-lifetime service means clients share the same PR caches, and a WS // mutation invalidates the HTTP diff cache that every client reads from. Layer.provide(Layer.succeed(PullRequestService.PullRequestService, pullRequests)), + Layer.provide(Layer.succeed(BridgetReader.BridgetReader, bridgetReader)), Layer.provide( SourceControlDiscovery.layer.pipe( Layer.provide( diff --git a/apps/web/src/components/BridgetPanel.test.tsx b/apps/web/src/components/BridgetPanel.test.tsx new file mode 100644 index 000000000000..f613e5ada835 --- /dev/null +++ b/apps/web/src/components/BridgetPanel.test.tsx @@ -0,0 +1,788 @@ +import { EnvironmentId, ProjectId, ThreadId } from "@t3tools/contracts"; +import { act, cloneElement, useSyncExternalStore, type ReactElement, type ReactNode } from "react"; +import { create, type ReactTestRenderer } from "react-test-renderer"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vite-plus/test"; + +import { BridgetPanel } from "./BridgetPanel"; + +type Target = { + environmentId: string; + input: { + threadId: string; + projectId: string; + action: string; + sharedThreadId?: string; + after?: string; + fromSeq?: number; + beforeSeq?: number; + toSeq?: number; + }; +}; +type View = { + data: unknown; + dataUpdatedAt: number | null; + error: string | null; + isPending: boolean; + isSuccess: boolean; + refresh: () => void; +}; +const resources = new Map(); +const listeners = new Set<() => void>(); +const requests: Target[] = []; +const measurements = new Map(); +const observers = new Set<{ callback: () => void; target: object | null }>(); +const controls = new Map }>(); +let activeElement: object | null = null; +const ownerDocument = { + get activeElement() { + return activeElement; + }, +}; +let timestamp = 0; +const keyOf = (target: Target) => JSON.stringify(target); +function resource(target: Target): View { + const key = keyOf(target); + const cached = resources.get(key); + if (cached) return cached; + requests.push(target); + const view: View = { + data: null, + dataUpdatedAt: null, + error: null, + isPending: true, + isSuccess: false, + refresh: () => { + requests.push(target); + resources.set(key, { ...resources.get(key)!, isPending: true }); + for (const listener of listeners) listener(); + }, + }; + resources.set(key, view); + return view; +} +vi.mock("~/state/orchestration", () => ({ + orchestrationEnvironment: { bridgetRead: (target: Target) => target }, +})); +vi.mock("~/state/query", () => ({ + useEnvironmentQuery: (target: Target | null) => { + const empty = { + data: null, + dataUpdatedAt: null, + error: null, + isPending: false, + isSuccess: false, + refresh: () => {}, + }; + if (target) resource(target); + return useSyncExternalStore( + (listener) => { + listeners.add(listener); + return () => listeners.delete(listener); + }, + () => (target ? resource(target) : empty), + ); + }, +})); +vi.mock("~/components/ui/button", () => ({ Button: "button" })); +vi.mock("~/components/ui/input", () => ({ Input: "input" })); +vi.mock("~/components/ui/tooltip", () => ({ + Tooltip: ({ children }: { children?: ReactNode }) => <>{children}, + TooltipTrigger: ({ render, children }: { render: ReactElement; children: ReactNode }) => + cloneElement(render, {}, children), + TooltipPopup: ({ children }: { children?: ReactNode }) => <>{children}, +})); +vi.mock("~/components/ui/scroll-area", () => ({ + ScrollArea: ({ children }: { children?: ReactNode }) =>
{children}
, +})); +const copy = vi.hoisted(() => vi.fn(async () => true)); +vi.mock("~/hooks/useCopyToClipboard", () => ({ writeTextToClipboard: copy })); + +const context = { + environmentId: EnvironmentId.make("env-A"), + threadId: ThreadId.make("t3-A"), + projectId: ProjectId.make("p-A"), + visible: true, +}; +let renderer: ReactTestRenderer | null = null; +function text(): string { + return JSON.stringify(renderer?.toJSON()); +} +function button(label: string) { + return renderer!.root + .findAllByType("button") + .find((node) => node.props["aria-label"] === label || node.children.includes(label))!; +} +async function mount(props = context) { + await act(async () => { + renderer = create(, { + createNodeMock: (element) => { + const props = element.props as { + "data-bridget-body"?: boolean; + "aria-label"?: string; + children?: ReactNode; + }; + if (element.type === "button" && props["aria-label"]) { + const control = { + focus: vi.fn(() => { + activeElement = control; + }), + }; + controls.set(props["aria-label"], control); + return control; + } + if (!props["data-bridget-body"]) return null; + const contents = String(props.children); + const measured = measurements.get(contents) ?? { clientHeight: 80, scrollHeight: 40 }; + measurements.set(contents, measured); + return Object.assign(measured, { ownerDocument }); + }, + }); + }); +} +async function respond(action: string, result: unknown, sharedThreadId?: string) { + const target = requests.findLast( + (request) => request.input.action === action && request.input.sharedThreadId === sharedThreadId, + )!; + await act(async () => { + resources.set(keyOf(target), { + ...resource(target), + data: { version: 1, subject: { agent_id: "agent-A", name: "Agent A" }, result }, + dataUpdatedAt: ++timestamp, + isPending: false, + isSuccess: true, + error: null, + }); + for (const listener of listeners) listener(); + }); +} +const thread = (id: string, title = id) => ({ + thread_id: id, + title, + creator_id: "agent-A", + state: "open", + last_seq: 6, + last_activity_at: 1791378000, + members: [ + { agent_id: "agent-A", name: "Alice" }, + { agent_id: "agent-B", name: null }, + ], +}); +const body = " ```typescript\r\n\tconst café = '你好 👩🏽‍💻';\r\n```\n fin "; +const entry = (seq: number, kind?: string) => ({ + seq, + message_id: `m-${seq}`, + author_id: "agent-A", + author_name: seq === 2 ? null : "Alice", + created_at: 1791378000, + body: seq === 1 ? body : `Texte ${seq}`, + notify: { mode: "all", targets: [] }, + reply_to_seq: null, + ...(kind ? { kind } : {}), +}); +async function open(id: string, closed = false) { + await act(async () => button(`Ouvrir ${id}`).props.onClick()); + await respond( + "show", + { + status: "shown", + ...thread(id), + state: closed ? "closed" : "open", + created_at: 1791378000, + closed_at: closed ? 1791379000 : null, + }, + id, + ); +} +async function page(id: string, from: number, through: number, snapshot = 6) { + await respond( + "history_recent", + { + status: "history_recent", + thread_id: id, + through_seq: through, + snapshot_seq: snapshot, + has_more: from > 1, + next_before_seq: from > 1 ? from - 1 : null, + entries: Array.from({ length: through - from + 1 }, (_, index) => + entry(through - index, ["history", "action", "blocker", "decision"][through - index - 1]), + ), + }, + id, + ); +} +beforeEach(() => { + resources.clear(); + requests.length = 0; + timestamp = 0; + copy.mockClear(); + measurements.clear(); + observers.clear(); + controls.clear(); + activeElement = null; + vi.stubGlobal("IS_REACT_ACT_ENVIRONMENT", true); + vi.stubGlobal( + "ResizeObserver", + class { + record: { callback: () => void; target: object | null }; + constructor(callback: () => void) { + this.record = { callback, target: null }; + } + observe(target: object) { + this.record.target = target; + observers.add(this.record); + } + disconnect() { + observers.delete(this.record); + } + }, + ); +}); +afterEach(async () => { + await act(async () => renderer?.unmount()); + renderer = null; + listeners.clear(); + vi.unstubAllGlobals(); +}); + +describe("Bridget native reader", () => { + it("requests a globally recent list and the latest history page, not an oldest-first window", async () => { + await mount(); + expect(requests[0]!.input.action).toBe("list_recent"); + await respond("list_recent", { + status: "listed_recent", + threads: [thread("recent")], + next_after: null, + }); + await open("recent"); + expect(requests.some((request) => request.input.action === "history_recent")).toBe(true); + expect(requests.some((request) => request.input.fromSeq === 1)).toBe(false); + }); + it("uses a short French timestamp and separates message metadata from its exact body", async () => { + await mount(); + await respond("list_recent", { + status: "listed_recent", + threads: [thread("one")], + next_after: null, + }); + await open("one"); + await page("one", 1, 2); + expect(renderer!.root.findAllByType("time").length).toBeGreaterThan(0); + expect(renderer!.root.findAllByType("details").length).toBeGreaterThan(0); + expect(text()).not.toContain("AM"); + expect(text()).not.toContain("PM"); + }); + it("replaces a duplicate thread observation and reorders all loaded threads by activity and ID", async () => { + await mount(); + await respond("list_recent", { + status: "listed_recent", + threads: [ + { ...thread("c"), last_activity_at: 200 }, + { ...thread("b"), last_activity_at: 100 }, + ], + next_after: "100:00000000-0000-0000-0000-000000000001", + }); + await act(async () => button("Autres fils").props.onClick()); + await respond("list_recent", { + status: "listed_recent", + threads: [ + { ...thread("a"), last_activity_at: 100 }, + { ...thread("b", "b nouveau"), last_activity_at: 50 }, + { ...thread("d"), last_activity_at: 50 }, + ], + next_after: null, + }); + expect( + renderer!.root + .findAllByType("button") + .filter((node) => node.props["aria-label"]?.startsWith("Ouvrir ")) + .map((node) => node.props["aria-label"]), + ).toEqual(["Ouvrir c", "Ouvrir a", "Ouvrir b nouveau", "Ouvrir d"]); + expect(requests.at(-1)!.input.after).toBe("100:00000000-0000-0000-0000-000000000001"); + }); + it("distinguishes a new empty thread's creation date from its last exchange", async () => { + await mount(); + await respond("list_recent", { + status: "listed_recent", + threads: [{ ...thread("empty"), last_seq: 0 }], + next_after: null, + }); + expect(text()).toContain("Créé · "); + const date = renderer!.root.findByType("time"); + expect(date.props.dateTime).toBe(new Date(1791378000 * 1000).toISOString()); + expect(date.props["aria-label"]).toMatch(/2026/); + expect(date.children.join("")).toMatch(/oct\./); + }); + it("offers expansion only for measured overflow and observes resize without accumulating observers", async () => { + measurements.set(body, { clientHeight: 80, scrollHeight: 40 }); + await mount(); + await respond("list_recent", { + status: "listed_recent", + threads: [thread("one")], + next_after: null, + }); + await open("one"); + await page("one", 1, 1, 1); + expect(button("Déplier le message 1")).toBeUndefined(); + expect(observers.size).toBe(1); + measurements.get(body)!.scrollHeight = 160; + await act(async () => { + for (const observer of observers) observer.callback(); + }); + expect(button("Déplier le message 1").props["aria-expanded"]).toBe(false); + await act(async () => button("Copier le message 1").props.onClick()); + expect(copy).toHaveBeenLastCalledWith(body, "message Bridget"); + await act(async () => button("Déplier le message 1").props.onClick()); + expect(button("Replier le message 1").props["aria-expanded"]).toBe(true); + expect(renderer!.root.findByProps({ "data-bridget-body": true }).children).toEqual([body]); + expect(observers.size).toBe(0); + await act(async () => button("Replier le message 1").props.onClick()); + expect(observers.size).toBe(1); + measurements.get(body)!.scrollHeight = 40; + await act(async () => { + for (const observer of observers) observer.callback(); + }); + expect(button("Déplier le message 1")).toBeUndefined(); + await act(async () => renderer!.update()); + expect(observers.size).toBe(0); + }); + it("searches the hidden tail of a collapsed plain message and copies the exact CRLF and Unicode body", async () => { + measurements.set(body, { clientHeight: 80, scrollHeight: 200 }); + await mount(); + await respond("list_recent", { + status: "listed_recent", + threads: [thread("one")], + next_after: null, + }); + await open("one"); + await page("one", 1, 1, 1); + const count = requests.length; + const search = renderer!.root.findByProps({ + "aria-label": "Rechercher dans les données chargées", + }); + await act(async () => search.props.onChange({ currentTarget: { value: "fin" } })); + expect(renderer!.root.findAllByType("article")).toHaveLength(1); + expect(button("Déplier le message 1").props["aria-expanded"]).toBe(false); + await act(async () => button("Copier le message 1").props.onClick()); + expect(copy).toHaveBeenLastCalledWith(body, "message Bridget"); + expect(requests).toHaveLength(count); + }); + it("keeps keyboard focus on a message control when resizing removes its expansion button", async () => { + measurements.set(body, { clientHeight: 80, scrollHeight: 200 }); + await mount(); + await respond("list_recent", { + status: "listed_recent", + threads: [thread("one")], + next_after: null, + }); + await open("one"); + await page("one", 1, 1, 1); + activeElement = controls.get("Déplier le message 1")!; + measurements.get(body)!.scrollHeight = 40; + await act(async () => { + for (const observer of observers) observer.callback(); + }); + expect(button("Déplier le message 1")).toBeUndefined(); + expect(controls.get("Copier le message 1")!.focus).toHaveBeenCalledWith({ + preventScroll: true, + }); + expect(activeElement).toBe(controls.get("Copier le message 1")); + }); + it("keeps correction coordinates and French kinds in the message details", async () => { + await mount(); + await respond("list_recent", { + status: "listed_recent", + threads: [thread("one")], + next_after: null, + }); + await open("one"); + await respond( + "history_recent", + { + status: "history_recent", + thread_id: "one", + through_seq: 9, + snapshot_seq: 9, + has_more: true, + next_before_seq: 8, + entries: [{ ...entry(9, "blocker"), supersedes_seq: 3, superseded_by_seq: null }], + }, + "one", + ); + const message = renderer!.root.findByType("article"); + const detail = message.findByType("details"); + expect( + JSON.stringify( + detail.children.map((node) => (typeof node === "string" ? node : node.children)), + ), + ).toContain("Blocage"); + expect(text()).toContain("Remplace #3"); + expect(renderer!.root.findByProps({ "data-bridget-body": true }).children).toEqual(["Texte 9"]); + }); + it("purges all loaded messages when an older page returns a different snapshot", async () => { + await mount(); + await respond("list_recent", { + status: "listed_recent", + threads: [thread("one")], + next_after: null, + }); + await open("one"); + await page("one", 5, 6); + await act(async () => button("Messages plus anciens").props.onClick()); + await page("one", 3, 4, 7); + expect(text()).toContain("incompatible"); + expect(renderer!.root.findAllByType("article")).toHaveLength(0); + expect(observers.size).toBe(0); + }); + it("loads only the selected T3 context and exposes member names with ID fallback", async () => { + await mount(); + expect( + requests.every( + (request) => + request.environmentId === "env-A" && + request.input.threadId === "t3-A" && + request.input.projectId === "p-A", + ), + ).toBe(true); + await respond("list_recent", { + status: "listed_recent", + threads: [thread("shared-A", "Planning")], + next_after: null, + }); + expect(text()).toContain("Planning"); + expect(text()).toContain("Alice"); + expect(text()).toContain("agent-B"); + expect( + requests.every((request) => + ["list_recent", "show", "history_recent"].includes(request.input.action), + ), + ).toBe(true); + }); + it("shows an empty list and permits a manual refresh after a non-retryable refusal", async () => { + await mount(); + await respond("list_recent", { status: "listed_recent", threads: [], next_after: null }); + expect(text()).toContain("Aucun fil Bridget"); + await act(async () => button("Rafraîchir").props.onClick()); + await respond("list_recent", { + status: "error", + code: "binding_unavailable", + detail: "Liaison Bridget indisponible.", + retryable: false, + }); + expect(text()).toContain("Liaison Bridget indisponible"); + expect(button("Rafraîchir").props.disabled).not.toBe(true); + }); + it("purges on close, rejects late A while in B and does not resurrect A after returning", async () => { + await mount(); + const old = requests[0]!; + await respond("list_recent", { + status: "listed_recent", + threads: [thread("shared-A", "Secret A")], + next_after: null, + }); + await act(async () => renderer!.update()); + expect(text()).not.toContain("Secret A"); + const other = { + ...context, + environmentId: EnvironmentId.make("env-B"), + threadId: ThreadId.make("t3-B"), + }; + await act(async () => renderer!.update()); + await act(async () => { + resources.set(keyOf(old), { + ...resource(old), + data: { + version: 1, + subject: null, + result: { + status: "listed_recent", + threads: [thread("shared-A", "Late A")], + next_after: null, + }, + }, + dataUpdatedAt: ++timestamp, + isPending: false, + }); + for (const listener of listeners) listener(); + }); + expect(text()).not.toContain("Late A"); + await act(async () => renderer!.update()); + expect(text()).not.toContain("Late A"); + expect(text()).not.toContain("Secret A"); + }); + it("refreshes a pending A request when returning through B before it has completed", async () => { + await mount(); + const original = requests.length; + await act(async () => + renderer!.update(), + ); + await act(async () => renderer!.update()); + expect(requests.filter((request) => request.environmentId === "env-A")).toHaveLength( + original + 1, + ); + expect(renderer!.root.findAllByProps({ "data-bridget-body": true })).toHaveLength(0); + }); + it("reads the latest 137-message snapshot and appends older pages without reversing the page", async () => { + await mount(); + await respond("list_recent", { + status: "listed_recent", + threads: [thread("one"), thread("two")], + next_after: null, + }); + await open("one"); + await page("one", 88, 137, 137); + expect(renderer!.root.findAllByType("article")[0]!.props["aria-label"]).toBe("Message 137"); + await act(async () => button("Messages plus anciens").props.onClick()); + await page("one", 38, 87, 137); + await act(async () => button("Messages plus anciens").props.onClick()); + await page("one", 1, 37, 137); + expect( + requests + .filter((request) => request.input.action === "history_recent") + .map((request) => request.input), + ).toEqual( + expect.arrayContaining([ + expect.objectContaining({ beforeSeq: 87, toSeq: 137 }), + expect.objectContaining({ beforeSeq: 37, toSeq: 137 }), + ]), + ); + expect(renderer!.root.findAllByProps({ "data-bridget-body": true })).toHaveLength(137); + expect(renderer!.root.findAllByType("article").map((node) => node.props["aria-label"])).toEqual( + Array.from({ length: 137 }, (_, index) => `Message ${137 - index}`), + ); + expect(button("Messages plus anciens")).toBeUndefined(); + await act(async () => button("Copier le message 1").props.onClick()); + expect(copy).toHaveBeenCalledWith(body, "message Bridget"); + for (const label of ["Historique", "Action", "Blocage", "Décision", "Message", "agent-A"]) + expect(text()).toContain(label); + await open("two", true); + await page("two", 1, 1, 1); + expect(text()).toContain("Fermé"); + expect(renderer!.root.findAllByProps({ "data-bridget-body": true })).toHaveLength(1); + expect(text()).not.toContain("Texte 6"); + }); + it("clears loaded bodies and members when a later history page refuses access", async () => { + await mount(); + await respond("list_recent", { + status: "listed_recent", + threads: [thread("one")], + next_after: null, + }); + await open("one"); + await page("one", 5, 6); + await act(async () => button("Messages plus anciens").props.onClick()); + await respond( + "history_recent", + { + status: "error", + code: "thread_unavailable", + detail: "Accès au fil refusé.", + retryable: false, + }, + "one", + ); + expect(text()).toContain("Accès au fil refusé"); + expect(renderer!.root.findAllByProps({ "data-bridget-body": true })).toHaveLength(0); + expect(text()).not.toContain("Alice"); + }); + it("searches only loaded titles, members, authors and bodies without requests on typing", async () => { + await mount(); + await respond("list_recent", { + status: "listed_recent", + threads: [thread("one"), thread("two")], + next_after: null, + }); + await open("one"); + await page("one", 1, 2); + const count = requests.length; + const search = renderer!.root.findByProps({ + "aria-label": "Rechercher dans les données chargées", + }); + await act(async () => search.props.onChange({ currentTarget: { value: "café" } })); + expect(text()).toContain("données chargées"); + expect(renderer!.root.findAllByProps({ "aria-label": "Ouvrir two" })).toHaveLength(0); + expect(renderer!.root.findAllByProps({ "data-bridget-body": true })).toHaveLength(1); + await act(async () => search.props.onChange({ currentTarget: { value: "inexistant" } })); + expect(text()).toContain("Aucun résultat"); + await act(async () => button("Effacer la recherche").props.onClick()); + expect(renderer!.root.findAllByProps({ "aria-label": "Ouvrir two" })).toHaveLength(1); + for (const value of ["one", "Alice", "agent-A"]) { + await act(async () => search.props.onChange({ currentTarget: { value } })); + expect(text()).toContain("one"); + } + expect(requests).toHaveLength(count); + }); + it("refreshes the list and current thread manually with a new snapshot and no background reads", async () => { + await mount(); + await respond("list_recent", { + status: "listed_recent", + threads: [thread("one")], + next_after: null, + }); + await open("one"); + await page("one", 1, 2); + const before = requests.length; + await act(async () => button("Rafraîchir").props.onClick()); + expect(renderer!.root.findAllByProps({ "data-bridget-body": true })).toHaveLength(0); + await respond("list_recent", { + status: "listed_recent", + threads: [thread("one")], + next_after: null, + }); + await respond( + "show", + { status: "shown", ...thread("one"), created_at: 1791378000, closed_at: null }, + "one", + ); + await page("one", 1, 1, 1); + expect(requests.length).toBeGreaterThan(before); + expect(renderer!.root.findAllByProps({ "data-bridget-body": true })).toHaveLength(1); + expect( + renderer!.root + .findAllByType("button") + .some((node) => node.children.includes("Messages plus anciens")), + ).toBe(false); + const settled = requests.length; + await act(async () => {}); + expect(requests).toHaveLength(settled); + }); + it("never appends a cached later page while a refreshed snapshot is being read", async () => { + await mount(); + await respond("list_recent", { + status: "listed_recent", + threads: [thread("one")], + next_after: null, + }); + await open("one"); + await page("one", 5, 6); + await act(async () => button("Messages plus anciens").props.onClick()); + await page("one", 3, 4); + await act(async () => button("Rafraîchir").props.onClick()); + await respond("list_recent", { + status: "listed_recent", + threads: [thread("one")], + next_after: null, + }); + await respond( + "show", + { status: "shown", ...thread("one"), created_at: 1791378000, closed_at: null }, + "one", + ); + await page("one", 5, 6); + await act(async () => button("Messages plus anciens").props.onClick()); + expect(text()).not.toContain("Texte 3"); + expect(renderer!.root.findAllByProps({ "data-bridget-body": true })).toHaveLength(2); + }); + it("does not query without a complete context and distinguishes an empty history from loading", async () => { + await act(async () => { + renderer = create(); + }); + expect(requests).toHaveLength(0); + expect(text()).toContain("Sélectionnez une conversation"); + await act(async () => renderer!.update()); + expect(text()).toContain("Chargement des fils"); + await respond("list_recent", { + status: "listed_recent", + threads: [thread("empty")], + next_after: null, + }); + await open("empty"); + expect(text()).toContain("Chargement de l’historique"); + await respond( + "history_recent", + { + status: "history_recent", + thread_id: "empty", + through_seq: 0, + snapshot_seq: 0, + has_more: false, + next_before_seq: null, + entries: [], + }, + "empty", + ); + expect(text()).toContain("Ce fil ne contient aucun message"); + }); + it.each([ + "unsupported_version", + "binding_unavailable", + "project_mismatch", + "thread_unavailable", + "response_too_large", + ])("shows %s without automatic retry and still offers manual refresh", async (code) => { + await mount(); + await respond("list_recent", { + status: "error", + code, + detail: `Refus ${code}.`, + retryable: false, + }); + expect(text()).toContain(`Refus ${code}`); + const settled = requests.length; + await act(async () => {}); + expect(requests).toHaveLength(settled); + expect(button("Rafraîchir").props.disabled).not.toBe(true); + }); + it("ignores an old thread history response after another thread is selected", async () => { + await mount(); + await respond("list_recent", { + status: "listed_recent", + threads: [thread("one"), thread("two")], + next_after: null, + }); + await open("one"); + await open("two"); + await page("one", 1, 6); + expect(renderer!.root.findAllByProps({ "data-bridget-body": true })).toHaveLength(0); + await page("two", 2, 2, 2); + expect(text()).toContain("Texte 2"); + expect(text()).not.toContain("Texte 6"); + }); + it("rejects a detail response for another shared thread instead of displaying its members", async () => { + await mount(); + await respond("list_recent", { + status: "listed_recent", + threads: [thread("one")], + next_after: null, + }); + await act(async () => button("Ouvrir one").props.onClick()); + await respond( + "show", + { + status: "shown", + ...thread("forged", "Foreign title"), + created_at: 1791378000, + closed_at: null, + }, + "one", + ); + expect(text()).not.toContain("Foreign title"); + expect(text()).toContain("incompatible"); + }); + it("clears all loaded data on a transport timeout and reports a failed copy", async () => { + await mount(); + await respond("list_recent", { + status: "listed_recent", + threads: [thread("one")], + next_after: null, + }); + await open("one"); + await page("one", 5, 6); + copy.mockRejectedValueOnce(new Error("clipboard unavailable")); + await act(async () => button("Copier le message 5").props.onClick()); + expect(text()).toContain("Copie indisponible"); + await act(async () => button("Messages plus anciens").props.onClick()); + const target = requests.findLast((request) => request.input.action === "history_recent")!; + await act(async () => { + resources.set(keyOf(target), { + ...resource(target), + error: "Bridget did not respond within six seconds.", + isPending: false, + isSuccess: false, + }); + for (const listener of listeners) listener(); + }); + expect(text()).toContain("six seconds"); + expect(text()).not.toContain("Alice"); + expect(renderer!.root.findAllByProps({ "data-bridget-body": true })).toHaveLength(0); + }); +}); diff --git a/apps/web/src/components/BridgetPanel.tsx b/apps/web/src/components/BridgetPanel.tsx new file mode 100644 index 000000000000..6cedd447120b --- /dev/null +++ b/apps/web/src/components/BridgetPanel.tsx @@ -0,0 +1,549 @@ +import type { BridgetHumanView, EnvironmentId, ProjectId, ThreadId } from "@t3tools/contracts"; +import { ChevronDown, ChevronUp, Copy, RefreshCw, X } from "lucide-react"; +import { useCallback, useEffect, useId, useRef, useState } from "react"; + +import { orchestrationEnvironment } from "~/state/orchestration"; +import { useEnvironmentQuery } from "~/state/query"; +import { Button } from "~/components/ui/button"; +import { ScrollArea } from "~/components/ui/scroll-area"; +import { writeTextToClipboard } from "~/hooks/useCopyToClipboard"; +import { cn } from "~/lib/utils"; +import { Input } from "~/components/ui/input"; +import { Tooltip, TooltipPopup, TooltipTrigger } from "~/components/ui/tooltip"; + +type Listed = Extract; +type History = Extract; +type Context = { environmentId: EnvironmentId; threadId: ThreadId; projectId: ProjectId }; + +// T3's timestamp tooltips contain English ordinals; this French panel keeps its labels local. +const shortDate = new Intl.DateTimeFormat("fr-FR", { + day: "2-digit", + month: "short", + hour: "2-digit", + minute: "2-digit", +}); +const exactDate = new Intl.DateTimeFormat("fr-FR", { + dateStyle: "full", + timeStyle: "long", +}); + +function BridgetDate({ seconds }: { seconds: number }) { + const date = new Date(seconds * 1000); + if (Number.isNaN(date.getTime())) return Date indisponible; + return ( + + } + > + {shortDate.format(date)} + + {exactDate.format(date)} + + ); +} + +export function BridgetPanel(props: { + environmentId: EnvironmentId | null; + threadId: ThreadId | null; + projectId: ProjectId | null; + visible: boolean; +}) { + if (!props.visible) return null; + if (!props.environmentId || !props.threadId || !props.projectId) { + return ( +

+ Sélectionnez une conversation T3 enregistrée dans un projet. +

+ ); + } + return ( + + ); +} + +function BridgetSession(context: Context) { + const [after, setAfter] = useState(); + const [revision, setRevision] = useState(0); + const [threads, setThreads] = useState([]); + const [selectedId, setSelectedId] = useState(null); + const [detailError, setDetailError] = useState(null); + const [search, setSearch] = useState(""); + const [loadedEntries, setLoadedEntries] = useState>({}); + const query = useEnvironmentQuery( + orchestrationEnvironment.bridgetRead({ + environmentId: context.environmentId, + input: { + threadId: context.threadId, + projectId: context.projectId, + action: "list_recent", + limit: 20, + ...(after ? { after } : {}), + }, + }), + ); + const [freshness, setFreshness] = useState({ after, revision, data: query.data }); + const sameRequest = freshness.after === after && freshness.revision === revision; + if (!sameRequest) setFreshness({ after, revision, data: query.data }); + const discarded = sameRequest ? freshness.data : query.data; + const refreshList = query.refresh; + useEffect(() => { + // Refresh on opening as well: the atom may retain an older pending request. + refreshList(); + }, [refreshList, revision]); + const result = !query.isPending && query.data !== discarded ? query.data?.result : null; + const error = + detailError ?? + (!query.isPending ? query.error : null) ?? + (result?.status === "error" + ? result.detail + : result && result.status !== "listed_recent" + ? "Réponse Bridget incompatible." + : null); + const loading = query.isPending || (!error && result === null); + const rememberHistory = useCallback((id: string, entries: History["entries"]) => { + setLoadedEntries((current) => ({ ...current, [id]: entries })); + }, []); + useEffect(() => { + if (error) { + setThreads([]); + setSelectedId(null); + setLoadedEntries({}); + return; + } + if (result?.status !== "listed_recent") return; + setThreads((current) => { + const byId = new Map(current.map((thread) => [thread.thread_id, thread])); + for (const thread of result.threads) byId.set(thread.thread_id, thread); + return [...byId.values()].sort( + (left, right) => + right.last_activity_at - left.last_activity_at || + (left.thread_id < right.thread_id ? -1 : left.thread_id > right.thread_id ? 1 : 0), + ); + }); + }, [error, result]); + const needle = search.trim().toLocaleLowerCase(); + // O(n + loaded text): search never creates a remote query. + const visibleThreads = needle + ? threads.filter((thread) => + [ + thread.title, + ...thread.members.flatMap((member) => [member.name ?? "", member.agent_id]), + ...(loadedEntries[thread.thread_id] ?? []).flatMap((entry) => [ + entry.body, + entry.author_name ?? "", + entry.author_id, + ]), + ].some((value) => value.toLocaleLowerCase().includes(needle)), + ) + : threads; + return ( +
+
+
+ Fils partagés +

Plus récents d’abord

+
+ +
+
+
+ setSearch(event.currentTarget.value)} + /> + {search ? ( + + ) : null} +
+

+ Recherche locale dans les données chargées : titres, membres, auteurs et messages. +

+
+ +
+ {error ? ( +

+ {error} +

+ ) : ( + <> + {loading ? ( +

+ Chargement des fils… +

+ ) : null} + {!loading && threads.length === 0 ? ( +

+ Aucun fil Bridget pour cet agent. +

+ ) : null} + {!loading && needle && visibleThreads.length === 0 ? ( +

+ Aucun résultat dans les données chargées. +

+ ) : null} + {visibleThreads.map((thread) => ( + + ))} + {result?.status === "listed_recent" && result.next_after ? ( + + ) : null} + + )} +
+ + {!error && selectedId ? ( + + ) : null} +
+ ); +} + +function BridgetThread({ + context, + sharedThreadId, + search, + onLoaded, + onError, +}: { + context: Context; + sharedThreadId: string; + search: string; + onLoaded: (id: string, entries: History["entries"]) => void; + onError: (detail: string) => void; +}) { + const [range, setRange] = useState<{ beforeSeq?: number; toSeq?: number }>({}); + const [history, setHistory] = useState(null); + const [copyStatus, setCopyStatus] = useState(null); + const shown = useEnvironmentQuery( + orchestrationEnvironment.bridgetRead({ + environmentId: context.environmentId, + input: { + threadId: context.threadId, + projectId: context.projectId, + action: "show", + sharedThreadId, + }, + }), + ); + const query = useEnvironmentQuery( + orchestrationEnvironment.bridgetRead({ + environmentId: context.environmentId, + input: { + threadId: context.threadId, + projectId: context.projectId, + action: "history_recent", + sharedThreadId, + limit: 50, + ...range, + }, + }), + ); + const [discardedShown] = useState(shown.data); + const [freshness, setFreshness] = useState({ range, data: query.data }); + if (freshness.range !== range) setFreshness({ range, data: query.data }); + const discardedHistory = freshness.range === range ? freshness.data : query.data; + const refreshDetail = shown.refresh; + const refreshHistory = query.refresh; + useEffect(() => { + refreshDetail(); + }, [refreshDetail]); + useEffect(() => { + refreshHistory(); + }, [refreshHistory]); + const shownResult = shown.data !== discardedShown && !shown.isPending ? shown.data?.result : null; + const page = !query.isPending && query.data !== discardedHistory ? query.data?.result : null; + const incompatible = + (shownResult && + shownResult.status !== "error" && + (shownResult.status !== "shown" || shownResult.thread_id !== sharedThreadId)) || + (page && + page.status !== "error" && + (page.status !== "history_recent" || + page.thread_id !== sharedThreadId || + (range.toSeq !== undefined && page.snapshot_seq !== range.toSeq))); + const error = + (!shown.isPending ? shown.error : null) ?? + (!query.isPending ? query.error : null) ?? + (shownResult?.status === "error" + ? shownResult.detail + : page?.status === "error" + ? page.detail + : incompatible + ? "Réponse Bridget incompatible avec ce fil." + : null); + useEffect(() => { + if (error) { + onError(error); + } + }, [error, onError]); + useEffect(() => { + if (error || page?.status !== "history_recent") return; + setHistory((current) => { + const sequences = new Set(current?.entries.map((entry) => entry.seq)); + return { + ...page, + entries: [ + ...(current?.entries ?? []), + ...page.entries.filter((entry) => !sequences.has(entry.seq)), + ], + }; + }); + }, [error, page]); + useEffect(() => { + if (history) onLoaded(sharedThreadId, history.entries); + }, [history, onLoaded, sharedThreadId]); + if (error) return null; + const detail = shownResult?.status === "shown" ? shownResult : null; + const entries = history?.entries ?? []; + const visibleEntries = search + ? entries.filter((entry) => + [entry.body, entry.author_name ?? "", entry.author_id].some((value) => + value.toLocaleLowerCase().includes(search), + ), + ) + : entries; + return ( +
+
+ {detail ? ( + <> +

{detail.title}

+

+ {detail.members.map((member) => member.name ?? member.agent_id).join(" · ")} +

+
+ Détails du fil +

+ {detail.state === "closed" ? "Fermé" : "Ouvert"} · Créé le{" "} + + {detail.closed_at !== null ? ( + <> + {" "} + · Fermé le + + ) : null} +

+
+ + ) : ( +

+ Chargement du fil… +

+ )} +
+ +
+ {visibleEntries.map((entry) => ( + + ))} + {query.isPending ? ( +

+ Chargement de l’historique… +

+ ) : null} + {history && history.entries.length === 0 ? ( +

+ Ce fil ne contient aucun message. +

+ ) : null} + {search && entries.length > 0 && visibleEntries.length === 0 ? ( +

+ Aucun résultat dans les messages chargés. +

+ ) : null} + {history?.has_more && history.next_before_seq !== null ? ( + + ) : null} + {copyStatus ? ( +

+ {copyStatus} +

+ ) : null} +
+ +
+ ); +} + +function BridgetMessage({ + entry, + onCopyStatus, +}: { + entry: History["entries"][number]; + onCopyStatus: (status: string) => void; +}) { + const [expanded, setExpanded] = useState(false); + const [overflows, setOverflows] = useState(false); + const bodyRef = useRef(null); + const toggleRef = useRef(null); + const copyRef = useRef(null); + const bodyId = useId(); + useEffect(() => { + const body = bodyRef.current; + if (!body || expanded || entry.body.length === 0) return; + const measure = () => { + const overflow = body.scrollHeight > body.clientHeight; + if ( + !overflow && + toggleRef.current !== null && + body.ownerDocument?.activeElement === toggleRef.current + ) { + copyRef.current?.focus({ preventScroll: true }); + } + setOverflows(overflow); + }; + measure(); + if (typeof ResizeObserver === "undefined") return; + const observer = new ResizeObserver(measure); + observer.observe(body); + return () => observer.disconnect(); + }, [expanded, entry.body]); + const kind = entry.kind + ? { history: "Historique", action: "Action", blocker: "Blocage", decision: "Décision" }[ + entry.kind + ] + : "Message"; + return ( +
+
+
+ + {entry.author_name ?? entry.author_id} + + + + +
+ +
+
+ {entry.body} +
+ {overflows || expanded ? ( +
+ +
+ ) : null} +
+ Détails +

+ {kind} · Message #{entry.seq} + {entry.supersedes_seq != null ? ` · Remplace #${entry.supersedes_seq}` : ""} + {entry.superseded_by_seq != null ? ` · Remplacé par #${entry.superseded_by_seq}` : ""} +

+
+
+ ); +} diff --git a/apps/web/src/components/ChatView.tsx b/apps/web/src/components/ChatView.tsx index a32c1473cda7..273297c78b9d 100644 --- a/apps/web/src/components/ChatView.tsx +++ b/apps/web/src/components/ChatView.tsx @@ -218,6 +218,7 @@ import { PullRequestDetailGhost } from "./pullRequest/PullRequestGhosts"; import { PullRequestsUnavailableState } from "./pullRequest/PullRequestsUnavailableState"; import { RightPanelTabs } from "./RightPanelTabs"; import { AgentsPanel } from "./AgentsPanel"; +import { BridgetPanel } from "./BridgetPanel"; import { LinkPullRequestDialogHost } from "./pullRequest/LinkPullRequestDialog"; import { ThreadPullRequestsPanel } from "./pullRequest/ThreadPullRequestsPanel"; import { useDeviceState } from "~/state/device"; @@ -4566,6 +4567,11 @@ export default function ChatView(props: ChatViewProps) { if (!activeThreadRef) return; useRightPanelStore.getState().open(activeThreadRef, "agents"); }, [activeThreadRef]); + const bridgetAvailable = isServerThread && activeProject !== null; + const addBridgetSurface = useCallback(() => { + if (!activeThreadRef || !bridgetAvailable) return; + useRightPanelStore.getState().open(activeThreadRef, "bridget"); + }, [activeThreadRef, bridgetAvailable]); const supportsThreadPullRequests = serverConfig?.environment.capabilities.threadPullRequests === true; const visiblePullRequests = visibleThreadPullRequests( @@ -9630,6 +9636,13 @@ export default function ChatView(props: ChatViewProps) { /> ) : renderedRightPanelSurface?.kind === "pull-requests" && activeThreadRef ? ( + ) : renderedRightPanelSurface?.kind === "bridget" ? ( + ) : renderedRightPanelSurface?.kind === "agents" ? ( @@ -10349,6 +10364,7 @@ export default function ChatView(props: ChatViewProps) { onAddPullRequest={addPullRequestSurface} onAddPullRequests={addPullRequestsSurface} onAddAgents={addAgentsSurface} + onAddBridget={addBridgetSurface} onAddDevice={addDeviceSurface} browserAvailable={isPreviewSupportedInRuntime()} terminalAvailable={activeProject !== null} @@ -10357,6 +10373,7 @@ export default function ChatView(props: ChatViewProps) { pullRequestAvailable={pullRequestSurfaceAvailable} pullRequestsAvailable={pullRequestsSurfaceAvailable} agentsAvailable + bridgetAvailable={bridgetAvailable} deviceAvailable={activeThreadRef !== null} liveAgentCount={agentPanelModel.liveCount} > diff --git a/apps/web/src/components/RightPanelTabs.test.tsx b/apps/web/src/components/RightPanelTabs.test.tsx index dfaaf171c429..8c88456ce658 100644 --- a/apps/web/src/components/RightPanelTabs.test.tsx +++ b/apps/web/src/components/RightPanelTabs.test.tsx @@ -1,7 +1,12 @@ -import { EnvironmentId, type ThreadPullRequestLink } from "@t3tools/contracts"; +// @vitest-environment jsdom +import { EnvironmentId, ThreadId, type ThreadPullRequestLink } from "@t3tools/contracts"; import type { DesktopPreviewFavicon, PreviewSessionSnapshot } from "@t3tools/contracts"; +import { scopeThreadRef } from "@t3tools/client-runtime/environment"; +import { act, type ComponentProps } from "react"; +import { createRoot } from "react-dom/client"; import { renderToStaticMarkup } from "react-dom/server"; -import { describe, expect, it } from "vite-plus/test"; +import { describe, expect, it, vi } from "vite-plus/test"; +import { selectThreadRightPanelState, useRightPanelStore } from "~/rightPanelStore"; import { RightPanelTabs, @@ -89,13 +94,14 @@ function overlay( }; } -function renderTabs( +function tabsElement( first: DesktopPreviewFavicon | null, second?: DesktopPreviewFavicon, audio?: { audible?: boolean; audioMuted?: boolean }, previewRuntimeTabId: ((tabId: string) => string) | null = (tabId) => `runtime:${tabId}`, + overrides?: Partial>, ) { - return renderToStaticMarkup( + return ( undefined} onAddFiles={() => undefined} onAddAgents={() => undefined} + onAddBridget={() => undefined} onAddDevice={() => undefined} liveAgentCount={0} browserAvailable @@ -132,13 +139,108 @@ function renderTabs( pullRequestAvailable={false} pullRequestsAvailable={false} agentsAvailable={false} + bridgetAvailable={false} deviceAvailable={false} + {...overrides} >
content
- , + ); } +function renderTabs(...args: Parameters) { + return renderToStaticMarkup(tabsElement(...args)); +} + +describe("Bridget native surface navigation", () => { + it("opens by keyboard, reuses the singleton from the plus menu and closes its native tab", async () => { + const host = document.createElement("div"); + document.body.append(host); + vi.stubGlobal( + "ResizeObserver", + class { + observe() {} + unobserve() {} + disconnect() {} + }, + ); + vi.stubGlobal("IS_REACT_ACT_ENVIRONMENT", true); + vi.stubGlobal("matchMedia", () => ({ + matches: false, + addEventListener() {}, + removeEventListener() {}, + })); + const originalGetAnimations = Element.prototype.getAnimations; + Element.prototype.getAnimations = () => []; + const ref = scopeThreadRef(EnvironmentId.make("native-bridget"), ThreadId.make("native-chat")); + useRightPanelStore.setState({ byThreadKey: {}, userActionRevisionByThreadKey: {} }); + const root = createRoot(host); + const draw = () => { + const state = selectThreadRightPanelState(useRightPanelStore.getState().byThreadKey, ref); + root.render( + tabsElement(null, undefined, undefined, null, { + mode: "sheet", + surfaces: state.surfaces, + activeSurfaceId: state.activeSurfaceId, + browserAvailable: false, + bridgetAvailable: true, + onAddBridget: () => { + useRightPanelStore.getState().open(ref, "bridget"); + draw(); + }, + onCloseSurface: (surface) => { + useRightPanelStore.getState().closeSurface(ref, surface.id); + draw(); + }, + }), + ); + }; + try { + await act(async () => draw()); + const launcher = document.querySelector("[aria-label='Open a surface']")!; + launcher.focus(); + expect(document.activeElement).toBe(launcher); + await act(async () => + launcher.dispatchEvent(new KeyboardEvent("keydown", { key: "ArrowDown", bubbles: true })), + ); + await act(async () => + launcher.dispatchEvent(new KeyboardEvent("keydown", { key: "Enter", bubbles: true })), + ); + expect( + selectThreadRightPanelState(useRightPanelStore.getState().byThreadKey, ref).surfaces, + ).toEqual([{ id: "bridget", kind: "bridget" }]); + const tab = document.querySelector("[data-active-tab='true']")!; + expect(tab.textContent).toBe("Bridget"); + const icon = tab.querySelector("span.bg-current")!; + expect(icon.style.maskImage).not.toBe(""); + expect(icon.classList.contains("bg-current")).toBe(true); + await act(async () => + document + .querySelector("button[aria-label='Add panel surface']")! + .click(), + ); + const item = [...document.querySelectorAll("[role='menuitem']")].find((node) => + node.textContent?.includes("Bridget"), + )!; + await act(async () => item.click()); + expect( + selectThreadRightPanelState(useRightPanelStore.getState().byThreadKey, ref).surfaces, + ).toHaveLength(1); + await act(async () => + document.querySelector("button[aria-label='Close Bridget']")!.click(), + ); + expect( + selectThreadRightPanelState(useRightPanelStore.getState().byThreadKey, ref).surfaces, + ).toEqual([]); + } finally { + await act(async () => root.unmount()); + host.remove(); + Element.prototype.getAnimations = originalGetAnimations; + vi.unstubAllGlobals(); + } + }); +}); + describe("RightPanelTabs preview favicon", () => { it("prefers a live capture and never asks Google about a private hostname", () => { const captured = renderTabs(favicon("data:image/png;base64,AAAA", "http://24x.xf.local/")); @@ -163,6 +265,16 @@ describe("RightPanelTabs preview favicon", () => { }); describe("surface shortcuts", () => { + it("allows the Bridget launcher letter without taking modified shortcuts", () => { + const bridget = { shortcut: "R", available: true, label: "Bridget" }; + expect(surfaceShortcutActionForKey([bridget], shortcutEvent("r"))).toBe(bridget); + expect( + surfaceShortcutActionForKey([bridget], shortcutEvent("r", { ctrlKey: true })), + ).toBeNull(); + expect( + surfaceShortcutActionForKey([{ ...bridget, available: false }], shortcutEvent("r")), + ).toBeNull(); + }); const actions = [ { shortcut: "B", available: true, label: "Browser" }, { shortcut: "D", available: false, label: "Diff" }, diff --git a/apps/web/src/components/RightPanelTabs.tsx b/apps/web/src/components/RightPanelTabs.tsx index 001342e48262..4a3ea8e758aa 100644 --- a/apps/web/src/components/RightPanelTabs.tsx +++ b/apps/web/src/components/RightPanelTabs.tsx @@ -78,6 +78,22 @@ import { previewBridge } from "./preview/previewBridge"; import { PierreEntryIcon } from "./chat/PierreEntryIcon"; import { resolvePullRequestState } from "./pullRequest/pullRequestPresentation"; import { PullRequestGlyph } from "~/components/pullRequest/pullRequestIcons"; +import bridgetLogoUrl from "~/assets/bridget-logo.svg"; + +function BridgetIcon({ className }: { className?: string }) { + return ( + + ); +} interface RightPanelTabsProps { mode: PreviewPanelMode; @@ -122,6 +138,7 @@ interface RightPanelTabsProps { onAddPullRequest: () => void; onAddPullRequests: () => void; onAddAgents: () => void; + onAddBridget: () => void; onAddDevice: () => void; browserAvailable: boolean; terminalAvailable: boolean; @@ -130,6 +147,7 @@ interface RightPanelTabsProps { pullRequestAvailable: boolean; pullRequestsAvailable: boolean; agentsAvailable: boolean; + bridgetAvailable: boolean; deviceAvailable: boolean; pullRequestStatusSeeds?: Readonly>; /** Running + waiting subagents; badges the Agents card in the empty state. */ @@ -161,6 +179,7 @@ const SURFACE_DISABLED_REASONS = { pullRequest: "This thread's branch has no pull request yet.", pullRequests: "No linked pull requests are available for this thread.", agents: "Agents are only available from a thread.", + bridget: "Bridget requires a saved project conversation.", device: "Devices are only available from a thread.", } as const; @@ -185,6 +204,7 @@ const SURFACE_UNAVAILABLE_HINTS = { pullRequest: "No pull request on this branch yet.", pullRequests: "No linked pull requests available.", agents: "Available from a thread.", + bridget: "Available from a saved project conversation.", device: "Available from a thread.", } as const; @@ -325,6 +345,7 @@ function RightPanelEmptyState(props: { onAddPullRequest: () => void; onAddPullRequests: () => void; onAddAgents: () => void; + onAddBridget: () => void; onAddDevice: () => void; browserAvailable: boolean; terminalAvailable: boolean; @@ -333,6 +354,7 @@ function RightPanelEmptyState(props: { pullRequestAvailable: boolean; pullRequestsAvailable: boolean; agentsAvailable: boolean; + bridgetAvailable: boolean; deviceAvailable: boolean; liveAgentCount: number; }) { @@ -403,6 +425,15 @@ function RightPanelEmptyState(props: { onClick: props.onAddAgents, badgeCount: props.liveAgentCount, }, + { + label: "Bridget", + icon: BridgetIcon, + shortcut: "R", + available: props.bridgetAvailable, + disabledReason: SURFACE_UNAVAILABLE_HINTS.bridget, + onClick: props.onAddBridget, + badgeCount: 0, + }, { label: "Device", description: "Watch an iOS Simulator or Android Emulator.", @@ -630,6 +661,8 @@ function surfaceTitle( return "Pull requests"; case "agents": return "Agents"; + case "bridget": + return "Bridget"; case "device": return surface.title ?? surface.target?.name ?? "Device"; case "preview": { @@ -715,6 +748,8 @@ function SurfaceIcon({ return ; case "agents": return ; + case "bridget": + return ; case "device": return surface.target?.platform === "ios" ? ( @@ -925,6 +960,14 @@ export function RightPanelTabs(props: RightPanelTabsProps) { disabledReason: SURFACE_DISABLED_REASONS.agents, onClick: props.onAddAgents, }, + { + label: "Bridget", + icon: BridgetIcon, + shortcut: "R", + available: props.bridgetAvailable, + disabledReason: SURFACE_DISABLED_REASONS.bridget, + onClick: props.onAddBridget, + }, { label: "Device", icon: Smartphone, @@ -1415,6 +1458,7 @@ export function RightPanelTabs(props: RightPanelTabsProps) { onAddPullRequest={props.onAddPullRequest} onAddPullRequests={props.onAddPullRequests} onAddAgents={props.onAddAgents} + onAddBridget={props.onAddBridget} onAddDevice={props.onAddDevice} browserAvailable={props.browserAvailable} terminalAvailable={props.terminalAvailable} @@ -1423,6 +1467,7 @@ export function RightPanelTabs(props: RightPanelTabsProps) { pullRequestAvailable={props.pullRequestAvailable} pullRequestsAvailable={props.pullRequestsAvailable} agentsAvailable={props.agentsAvailable} + bridgetAvailable={props.bridgetAvailable} deviceAvailable={props.deviceAvailable} liveAgentCount={props.liveAgentCount} /> diff --git a/apps/web/src/hooks/useCopyToClipboard.test.ts b/apps/web/src/hooks/useCopyToClipboard.test.ts index 49e19c0766f1..cd81aba5ff8e 100644 --- a/apps/web/src/hooks/useCopyToClipboard.test.ts +++ b/apps/web/src/hooks/useCopyToClipboard.test.ts @@ -1,3 +1,6 @@ +// @vitest-environment jsdom + +import * as NodeBuffer from "node:buffer"; import { afterEach, describe, expect, it, vi } from "vite-plus/test"; import { COMPOSER_CONTEXT_CLIPBOARD_MIME } from "@t3tools/shared/composerContextClipboard"; @@ -10,6 +13,7 @@ import { describe("writeTextToClipboard", () => { it("reserves plain text even when an extra flavor attempts to replace it", async () => { + vi.stubGlobal("Blob", NodeBuffer.Blob); const write = vi.fn().mockResolvedValue(undefined); vi.stubGlobal("window", {}); vi.stubGlobal("navigator", { clipboard: { write, writeText: vi.fn() } }); @@ -33,6 +37,7 @@ describe("writeTextToClipboard", () => { }); it("keeps caller-built rich HTML when attaching a context fragment", async () => { + vi.stubGlobal("Blob", NodeBuffer.Blob); const write = vi.fn().mockResolvedValue(undefined); vi.stubGlobal("window", {}); vi.stubGlobal("navigator", { clipboard: { write, writeText: vi.fn() } }); @@ -130,6 +135,55 @@ describe("writeTextToClipboard", () => { }, ); + it.each(["plain", "custom flavors"] as const)( + "preserves CRLF, Unicode and spaces in the HTTP fallback with %s", + async (format) => { + const value = " début\r\n```ts\r\nconst café = '雪 🦋'; \r\n```\r\n\tfin "; + const flavors = format === "plain" ? undefined : { "text/html": "
rich
" }; + const previousExecCommand = Object.getOwnPropertyDescriptor(document, "execCommand"); + const previouslyFocused = document.createElement("button"); + document.body.appendChild(previouslyFocused); + previouslyFocused.focus(); + const data = new Map(); + const setData = vi.fn((type: string, contents: string) => data.set(type, contents)); + const event = new Event("copy", { bubbles: true, cancelable: true }); + Object.defineProperty(event, "clipboardData", { value: { setData } }); + let textarea: HTMLTextAreaElement | undefined; + const execCommand = vi.fn(() => { + const active = document.activeElement; + if (!(active instanceof HTMLTextAreaElement)) throw new Error("No copy selection"); + textarea = active; + // A real textarea normalizes CRLF. Only the copy event can preserve the source. + expect(active.value).toBe(value.replaceAll("\r\n", "\n")); + active.dispatchEvent(event); + return true; + }); + Object.defineProperty(document, "execCommand", { configurable: true, value: execCommand }); + vi.stubGlobal("navigator", {}); + + try { + await expect(writeTextToClipboard(value, "message Bridget", flavors)).resolves.toBe(true); + expect(execCommand).toHaveBeenCalledWith("copy"); + expect(data.get("text/plain")).toBe(value); + expect(event.defaultPrevented).toBe(true); + expect(data.get("text/html")).toBe(flavors?.["text/html"]); + expect(document.activeElement).toBe(previouslyFocused); + expect(textarea?.isConnected).toBe(false); + setData.mockClear(); + textarea?.dispatchEvent(event); + expect(setData).not.toHaveBeenCalled(); + } finally { + textarea?.remove(); + previouslyFocused.remove(); + if (previousExecCommand) { + Object.defineProperty(document, "execCommand", previousExecCommand); + } else { + Reflect.deleteProperty(document, "execCommand"); + } + } + }, + ); + it("uses the Clipboard API without touching the fallback when it is available", async () => { const writeText = vi.fn().mockResolvedValue(undefined); const execCommand = vi.fn(); diff --git a/apps/web/src/hooks/useCopyToClipboard.ts b/apps/web/src/hooks/useCopyToClipboard.ts index 15fd455c75d7..da8a9c54e1d1 100644 --- a/apps/web/src/hooks/useCopyToClipboard.ts +++ b/apps/web/src/hooks/useCopyToClipboard.ts @@ -70,10 +70,13 @@ function writeTextWithExecCommand( const previouslyFocused = document.activeElement; const copy = (event: ClipboardEvent) => { - if (!extraFlavors || !event.clipboardData) return; + if (!event.clipboardData) return; + // The textarea normalizes line endings; write the unmodified source instead. event.clipboardData.setData("text/plain", value); - for (const [type, data] of Object.entries(extraFlavors)) - event.clipboardData.setData(type, data); + if (extraFlavors) { + for (const [type, data] of Object.entries(extraFlavors)) + event.clipboardData.setData(type, data); + } event.preventDefault(); }; document.body.appendChild(textarea); diff --git a/apps/web/src/rightPanelStore.test.ts b/apps/web/src/rightPanelStore.test.ts index c73a2a009c16..bfbb9ec4b4c9 100644 --- a/apps/web/src/rightPanelStore.test.ts +++ b/apps/web/src/rightPanelStore.test.ts @@ -21,6 +21,31 @@ beforeEach(() => { }); describe("rightPanelStore", () => { + it("keeps Bridget singleton scoped by environment and conversation through migration", () => { + const store = useRightPanelStore.getState(); + store.open(refA, "bridget"); + store.open(refA, "bridget"); + const remote = scopeThreadRef("env-2" as EnvironmentId, refA.threadId); + store.open(remote, "bridget"); + expect( + selectThreadRightPanelState(useRightPanelStore.getState().byThreadKey, refA).surfaces, + ).toEqual([{ id: "bridget", kind: "bridget" }]); + const migrated = migratePersistedRightPanelState({ + byThreadKey: useRightPanelStore.getState().byThreadKey, + }); + useRightPanelStore.setState(migrated); + store.closeSurface(refA, "bridget"); + expect( + selectThreadRightPanelState(useRightPanelStore.getState().byThreadKey, refA).surfaces, + ).toEqual([]); + expect( + selectThreadRightPanelState(useRightPanelStore.getState().byThreadKey, remote) + .activeSurfaceId, + ).toBe("bridget"); + expect( + selectThreadRightPanelState(useRightPanelStore.getState().byThreadKey, refB).surfaces, + ).toEqual([]); + }); it("gives each host/device its own tab and preserves renamed tabs", () => { const store = useRightPanelStore.getState(); const android = { diff --git a/apps/web/src/rightPanelStore.ts b/apps/web/src/rightPanelStore.ts index 75f6d2ffe2b0..9f697c0c2c7b 100644 --- a/apps/web/src/rightPanelStore.ts +++ b/apps/web/src/rightPanelStore.ts @@ -29,6 +29,7 @@ const RIGHT_PANEL_KINDS = [ "pull-request", "pull-requests", "agents", + "bridget", ] as const; export type RightPanelKind = (typeof RIGHT_PANEL_KINDS)[number]; @@ -85,7 +86,8 @@ export type RightPanelSurface = } /** The thread's linked pull requests, one singleton tab beside any number of `pull-request` tabs. */ | { id: "pull-requests"; kind: "pull-requests" } - | { id: "agents"; kind: "agents" }; + | { id: "agents"; kind: "agents" } + | { id: "bridget"; kind: "bridget" }; const RIGHT_PANEL_STORAGE_KEY = "t3code:right-panel-state:v2"; // v9 removed the "plan" surface kind (plans render inline in the transcript). @@ -191,6 +193,8 @@ const singletonSurface = ( return { id: "pull-requests", kind }; case "agents": return { id: "agents", kind }; + case "bridget": + return { id: "bridget", kind }; case "device": return { id: "device", kind }; } diff --git a/apps/web/src/routes/_chat.pull-requests.tsx b/apps/web/src/routes/_chat.pull-requests.tsx index 1b7239a0bef5..70cd2d1f623c 100644 --- a/apps/web/src/routes/_chat.pull-requests.tsx +++ b/apps/web/src/routes/_chat.pull-requests.tsx @@ -2114,6 +2114,7 @@ function PullRequestsRouteView() { onAddPullRequest={() => undefined} onAddPullRequests={() => undefined} onAddAgents={() => undefined} + onAddBridget={() => undefined} onAddDevice={() => undefined} browserAvailable={false} terminalAvailable={false} @@ -2122,6 +2123,7 @@ function PullRequestsRouteView() { pullRequestAvailable={false} pullRequestsAvailable={false} agentsAvailable={false} + bridgetAvailable={false} deviceAvailable={false} liveAgentCount={0} pullRequestStatusSeeds={listedPullRequestTabStatuses} diff --git a/packages/client-runtime/src/state/orchestration.test.ts b/packages/client-runtime/src/state/orchestration.test.ts new file mode 100644 index 000000000000..3dfc2b2338c9 --- /dev/null +++ b/packages/client-runtime/src/state/orchestration.test.ts @@ -0,0 +1,249 @@ +import { + BridgetReadError, + EnvironmentId, + ProjectId, + ThreadId, + WS_METHODS, + type BridgetHumanView, + type BridgetReadInput, +} from "@t3tools/contracts"; +import { expect, it } from "@effect/vitest"; +import * as Effect from "effect/Effect"; +import * as Deferred from "effect/Deferred"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as Stream from "effect/Stream"; +import * as SubscriptionRef from "effect/SubscriptionRef"; +import * as Schema from "effect/Schema"; +import { AsyncResult, Atom, AtomRegistry } from "effect/unstable/reactivity"; +import { + AVAILABLE_CONNECTION_STATE, + PrimaryConnectionTarget, + type SupervisorConnectionState, + type PreparedConnection, +} from "../connection/model.ts"; +import { EnvironmentRegistry } from "../connection/registry.ts"; +import { EnvironmentSupervisor } from "../connection/supervisor.ts"; +import type { RpcSession } from "../rpc/session.ts"; +import { createOrchestrationEnvironmentAtoms } from "./orchestration.ts"; + +const isBridgetReadError = Schema.is(BridgetReadError); + +it.effect("routes human reads by environment and refreshes only the selected atom", () => + Effect.scoped( + Effect.gen(function* () { + const calls: Array<{ environmentId: string; input: BridgetReadInput }> = []; + let refusal = false; + const supervisors = new Map(); + for (const id of ["environment-a", "environment-b"]) { + const environmentId = EnvironmentId.make(id); + const session = { + client: { + [WS_METHODS.bridgetRead]: (input: BridgetReadInput) => + Effect.suspend(() => { + calls.push({ environmentId, input }); + return refusal + ? Effect.fail(new BridgetReadError({ code: "context_missing" })) + : Effect.succeed({ + version: 1, + subject: null, + result: { status: "listed", threads: [], next_after: null }, + } satisfies BridgetHumanView); + }), + }, + } as unknown as RpcSession; + supervisors.set( + environmentId, + EnvironmentSupervisor.of({ + target: new PrimaryConnectionTarget({ + environmentId, + label: id, + httpBaseUrl: `https://${id}.test`, + wsBaseUrl: `wss://${id}.test`, + }), + state: yield* SubscriptionRef.make({ + ...AVAILABLE_CONNECTION_STATE, + phase: "connected", + }), + session: yield* SubscriptionRef.make(Option.some(session)), + prepared: yield* SubscriptionRef.make(Option.none()), + connect: Effect.void, + disconnect: Effect.void, + retryNow: Effect.void, + }), + ); + } + const run: EnvironmentRegistry["Service"]["run"] = (id, effect) => + Effect.provideService(effect, EnvironmentSupervisor, supervisors.get(id)!); + const followStream: EnvironmentRegistry["Service"]["followStream"] = (id, stream) => + Stream.provideService(stream, EnvironmentSupervisor, supervisors.get(id)!); + const stateChanges: EnvironmentRegistry["Service"]["stateChanges"] = (id) => + SubscriptionRef.changes(supervisors.get(id)!.state); + const registryService = EnvironmentRegistry.of({ + run, + followStream, + stateChanges, + } as unknown as EnvironmentRegistry["Service"]); + const runtime = Atom.runtime(Layer.succeed(EnvironmentRegistry, registryService)); + const family = createOrchestrationEnvironmentAtoms(runtime).bridgetRead; + const input = { + threadId: ThreadId.make("8b09a229-dc14-4b38-91b5-2bf0e2a294ac"), + projectId: ProjectId.make("project"), + action: "list", + } as const; + const atomA = family({ environmentId: EnvironmentId.make("environment-a"), input }); + const atomB = family({ environmentId: EnvironmentId.make("environment-b"), input }); + const registry = AtomRegistry.make(); + const unmountA = registry.mount(atomA); + const unmountB = registry.mount(atomB); + yield* Effect.addFinalizer(() => + Effect.sync(() => { + unmountA(); + unmountB(); + registry.dispose(); + }), + ); + yield* AtomRegistry.getResult(registry, atomA, { suspendOnWaiting: true }); + yield* AtomRegistry.getResult(registry, atomB, { suspendOnWaiting: true }); + expect(calls.map((call) => call.environmentId).sort()).toEqual([ + "environment-a", + "environment-b", + ]); + for (const call of calls) expect(call.input).toEqual(input); + registry.refresh(atomA); + yield* AtomRegistry.getResult(registry, atomA, { suspendOnWaiting: true }); + expect(calls.filter((call) => call.environmentId === "environment-a")).toHaveLength(2); + expect(calls.filter((call) => call.environmentId === "environment-b")).toHaveLength(1); + refusal = true; + registry.refresh(atomA); + const error = yield* AtomRegistry.getResult(registry, atomA, { suspendOnWaiting: true }).pipe( + Effect.flip, + ); + expect(isBridgetReadError(error)).toBe(true); + }), + ), +); + +it.effect( + "refreshes a pending native query after A to B to A and discards its late old response", + () => + Effect.scoped( + Effect.gen(function* () { + const environmentId = EnvironmentId.make("pending-refresh-environment"); + const threadA = ThreadId.make("8b09a229-dc14-4b38-91b5-2bf0e2a294ac"); + const threadB = ThreadId.make("3503a3ce-cb97-46eb-b17b-339377b0c1ce"); + const oldResponse = yield* Deferred.make(); + const newResponse = yield* Deferred.make(); + const oldStarted = yield* Deferred.make(); + const newStarted = yield* Deferred.make(); + const oldSettled = yield* Deferred.make(); + const calls: ThreadId[] = []; + const view = (name: string): BridgetHumanView => ({ + version: 1, + subject: { agent_id: threadA, name }, + result: { status: "listed", threads: [], next_after: null }, + }); + let aCalls = 0; + const session = { + client: { + [WS_METHODS.bridgetRead]: (input: BridgetReadInput) => + Effect.gen(function* () { + calls.push(input.threadId); + if (input.threadId === threadB) return view("B"); + aCalls += 1; + if (aCalls === 1) { + yield* Deferred.succeed(oldStarted, undefined); + // Simulate an RPC response that can complete after its caller was cancelled. + return yield* Deferred.await(oldResponse).pipe( + Effect.uninterruptible, + Effect.ensuring(Deferred.succeed(oldSettled, undefined)), + ); + } + yield* Deferred.succeed(newStarted, undefined); + return yield* Deferred.await(newResponse); + }), + }, + } as unknown as RpcSession; + const supervisor = EnvironmentSupervisor.of({ + target: new PrimaryConnectionTarget({ + environmentId, + label: "Pending refresh", + httpBaseUrl: "https://pending.test", + wsBaseUrl: "wss://pending.test", + }), + state: yield* SubscriptionRef.make({ + ...AVAILABLE_CONNECTION_STATE, + phase: "connected", + }), + session: yield* SubscriptionRef.make(Option.some(session)), + prepared: yield* SubscriptionRef.make(Option.none()), + connect: Effect.void, + disconnect: Effect.void, + retryNow: Effect.void, + }); + const run: EnvironmentRegistry["Service"]["run"] = (_id, effect) => + Effect.provideService(effect, EnvironmentSupervisor, supervisor); + const followStream: EnvironmentRegistry["Service"]["followStream"] = (_id, stream) => + Stream.provideService(stream, EnvironmentSupervisor, supervisor); + const stateChanges: EnvironmentRegistry["Service"]["stateChanges"] = () => + SubscriptionRef.changes(supervisor.state); + const registryService = EnvironmentRegistry.of({ + run, + followStream, + stateChanges, + } as unknown as EnvironmentRegistry["Service"]); + const runtime = Atom.runtime(Layer.succeed(EnvironmentRegistry, registryService)); + const family = createOrchestrationEnvironmentAtoms(runtime).bridgetRead; + const inputA = { + threadId: threadA, + projectId: ProjectId.make("project"), + action: "list", + } as const; + const inputB = { ...inputA, threadId: threadB }; + const atomA = family({ environmentId, input: inputA }); + const atomB = family({ environmentId, input: inputB }); + const registry = AtomRegistry.make(); + const publishedA: Array = []; + const unsubscribe = registry.subscribe( + atomA, + (result) => { + if (AsyncResult.isSuccess(result)) publishedA.push(result.value.subject?.name); + }, + { immediate: true }, + ); + let unmount = registry.mount(atomA); + yield* Effect.addFinalizer(() => + Effect.sync(() => { + unmount(); + unsubscribe(); + registry.dispose(); + }), + ); + yield* Deferred.await(oldStarted); + unmount(); + unmount = registry.mount(atomB); + expect( + (yield* AtomRegistry.getResult(registry, atomB, { suspendOnWaiting: true })).subject + ?.name, + ).toBe("B"); + unmount(); + unmount = registry.mount(atomA); + registry.refresh(atomA); + yield* Deferred.await(newStarted); + yield* Deferred.succeed(newResponse, view("new A")); + expect( + (yield* AtomRegistry.getResult(registry, atomA, { suspendOnWaiting: true })).subject + ?.name, + ).toBe("new A"); + yield* Deferred.succeed(oldResponse, view("old A")); + yield* Deferred.await(oldSettled); + expect( + (yield* AtomRegistry.getResult(registry, atomA, { suspendOnWaiting: true })).subject + ?.name, + ).toBe("new A"); + expect(publishedA).not.toContain("old A"); + expect(publishedA).toContain("new A"); + expect(calls).toEqual([threadA, threadB, threadA]); + }), + ), +); diff --git a/packages/client-runtime/src/state/orchestration.ts b/packages/client-runtime/src/state/orchestration.ts index ba80275bffb3..ad71218597c7 100644 --- a/packages/client-runtime/src/state/orchestration.ts +++ b/packages/client-runtime/src/state/orchestration.ts @@ -1,4 +1,4 @@ -import { ORCHESTRATION_WS_METHODS } from "@t3tools/contracts"; +import { ORCHESTRATION_WS_METHODS, WS_METHODS } from "@t3tools/contracts"; import { Atom } from "effect/unstable/reactivity"; import { createEnvironmentRpcQueryAtomFamily } from "./runtime.ts"; @@ -8,6 +8,12 @@ export function createOrchestrationEnvironmentAtoms( runtime: Atom.AtomRuntime, ) { return { + bridgetRead: createEnvironmentRpcQueryAtomFamily(runtime, { + label: "environment-data:orchestration:bridget-read", + tag: WS_METHODS.bridgetRead, + staleTimeMs: 3_000, + idleTtlMs: 10_000, + }), turnDiff: createEnvironmentRpcQueryAtomFamily(runtime, { label: "environment-data:orchestration:turn-diff", tag: ORCHESTRATION_WS_METHODS.getTurnDiff, diff --git a/packages/contracts/src/bridget.test.ts b/packages/contracts/src/bridget.test.ts new file mode 100644 index 000000000000..0e6d31de43fd --- /dev/null +++ b/packages/contracts/src/bridget.test.ts @@ -0,0 +1,256 @@ +import { describe, expect, it } from "vite-plus/test"; +import * as Schema from "effect/Schema"; +import { BridgetHumanView, BridgetReadInput } from "./bridget.ts"; + +const threadId = "8b09a229-dc14-4b38-91b5-2bf0e2a294ac"; +const sharedThreadId = "3503a3ce-cb97-46eb-b17b-339377b0c1ce"; +const base = { threadId, projectId: "project-1" }; +const decodeInput = Schema.decodeUnknownSync(BridgetReadInput); +const decodeView = Schema.decodeUnknownSync(BridgetHumanView); + +describe("Bridget human read contract", () => { + const recentCursor = `1791331200:${sharedThreadId}`; + const recentThread = { + thread_id: sharedThreadId, + title: "Recent", + creator_id: threadId, + state: "open", + last_seq: 3, + members: [], + last_activity_at: 1791331200, + }; + const recentEntry = (seq: number) => ({ + seq, + message_id: sharedThreadId, + author_id: threadId, + author_name: "Agent", + created_at: 1791331200, + body: " é🙂\nexact body\n", + notify: { mode: "none", targets: [] }, + reply_to_seq: null, + }); + const recentHistory = { + status: "history_recent", + thread_id: sharedThreadId, + entries: [recentEntry(3), recentEntry(2)], + snapshot_seq: 3, + through_seq: 3, + has_more: true, + next_before_seq: 1, + }; + it("accepts independent recent variants and keeps their exact bodies", () => { + expect( + decodeInput({ ...base, action: "list_recent", after: recentCursor, limit: 100 }).action, + ).toBe("list_recent"); + expect( + decodeInput({ ...base, action: "history_recent", sharedThreadId, beforeSeq: 0, toSeq: 3 }) + .action, + ).toBe("history_recent"); + expect( + decodeView({ + version: 1, + subject: null, + result: { + status: "listed_recent", + threads: [recentThread], + next_after: recentCursor, + }, + }).result.status, + ).toBe("listed_recent"); + const result = decodeView({ version: 1, subject: null, result: recentHistory }).result; + expect(result.status === "history_recent" && result.entries[0]?.body).toBe(recentEntry(3).body); + }); + it.each(["01791331200", "+1791331200", "-1", "9007199254740992", "1791331200 "])( + "rejects noncanonical or unsafe cursor timestamp %s", + (timestamp) => { + expect(() => + decodeInput({ ...base, action: "list_recent", after: `${timestamp}:${sharedThreadId}` }), + ).toThrow(); + }, + ); + it.each([ + { action: "list_recent", after: `${recentCursor}:extra` }, + { action: "list_recent", after: recentCursor.toUpperCase() }, + { action: "list_recent", beforeSeq: 1 }, + { action: "list", after: recentCursor }, + { action: "history_recent", sharedThreadId, fromSeq: 1 }, + { action: "history_recent", sharedThreadId, beforeSeq: -1 }, + { action: "history_recent", sharedThreadId, toSeq: Number.MAX_SAFE_INTEGER + 1 }, + { action: "history_recent", sharedThreadId, socket: "/forged" }, + ])("rejects recent cross-action or injected options %j", (input) => { + expect(() => decodeInput({ ...base, ...input })).toThrow(); + }); + it.each([ + { entries: [recentEntry(2), recentEntry(3)] }, + { entries: [recentEntry(3), recentEntry(3)] }, + { through_seq: 2 }, + { through_seq: 4 }, + { next_before_seq: 2 }, + { next_before_seq: null }, + { has_more: false }, + { entries: [] }, + { entries: [recentEntry(1)], next_before_seq: 0 }, + ])("rejects incoherent recent pagination %j", (change) => { + expect(() => + decodeView({ version: 1, subject: null, result: { ...recentHistory, ...change } }), + ).toThrow(); + }); + it("accepts terminal sequence one and an empty zero-bound page without continuation", () => { + for (const change of [ + { entries: [recentEntry(1)] }, + { entries: [], snapshot_seq: 0, through_seq: 0 }, + ]) + expect( + decodeView({ + version: 1, + subject: null, + result: { + ...recentHistory, + ...change, + has_more: false, + next_before_seq: null, + }, + }).result.status, + ).toBe("history_recent"); + }); + it("accepts an exhausted sparse history without inventing older rows", () => { + expect( + decodeView({ + version: 1, + subject: null, + result: { + ...recentHistory, + entries: [recentEntry(5)], + snapshot_seq: 5, + through_seq: 5, + has_more: false, + next_before_seq: null, + }, + }).result.status, + ).toBe("history_recent"); + }); + it("rejects missing activity, out-of-order lists and a cursor unrelated to the emitted page", () => { + for (const change of [ + { threads: [{ ...recentThread, last_activity_at: undefined }] }, + { threads: [{ ...recentThread, last_activity_at: Number.MAX_SAFE_INTEGER + 1 }] }, + { + threads: [ + recentThread, + { ...recentThread, thread_id: threadId, last_activity_at: 1791331201 }, + ], + }, + { threads: [recentThread, recentThread] }, + { next_after: `1791331201:${sharedThreadId}` }, + { threads: [] }, + ]) + expect(() => + decodeView({ + version: 1, + subject: null, + result: { + status: "listed_recent", + threads: [recentThread], + next_after: recentCursor, + ...change, + }, + }), + ).toThrow(); + }); + it("accepts equal activity UUID ASC and rejects UUID DESC", () => { + const second = { ...recentThread, thread_id: threadId }; + const result = { + status: "listed_recent", + threads: [recentThread, second], + next_after: `1791331200:${threadId}`, + }; + expect(decodeView({ version: 1, subject: null, result }).result.status).toBe("listed_recent"); + expect(() => + decodeView({ + version: 1, + subject: null, + result: { ...result, threads: [second, recentThread], next_after: null }, + }), + ).toThrow(); + }); + it("accepts the three bounded read actions", () => { + expect(decodeInput({ ...base, action: "list", limit: 100 })).toMatchObject({ action: "list" }); + expect(decodeInput({ ...base, action: "show", sharedThreadId })).toMatchObject({ + action: "show", + }); + expect( + decodeInput({ ...base, action: "history", sharedThreadId, limit: 200, fromSeq: 1, toSeq: 2 }), + ).toMatchObject({ action: "history" }); + }); + + it.each(["root", "agentId", "socket", "command", "credentials", "environmentId"])( + "refuses browser-selected authority: %s", + (field) => { + expect(() => decodeInput({ ...base, action: "list", [field]: "/forged" })).toThrow(); + }, + ); + + it.each([ + { action: "read" }, + { action: "list", limit: 0 }, + { action: "list", limit: 101 }, + { action: "list", fromSeq: 1 }, + { action: "show" }, + { action: "show", sharedThreadId: "forged" }, + { action: "history", sharedThreadId, limit: 201 }, + { action: "history", sharedThreadId, fromSeq: 0 }, + { action: "history", sharedThreadId, fromSeq: 3, toSeq: 2 }, + ])("refuses invalid or cross-action options %j", (input) => { + expect(() => decodeInput({ ...base, ...input })).toThrow(); + }); + + it("preserves exact Unicode bodies and legacy nullable fields", () => { + const body = " Ligne é🙂\n\tseconde ligne\n"; + const view = decodeView({ + version: 1, + subject: { agent_id: threadId, name: null }, + result: { + status: "history", + from_seq: 1, + thread_id: sharedThreadId, + entries: [ + { + seq: 1, + message_id: sharedThreadId, + author_id: threadId, + author_name: null, + created_at: 1791331200, + body, + notify: { mode: "all", targets: [] }, + reply_to_seq: null, + kind: null, + supersedes_seq: null, + superseded_by_seq: null, + }, + ], + snapshot_seq: 1, + through_seq: 1, + has_more: false, + next_from_seq: null, + }, + }); + expect(view.result.status === "history" && view.result.entries[0]?.body).toBe(body); + }); + + it("rejects unsupported versions and agent-only projection fields", () => { + const view = { + version: 1, + subject: null, + result: { + status: "error", + code: "binding_unavailable", + detail: "Unavailable", + retryable: false, + }, + }; + expect(decodeView(view).result.status).toBe("error"); + expect(() => decodeView({ ...view, version: 2 })).toThrow(); + expect(() => decodeView({ ...view, credentials: "secret" })).toThrow(); + expect(() => decodeView({ ...view, result: { ...view.result, own_acked_seq: 1 } })).toThrow(); + expect(() => decodeView({ ...view, result: { ...view.result, code: "unknown" } })).toThrow(); + }); +}); diff --git a/packages/contracts/src/bridget.ts b/packages/contracts/src/bridget.ts new file mode 100644 index 000000000000..6a8193dd326a --- /dev/null +++ b/packages/contracts/src/bridget.ts @@ -0,0 +1,240 @@ +import * as Schema from "effect/Schema"; +import * as SchemaParser from "effect/SchemaParser"; +import { NonNegativeInt, PositiveInt, ProjectId, ThreadId } from "./baseSchemas.ts"; + +const Uuid = Schema.String.check(Schema.isUUID()); +const Name = Schema.NullOr(Schema.String.check(Schema.isMaxLength(256))); +const Sequence = NonNegativeInt.check(Schema.isLessThanOrEqualTo(Number.MAX_SAFE_INTEGER)); +const RecentCursor = Schema.String.check( + Schema.isMaxLength(128), + Schema.makeFilter((cursor) => { + const match = + /^(0|[1-9][0-9]*):[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/.exec(cursor); + return match !== null && Number.isSafeInteger(Number(match[1])); + }), +); +const context = { threadId: ThreadId.check(Schema.isUUID()), projectId: ProjectId }; + +const ReadInput = Schema.Union([ + Schema.Struct({ + ...context, + action: Schema.Literal("list"), + after: Schema.optional(Uuid), + limit: Schema.optional(PositiveInt.check(Schema.isLessThanOrEqualTo(100))), + }), + Schema.Struct({ ...context, action: Schema.Literal("show"), sharedThreadId: Uuid }), + Schema.Struct({ + ...context, + action: Schema.Literal("list_recent"), + after: Schema.optional(RecentCursor), + limit: Schema.optional(PositiveInt.check(Schema.isLessThanOrEqualTo(100))), + }), + Schema.Struct({ + ...context, + action: Schema.Literal("history_recent"), + sharedThreadId: Uuid, + limit: Schema.optional(PositiveInt.check(Schema.isLessThanOrEqualTo(200))), + beforeSeq: Schema.optional(Sequence), + toSeq: Schema.optional(Sequence), + }), + Schema.Struct({ + ...context, + action: Schema.Literal("history"), + sharedThreadId: Uuid, + limit: Schema.optional(PositiveInt.check(Schema.isLessThanOrEqualTo(200))), + fromSeq: Schema.optional( + PositiveInt.check(Schema.isLessThanOrEqualTo(Number.MAX_SAFE_INTEGER)), + ), + toSeq: Schema.optional(Sequence), + }).check( + Schema.makeFilter( + (input) => + input.fromSeq === undefined || input.toSeq === undefined || input.fromSeq <= input.toSeq, + ), + ), +]); +// RPC decoders otherwise discard unknown authority fields. Enforce closed parsing at the schema boundary. +export const BridgetReadInput = Schema.declareConstructor< + typeof ReadInput.Type, + typeof ReadInput.Encoded +>()([ReadInput], ([codec]) => { + const decode = SchemaParser.decodeUnknownEffect(codec, { onExcessProperty: "error" }); + return (input) => decode(input); +}); +export type BridgetReadInput = typeof BridgetReadInput.Type; + +const Member = Schema.Struct({ agent_id: Uuid, name: Name }); +const members = Schema.Array(Member).check(Schema.isMaxLength(16)); +const ThreadSummary = Schema.Struct({ + thread_id: Uuid, + title: Schema.String.check(Schema.isMaxLength(512)), + creator_id: Uuid, + state: Schema.Literals(["open", "closed"]), + last_seq: Sequence, + members, +}); +const Entry = Schema.Struct({ + seq: PositiveInt.check(Schema.isLessThanOrEqualTo(Number.MAX_SAFE_INTEGER)), + message_id: Uuid, + author_id: Uuid, + author_name: Name, + created_at: Schema.Int, + body: Schema.String.check(Schema.isMaxLength(60 * 1024)), + notify: Schema.Struct({ + mode: Schema.Literals(["none", "targets", "all"]), + targets: Schema.Array(Uuid).check(Schema.isMaxLength(16)), + }), + reply_to_seq: Schema.NullOr(Sequence), + kind: Schema.optional( + Schema.NullOr(Schema.Literals(["history", "action", "blocker", "decision"])), + ), + supersedes_seq: Schema.optional(Schema.NullOr(Sequence)), + superseded_by_seq: Schema.optional(Schema.NullOr(Sequence)), +}); +const HumanErrorCode = Schema.Literals([ + "unsupported_version", + "binding_unavailable", + "project_mismatch", + "thread_unavailable", + "invalid_request", + "storage_unavailable", + "response_too_large", +]); + +const HumanView = Schema.Struct({ + version: Schema.Literal(1), + subject: Schema.NullOr(Member), + result: Schema.Union([ + Schema.Struct({ + status: Schema.Literal("listed"), + threads: Schema.Array(ThreadSummary).check(Schema.isMaxLength(100)), + next_after: Schema.NullOr(Uuid), + }), + Schema.Struct({ + status: Schema.Literal("shown"), + ...ThreadSummary.fields, + created_at: Schema.Int, + closed_at: Schema.NullOr(Schema.Int), + }), + Schema.Struct({ + status: Schema.Literal("listed_recent"), + threads: Schema.Array( + Schema.Struct({ + ...ThreadSummary.fields, + last_activity_at: Sequence, + }), + ).check(Schema.isMaxLength(100)), + next_after: Schema.NullOr(RecentCursor), + }).check( + Schema.makeFilter((result) => { + const seen = new Set(); + for (let index = 0; index < result.threads.length; index++) { + const current = result.threads[index]!; + if (seen.has(current.thread_id)) return false; + seen.add(current.thread_id); + const previous = result.threads[index - 1]; + if ( + previous && + (previous.last_activity_at < current.last_activity_at || + (previous.last_activity_at === current.last_activity_at && + previous.thread_id >= current.thread_id)) + ) + return false; + } + const last = result.threads.at(-1); + return ( + result.next_after === null || + (last !== undefined && result.next_after === `${last.last_activity_at}:${last.thread_id}`) + ); + }), + ), + Schema.Struct({ + status: Schema.Literal("history_recent"), + thread_id: Uuid, + entries: Schema.Array(Entry).check(Schema.isMaxLength(200)), + snapshot_seq: Sequence, + through_seq: Sequence, + has_more: Schema.Boolean, + next_before_seq: Schema.NullOr(Sequence), + }).check( + Schema.makeFilter((result) => { + if (result.through_seq > result.snapshot_seq) return false; + for (let index = 0; index < result.entries.length; index++) { + const current = result.entries[index]!; + const previous = result.entries[index - 1]; + if (current.seq > result.through_seq || (previous && previous.seq <= current.seq)) + return false; + } + const last = result.entries.at(-1); + return result.has_more + ? last !== undefined && last.seq > 1 && result.next_before_seq === last.seq - 1 + : result.next_before_seq === null; + }), + ), + Schema.Struct({ + status: Schema.Literal("history"), + thread_id: Uuid, + entries: Schema.Array(Entry).check(Schema.isMaxLength(200)), + snapshot_seq: Sequence, + through_seq: Sequence, + has_more: Schema.Boolean, + next_from_seq: Schema.NullOr(Sequence), + from_seq: PositiveInt.check(Schema.isLessThanOrEqualTo(Number.MAX_SAFE_INTEGER)), + }), + Schema.Struct({ + status: Schema.Literal("error"), + code: HumanErrorCode, + detail: Schema.String.check(Schema.isMaxLength(512)), + retryable: Schema.Boolean, + }), + ]), +}); +export const BridgetHumanView = Schema.declareConstructor< + typeof HumanView.Type, + typeof HumanView.Encoded +>()([HumanView], ([codec]) => { + const decode = SchemaParser.decodeUnknownEffect(codec, { onExcessProperty: "error" }); + return (input) => decode(input); +}); +export type BridgetHumanView = typeof BridgetHumanView.Type; + +const BridgetReadErrorCode = Schema.Literals([ + "unavailable", + "timeout", + "invalid_output", + "output_limit", + "context_missing", + "project_mismatch", + "unsupported_version", + "invalid_request", + "command_failed", + "storage_unavailable", +]); +export class BridgetReadError extends Schema.TaggedError()("BridgetReadError", { + code: BridgetReadErrorCode, +}) { + override get message(): string { + switch (this.code) { + case "unavailable": + return "Bridget is unavailable in this environment."; + case "timeout": + return "Bridget did not respond within six seconds."; + case "invalid_output": + return "Bridget returned an invalid human thread response."; + case "output_limit": + return "The Bridget response exceeded the read limit."; + case "context_missing": + return "The selected conversation or project is unavailable."; + case "project_mismatch": + return "The selected conversation does not belong to this project."; + case "unsupported_version": + return "The Bridget human thread protocol is incompatible."; + case "invalid_request": + return "The Bridget read request is invalid."; + case "command_failed": + return "The Bridget read command failed."; + case "storage_unavailable": + return "The conversation context could not be read."; + } + } +} diff --git a/packages/contracts/src/index.ts b/packages/contracts/src/index.ts index 978a0459e69b..def853e10b43 100644 --- a/packages/contracts/src/index.ts +++ b/packages/contracts/src/index.ts @@ -1,4 +1,5 @@ export * from "./baseSchemas.ts"; +export * from "./bridget.ts"; export * from "./assistantCitations.ts"; export * from "./composerContext.ts"; export * from "./composerContextClipboard.ts"; diff --git a/packages/contracts/src/rpc.ts b/packages/contracts/src/rpc.ts index c1b25459048e..be283a86863e 100644 --- a/packages/contracts/src/rpc.ts +++ b/packages/contracts/src/rpc.ts @@ -8,6 +8,7 @@ import { import * as Schema from "effect/Schema"; import * as Rpc from "effect/unstable/rpc/Rpc"; import * as RpcGroup from "effect/unstable/rpc/RpcGroup"; +import { BridgetHumanView, BridgetReadInput, BridgetReadError } from "./bridget.ts"; import { NonNegativeInt, TrimmedNonEmptyString } from "./baseSchemas.ts"; import { CodexAuthCallbackInput, @@ -287,6 +288,7 @@ import { import { VcsError } from "./vcs.ts"; export const WS_METHODS = { + bridgetRead: "bridget.read", // Project registry methods projectsList: "projects.list", projectsAdd: "projects.add", @@ -993,6 +995,12 @@ const WsProjectsReadFileRpc = Rpc.make(WS_METHODS.projectsReadFile, { error: Schema.Union([ProjectReadFileError, EnvironmentAuthorizationError]), }); +const WsBridgetReadRpc = Rpc.make(WS_METHODS.bridgetRead, { + payload: BridgetReadInput, + success: BridgetHumanView, + error: Schema.Union([BridgetReadError, EnvironmentAuthorizationError]), +}); + const WsProjectsWriteFileRpc = Rpc.make(WS_METHODS.projectsWriteFile, { payload: ProjectWriteFileInput, success: ProjectWriteFileResult, @@ -1450,6 +1458,7 @@ const WsSubscribeResourceTelemetryRpc = Rpc.make(WS_METHODS.subscribeResourceTel }); export const WsRpcGroup = RpcGroup.make( + WsBridgetReadRpc, WsServerProbeRpc, WsServerGetConfigRpc, WsServerRefreshProvidersRpc,