diff --git a/apps/server/src/server.ts b/apps/server/src/server.ts index 0f9a8be27ca..3ce761f30bb 100644 --- a/apps/server/src/server.ts +++ b/apps/server/src/server.ts @@ -49,7 +49,7 @@ import { ProviderRuntimeIngestionLive } from "./orchestration/Layers/ProviderRun import { ProviderCommandReactorLive } from "./orchestration/Layers/ProviderCommandReactor.ts"; import { CheckpointReactorLive } from "./orchestration/Layers/CheckpointReactor.ts"; import { ThreadDeletionReactorLive } from "./orchestration/Layers/ThreadDeletionReactor.ts"; -import { T3xLayerLive } from "./t3x/index.ts"; // t3x fork seam — all fork-local features fan in here +import { T3xLayerLive, T3xRoutesLive } from "./t3x/index.ts"; // t3x fork seam — all fork-local features fan in here import * as AgentAwarenessRelay from "./relay/AgentAwarenessRelay.ts"; import { hasCloudPublicConfig } from "./cloud/publicConfig.ts"; import { ProviderRegistryLive } from "./provider/Layers/ProviderRegistry.ts"; @@ -361,6 +361,7 @@ export const makeRoutesLayer = Layer.mergeAll( Layer.provide(environmentAuthenticatedAuthLayer), ), otlpTracesProxyRouteLayer, + T3xRoutesLive, // t3x fork seam — all fork-local HTTP routes fan in here assetRouteLayer, staticAndDevRouteLayer, websocketRpcRouteLayer, diff --git a/apps/server/src/t3x/autoResume/Reactor.test.ts b/apps/server/src/t3x/autoResume/Reactor.test.ts index dd05fb687b6..21f0f1b778a 100644 --- a/apps/server/src/t3x/autoResume/Reactor.test.ts +++ b/apps/server/src/t3x/autoResume/Reactor.test.ts @@ -255,4 +255,51 @@ describe("AutoResumeReactor (integration)", () => { }).pipe(Effect.provide(AutoResumeReactorLive.pipe(Layer.provideMerge(deps)))); }).pipe(Effect.scoped, Effect.provide(Layer.mergeAll(NodeServices.layer, TestClock.layer()))), ); + + it.effect("does NOT schedule for a thread whose auto-resume is switched off", () => + Effect.gen(function* () { + const { dispatched, deps, store } = yield* harness(readModel({}), [rejectedEvent(100)]); + yield* store.setEnabled("thread-1", false); + + yield* Effect.gen(function* () { + yield* settle; // detection runs against the pre-loaded rejection + + assert.strictEqual( + (yield* store.listPending).length, + 0, + "a disabled thread must never schedule a resume", + ); + // Disabling is a deliberate user action, so it must not post timeline noise either. + assert.notInclude(types(yield* Ref.get(dispatched)), "thread.activity.append"); + + yield* advancePastResume; + assert.notInclude(types(yield* Ref.get(dispatched)), "thread.turn.start"); + }).pipe(Effect.provide(AutoResumeReactorLive.pipe(Layer.provideMerge(deps)))); + }).pipe(Effect.scoped, Effect.provide(Layer.mergeAll(NodeServices.layer, TestClock.layer()))), + ); + + it.effect("cancels an already-scheduled resume when the thread is switched off mid-wait", () => + Effect.gen(function* () { + const { dispatched, deps, store } = yield* harness(readModel({}), [rejectedEvent(100)]); + + yield* Effect.gen(function* () { + yield* settle; + assert.strictEqual((yield* store.listPending).length, 1, "precondition: it scheduled"); + + // The switch is flipped off *after* scheduling but *before* the window reopens. + // fireOne must re-read the record rather than trust the scheduling-time value. + yield* store.setEnabled("thread-1", false); + + yield* advancePastResume; + + assert.notInclude(types(yield* Ref.get(dispatched)), "thread.turn.start"); + assert.strictEqual((yield* store.listPending).length, 0, "pending must be cleared"); + assert.strictEqual( + yield* store.countFiredSince("thread-1", 0), + 0, + "a cancellation must not burn one of the 24h attempts", + ); + }).pipe(Effect.provide(AutoResumeReactorLive.pipe(Layer.provideMerge(deps)))); + }).pipe(Effect.scoped, Effect.provide(Layer.mergeAll(NodeServices.layer, TestClock.layer()))), + ); }); diff --git a/apps/server/src/t3x/autoResume/Reactor.ts b/apps/server/src/t3x/autoResume/Reactor.ts index 6dbca7247b0..9c106de421b 100644 --- a/apps/server/src/t3x/autoResume/Reactor.ts +++ b/apps/server/src/t3x/autoResume/Reactor.ts @@ -133,6 +133,9 @@ const makeSupervisor = Effect.gen(function* () { const threadId = event.threadId; const record = yield* store.getThread(threadId); + // Per-thread switch. A thread the user turned off never schedules, and posts no + // timeline note: they disabled it deliberately, so an activity row would be noise. + if (!record.enabled) return; const nowMs = yield* Clock.currentTimeMillis; const firedRecently = yield* store.countFiredSince(threadId, nowMs - BACKOFF_LOOKBACK_MS); const firedInCapWindow = yield* store.countFiredSince(threadId, nowMs - CAP_WINDOW_MS); @@ -183,6 +186,21 @@ const makeSupervisor = Effect.gen(function* () { return; } + // The user can switch a thread off *after* its resume was scheduled, so re-read the + // record here rather than trusting the value seen at scheduling time, and treat a + // disabled thread as a cancellation (drop the pending resume, say so once). + const record = yield* store.getThread(pending.threadId); + if (!record.enabled) { + yield* store.clearPending(pending.threadId); + yield* appendActivity( + thread.id, + "info", + "t3x.auto-resume.cancelled", + "Auto-resume cancelled: turned off for this thread.", + ); + return; + } + const reason = cancelReason(thread, pending.baseline); if (reason) { yield* store.clearPending(pending.threadId); diff --git a/apps/server/src/t3x/autoResume/http.test.ts b/apps/server/src/t3x/autoResume/http.test.ts new file mode 100644 index 00000000000..1e3eafd25af --- /dev/null +++ b/apps/server/src/t3x/autoResume/http.test.ts @@ -0,0 +1,216 @@ +// @effect-diagnostics nodeBuiltinImport:off +/** + * Route-level tests for `/api/t3x/auto-resume`. + * + * Serves ONLY the fork's route layer (not the whole `makeRoutesLayer`) over a real HTTP + * server on an ephemeral port, with `EnvironmentAuth` mocked. + * + * Scope note: because auth is mocked, these tests do NOT prove that the route's + * `authenticateWithOperateScope` faithfully mirrors upstream's private + * `authenticateRawRouteWithScope` — that remains a logic mirror tracked in SEAMS.md. + * What they DO prove is that a rejected credential and a missing scope each render as a + * 401/403 through the route's `catchTags` rather than escaping as a 500 or an unhandled + * defect, and that the GET/POST contract round-trips through a real store. + */ +import * as NodeHttpServer from "@effect/platform-node/NodeHttpServer"; +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { assert, describe, it } from "@effect/vitest"; +import { AuthOrchestrationOperateScope } from "@t3tools/contracts"; +import * as Effect from "effect/Effect"; +import * as FileSystem from "effect/FileSystem"; +import * as Layer from "effect/Layer"; +import { + FetchHttpClient, + HttpClient, + HttpClientRequest, + HttpRouter, + HttpServer, +} from "effect/unstable/http"; +import * as NodeHttp from "node:http"; +import * as NodePath from "node:path"; + +import * as EnvironmentAuth from "../../auth/EnvironmentAuth.ts"; +import { autoResumeRouteLayer } from "./http.ts"; +import { AutoResumeStore, makeAutoResumeStore } from "./state.ts"; + +const PATH = "/api/t3x/auto-resume"; + +const authedSession = (scopes: ReadonlyArray) => + ({ + sessionId: "test-session", + subject: "test", + method: "bearer-access-token", + scopes, + }) as unknown as EnvironmentAuth.AuthenticatedSession; + +/** Mock auth that succeeds with the given scopes. */ +const authOk = (scopes: ReadonlyArray = [AuthOrchestrationOperateScope]) => + Layer.mock(EnvironmentAuth.EnvironmentAuth)({ + authenticateHttpRequest: () => Effect.succeed(authedSession(scopes)), + }); + +/** + * Mock auth that rejects the credential. + * + * Must be a member of the `ServerAuthCredentialError` union + * (`ServerAuthMissingCredentialError | ServerAuthInvalidCredentialError`) — that union is + * what `isServerAuthCredentialError` guards on. An error outside it (and outside + * `ServerAuthInternalError`) matches neither `catchIf` branch and surfaces as a 500. That + * is equally true of upstream's OTLP route, which uses the identical two branches, so it + * is a shared property rather than a fork divergence. + */ +const authRejects = Layer.mock(EnvironmentAuth.EnvironmentAuth)({ + authenticateHttpRequest: () => + Effect.fail(new EnvironmentAuth.ServerAuthInvalidCredentialError({})), +}); + +const baseUrl = (pathname: string) => + Effect.gen(function* () { + const server = yield* HttpServer.HttpServer; + const address = server.address as HttpServer.TcpAddress; + return `http://127.0.0.1:${address.port}${pathname}`; + }); + +const getJson = (pathname: string) => + Effect.gen(function* () { + const client = yield* HttpClient.HttpClient; + return yield* client.get(yield* baseUrl(pathname)); + }); + +const postJson = (pathname: string, body: unknown) => + Effect.gen(function* () { + const client = yield* HttpClient.HttpClient; + const request = HttpClientRequest.bodyJsonUnsafe( + HttpClientRequest.post(yield* baseUrl(pathname)), + body, + ); + return yield* client.execute(request); + }); + +/** Serves the route over an ephemeral port with the given auth layer and a real store. */ +const withRoute = ( + authLayer: Layer.Layer, + body: Effect.Effect, +): Promise => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const root = yield* fs.makeTempDirectoryScoped({ prefix: "t3x-http-" }); + const storeLayer = Layer.effect( + AutoResumeStore, + makeAutoResumeStore(NodePath.join(root, "state.json")), + ); + + const served = HttpRouter.serve(autoResumeRouteLayer, { + disableListenLog: true, + disableLogger: true, + }).pipe( + Layer.provide(storeLayer), + Layer.provide(authLayer), + // provideMerge, not provide: the test body needs HttpServer in context to read the + // ephemeral port off `server.address`. + Layer.provideMerge(NodeHttpServer.layer(() => NodeHttp.createServer(), { port: 0 })), + ); + + return yield* body.pipe(Effect.provide(served), Effect.provide(FetchHttpClient.layer)); + }).pipe(Effect.scoped, Effect.provide(NodeServices.layer), Effect.runPromise); + +describe("/api/t3x/auto-resume", () => { + it("rejects a bad credential with 401/403 instead of leaking a 500", () => + withRoute( + authRejects, + Effect.gen(function* () { + const res = yield* getJson(`${PATH}?threadId=thread-a`); + assert.isTrue( + res.status === 401 || res.status === 403, + `expected 401/403 for a rejected credential, got ${res.status}`, + ); + }), + )); + + it("rejects a session lacking the operate scope", () => + withRoute( + authOk([]), + Effect.gen(function* () { + const res = yield* getJson(`${PATH}?threadId=thread-a`); + assert.isTrue( + res.status === 401 || res.status === 403, + `expected 401/403 without the operate scope, got ${res.status}`, + ); + }), + )); + + it("GET defaults a never-seen thread to enabled", () => + withRoute( + authOk(), + Effect.gen(function* () { + const res = yield* getJson(`${PATH}?threadId=fresh`); + assert.strictEqual(res.status, 200); + const body = (yield* res.json) as Record; + assert.strictEqual(body.enabled, true); + assert.strictEqual(body.overridePrompt, null); + assert.strictEqual(body.pending, null); + }), + )); + + it("GET without a threadId is a 400, not a crash", () => + withRoute( + authOk(), + Effect.gen(function* () { + const res = yield* getJson(PATH); + assert.strictEqual(res.status, 400); + }), + )); + + it("POST patches one field without clobbering the other", () => + withRoute( + authOk(), + Effect.gen(function* () { + const off = yield* postJson(PATH, { threadId: "t1", enabled: false }); + assert.strictEqual(off.status, 200); + assert.strictEqual(((yield* off.json) as Record).enabled, false); + + // Writing only overridePrompt must not resurrect `enabled`. + const text = yield* postJson(PATH, { threadId: "t1", overridePrompt: "keep going" }); + const textBody = (yield* text.json) as Record; + assert.strictEqual(textBody.overridePrompt, "keep going"); + assert.strictEqual(textBody.enabled, false, "a prompt write must not clobber enabled"); + + // …and writing only `enabled` must not wipe the prompt. + const on = yield* postJson(PATH, { threadId: "t1", enabled: true }); + const onBody = (yield* on.json) as Record; + assert.strictEqual(onBody.enabled, true); + assert.strictEqual(onBody.overridePrompt, "keep going", "toggle must not clobber prompt"); + }), + )); + + // This route is what makes threadId caller-controlled; before it, ids only ever came + // from provider events. A prototype-chain id used to resolve on Object.prototype, which + // 500'd on read and persisted a malformed record that wiped the whole state file on the + // next boot. Pinned here at the HTTP boundary, not just in the store. + it("handles a prototype-chain threadId as an ordinary unknown thread", () => + withRoute( + authOk(), + Effect.gen(function* () { + for (const hostile of ["constructor", "__proto__", "toString"]) { + const res = yield* getJson(`${PATH}?threadId=${encodeURIComponent(hostile)}`); + assert.strictEqual(res.status, 200, `GET threadId=${hostile} must not 500`); + const body = (yield* res.json) as Record; + assert.strictEqual(body.enabled, true); + assert.strictEqual(body.pending, null); + } + + const wrote = yield* postJson(PATH, { threadId: "__proto__", enabled: false }); + assert.strictEqual(wrote.status, 200, "POST threadId=__proto__ must not 500"); + assert.strictEqual(((yield* wrote.json) as Record).enabled, false); + }), + )); + + it("POST with a malformed body is a 400, not a 500", () => + withRoute( + authOk(), + Effect.gen(function* () { + const res = yield* postJson(PATH, { nope: true }); + assert.strictEqual(res.status, 400); + }), + )); +}); diff --git a/apps/server/src/t3x/autoResume/http.ts b/apps/server/src/t3x/autoResume/http.ts new file mode 100644 index 00000000000..5c62d3edd7f --- /dev/null +++ b/apps/server/src/t3x/autoResume/http.ts @@ -0,0 +1,165 @@ +/** + * Fork-owned raw HTTP route backing the per-thread auto-resume control. + * + * GET /api/t3x/auto-resume?threadId=… -> { enabled, overridePrompt, pending } + * POST /api/t3x/auto-resume -> same shape, after applying the write + * + * Deliberately a *raw* route rather than a WS-RPC method: adding an RPC would force edits + * to `@t3tools/contracts` (`WsRpcGroup`, `WS_METHODS`) plus `apps/server/src/ws.ts` and its + * scope map — several hot upstream files. A raw route costs exactly one additive line in + * `server.ts`'s route list instead. See docs/t3x/SEAMS.md. + * + * @module t3x/autoResume/http + */ + +import { AuthOrchestrationOperateScope } from "@t3tools/contracts"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as Schema from "effect/Schema"; +import { + HttpRouter, + HttpServerRequest, + HttpServerRespondable, + HttpServerResponse, +} from "effect/unstable/http"; + +import * as EnvironmentAuth from "../../auth/EnvironmentAuth.ts"; +import { + failEnvironmentAuthInvalid, + failEnvironmentInternal, + failEnvironmentScopeRequired, +} from "../../auth/http.ts"; +import { AutoResumeStore, type AutoResumeStoreShape } from "./state.ts"; + +export const AUTO_RESUME_ROUTE_PATH = "/api/t3x/auto-resume"; + +/** + * MIRROR of the module-private `authenticateRawRouteWithScope` in `apps/server/src/http.ts` + * (which the OTLP proxy route uses). It is not exported, so importing it would mean editing + * an upstream file; the fork replicates the ~15 lines instead. Registered as a logic mirror + * in docs/t3x/SEAMS.md — if upstream changes how raw routes authenticate, this must follow. + * + * Operate (not read) scope: these endpoints mutate scheduling behaviour. + */ +const authenticateWithOperateScope = Effect.gen(function* () { + const request = yield* HttpServerRequest.HttpServerRequest; + const serverAuth = yield* EnvironmentAuth.EnvironmentAuth; + const session = yield* serverAuth.authenticateHttpRequest(request).pipe( + Effect.catchIf(EnvironmentAuth.isServerAuthCredentialError, (error) => + failEnvironmentAuthInvalid(EnvironmentAuth.serverAuthCredentialReason(error)), + ), + Effect.catchIf(EnvironmentAuth.isServerAuthInternalError, (error) => + failEnvironmentInternal("internal_error", error), + ), + ); + if (!session.scopes.includes(AuthOrchestrationOperateScope)) { + return yield* failEnvironmentScopeRequired(AuthOrchestrationOperateScope); + } +}); + +const WriteBody = Schema.Struct({ + threadId: Schema.String, + enabled: Schema.optionalKey(Schema.Boolean), + overridePrompt: Schema.optionalKey(Schema.NullOr(Schema.String)), +}); + +const decodeWriteBody = Schema.decodeUnknownEffect(WriteBody); + +/** + * The wire shape shared by GET and POST responses. + * + * Takes the store as an argument instead of `yield*`-ing it from context. That is what + * keeps `AutoResumeStore` out of the *handler's* requirements — see `autoResumeRouteLayer`. + */ +const readThreadState = (store: AutoResumeStoreShape, threadId: string) => + Effect.gen(function* () { + const record = yield* store.getThread(threadId); + return { + enabled: record.enabled, + overridePrompt: record.overridePrompt, + pending: + record.pending === null + ? null + : { resumeAtMs: record.pending.resumeAtMs, reason: record.pending.reason }, + }; + }); + +const makeGetRoute = (store: AutoResumeStoreShape) => + HttpRouter.add( + "GET", + AUTO_RESUME_ROUTE_PATH, + Effect.gen(function* () { + yield* authenticateWithOperateScope; + const request = yield* HttpServerRequest.HttpServerRequest; + const url = HttpServerRequest.toURL(request); + if (Option.isNone(url)) { + return HttpServerResponse.text("Bad Request", { status: 400 }); + } + const threadId = url.value.searchParams.get("threadId"); + if (threadId === null || threadId === "") { + return HttpServerResponse.text("Missing threadId", { status: 400 }); + } + return HttpServerResponse.jsonUnsafe(yield* readThreadState(store, threadId)); + }).pipe( + Effect.catchTags({ + EnvironmentAuthInvalidError: HttpServerRespondable.toResponse, + EnvironmentInternalError: HttpServerRespondable.toResponse, + EnvironmentScopeRequiredError: HttpServerRespondable.toResponse, + }), + ), + ); + +const makePostRoute = (store: AutoResumeStoreShape) => + HttpRouter.add( + "POST", + AUTO_RESUME_ROUTE_PATH, + Effect.gen(function* () { + yield* authenticateWithOperateScope; + const request = yield* HttpServerRequest.HttpServerRequest; + + const body = yield* decodeWriteBody(yield* request.json).pipe( + Effect.map((decoded): typeof WriteBody.Type | null => decoded), + // A malformed body is a client error, not a 500. + Effect.orElseSucceed(() => null), + ); + if (body === null || body.threadId === "") { + return HttpServerResponse.text("Invalid body", { status: 400 }); + } + + // Both fields are optional so the UI can PATCH one without clobbering the other: + // the toggle must not wipe a resume message the user is mid-way through typing. + if (body.enabled !== undefined) { + yield* store.setEnabled(body.threadId, body.enabled); + } + if (body.overridePrompt !== undefined) { + yield* store.setOverridePrompt(body.threadId, body.overridePrompt); + } + + return HttpServerResponse.jsonUnsafe(yield* readThreadState(store, body.threadId)); + }).pipe( + Effect.catchTags({ + EnvironmentAuthInvalidError: HttpServerRespondable.toResponse, + EnvironmentInternalError: HttpServerRespondable.toResponse, + EnvironmentScopeRequiredError: HttpServerRespondable.toResponse, + }), + ), + ); + +/** + * Mounted from `server.ts`'s route list — the fork's only route seam. + * + * `Layer.unwrap` resolves `AutoResumeStore` once, at *layer construction*, and the handlers + * close over the resulting value. This is load-bearing: if the handlers `yield*`-ed the store + * from context instead, that requirement would propagate out through `HttpRouter` into the + * type of upstream's `makeRoutesLayer`, and every existing `server.test.ts` / `bin.test.ts` + * case that builds it would stop type-checking. A fork change must never widen an upstream + * signature. Here the requirement lands on the *layer* instead, where `t3x/index.ts` + * discharges it with `Layer.provide(AutoResumeStoreLive)`. + */ +export const autoResumeRouteLayer = Layer.unwrap( + Effect.gen(function* () { + const store = yield* AutoResumeStore; + return Layer.merge(makeGetRoute(store), makePostRoute(store)); + }), +); diff --git a/apps/server/src/t3x/autoResume/sharing.test.ts b/apps/server/src/t3x/autoResume/sharing.test.ts new file mode 100644 index 00000000000..cf208248ee2 --- /dev/null +++ b/apps/server/src/t3x/autoResume/sharing.test.ts @@ -0,0 +1,85 @@ +// @effect-diagnostics nodeBuiltinImport:off +/** + * Pins the layer-memoisation assumption that the auto-resume UI depends on. + * + * `t3x/index.ts` gives the reactor (`T3xLayerLive`) and the HTTP route (`T3xRoutesLive`) + * their store by `Layer.provide`-ing the *same* `AutoResumeStoreLive` value to each, + * independently. That is deliberate — leaving `AutoResumeStore` as an open requirement on + * the route would widen upstream's `makeRoutesLayer` signature and break its tests. + * + * It only works because Effect memoises layer construction per build, so both consumers + * receive one instance. If that ever stopped holding, the route would mutate its own + * in-memory copy of the state while the reactor kept reading a stale one: toggling + * auto-resume off in the UI would appear to work and the reactor would resume anyway. + * That failure is completely silent, hence this test. + * + * SCOPE, stated honestly: this builds its own store layer and two probe services rather + * than importing `T3xLayerLive` / `T3xRoutesLive`. It therefore guards the *assumption* + * (one layer value provided to two independent consumers yields one instance) for the + * composition shape `t3x/index.ts` uses — it does not exercise that file's actual graph, + * and would stay green if someone gave each consumer its own store layer. Importing the + * real layers here is not practical: both are `Layer.provide`d shut precisely so they + * leak no requirement, so neither exposes the store for a test to compare. + */ +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { assert, describe, it } from "@effect/vitest"; +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 NodePath from "node:path"; + +import { AutoResumeStore, type AutoResumeStoreShape, makeAutoResumeStore } from "./state.ts"; + +// Two independent consumers, mirroring the reactor and the route. +class ProbeA extends Context.Service()( + "t3/t3x/autoResume/sharing.test/ProbeA", +) {} +class ProbeB extends Context.Service()( + "t3/t3x/autoResume/sharing.test/ProbeB", +) {} + +describe("AutoResumeStore layer sharing", () => { + it("hands the same store instance to two consumers that each provide it independently", () => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const root = yield* fs.makeTempDirectoryScoped({ prefix: "t3x-share-" }); + const statePath = NodePath.join(root, "state.json"); + + // Exactly the shape used in t3x/index.ts: ONE layer value, provided to each consumer. + const StoreLive = Layer.effect(AutoResumeStore, makeAutoResumeStore(statePath)); + + const ProbeALive = Layer.effect( + ProbeA, + Effect.gen(function* () { + return { store: yield* AutoResumeStore }; + }), + ).pipe(Layer.provide(StoreLive)); + + const ProbeBLive = Layer.effect( + ProbeB, + Effect.gen(function* () { + return { store: yield* AutoResumeStore }; + }), + ).pipe(Layer.provide(StoreLive)); + + yield* Effect.gen(function* () { + const a = yield* ProbeA; + const b = yield* ProbeB; + + assert.strictEqual(a.store, b.store, "both consumers must share one store instance"); + + // Behavioural proof, not just reference equality: a write through one consumer is + // immediately visible through the other. This is the property the UI depends on. + yield* a.store.setEnabled("thread-a", false); + assert.strictEqual( + (yield* b.store.getThread("thread-a")).enabled, + false, + "a toggle written via one consumer must be visible to the other", + ); + + yield* b.store.setOverridePrompt("thread-a", "from b"); + assert.strictEqual((yield* a.store.getThread("thread-a")).overridePrompt, "from b"); + }).pipe(Effect.provide(Layer.merge(ProbeALive, ProbeBLive))); + }).pipe(Effect.scoped, Effect.provide(NodeServices.layer), Effect.runPromise)); +}); diff --git a/apps/server/src/t3x/autoResume/state.test.ts b/apps/server/src/t3x/autoResume/state.test.ts index a5c769a0228..098e0f68ba1 100644 --- a/apps/server/src/t3x/autoResume/state.test.ts +++ b/apps/server/src/t3x/autoResume/state.test.ts @@ -127,4 +127,123 @@ describe("AutoResumeStore", () => { const stat = yield* fs.stat(dirPath); assert.strictEqual(stat.type, "Directory"); }).pipe(Effect.scoped, Effect.provide(NodeServices.layer), Effect.runPromise)); + + it("defaults enabled to true for a thread that has no record yet", () => + withStore((store) => + Effect.gen(function* () { + const record = yield* store.getThread("never-seen"); + assert.strictEqual(record.enabled, true); + assert.strictEqual(record.overridePrompt, null); + }), + )); + + // `threads` is a plain object, so these ids resolve on Object.prototype and are truthy. + // A `?? EMPTY_RECORD` lookup therefore returns a prototype method typed as a + // ThreadRecord: reads blow up on `record.pending`, and writes persist a record with no + // required keys, which fails the whole-file decode on the next boot and wipes every + // pending resume. The HTTP route takes threadId straight from the caller, so this is + // reachable input, not a curiosity. + for (const hostile of ["constructor", "toString", "valueOf", "hasOwnProperty", "__proto__"]) { + it(`treats the prototype-chain threadId "${hostile}" as an absent record`, () => + withStore((store) => + Effect.gen(function* () { + const record = yield* store.getThread(hostile); + assert.strictEqual(record.enabled, true); + assert.strictEqual(record.pending, null, "must be null, not undefined"); + assert.strictEqual(record.overridePrompt, null); + assert.deepStrictEqual([...record.firedAtMs], []); + }), + )); + } + + it("a write under a prototype-chain threadId does not corrupt other threads' state", () => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const root = yield* fs.makeTempDirectoryScoped({ prefix: "t3x-state-" }); + const path = NodePath.join(root, "state.json"); + + const store1 = yield* makeAutoResumeStore(path); + yield* store1.schedule(pending("real-thread")); + yield* store1.recordFired("other-thread", 42); + + // The hostile write must persist a well-formed record like any other. + yield* store1.setEnabled("__proto__", false); + yield* store1.setOverridePrompt("constructor", "x"); + + // Everything must still decode after a restart — the failure mode this guards is a + // whole-file decode failure collapsing to EMPTY_STATE. + const store2 = yield* makeAutoResumeStore(path); + assert.strictEqual( + (yield* store2.listPending).length, + 1, + "the real thread's pending resume must survive", + ); + assert.strictEqual( + yield* store2.countFiredSince("other-thread", 0), + 1, + "fired history behind the 24h cap must survive", + ); + assert.strictEqual((yield* store2.getThread("__proto__")).enabled, false); + }).pipe(Effect.scoped, Effect.provide(NodeServices.layer), Effect.runPromise)); + + it("setEnabled and setOverridePrompt round-trip and survive a restart", () => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const root = yield* fs.makeTempDirectoryScoped({ prefix: "t3x-state-" }); + const path = NodePath.join(root, "state.json"); + + const store1 = yield* makeAutoResumeStore(path); + yield* store1.setEnabled("thread-a", false); + yield* store1.setOverridePrompt("thread-a", "keep going"); + + const before = yield* store1.getThread("thread-a"); + assert.strictEqual(before.enabled, false); + assert.strictEqual(before.overridePrompt, "keep going"); + + const store2 = yield* makeAutoResumeStore(path); + const after = yield* store2.getThread("thread-a"); + assert.strictEqual(after.enabled, false); + assert.strictEqual(after.overridePrompt, "keep going"); + + // Clearing the override falls back to the configured default. + yield* store2.setOverridePrompt("thread-a", null); + assert.strictEqual((yield* store2.getThread("thread-a")).overridePrompt, null); + // …and toggling back on does not disturb the override. + yield* store2.setEnabled("thread-a", true); + assert.strictEqual((yield* store2.getThread("thread-a")).enabled, true); + }).pipe(Effect.scoped, Effect.provide(NodeServices.layer), Effect.runPromise)); + + it("decodes a pre-`enabled` state file as enabled without dropping pending or fired history", () => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const root = yield* fs.makeTempDirectoryScoped({ prefix: "t3x-state-" }); + const path = NodePath.join(root, "legacy.json"); + + // Exactly the shape written before `enabled` existed: no `enabled` key at all. + // `enabled` is a *required* key on the Type side, so if it were not declared with a + // decoding default the whole-file decode would fail — and the boot path turns a decode + // failure into EMPTY_STATE, silently destroying the pending resume + fired history + // below. This test is the regression guard for that data-loss path. + // Written as a literal so the test pins the exact historical on-disk format + // rather than whatever the current encoder happens to produce. + yield* fs.writeFileString( + path, + `{"version":1,"threads":{"thread-a":{` + + `"pending":{"threadId":"thread-a","resumeAtMs":4242,"reason":"five_hour",` + + `"scheduledAtMs":7,"baseline":{"newestUserMessageId":"m1","latestTurnId":"t1"}},` + + `"firedAtMs":[111,222],"overridePrompt":"legacy text"}}}`, + ); + + const store = yield* makeAutoResumeStore(path); + + const record = yield* store.getThread("thread-a"); + assert.strictEqual(record.enabled, true, "missing `enabled` must decode as on"); + assert.strictEqual(record.overridePrompt, "legacy text"); + + // The rest of the record must survive intact (i.e. the file did not collapse to empty). + const list = yield* store.listPending; + assert.strictEqual(list.length, 1); + assert.strictEqual(list[0]!.resumeAtMs, 4_242); + assert.strictEqual(yield* store.countFiredSince("thread-a", 0), 2); + }).pipe(Effect.scoped, Effect.provide(NodeServices.layer), Effect.runPromise)); }); diff --git a/apps/server/src/t3x/autoResume/state.ts b/apps/server/src/t3x/autoResume/state.ts index e71c970394d..74384e92d55 100644 --- a/apps/server/src/t3x/autoResume/state.ts +++ b/apps/server/src/t3x/autoResume/state.ts @@ -40,8 +40,17 @@ export type PendingResume = typeof PendingResume.Type; const ThreadRecord = Schema.Struct({ pending: Schema.NullOr(PendingResume), firedAtMs: Schema.Array(Schema.Number), - /** Optional per-thread resume prompt override (settable by a future UI/CLI). */ + /** Optional per-thread resume prompt override (settable by the UI). */ overridePrompt: Schema.NullOr(Schema.String), + /** + * Per-thread auto-resume switch (default on). + * + * Optional *on disk* with a decode-time default of `true`: state files written + * before this field existed must still decode. A missing **required** key would + * fail the whole-file decode, and the boot path turns a decode failure into + * `EMPTY_STATE` — silently dropping every pending resume and the fired history. + */ + enabled: Schema.Boolean.pipe(Schema.withDecodingDefaultKey(Effect.succeed(true))), }); type ThreadRecord = typeof ThreadRecord.Type; @@ -56,6 +65,7 @@ const EMPTY_RECORD: ThreadRecord = { pending: null, firedAtMs: [], overridePrompt: null, + enabled: true, }; const decodeState = Schema.decodeUnknownEffect(Schema.fromJsonString(AutoResumeState)); @@ -69,6 +79,13 @@ export interface AutoResumeStoreShape { readonly recordFired: (threadId: string, atMs: number) => Effect.Effect; /** How many resumes fired for a thread since `sinceMs` (for caps + backoff). */ readonly countFiredSince: (threadId: string, sinceMs: number) => Effect.Effect; + /** Turn auto-resume on/off for a single thread (the UI toggle). */ + readonly setEnabled: (threadId: string, enabled: boolean) => Effect.Effect; + /** Set the per-thread resume text; `null` falls back to the configured default. */ + readonly setOverridePrompt: ( + threadId: string, + overridePrompt: string | null, + ) => Effect.Effect; } export class AutoResumeStore extends Context.Service()( @@ -143,8 +160,19 @@ export const makeAutoResumeStore = ( return persist(next).pipe(Effect.as(next)); }); + // `Object.hasOwn`, NOT `state.threads[threadId] ?? EMPTY_RECORD`. + // + // `threads` is a plain object, so a threadId of `constructor` / `toString` / + // `valueOf` / `__proto__` resolves on Object.prototype and is *truthy* — `??` never + // falls through, and the caller gets a prototype method typed as a ThreadRecord. + // Two things then break, and the HTTP route makes threadId caller-controlled: + // * reads dereference `record.pending` (undefined, not null) -> TypeError -> 500; + // * writes spread a prototype object, which has no own enumerable properties, and + // persist `{"enabled":false}` — missing required keys. The next boot fails the + // WHOLE-file decode, and that path collapses to EMPTY_STATE, silently destroying + // every pending resume and the fired history behind the 24h cap. const recordFor = (state: AutoResumeState, threadId: string): ThreadRecord => - state.threads[threadId] ?? EMPTY_RECORD; + Object.hasOwn(state.threads, threadId) ? state.threads[threadId]! : EMPTY_RECORD; return { listPending: SynchronizedRef.get(ref).pipe( @@ -202,5 +230,23 @@ export const makeAutoResumeStore = ( (state) => recordFor(state, threadId).firedAtMs.filter((t) => t >= sinceMs).length, ), ), + + setEnabled: (threadId, enabled) => + mutate((state) => ({ + ...state, + threads: { + ...state.threads, + [threadId]: { ...recordFor(state, threadId), enabled }, + }, + })), + + setOverridePrompt: (threadId, overridePrompt) => + mutate((state) => ({ + ...state, + threads: { + ...state.threads, + [threadId]: { ...recordFor(state, threadId), overridePrompt }, + }, + })), } satisfies AutoResumeStoreShape; }); diff --git a/apps/server/src/t3x/index.ts b/apps/server/src/t3x/index.ts index 174523dc418..f98bc956f4d 100644 --- a/apps/server/src/t3x/index.ts +++ b/apps/server/src/t3x/index.ts @@ -19,6 +19,7 @@ import * as Layer from "effect/Layer"; import * as Path from "effect/Path"; import { ServerConfig } from "../config.ts"; +import { autoResumeRouteLayer } from "./autoResume/http.ts"; import { AutoResumeReactorLive } from "./autoResume/Reactor.ts"; import { AutoResumeStore, makeAutoResumeStore } from "./autoResume/state.ts"; @@ -40,3 +41,29 @@ const AutoResumeStoreLive = Layer.effect( * this one layer. */ export const T3xLayerLive = AutoResumeReactorLive.pipe(Layer.provide(AutoResumeStoreLive)); + +/** + * All fork-local HTTP routes, fanned in here for the same reason as `T3xLayerLive`: + * `server.ts` adds ONE entry to its route list and imports it from this same module, so + * adding future routes never grows the upstream seam. + * + * The store is `provide`d here rather than left as an open requirement. That matters for + * two reasons: + * + * 1. **No seam leak.** An unsatisfied `AutoResumeStore` would surface in the type of + * upstream's `makeRoutesLayer` and fail every existing `server.test.ts` case — a fork + * change must never widen an upstream signature. + * 2. **Still one instance.** Both this layer and `T3xLayerLive` provide the *same* + * module-level `AutoResumeStoreLive` value. Effect memoises layer construction per + * build keyed on layer identity, and `provideWith` / `mergeAllEffect` / + * `HttpRouter.serve` all thread the same `MemoMap`, so the reactor and the route share + * one store rather than racing two in-memory copies over a single file. + * + * If that ever stopped holding, the route would mutate its own copy while the reactor + * read a stale one: switching auto-resume off in the UI would look like it worked and + * the thread would resume anyway. `autoResume/sharing.test.ts` pins the memoisation + * behaviour this relies on — note it exercises the composition *shape* with its own + * probes rather than importing these two layers, so treat it as a guard on the + * assumption, not proof of this file's graph. + */ +export const T3xRoutesLive = autoResumeRouteLayer.pipe(Layer.provide(AutoResumeStoreLive)); diff --git a/apps/web/src/routes/_chat.$environmentId.$threadId.tsx b/apps/web/src/routes/_chat.$environmentId.$threadId.tsx index 7dc6702b4ec..ffbd45a8067 100644 --- a/apps/web/src/routes/_chat.$environmentId.$threadId.tsx +++ b/apps/web/src/routes/_chat.$environmentId.$threadId.tsx @@ -9,6 +9,7 @@ import { SidebarInset } from "~/components/ui/sidebar"; import { useEnvironmentThreadRefs, useThreadDetail, useThreadShell } from "../state/entities"; import { useEnvironmentQuery } from "../state/query"; import { environmentShell } from "../state/shell"; +import { AutoResumeOverlay } from "../t3x/AutoResumeOverlay"; function ChatThreadRouteView() { const navigate = useNavigate(); @@ -68,6 +69,7 @@ function ChatThreadRouteView() { threadId={threadRef.threadId} routeKind="server" /> + ); } diff --git a/apps/web/src/t3x/AutoResumeOverlay.tsx b/apps/web/src/t3x/AutoResumeOverlay.tsx new file mode 100644 index 00000000000..ee4e6bfcf0d --- /dev/null +++ b/apps/web/src/t3x/AutoResumeOverlay.tsx @@ -0,0 +1,382 @@ +import * as Effect from "effect/Effect"; +import * as ManagedRuntime from "effect/ManagedRuntime"; +import { HttpClient, HttpClientRequest } from "effect/unstable/http"; +import { ChevronDownIcon, ChevronUpIcon } from "lucide-react"; +import { useCallback, useEffect, useId, useRef, useState } from "react"; + +import { Label } from "~/components/ui/label"; +import { Switch } from "~/components/ui/switch"; +import { Textarea } from "~/components/ui/textarea"; +import { primaryEnvironmentHttpLayer } from "~/environments/primary/httpLayer"; +import { resolvePrimaryEnvironmentHttpUrl } from "~/environments/primary/target"; +import { cn } from "~/lib/utils"; + +const AUTO_RESUME_PATH = "/api/t3x/auto-resume"; +const POLL_INTERVAL_MS = 30_000; +const PROMPT_DEBOUNCE_MS = 600; + +export interface AutoResumeThreadRef { + readonly environmentId: string; + readonly threadId: string; +} + +interface AutoResumePending { + readonly resumeAtMs: number; + readonly reason: string; +} + +interface AutoResumeState { + readonly enabled: boolean; + readonly overridePrompt: string | null; + readonly pending: AutoResumePending | null; +} + +interface AutoResumeWrite { + readonly threadId: string; + readonly enabled?: boolean; + readonly overridePrompt?: string | null; +} + +/** + * `/api/t3x/auto-resume` is a raw route, so it has to be called the same way + * `observability/clientTracing.ts` calls `/api/observability/v1/traces`: build the URL with + * `resolvePrimaryEnvironmentHttpUrl` and run over `primaryEnvironmentHttpLayer`, which is the only + * place in the web app that knows how to authenticate the primary environment (session cookies for + * a same-origin browser primary, desktop bearer token otherwise). + */ +const autoResumeRuntime = ManagedRuntime.make(primaryEnvironmentHttpLayer); + +function isJsonObject(value: unknown): value is Record { + return typeof value === "object" && value !== null && !Array.isArray(value); +} + +function parseAutoResumePending(value: unknown): AutoResumePending | null { + if (!isJsonObject(value)) { + return null; + } + const { resumeAtMs, reason } = value; + if (typeof resumeAtMs !== "number" || !Number.isFinite(resumeAtMs)) { + return null; + } + return { resumeAtMs, reason: typeof reason === "string" ? reason : "" }; +} + +function parseAutoResumeState(value: unknown): AutoResumeState | null { + if (!isJsonObject(value)) { + return null; + } + const { enabled, overridePrompt, pending } = value; + if (typeof enabled !== "boolean") { + return null; + } + return { + enabled, + overridePrompt: + typeof overridePrompt === "string" && overridePrompt !== "" ? overridePrompt : null, + pending: parseAutoResumePending(pending), + }; +} + +/** + * Auto-resume is an enhancement layered over the thread view: any failure (401, route not deployed, + * offline) resolves to `null` so the overlay simply disappears instead of degrading the chat. + */ +async function runAutoResumeRequest( + effect: Effect.Effect, +): Promise { + try { + return await autoResumeRuntime.runPromise(effect); + } catch { + return null; + } +} + +function readAutoResumeState(threadId: string): Promise { + const url = resolvePrimaryEnvironmentHttpUrl(AUTO_RESUME_PATH, { threadId }); + return runAutoResumeRequest( + Effect.gen(function* () { + const response = yield* HttpClient.get(url); + if (response.status !== 200) { + return null; + } + return parseAutoResumeState(yield* response.json); + }), + ); +} + +function writeAutoResumeState(body: AutoResumeWrite): Promise { + const url = resolvePrimaryEnvironmentHttpUrl(AUTO_RESUME_PATH); + return runAutoResumeRequest( + Effect.gen(function* () { + const response = yield* HttpClient.execute( + HttpClientRequest.bodyJsonUnsafe(HttpClientRequest.post(url), body), + ); + if (response.status !== 200) { + return null; + } + return parseAutoResumeState(yield* response.json); + }), + ); +} + +const nextAttemptFormatter = new Intl.DateTimeFormat(undefined, { + hour: "numeric", + minute: "2-digit", +}); + +function formatNextAttempt(resumeAtMs: number): string { + return nextAttemptFormatter.format(new Date(resumeAtMs)); +} + +export function formatAutoResumeStatus(state: AutoResumeState): string { + if (!state.enabled) { + return "Auto-resume: off"; + } + if (state.pending === null) { + return "Auto-resume: on"; + } + return `Auto-resume: on · next attempt ~${formatNextAttempt(state.pending.resumeAtMs)}`; +} + +function normalizeOverridePrompt(value: string): string | null { + return value.trim() === "" ? null : value; +} + +interface AutoResumeOverlayProps { + readonly threadRef: AutoResumeThreadRef; +} + +/** + * Floating per-thread control for auto-resume (auto-continuing a thread after a usage-limit pause). + * Renders nothing until the server confirms the feature is reachable for this thread. + */ +export function AutoResumeOverlay({ threadRef }: AutoResumeOverlayProps) { + const threadId = threadRef.threadId; + const switchId = useId(); + const [state, setState] = useState(null); + const [promptDraft, setPromptDraft] = useState(null); + const [expanded, setExpanded] = useState(false); + + const mountedRef = useRef(true); + const threadIdRef = useRef(threadId); + const inFlightWritesRef = useRef(0); + const promptTimerRef = useRef(null); + // The edit a debounced write is waiting to send, so it can be flushed rather than + // dropped when the thread changes mid-debounce. + const pendingPromptRef = useRef<{ threadId: string; value: string } | null>(null); + + useEffect(() => { + threadIdRef.current = threadId; + }, [threadId]); + + useEffect(() => { + mountedRef.current = true; + return () => { + mountedRef.current = false; + if (promptTimerRef.current !== null) { + window.clearTimeout(promptTimerRef.current); + promptTimerRef.current = null; + } + }; + }, []); + + useEffect(() => { + let cancelled = false; + setState(null); + setPromptDraft(null); + setExpanded(false); + + const load = async () => { + // Never let a poll response stomp an optimistic value that has not round-tripped yet. + if (inFlightWritesRef.current > 0) { + return; + } + const next = await readAutoResumeState(threadId); + if (cancelled || next === null) { + return; + } + setState(next); + setPromptDraft((current) => current ?? next.overridePrompt ?? ""); + }; + + void load(); + const intervalId = window.setInterval(() => void load(), POLL_INTERVAL_MS); + const handleFocus = () => void load(); + window.addEventListener("focus", handleFocus); + + return () => { + cancelled = true; + window.clearInterval(intervalId); + window.removeEventListener("focus", handleFocus); + }; + }, [threadId]); + + const applyWriteResult = useCallback( + (requestThreadId: string, fallback: AutoResumeState | null, next: AutoResumeState | null) => { + if (!mountedRef.current || threadIdRef.current !== requestThreadId) { + return; + } + if (next === null && fallback === null) { + return; + } + setState(next ?? fallback); + }, + [], + ); + + /** + * Issues a write and guarantees the in-flight counter is released exactly once. + * + * The counter gates polling, so a leaked increment stops the overlay refreshing for the + * life of the component. `writeAutoResumeState` swallows its own async failures, but URL + * construction happens before the promise exists and can throw synchronously — hence the + * try/catch around the release rather than relying on `.then` alone. + */ + const submitWrite = useCallback( + (requestThreadId: string, fallback: AutoResumeState | null, body: AutoResumeWrite) => { + inFlightWritesRef.current += 1; + let released = false; + const release = () => { + if (released) return; + released = true; + inFlightWritesRef.current -= 1; + }; + try { + void writeAutoResumeState(body).then( + (next) => { + release(); + applyWriteResult(requestThreadId, fallback, next); + }, + () => release(), + ); + } catch { + release(); + } + }, + [applyWriteResult], + ); + + /** + * Sends a debounced resume-message edit immediately. + * + * Called when the thread changes or the overlay unmounts. Without it the pending timer + * is simply dropped: typing in one thread and switching within the debounce window lost + * the edit silently, because the next keystroke in the new thread cleared the old timer. + * Fire-and-forget — it deliberately carries the ORIGINATING threadId and never touches + * state, since the component may be unmounting. + */ + const flushPendingPrompt = useCallback(() => { + if (promptTimerRef.current !== null) { + window.clearTimeout(promptTimerRef.current); + promptTimerRef.current = null; + } + const pending = pendingPromptRef.current; + pendingPromptRef.current = null; + if (pending === null) { + return; + } + try { + void writeAutoResumeState({ + threadId: pending.threadId, + overridePrompt: normalizeOverridePrompt(pending.value), + }); + } catch { + // A dropped flush is not worth surfacing; the value is still in the textbox. + } + }, []); + + // Flush on thread change AND on unmount (this cleanup runs for both). + useEffect(() => () => flushPendingPrompt(), [threadId, flushPendingPrompt]); + + const handleToggle = useCallback( + (checked: boolean) => { + const previous = state; + if (previous === null) { + return; + } + // Optimistic: `applyWriteResult` puts `previous` back if the write fails. + setState({ ...previous, enabled: checked }); + submitWrite(threadId, previous, { threadId, enabled: checked }); + }, + [state, submitWrite, threadId], + ); + + const handlePromptChange = useCallback( + (value: string) => { + setPromptDraft(value); + pendingPromptRef.current = { threadId, value }; + if (promptTimerRef.current !== null) { + window.clearTimeout(promptTimerRef.current); + } + promptTimerRef.current = window.setTimeout(() => { + promptTimerRef.current = null; + pendingPromptRef.current = null; + submitWrite(threadId, null, { + threadId, + overridePrompt: normalizeOverridePrompt(value), + }); + }, PROMPT_DEBOUNCE_MS); + }, + [submitWrite, threadId], + ); + + if (state === null) { + return null; + } + + const pending = state.pending; + + return ( +
+ + + {expanded ? ( +
+
+ + +
+ + {pending !== null ? ( +

+ {pending.reason === "" ? "Paused" : `Paused: ${pending.reason}`} · next attempt ~ + {formatNextAttempt(pending.resumeAtMs)} +

+ ) : null} + +