From 90cc6e71b53c38cd8c96db02be9d57423eff601e Mon Sep 17 00:00:00 2001 From: Raj D <25481060+radroid@users.noreply.github.com> Date: Fri, 24 Jul 2026 10:16:31 -0400 Subject: [PATCH 1/6] feat(t3x): per-thread auto-resume switch (server state + gating) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Phase 1 of docs/superpowers/specs/2026-07-24-t3x-autoresume-ui-design.md. state.ts - ThreadRecord gains `enabled`, plus `setEnabled` / `setOverridePrompt` mutations (overridePrompt was already stored but had no setter). - `enabled` is declared with `Schema.withDecodingDefaultKey(Effect.succeed(true))`, NOT as a required key. The boot path turns a decode failure into EMPTY_STATE, so a required key missing from an older state file would silently destroy every pending resume and the fired history. A regression test pins the legacy on-disk format. Note: the design doc specified `Schema.optionalWith`, which does not exist in this repo's Effect (4.0.0-beta.78). `withDecodingDefaultKey` is the equivalent there. Reactor.ts - Detection: a disabled thread never schedules, and posts no timeline note — the user turned it off deliberately, so an activity row would be noise. - fireOne: re-reads the record instead of trusting the value seen at scheduling time, so a thread switched off *mid-wait* cancels rather than fires, and does not burn one of its 24h attempts. Tests 50 -> 52 green; server typecheck clean. Both new Reactor tests were verified to FAIL when the gates are neutered, so they are real guards rather than vacuous. Co-Authored-By: Claude Opus 4.8 --- .../server/src/t3x/autoResume/Reactor.test.ts | 47 +++++++++++++ apps/server/src/t3x/autoResume/Reactor.ts | 18 +++++ apps/server/src/t3x/autoResume/state.test.ts | 70 +++++++++++++++++++ apps/server/src/t3x/autoResume/state.ts | 37 +++++++++- 4 files changed, 171 insertions(+), 1 deletion(-) 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/state.test.ts b/apps/server/src/t3x/autoResume/state.test.ts index a5c769a0228..8a66ef5b349 100644 --- a/apps/server/src/t3x/autoResume/state.test.ts +++ b/apps/server/src/t3x/autoResume/state.test.ts @@ -127,4 +127,74 @@ 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); + }), + )); + + 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..1fc0496c491 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()( @@ -202,5 +219,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; }); From cfe4e6293c25e2f24fad7a92b124d0d4e78fc405 Mon Sep 17 00:00:00 2001 From: Raj D <25481060+radroid@users.noreply.github.com> Date: Fri, 24 Jul 2026 10:39:44 -0400 Subject: [PATCH 2/6] feat(t3x): authenticated /api/t3x/auto-resume route MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Phase 2 of docs/superpowers/specs/2026-07-24-t3x-autoresume-ui-design.md. GET /api/t3x/auto-resume?threadId=… -> { enabled, overridePrompt, pending } POST /api/t3x/auto-resume -> same shape, after applying the write A raw route, not a WS-RPC method: an RPC would mean editing @t3tools/contracts, ws.ts and its scope map — several hot upstream files. This costs one additive line in server.ts's route list, and it is imported from t3x/index.ts alongside the existing T3xLayerLive, so server.ts gains no new import line. Auth mirrors the module-private `authenticateRawRouteWithScope` used by the OTLP proxy route (operate scope, since these endpoints change scheduling behaviour). Importing it would require exporting it from an upstream file, so the fork replicates it — recorded as a logic mirror in SEAMS.md. The layer shape here is load-bearing. Handlers close over the store resolved by `Layer.unwrap` at layer-construction time rather than `yield*`-ing it from context. Doing the latter propagated an `AutoResumeStore` requirement out through HttpRouter into the type of upstream's `makeRoutesLayer`, breaking 222 checks in server.test.ts and 34 in bin.test.ts. A fork change must never widen an upstream signature. Server typecheck is back to 0 errors. Both T3xLayerLive and T3xRoutesLive `Layer.provide` the same AutoResumeStoreLive value, so Effect's per-build memoisation gives the reactor and the route ONE store. sharing.test.ts pins that: if it regressed, the route would mutate its own copy while the reactor read a stale one — the UI toggle would look like it worked and the thread would resume anyway, silently. Tests 52 -> 53 green. Co-Authored-By: Claude Opus 4.8 --- apps/server/src/server.ts | 3 +- apps/server/src/t3x/autoResume/http.ts | 165 ++++++++++++++++++ .../server/src/t3x/autoResume/sharing.test.ts | 77 ++++++++ apps/server/src/t3x/index.ts | 19 ++ 4 files changed, 263 insertions(+), 1 deletion(-) create mode 100644 apps/server/src/t3x/autoResume/http.ts create mode 100644 apps/server/src/t3x/autoResume/sharing.test.ts 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/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..90d451091bb --- /dev/null +++ b/apps/server/src/t3x/autoResume/sharing.test.ts @@ -0,0 +1,77 @@ +// @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. + */ +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/index.ts b/apps/server/src/t3x/index.ts index 174523dc418..8b8c6657914 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,21 @@ 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* + * `AutoResumeStoreLive` value, and Effect memoises layer construction per build, so + * the reactor and the route share a single store rather than racing two in-memory + * copies over one file. `index.test.ts` pins that behaviour. + */ +export const T3xRoutesLive = autoResumeRouteLayer.pipe(Layer.provide(AutoResumeStoreLive)); From 39ca69c5b15c423b76bd39a3820bcd0c2d349125 Mon Sep 17 00:00:00 2001 From: Raj D <25481060+radroid@users.noreply.github.com> Date: Fri, 24 Jul 2026 10:42:04 -0400 Subject: [PATCH 3/6] feat(t3x): per-thread auto-resume overlay in the thread view MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Phase 3 of docs/superpowers/specs/2026-07-24-t3x-autoresume-ui-design.md. A collapsed status pill in the thread view ("Auto-resume: on · next attempt ~3:47 PM") that expands to a switch and a resume-message textbox, backed by GET/POST /api/t3x/auto-resume. Web seam is exactly +1 import and +1 JSX element in the server-thread route, rendered as a sibling of inside . Everything else is new code under apps/web/src/t3x/. Requests go through `primaryEnvironmentHttpLayer` rather than a hand-rolled fetch. This is not incidental: the same web bundle runs both in the browser and inside the Electron renderer (t3code://app), and those authenticate differently — session cookies with credentials:"include" for a same-origin browser, a desktop bearer token with credentials:"omit" otherwise. The deciding helper (`isSameOriginBrowserPrimary`) is module-private, so a hand-rolled fetch would have to duplicate it; getting it wrong would 401 in the macOS app only, and because the overlay degrades silently the control would simply never appear there. Any failure (401, route absent, offline) resolves to null and renders nothing, so a broken auto-resume backend can never degrade the chat view. Positioned top-right: the composer is itself a full-width absolutely-positioned overlay, so a bottom-right card would sit on top of it. A poll is skipped while a write is unacked so it cannot stomp an optimistic value, and a thread-id ref stops a late response from a previous thread being applied to the current one. Note: ManagedRuntime is created at module level (clientTracing.ts builds its own inside a function). Deliberate — `make` is lazy, so this shares one HTTP client across thread views instead of rebuilding per mount. Web typecheck clean. Co-Authored-By: Claude Opus 4.8 --- .../routes/_chat.$environmentId.$threadId.tsx | 2 + apps/web/src/t3x/AutoResumeOverlay.tsx | 320 ++++++++++++++++++ 2 files changed, 322 insertions(+) create mode 100644 apps/web/src/t3x/AutoResumeOverlay.tsx 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..dc72d136d1b --- /dev/null +++ b/apps/web/src/t3x/AutoResumeOverlay.tsx @@ -0,0 +1,320 @@ +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); + + 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) => { + inFlightWritesRef.current -= 1; + if (!mountedRef.current || threadIdRef.current !== requestThreadId) { + return; + } + if (next === null && fallback === null) { + return; + } + setState(next ?? fallback); + }, + [], + ); + + 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 }); + inFlightWritesRef.current += 1; + void writeAutoResumeState({ threadId, enabled: checked }).then((next) => { + applyWriteResult(threadId, previous, next); + }); + }, + [applyWriteResult, state, threadId], + ); + + const handlePromptChange = useCallback( + (value: string) => { + setPromptDraft(value); + if (promptTimerRef.current !== null) { + window.clearTimeout(promptTimerRef.current); + } + promptTimerRef.current = window.setTimeout(() => { + promptTimerRef.current = null; + inFlightWritesRef.current += 1; + void writeAutoResumeState({ + threadId, + overridePrompt: normalizeOverridePrompt(value), + }).then((next) => { + applyWriteResult(threadId, null, next); + }); + }, PROMPT_DEBOUNCE_MS); + }, + [applyWriteResult, 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} + +