From f77d0a49caf1b735f8c237ba76624da9e6d6f98b Mon Sep 17 00:00:00 2001 From: Jules-Astier <123639449+Jules-Astier@users.noreply.github.com> Date: Wed, 22 Jul 2026 12:07:33 +0100 Subject: [PATCH 1/3] feat(codex): add host-managed wait tool --- .../provider/Layers/CodexDynamicTools.test.ts | 173 +++++++++++++++++ .../src/provider/Layers/CodexDynamicTools.ts | 150 +++++++++++++++ .../Layers/CodexSessionRuntime.test.ts | 75 +++++++- .../provider/Layers/CodexSessionRuntime.ts | 47 ++++- docs/project/deferred-thread-resume.md | 174 ++++++++++++++++++ docs/providers/codex.md | 12 ++ 6 files changed, 620 insertions(+), 11 deletions(-) create mode 100644 apps/server/src/provider/Layers/CodexDynamicTools.test.ts create mode 100644 apps/server/src/provider/Layers/CodexDynamicTools.ts create mode 100644 docs/project/deferred-thread-resume.md diff --git a/apps/server/src/provider/Layers/CodexDynamicTools.test.ts b/apps/server/src/provider/Layers/CodexDynamicTools.test.ts new file mode 100644 index 00000000000..21d87179349 --- /dev/null +++ b/apps/server/src/provider/Layers/CodexDynamicTools.test.ts @@ -0,0 +1,173 @@ +import * as NodeAssert from "node:assert/strict"; + +import { it } from "@effect/vitest"; +import * as Deferred from "effect/Deferred"; +import * as Effect from "effect/Effect"; +import * as Fiber from "effect/Fiber"; +import * as TestClock from "effect/testing/TestClock"; +import { describe } from "vite-plus/test"; +import type * as EffectCodexSchema from "effect-codex-app-server/schema"; + +import { + handleT3CodexDynamicToolCall, + makeT3CodexDynamicToolWaitRegistry, + T3_CODEX_DYNAMIC_TOOLS, + T3_CODEX_DYNAMIC_TOOL_NAMESPACE, + T3_CODEX_WAIT_MAX_DURATION_MS, + T3_CODEX_WAIT_MIN_DURATION_MS, + T3_CODEX_WAIT_TOOL_NAME, +} from "./CodexDynamicTools.ts"; + +function waitCall( + arguments_: unknown, + overrides: Partial = {}, +): EffectCodexSchema.DynamicToolCallParams { + return { + arguments: arguments_, + callId: "wait-call-1", + namespace: T3_CODEX_DYNAMIC_TOOL_NAMESPACE, + threadId: "provider-thread-1", + tool: T3_CODEX_WAIT_TOOL_NAME, + turnId: "provider-turn-1", + ...overrides, + }; +} + +describe("T3 Codex dynamic tools", () => { + it("publishes a bounded namespaced wait tool", () => { + const namespace = T3_CODEX_DYNAMIC_TOOLS[0]; + const wait = namespace?.tools[0]; + + NodeAssert.equal(namespace?.type, "namespace"); + NodeAssert.equal(namespace?.name, T3_CODEX_DYNAMIC_TOOL_NAMESPACE); + NodeAssert.equal(wait?.type, "function"); + NodeAssert.equal(wait?.name, T3_CODEX_WAIT_TOOL_NAME); + NodeAssert.deepStrictEqual(wait?.inputSchema, { + type: "object", + properties: { + durationMs: { + type: "integer", + minimum: T3_CODEX_WAIT_MIN_DURATION_MS, + maximum: T3_CODEX_WAIT_MAX_DURATION_MS, + description: "How long T3 should wait, in milliseconds.", + }, + reason: { + type: "string", + maxLength: 500, + description: "Optional short explanation of the external work being awaited.", + }, + }, + required: ["durationMs"], + additionalProperties: false, + }); + }); + + it.effect("does not complete before the requested duration", () => + Effect.gen(function* () { + const fiber = yield* handleT3CodexDynamicToolCall( + waitCall({ durationMs: T3_CODEX_WAIT_MIN_DURATION_MS, reason: "CI is running" }), + ).pipe(Effect.forkChild); + + yield* TestClock.adjust(T3_CODEX_WAIT_MIN_DURATION_MS - 1); + NodeAssert.equal(fiber.pollUnsafe(), undefined); + + yield* TestClock.adjust(1); + NodeAssert.deepStrictEqual(yield* Fiber.join(fiber), { + success: true, + contentItems: [ + { + type: "inputText", + text: `Wait completed after ${T3_CODEX_WAIT_MIN_DURATION_MS} ms.`, + }, + ], + }); + }).pipe(Effect.provide(TestClock.layer())), + ); + + it.effect("cancels an active wait", () => + Effect.gen(function* () { + const cancelled = yield* Deferred.make(); + const fiber = yield* handleT3CodexDynamicToolCall( + waitCall({ durationMs: T3_CODEX_WAIT_MAX_DURATION_MS }), + Deferred.await(cancelled), + ).pipe(Effect.forkChild); + + yield* Effect.yieldNow; + yield* Deferred.succeed(cancelled, undefined); + + NodeAssert.deepStrictEqual(yield* Fiber.join(fiber), { + success: false, + contentItems: [ + { + type: "inputText", + text: "Wait cancelled because the turn was interrupted or the session closed.", + }, + ], + }); + }).pipe(Effect.provide(TestClock.layer())), + ); + + it.effect("cancels waits by turn and cancels all remaining waits on shutdown", () => + Effect.gen(function* () { + const registry = yield* makeT3CodexDynamicToolWaitRegistry(); + const first = yield* registry + .handle( + waitCall( + { durationMs: T3_CODEX_WAIT_MAX_DURATION_MS }, + { callId: "wait-call-1", turnId: "provider-turn-1" }, + ), + ) + .pipe(Effect.forkChild); + const second = yield* registry + .handle( + waitCall( + { durationMs: T3_CODEX_WAIT_MAX_DURATION_MS }, + { callId: "wait-call-2", turnId: "provider-turn-2" }, + ), + ) + .pipe(Effect.forkChild); + + yield* Effect.yieldNow; + yield* registry.cancelTurn("provider-turn-1"); + + NodeAssert.equal((yield* Fiber.join(first)).success, false); + NodeAssert.equal(second.pollUnsafe(), undefined); + + yield* registry.cancelAll; + NodeAssert.equal((yield* Fiber.join(second)).success, false); + }).pipe(Effect.provide(TestClock.layer())), + ); + + it.effect("rejects malformed and out-of-range arguments without sleeping", () => + Effect.gen(function* () { + for (const arguments_ of [ + {}, + { durationMs: T3_CODEX_WAIT_MIN_DURATION_MS - 1 }, + { durationMs: T3_CODEX_WAIT_MAX_DURATION_MS + 1 }, + { durationMs: 1_000.5 }, + { durationMs: "1000" }, + ]) { + const response = yield* handleT3CodexDynamicToolCall(waitCall(arguments_)); + NodeAssert.equal(response.success, false); + NodeAssert.match( + response.contentItems[0]?.type === "inputText" ? response.contentItems[0].text : "", + /Invalid wait arguments/, + ); + } + }), + ); + + it.effect("rejects calls outside the T3 namespace", () => + Effect.gen(function* () { + const response = yield* handleT3CodexDynamicToolCall( + waitCall({ durationMs: T3_CODEX_WAIT_MIN_DURATION_MS }, { namespace: "other" }), + ); + + NodeAssert.equal(response.success, false); + NodeAssert.match( + response.contentItems[0]?.type === "inputText" ? response.contentItems[0].text : "", + /Unknown T3 Code dynamic tool/, + ); + }), + ); +}); diff --git a/apps/server/src/provider/Layers/CodexDynamicTools.ts b/apps/server/src/provider/Layers/CodexDynamicTools.ts new file mode 100644 index 00000000000..d09532cb5aa --- /dev/null +++ b/apps/server/src/provider/Layers/CodexDynamicTools.ts @@ -0,0 +1,150 @@ +import * as Deferred from "effect/Deferred"; +import * as Duration from "effect/Duration"; +import * as Effect from "effect/Effect"; +import * as Ref from "effect/Ref"; +import * as Result from "effect/Result"; +import * as Schema from "effect/Schema"; +import type * as EffectCodexSchema from "effect-codex-app-server/schema"; + +export const T3_CODEX_DYNAMIC_TOOL_NAMESPACE = "t3"; +export const T3_CODEX_WAIT_TOOL_NAME = "wait"; +export const T3_CODEX_WAIT_MIN_DURATION_MS = 1_000; +export const T3_CODEX_WAIT_MAX_DURATION_MS = 3_600_000; + +const T3CodexWaitArguments = Schema.Struct({ + durationMs: Schema.Int.check(Schema.isGreaterThanOrEqualTo(T3_CODEX_WAIT_MIN_DURATION_MS)).check( + Schema.isLessThanOrEqualTo(T3_CODEX_WAIT_MAX_DURATION_MS), + ), + reason: Schema.optionalKey(Schema.String.check(Schema.isMaxLength(500))), +}); +const decodeT3CodexWaitArguments = Schema.decodeUnknownEffect(T3CodexWaitArguments); + +export const T3_CODEX_DYNAMIC_TOOLS = [ + { + type: "namespace", + name: T3_CODEX_DYNAMIC_TOOL_NAMESPACE, + description: "T3 Code host lifecycle tools that can pause work without model polling.", + tools: [ + { + type: "function", + name: T3_CODEX_WAIT_TOOL_NAME, + description: + "Wait inside T3 Code without using model inference. Use this for an intentional idle period while external work is running, then check the work once after the wait completes. The wait only lasts while this live T3 provider session remains connected; it is cancelled if the turn is interrupted or the session closes.", + inputSchema: { + type: "object", + properties: { + durationMs: { + type: "integer", + minimum: T3_CODEX_WAIT_MIN_DURATION_MS, + maximum: T3_CODEX_WAIT_MAX_DURATION_MS, + description: "How long T3 should wait, in milliseconds.", + }, + reason: { + type: "string", + maxLength: 500, + description: "Optional short explanation of the external work being awaited.", + }, + }, + required: ["durationMs"], + additionalProperties: false, + }, + }, + ], + }, +] as const satisfies ReadonlyArray; + +function textResponse(success: boolean, text: string): EffectCodexSchema.DynamicToolCallResponse { + return { + success, + contentItems: [{ type: "inputText", text }], + }; +} + +export const handleT3CodexDynamicToolCall = Effect.fn("handleT3CodexDynamicToolCall")(function* ( + payload: EffectCodexSchema.DynamicToolCallParams, + cancelled: Effect.Effect = Effect.never, +): Effect.fn.Return { + if ( + payload.namespace !== T3_CODEX_DYNAMIC_TOOL_NAMESPACE || + payload.tool !== T3_CODEX_WAIT_TOOL_NAME + ) { + return textResponse( + false, + `Unknown T3 Code dynamic tool: ${payload.namespace ?? ""}.${payload.tool}`, + ); + } + + const decoded = yield* Effect.result(decodeT3CodexWaitArguments(payload.arguments)); + if (Result.isFailure(decoded)) { + return textResponse( + false, + `Invalid wait arguments. durationMs must be an integer from ${T3_CODEX_WAIT_MIN_DURATION_MS} through ${T3_CODEX_WAIT_MAX_DURATION_MS}.`, + ); + } + + const outcome = yield* Effect.sleep(Duration.millis(decoded.success.durationMs)).pipe( + Effect.as("elapsed" as const), + Effect.raceFirst(cancelled.pipe(Effect.as("cancelled" as const))), + ); + + return outcome === "elapsed" + ? textResponse(true, `Wait completed after ${decoded.success.durationMs} ms.`) + : textResponse(false, "Wait cancelled because the turn was interrupted or the session closed."); +}); + +interface PendingWait { + readonly turnId: string; + readonly cancelled: Deferred.Deferred; +} + +export interface T3CodexDynamicToolWaitRegistry { + readonly handle: ( + payload: EffectCodexSchema.DynamicToolCallParams, + ) => Effect.Effect; + readonly cancelTurn: (turnId: string) => Effect.Effect; + readonly cancelAll: Effect.Effect; +} + +export const makeT3CodexDynamicToolWaitRegistry = Effect.fn("makeT3CodexDynamicToolWaitRegistry")( + function* (): Effect.fn.Return { + const pendingRef = yield* Ref.make(new Map()); + + const cancelWhere = (predicate: (pending: PendingWait) => boolean) => + Ref.get(pendingRef).pipe( + Effect.flatMap((pendingWaits) => + Effect.forEach( + Array.from(pendingWaits.values()), + (pending) => + predicate(pending) + ? Deferred.succeed(pending.cancelled, undefined).pipe(Effect.ignore) + : Effect.void, + { discard: true }, + ), + ), + ); + + return { + handle: (payload) => + Effect.gen(function* () { + const cancelled = yield* Deferred.make(); + yield* Ref.update(pendingRef, (current) => { + const next = new Map(current); + next.set(payload.callId, { turnId: payload.turnId, cancelled }); + return next; + }); + + return yield* handleT3CodexDynamicToolCall(payload, Deferred.await(cancelled)).pipe( + Effect.ensuring( + Ref.update(pendingRef, (current) => { + const next = new Map(current); + next.delete(payload.callId); + return next; + }), + ), + ); + }), + cancelTurn: (turnId) => cancelWhere((pending) => pending.turnId === turnId), + cancelAll: cancelWhere(() => true), + }; + }, +); diff --git a/apps/server/src/provider/Layers/CodexSessionRuntime.test.ts b/apps/server/src/provider/Layers/CodexSessionRuntime.test.ts index 119fa36303a..2f89fd9cf35 100644 --- a/apps/server/src/provider/Layers/CodexSessionRuntime.test.ts +++ b/apps/server/src/provider/Layers/CodexSessionRuntime.test.ts @@ -14,6 +14,7 @@ import { CODEX_PLAN_MODE_DEVELOPER_INSTRUCTIONS, } from "../CodexDeveloperInstructions.ts"; import { codexSessionAppServerArgs } from "./codexLaunchArgs.ts"; +import { T3_CODEX_DYNAMIC_TOOLS } from "./CodexDynamicTools.ts"; import { buildTurnStartParams, hasConfiguredMcpServer, @@ -48,18 +49,22 @@ function makeThreadOpenResponse( modelProvider: "openai", approvalPolicy: "never", approvalsReviewer: "user", - sandbox: { type: "danger-full-access" }, + sandbox: { type: "dangerFullAccess" }, thread: { id: threadId, - createdAt: "2026-04-18T00:00:00.000Z", - source: { session: "cli" }, + cliVersion: "0.0.0-test", + createdAt: 1_776_470_400, + cwd: "/tmp/project", + ephemeral: false, + modelProvider: "openai", + preview: "", + sessionId: "session-1", + source: "appServer", turns: [], - status: { - state: "idle", - activeFlags: [], - }, + status: { type: "idle" }, + updatedAt: 1_776_470_400, }, - } as unknown as CodexRpc.ClientRequestResponsesByMethod["thread/start"]; + }; } describe("buildTurnStartParams", () => { @@ -365,11 +370,55 @@ describe("isRecoverableThreadResumeError", () => { }); describe("openCodexThread", () => { + it.effect("starts a fresh thread with the T3 dynamic tools", () => + Effect.gen(function* () { + const calls: Array<{ method: string; payload: unknown }> = []; + const client = { + raw: { + request: (method: string, payload?: unknown) => { + calls.push({ method, payload }); + return Effect.succeed(makeThreadOpenResponse("fresh-thread")); + }, + }, + request: ( + _method: M, + _payload: CodexRpc.ClientRequestParamsByMethod[M], + ) => Effect.die("typed request path should not be used for a fresh thread"), + }; + + const opened = yield* openCodexThread({ + client, + threadId: ThreadId.make("thread-1"), + runtimeMode: "full-access", + cwd: "/tmp/project", + requestedModel: "gpt-5.3-codex", + serviceTier: undefined, + resumeThreadId: undefined, + }); + + NodeAssert.equal(opened.thread.id, "fresh-thread"); + NodeAssert.equal(calls.length, 1); + const freshStart = calls[0]; + NodeAssert.ok(freshStart); + NodeAssert.equal(freshStart.method, "thread/start"); + NodeAssert.deepStrictEqual( + (freshStart.payload as { readonly dynamicTools?: unknown }).dynamicTools, + T3_CODEX_DYNAMIC_TOOLS, + ); + }), + ); + it.effect("falls back to thread/start when resume fails recoverably", () => Effect.gen(function* () { const calls: Array<{ method: "thread/start" | "thread/resume"; payload: unknown }> = []; const started = makeThreadOpenResponse("fresh-thread"); const client = { + raw: { + request: (method: string, payload?: unknown) => { + calls.push({ method: method as "thread/start", payload }); + return Effect.succeed(started); + }, + }, request: ( method: M, payload: CodexRpc.ClientRequestParamsByMethod[M], @@ -402,12 +451,22 @@ describe("openCodexThread", () => { calls.map((call) => call.method), ["thread/resume", "thread/start"], ); + const fallbackStart = calls[1]; + NodeAssert.ok(fallbackStart); + NodeAssert.deepStrictEqual( + (fallbackStart.payload as { readonly dynamicTools?: unknown }).dynamicTools, + T3_CODEX_DYNAMIC_TOOLS, + ); }), ); it.effect("propagates non-recoverable resume failures", () => Effect.gen(function* () { const client = { + raw: { + request: (_method: string, _payload?: unknown) => + Effect.succeed(makeThreadOpenResponse("fresh-thread")), + }, request: ( method: M, _payload: CodexRpc.ClientRequestParamsByMethod[M], diff --git a/apps/server/src/provider/Layers/CodexSessionRuntime.ts b/apps/server/src/provider/Layers/CodexSessionRuntime.ts index 5a81e915e34..4fd1cc847b7 100644 --- a/apps/server/src/provider/Layers/CodexSessionRuntime.ts +++ b/apps/server/src/provider/Layers/CodexSessionRuntime.ts @@ -36,10 +36,14 @@ import * as CodexRpc from "effect-codex-app-server/rpc"; import * as EffectCodexSchema from "effect-codex-app-server/schema"; import { buildCodexInitializeParams } from "./CodexProvider.ts"; +import { makeT3CodexDynamicToolWaitRegistry, T3_CODEX_DYNAMIC_TOOLS } from "./CodexDynamicTools.ts"; import { codexSessionAppServerArgs } from "./codexLaunchArgs.ts"; import { expandHomePath } from "../../pathExpansion.ts"; import { buildCodexDeveloperInstructions } from "../CodexDeveloperInstructions.ts"; const decodeV2TurnStartResponse = Schema.decodeUnknownEffect(EffectCodexSchema.V2TurnStartResponse); +const decodeV2ThreadStartResponse = Schema.decodeUnknownEffect( + EffectCodexSchema.V2ThreadStartResponse, +); const PROVIDER = ProviderDriverKind.make("codex"); @@ -285,12 +289,16 @@ function runtimeModeToThreadConfig(input: RuntimeMode): { } } +type CodexThreadStartParamsWithDynamicTools = EffectCodexSchema.V2ThreadStartParams & { + readonly dynamicTools: ReadonlyArray; +}; + function buildThreadStartParams(input: { readonly cwd: string; readonly runtimeMode: RuntimeMode; readonly model: string | undefined; readonly serviceTier: CodexServiceTier | undefined; -}): EffectCodexSchema.V2ThreadStartParams { +}): CodexThreadStartParamsWithDynamicTools { const config = runtimeModeToThreadConfig(input.runtimeMode); return { cwd: input.cwd, @@ -298,6 +306,7 @@ function buildThreadStartParams(input: { sandbox: config.sandbox, ...(input.model ? { model: input.model } : {}), ...(input.serviceTier ? { serviceTier: input.serviceTier } : {}), + dynamicTools: T3_CODEX_DYNAMIC_TOOLS, }; } @@ -433,12 +442,39 @@ type CodexThreadOpenResponse = type CodexThreadOpenMethod = "thread/start" | "thread/resume"; interface CodexThreadOpenClient { + readonly raw: { + readonly request: ( + method: string, + payload?: unknown, + ) => Effect.Effect; + }; readonly request: ( method: M, payload: CodexRpc.ClientRequestParamsByMethod[M], ) => Effect.Effect; } +function startCodexThread( + client: CodexThreadOpenClient, + params: CodexThreadStartParamsWithDynamicTools, +): Effect.Effect< + CodexRpc.ClientRequestResponsesByMethod["thread/start"], + CodexErrors.CodexAppServerError +> { + return client.raw.request("thread/start", params).pipe( + Effect.flatMap(decodeV2ThreadStartResponse), + Effect.mapError((error) => + Schema.isSchemaError(error) + ? CodexErrors.CodexAppServerProtocolParseError.fromSchemaError( + "decode-response-payload", + error, + { method: "thread/start" }, + ) + : error, + ), + ); +} + export const openCodexThread = (input: { readonly client: CodexThreadOpenClient; readonly threadId: ThreadId; @@ -457,7 +493,7 @@ export const openCodexThread = (input: { }); if (resumeThreadId === undefined) { - return input.client.request("thread/start", startParams); + return startCodexThread(input.client, startParams); } return input.client @@ -473,7 +509,7 @@ export const openCodexThread = (input: { resumeThreadId, recoverable: true, cause: error, - }).pipe(Effect.andThen(input.client.request("thread/start", startParams))), + }).pipe(Effect.andThen(startCodexThread(input.client, startParams))), ), ); }; @@ -707,6 +743,7 @@ export const makeCodexSessionRuntime = ( const pendingApprovalsRef = yield* Ref.make(new Map()); const approvalCorrelationsRef = yield* Ref.make(new Map()); const pendingUserInputsRef = yield* Ref.make(new Map()); + const dynamicToolWaitRegistry = yield* makeT3CodexDynamicToolWaitRegistry(); const collabReceiverTurnsRef = yield* Ref.make(new Map()); const closedRef = yield* Ref.make(false); @@ -1113,6 +1150,8 @@ export const makeCodexSessionRuntime = ( }), ); + yield* client.handleServerRequest("item/tool/call", dynamicToolWaitRegistry.handle); + yield* client.handleUnknownServerRequest((method) => Effect.fail(CodexErrors.CodexAppServerRequestError.methodNotFound(method)), ); @@ -1245,6 +1284,7 @@ export const makeCodexSessionRuntime = ( } yield* settlePendingApprovals("cancel"); yield* settlePendingUserInputs({}); + yield* dynamicToolWaitRegistry.cancelAll; yield* updateSession(sessionRef, { status: "closed", activeTurnId: undefined, @@ -1320,6 +1360,7 @@ export const makeCodexSessionRuntime = ( if (!effectiveTurnId) { return; } + yield* dynamicToolWaitRegistry.cancelTurn(effectiveTurnId); yield* client.request("turn/interrupt", { threadId: providerThreadId, turnId: effectiveTurnId, diff --git a/docs/project/deferred-thread-resume.md b/docs/project/deferred-thread-resume.md new file mode 100644 index 00000000000..f734f94a443 --- /dev/null +++ b/docs/project/deferred-thread-resume.md @@ -0,0 +1,174 @@ +# Deferred thread resume + +## Summary + +Long-running external work currently encourages an agent to poll: sleep, spend another inference +turn checking status, and repeat. T3 should provide host-owned waiting primitives so elapsed time +does not consume model inference. + +This project is split into two deliberately different capabilities: + +1. **Live wait** — a tool call remains open while the current provider turn and T3 server process + remain alive. The host completes the tool call after a delay or cancellation. This is the first + implementation milestone and the scope of the initial PR. +2. **Durable resume** — the current turn ends, T3 persists a waitpoint, and a scheduler starts a new + turn after a timer or external event. This is required for restarts, webhook triggers, and + long-lived watches such as pull-request CI. It is a follow-up because its lifecycle and delivery + semantics are materially different from a blocking tool call. + +## Problem + +Agents commonly monitor commands, CI, deployments, and review bots by polling every few seconds. +Most polls provide no new information, but each one wakes the model and consumes inference. T3 +already owns provider sessions and their UI lifecycle, so it is the right layer to wait without +asking the model to reason again. + +## Goals + +- Let a Codex agent wait from 1 second to 1 hour without periodic model inference. +- Show the wait through the existing dynamic-tool lifecycle in the thread timeline. +- Cancel a live wait promptly when its turn is interrupted or its provider session closes. +- Keep the first change isolated to the Codex provider runtime and compatible with app-server + versions that ignore unknown experimental `thread/start` fields. +- Define the durable timer/trigger architecture before extending the feature to external events. + +## Non-goals for the initial PR + +- Surviving a T3 process restart. +- Waking a completed thread from a webhook or WebSocket event. +- Polling GitHub on behalf of an agent. +- Provider-neutral tool injection. Claude, OpenCode, and ACP need equivalent provider hooks before + they can expose the same capability. +- A new bespoke waiting UI. Codex already emits dynamic-tool started/completed events and T3 already + renders them. + +## User and agent experience + +T3 registers a namespaced `t3.wait` dynamic tool when it starts a new Codex thread. The tool accepts: + +- `durationMs`: integer duration from 1,000 through 3,600,000 milliseconds. +- `reason`: optional short explanation shown in the tool arguments. + +The tool description tells the agent to use it for intentional idle periods while external work is +running, and not for ordinary command execution or sub-second delays. While it is open, the thread +remains visibly active. On expiry it returns the elapsed duration and the agent may perform one +status check. Interrupting the turn cancels the wait before T3 sends the provider interrupt request. + +## Technical approach: live wait + +### Provider registration + +Codex app-server supports experimental `dynamicTools` on `thread/start` and sends calls to the host +as `item/tool/call` server requests. T3 already initializes app-server with experimental API support. +The checked-in generated schema exposes the dynamic-tool specification and call types but currently +omits the experimental `dynamicTools` property from `V2ThreadStartParams`, so thread creation uses +the raw protocol request and decodes the response with the existing generated response schema. + +The field is only sent on fresh `thread/start`. A resumed provider thread keeps the tool catalog +with which it was created; old threads therefore gain the tool after a recoverable resume fallback +or when the user starts a new thread. + +### Execution and cancellation + +The handler validates the namespace, tool name, and bounded duration. It races an Effect sleep +against a per-turn cancellation deferred. The deferred is completed before `turn/interrupt` and all +outstanding wait deferreds are completed during session shutdown. + +The response uses Codex's normal `DynamicToolCallResponse`: successful expiry returns +`success: true`; invalid input, unknown tools, and cancellation return `success: false` with a short +text result. No timer loop and no model request occurs while the Effect sleep is pending. + +### Compatibility + +`dynamicTools` is an experimental app-server field. Older app-server versions deserialize unknown +fields permissively, so thread creation continues and the tool is simply unavailable. Current +versions advertise the tool to the model and issue `item/tool/call` requests. T3's existing Codex +adapter maps `dynamicToolCall` lifecycle items to canonical `dynamic_tool_call` activities, and the +web timeline already renders those activities. + +## Follow-up architecture: durable resume + +A durable implementation should not keep a provider request or inference turn open. It needs a +persisted waitpoint owned by orchestration: + +```text +agent registers waitpoint -> current turn settles -> scheduler sleeps outside inference + ^ | + | v + timer / webhook / provider event -> claim waitpoint -> start one continuation turn +``` + +Suggested waitpoint fields: + +- id, T3 thread id, provider instance id, and original turn id +- kind: `timer`, `webhook`, or provider-specific event +- condition payload and optional deadline +- continuation prompt/template +- state: `pending`, `claimed`, `delivered`, `cancelled`, or `expired` +- idempotency key, attempt count, timestamps, and last error + +Delivery should be at-least-once with an atomic claim and idempotent continuation key. On startup, +the scheduler reloads pending waitpoints and re-arms timers. External triggers should enter through +authenticated, scoped adapters rather than exposing a generic unauthenticated callback. A GitHub PR +watcher can then translate check-suite, review, and pull-request webhooks into normalized waitpoint +signals. The sidebar/thread lifecycle should add an explicit `waiting` state only for this durable +mode; a live wait remains an active tool call. + +## Alternatives considered + +- **Agent-side polling:** no server work, but repeatedly consumes inference and is the behavior this + feature is intended to replace. +- **MCP wait server:** portable across clients, but T3 would lose direct lifecycle ownership and + cancellation unless it also hosted and correlated the MCP server. Codex dynamic tools give T3 a + smaller first integration. +- **Implement durable wake immediately:** covers timers and webhooks, but requires persistence, + scheduler recovery, continuation policy, UI state, authorization, and idempotency in one change. + Keeping live and durable semantics separate makes the first PR reviewable and testable. +- **Keep a WebSocket open for every trigger:** useful as an adapter transport, but not sufficient as + the source of truth because sockets do not survive process or network failure. + +## Test plan + +Focused automated coverage: + +- Thread start sends the namespaced dynamic-tool schema through the raw request path and decodes the + response. +- Recoverable resume fallback starts a fresh thread with the tool schema. +- Valid waits remain pending until the Effect test clock reaches the requested duration. +- Duration bounds, malformed arguments, wrong namespaces, and unknown tool names fail without sleep. +- Cancellation wins the race and returns a failed/cancelled tool result. +- Existing provider adapter tests continue to verify canonical dynamic-tool activity mapping. + +Targeted validation: + +- Format and lint changed files. +- Run the focused Codex runtime/dynamic-tool tests. +- Typecheck the server package. +- Manually start a Codex thread against a compatible app-server, request a short wait, confirm one + tool activity appears and completes, then repeat while interrupting the turn. + +## Rollout and observability + +The initial feature has no migration and no feature flag. App-server compatibility degrades by +omitting the tool rather than breaking the session. Existing provider event ingestion records the +dynamic-tool lifecycle. A follow-up may add wait duration/cancellation metrics once the contract is +used beyond Codex. + +## Risks + +- Experimental app-server protocol changes: isolate the raw field construction and response decode + so regeneration can replace it cleanly. +- Excessive waits: enforce a one-hour upper bound. +- Interrupted waits continuing in the background: correlate cancellation by provider turn and settle + it before sending the interrupt request. +- Confusing live and durable behavior: name and document the initial tool as a live wait and do not + claim restart or webhook guarantees. + +## Definition of done for the initial PR + +- A new Codex thread receives the `t3.wait` tool. +- A valid call sleeps in T3 and completes without model polling. +- Interrupt and session close cancel outstanding waits. +- Focused tests and server typecheck pass. +- The provider documentation explains the capability and its live-session limitation. +- The PR explicitly links this design and leaves durable waitpoints as follow-up scope. diff --git a/docs/providers/codex.md b/docs/providers/codex.md index cc9e84e484a..e9e68d2c228 100644 --- a/docs/providers/codex.md +++ b/docs/providers/codex.md @@ -27,6 +27,18 @@ Log in with Codex normally: codex login ``` +## Waiting Without Polling + +New Codex threads receive a T3-hosted `t3.wait` tool. An agent can use it to wait from one second to +one hour while an external command, CI job, deployment, or other task is running. The timer runs in +T3 Code and does not repeatedly wake the model, so the agent can perform one status check after the +wait instead of spending inference on frequent polling. + +This is a live-session wait. It is cancelled if the current turn is interrupted, the provider +session closes, or T3 Code stops. It does not wake a completed thread after an app restart and does +not subscribe to external webhooks. Durable timers and event-triggered continuation are described in +the [deferred thread resume design](../project/deferred-thread-resume.md). + ## I Want Work And Personal Codex Accounts Use one real Codex home and one shadow home. From e9720c427d0b0d04fd92dabd8035f64ddd3d18ed Mon Sep 17 00:00:00 2001 From: Jules-Astier <123639449+Jules-Astier@users.noreply.github.com> Date: Wed, 22 Jul 2026 12:12:58 +0100 Subject: [PATCH 2/3] fix(codex): cancel late wait registrations --- .../provider/Layers/CodexDynamicTools.test.ts | 34 +++++++++ .../src/provider/Layers/CodexDynamicTools.ts | 71 ++++++++++++------- 2 files changed, 81 insertions(+), 24 deletions(-) diff --git a/apps/server/src/provider/Layers/CodexDynamicTools.test.ts b/apps/server/src/provider/Layers/CodexDynamicTools.test.ts index 21d87179349..9a6d5a49222 100644 --- a/apps/server/src/provider/Layers/CodexDynamicTools.test.ts +++ b/apps/server/src/provider/Layers/CodexDynamicTools.test.ts @@ -138,6 +138,40 @@ describe("T3 Codex dynamic tools", () => { }).pipe(Effect.provide(TestClock.layer())), ); + it.effect("immediately cancels waits registered after their turn was interrupted", () => + Effect.gen(function* () { + const registry = yield* makeT3CodexDynamicToolWaitRegistry(); + yield* registry.cancelTurn("provider-turn-1"); + + const response = yield* registry.handle( + waitCall({ durationMs: T3_CODEX_WAIT_MAX_DURATION_MS }, { turnId: "provider-turn-1" }), + ); + + NodeAssert.equal(response.success, false); + NodeAssert.match( + response.contentItems[0]?.type === "inputText" ? response.contentItems[0].text : "", + /Wait cancelled/, + ); + }).pipe(Effect.provide(TestClock.layer())), + ); + + it.effect("immediately cancels waits registered after session shutdown", () => + Effect.gen(function* () { + const registry = yield* makeT3CodexDynamicToolWaitRegistry(); + yield* registry.cancelAll; + + const response = yield* registry.handle( + waitCall({ durationMs: T3_CODEX_WAIT_MAX_DURATION_MS }), + ); + + NodeAssert.equal(response.success, false); + NodeAssert.match( + response.contentItems[0]?.type === "inputText" ? response.contentItems[0].text : "", + /Wait cancelled/, + ); + }).pipe(Effect.provide(TestClock.layer())), + ); + it.effect("rejects malformed and out-of-range arguments without sleeping", () => Effect.gen(function* () { for (const arguments_ of [ diff --git a/apps/server/src/provider/Layers/CodexDynamicTools.ts b/apps/server/src/provider/Layers/CodexDynamicTools.ts index d09532cb5aa..4ccc8ca8d96 100644 --- a/apps/server/src/provider/Layers/CodexDynamicTools.ts +++ b/apps/server/src/provider/Layers/CodexDynamicTools.ts @@ -97,6 +97,12 @@ interface PendingWait { readonly cancelled: Deferred.Deferred; } +interface WaitRegistryState { + readonly pending: Map; + readonly cancelledTurnIds: Set; + readonly closed: boolean; +} + export interface T3CodexDynamicToolWaitRegistry { readonly handle: ( payload: EffectCodexSchema.DynamicToolCallParams, @@ -107,44 +113,61 @@ export interface T3CodexDynamicToolWaitRegistry { export const makeT3CodexDynamicToolWaitRegistry = Effect.fn("makeT3CodexDynamicToolWaitRegistry")( function* (): Effect.fn.Return { - const pendingRef = yield* Ref.make(new Map()); + const stateRef = yield* Ref.make({ + pending: new Map(), + cancelledTurnIds: new Set(), + closed: false, + }); - const cancelWhere = (predicate: (pending: PendingWait) => boolean) => - Ref.get(pendingRef).pipe( - Effect.flatMap((pendingWaits) => - Effect.forEach( - Array.from(pendingWaits.values()), - (pending) => - predicate(pending) - ? Deferred.succeed(pending.cancelled, undefined).pipe(Effect.ignore) - : Effect.void, - { discard: true }, - ), - ), + const completeCancellations = (pendingWaits: ReadonlyArray) => + Effect.forEach( + pendingWaits, + (pending) => Deferred.succeed(pending.cancelled, undefined).pipe(Effect.ignore), + { discard: true }, ); return { handle: (payload) => Effect.gen(function* () { const cancelled = yield* Deferred.make(); - yield* Ref.update(pendingRef, (current) => { - const next = new Map(current); - next.set(payload.callId, { turnId: payload.turnId, cancelled }); - return next; + const registered = yield* Ref.modify(stateRef, (current) => { + if (current.closed || current.cancelledTurnIds.has(payload.turnId)) { + return [false, current] as const; + } + const pending = new Map(current.pending); + pending.set(payload.callId, { turnId: payload.turnId, cancelled }); + return [true, { ...current, pending }] as const; }); - return yield* handleT3CodexDynamicToolCall(payload, Deferred.await(cancelled)).pipe( + return yield* handleT3CodexDynamicToolCall( + payload, + registered ? Deferred.await(cancelled) : Effect.void, + ).pipe( Effect.ensuring( - Ref.update(pendingRef, (current) => { - const next = new Map(current); - next.delete(payload.callId); - return next; + Ref.update(stateRef, (current) => { + if (!current.pending.has(payload.callId)) { + return current; + } + const pending = new Map(current.pending); + pending.delete(payload.callId); + return { ...current, pending }; }), ), ); }), - cancelTurn: (turnId) => cancelWhere((pending) => pending.turnId === turnId), - cancelAll: cancelWhere(() => true), + cancelTurn: (turnId) => + Ref.modify(stateRef, (current) => { + const cancelledTurnIds = new Set(current.cancelledTurnIds); + cancelledTurnIds.add(turnId); + const pending = Array.from(current.pending.values()).filter( + (wait) => wait.turnId === turnId, + ); + return [pending, { ...current, cancelledTurnIds }] as const; + }).pipe(Effect.flatMap(completeCancellations)), + cancelAll: Ref.modify(stateRef, (current) => [ + Array.from(current.pending.values()), + { ...current, closed: true }, + ]).pipe(Effect.flatMap(completeCancellations)), }; }, ); From 35dba04fc5810e3c8f9a45512dde13f897075ae3 Mon Sep 17 00:00:00 2001 From: Jules-Astier <123639449+Jules-Astier@users.noreply.github.com> Date: Wed, 22 Jul 2026 13:13:11 +0100 Subject: [PATCH 3/3] feat(github): add durable PR waitpoints --- .../src/github/GitHubPullRequestProbe.test.ts | 107 +++++ .../src/github/GitHubPullRequestProbe.ts | 231 ++++++++++ .../GitHubWaitpointRegistration.test.ts | 102 +++++ .../src/github/GitHubWaitpointRegistration.ts | 145 +++++++ .../src/github/GitHubWaitpointRuntime.ts | 43 ++ .../GitHubWaitpointThreadGateway.test.ts | 70 +++ .../src/github/GitHubWaitpointWorker.test.ts | 153 +++++++ .../src/github/GitHubWaitpointWorker.ts | 303 +++++++++++++ apps/server/src/persistence/Errors.ts | 1 + .../src/persistence/GitHubWaitpoints.test.ts | 134 ++++++ .../src/persistence/GitHubWaitpoints.ts | 398 ++++++++++++++++++ apps/server/src/persistence/Migrations.ts | 2 + .../Migrations/034_GitHubWaitpoints.test.ts | 51 +++ .../Migrations/034_GitHubWaitpoints.ts | 35 ++ .../src/provider/Drivers/CodexDriver.ts | 3 + .../src/provider/Layers/CodexAdapter.ts | 5 + .../provider/Layers/CodexDynamicTools.test.ts | 52 ++- .../src/provider/Layers/CodexDynamicTools.ts | 131 +++++- .../provider/Layers/CodexSessionRuntime.ts | 12 +- apps/server/src/server.ts | 8 +- docs/project/deferred-thread-resume.md | 74 +++- docs/providers/codex.md | 15 +- 22 files changed, 2059 insertions(+), 16 deletions(-) create mode 100644 apps/server/src/github/GitHubPullRequestProbe.test.ts create mode 100644 apps/server/src/github/GitHubPullRequestProbe.ts create mode 100644 apps/server/src/github/GitHubWaitpointRegistration.test.ts create mode 100644 apps/server/src/github/GitHubWaitpointRegistration.ts create mode 100644 apps/server/src/github/GitHubWaitpointRuntime.ts create mode 100644 apps/server/src/github/GitHubWaitpointThreadGateway.test.ts create mode 100644 apps/server/src/github/GitHubWaitpointWorker.test.ts create mode 100644 apps/server/src/github/GitHubWaitpointWorker.ts create mode 100644 apps/server/src/persistence/GitHubWaitpoints.test.ts create mode 100644 apps/server/src/persistence/GitHubWaitpoints.ts create mode 100644 apps/server/src/persistence/Migrations/034_GitHubWaitpoints.test.ts create mode 100644 apps/server/src/persistence/Migrations/034_GitHubWaitpoints.ts diff --git a/apps/server/src/github/GitHubPullRequestProbe.test.ts b/apps/server/src/github/GitHubPullRequestProbe.test.ts new file mode 100644 index 00000000000..7dbcae47f8a --- /dev/null +++ b/apps/server/src/github/GitHubPullRequestProbe.test.ts @@ -0,0 +1,107 @@ +import { assert, it } from "@effect/vitest"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Ref from "effect/Ref"; + +import { GitHubCli } from "../sourceControl/GitHubCli.ts"; +import { + GitHubPullRequestProbe, + evaluateGitHubWaitpoint, + layer, +} from "./GitHubPullRequestProbe.ts"; + +const rawSnapshot = JSON.stringify({ + comments: [{ id: "comment-1", createdAt: "2026-07-22T11:08:34Z" }], + headRefOid: "abc123", + mergedAt: null, + reviews: [{ id: "review-1", submittedAt: "2026-07-22T11:10:03Z", state: "COMMENTED" }], + state: "OPEN", + statusCheckRollup: [ + { __typename: "CheckRun", name: "test", status: "COMPLETED", conclusion: "SUCCESS" }, + { __typename: "StatusContext", context: "preview", state: "PENDING" }, + ], + updatedAt: "2026-07-22T11:16:04Z", + url: "https://github.com/pingdotgg/t3code/pull/4262", +}); + +it.effect("reads and normalizes the narrow GitHub pull-request watch snapshot", () => + Effect.gen(function* () { + const calls = yield* Ref.make>([]); + const gitHubCli = GitHubCli.of({ + execute: (input: Parameters[0]) => + Ref.update(calls, (current) => [...current, ...input.args]).pipe( + Effect.as({ stdout: rawSnapshot, stderr: "", exitCode: 0 }), + ), + } as unknown as GitHubCli["Service"]); + + const snapshot = yield* Effect.gen(function* () { + const probe = yield* GitHubPullRequestProbe; + return yield* probe.get({ + cwd: "/tmp/project", + repository: "pingdotgg/t3code", + pullRequestNumber: 4262, + }); + }).pipe(Effect.provide(layer.pipe(Layer.provide(Layer.succeed(GitHubCli, gitHubCli))))); + + assert.deepStrictEqual(snapshot, { + url: "https://github.com/pingdotgg/t3code/pull/4262", + state: "open", + headSha: "abc123", + mergedAt: null, + updatedAt: "2026-07-22T11:16:04.000Z", + checks: [ + { name: "test", status: "completed", conclusion: "success" }, + { name: "preview", status: "pending", conclusion: null }, + ], + reviewActivity: [ + { id: "comment-1", occurredAt: "2026-07-22T11:08:34.000Z" }, + { id: "review-1", occurredAt: "2026-07-22T11:10:03.000Z" }, + ], + }); + assert.deepStrictEqual(yield* Ref.get(calls), [ + "pr", + "view", + "4262", + "--repo", + "pingdotgg/t3code", + "--json", + "headRefOid,state,mergedAt,updatedAt,statusCheckRollup,comments,reviews,url", + ]); + }), +); + +it("wakes only when the selected GitHub condition changes", () => { + const baseline: import("./GitHubPullRequestProbe.ts").GitHubPullRequestSnapshot = { + url: "https://github.com/pingdotgg/t3code/pull/4262", + state: "open" as const, + headSha: "abc123", + mergedAt: null, + updatedAt: "2026-07-22T11:16:04.000Z", + checks: [{ name: "test", status: "pending", conclusion: null }], + reviewActivity: [{ id: "review-1", occurredAt: "2026-07-22T11:10:03.000Z" }], + }; + + assert.isFalse(evaluateGitHubWaitpoint("checks_settled", baseline, baseline).satisfied); + assert.isTrue( + evaluateGitHubWaitpoint("checks_settled", baseline, { + ...baseline, + checks: [{ name: "test", status: "completed", conclusion: "failure" }], + }).satisfied, + ); + assert.isTrue( + evaluateGitHubWaitpoint("new_review_activity", baseline, { + ...baseline, + reviewActivity: [ + ...baseline.reviewActivity, + { id: "comment-2", occurredAt: "2026-07-22T12:00:00.000Z" }, + ], + }).satisfied, + ); + assert.isTrue( + evaluateGitHubWaitpoint("pull_request_closed", baseline, { + ...baseline, + state: "merged", + mergedAt: "2026-07-22T12:00:00.000Z", + }).satisfied, + ); +}); diff --git a/apps/server/src/github/GitHubPullRequestProbe.ts b/apps/server/src/github/GitHubPullRequestProbe.ts new file mode 100644 index 00000000000..0694f3b7844 --- /dev/null +++ b/apps/server/src/github/GitHubPullRequestProbe.ts @@ -0,0 +1,231 @@ +/** + * Narrow GitHub CLI adapter used by durable waitpoints. + * + * Raw `gh pr view` output is normalized here so registration and scheduling + * depend on a stable snapshot rather than GitHub's GraphQL-shaped payload. + */ +import * as Context from "effect/Context"; +import * as DateTime from "effect/DateTime"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Schema from "effect/Schema"; + +import { GitHubCli, type GitHubCliError } from "../sourceControl/GitHubCli.ts"; +import type { GitHubWaitpointCondition } from "../persistence/GitHubWaitpoints.ts"; + +const WATCH_FIELDS = "headRefOid,state,mergedAt,updatedAt,statusCheckRollup,comments,reviews,url"; + +const RawCheck = Schema.Struct({ + __typename: Schema.optional(Schema.String), + name: Schema.optional(Schema.String), + context: Schema.optional(Schema.String), + status: Schema.optional(Schema.String), + state: Schema.optional(Schema.String), + conclusion: Schema.optional(Schema.NullOr(Schema.String)), +}); +const RawReviewActivity = Schema.Struct({ + id: Schema.String, + createdAt: Schema.optional(Schema.String), + submittedAt: Schema.optional(Schema.String), +}); +const RawSnapshot = Schema.Struct({ + comments: Schema.Array(RawReviewActivity), + headRefOid: Schema.String, + mergedAt: Schema.NullOr(Schema.String), + reviews: Schema.Array(RawReviewActivity), + state: Schema.String, + statusCheckRollup: Schema.Array(RawCheck), + updatedAt: Schema.String, + url: Schema.String, +}); +const decodeRawSnapshot = Schema.decodeUnknownEffect(Schema.fromJsonString(RawSnapshot)); + +export const GitHubPullRequestSnapshot = Schema.Struct({ + url: Schema.String, + state: Schema.Literals(["open", "closed", "merged"]), + headSha: Schema.String, + mergedAt: Schema.NullOr(Schema.String), + updatedAt: Schema.String, + checks: Schema.Array( + Schema.Struct({ + name: Schema.String, + status: Schema.Literals(["pending", "completed"]), + conclusion: Schema.NullOr(Schema.String), + }), + ), + reviewActivity: Schema.Array( + Schema.Struct({ + id: Schema.String, + occurredAt: Schema.String, + }), + ), +}); +export type GitHubPullRequestSnapshot = typeof GitHubPullRequestSnapshot.Type; + +export class GitHubPullRequestProbeDecodeError extends Schema.TaggedErrorClass()( + "GitHubPullRequestProbeDecodeError", + { + repository: Schema.String, + pullRequestNumber: Schema.Int, + cause: Schema.Defect(), + }, +) { + override get message(): string { + return `GitHub returned an invalid watch snapshot for ${this.repository}#${this.pullRequestNumber}.`; + } +} + +export type GitHubPullRequestProbeError = GitHubCliError | GitHubPullRequestProbeDecodeError; + +export interface GitHubPullRequestProbeInput { + readonly cwd: string; + readonly repository: string; + readonly pullRequestNumber: number; +} + +export class GitHubPullRequestProbe extends Context.Service< + GitHubPullRequestProbe, + { + readonly get: ( + input: GitHubPullRequestProbeInput, + ) => Effect.Effect; + } +>()("t3/github/GitHubPullRequestProbe") {} + +function iso(value: string): string { + return DateTime.formatIso(DateTime.makeUnsafe(value)); +} + +function normalizeState( + state: string, + mergedAt: string | null, +): GitHubPullRequestSnapshot["state"] { + if (mergedAt !== null || state.toUpperCase() === "MERGED") return "merged"; + return state.toUpperCase() === "OPEN" ? "open" : "closed"; +} + +function normalizeCheck(check: typeof RawCheck.Type): GitHubPullRequestSnapshot["checks"][number] { + const rawStatus = (check.status ?? check.state ?? "PENDING").toUpperCase(); + const completed = + rawStatus === "COMPLETED" || + rawStatus === "SUCCESS" || + rawStatus === "FAILURE" || + rawStatus === "ERROR"; + const conclusion = + check.conclusion ?? (completed && rawStatus !== "COMPLETED" ? rawStatus : null); + return { + name: check.name ?? check.context ?? "GitHub check", + status: completed ? "completed" : "pending", + conclusion: conclusion?.toLowerCase() ?? null, + }; +} + +function normalizeActivity( + activity: typeof RawReviewActivity.Type, +): GitHubPullRequestSnapshot["reviewActivity"][number] | undefined { + const occurredAt = activity.createdAt ?? activity.submittedAt; + return occurredAt === undefined ? undefined : { id: activity.id, occurredAt: iso(occurredAt) }; +} + +function normalizeSnapshot(raw: typeof RawSnapshot.Type): GitHubPullRequestSnapshot { + const mergedAt = raw.mergedAt === null ? null : iso(raw.mergedAt); + const reviewActivity = [...raw.comments, ...raw.reviews] + .map(normalizeActivity) + .filter((activity) => activity !== undefined) + .sort( + (left, right) => + left.occurredAt.localeCompare(right.occurredAt) || left.id.localeCompare(right.id), + ); + return { + url: raw.url, + state: normalizeState(raw.state, mergedAt), + headSha: raw.headRefOid, + mergedAt, + updatedAt: iso(raw.updatedAt), + checks: raw.statusCheckRollup.map(normalizeCheck), + reviewActivity, + }; +} + +export interface GitHubWaitpointEvaluation { + readonly satisfied: boolean; + readonly summary: string; +} + +export function evaluateGitHubWaitpoint( + condition: GitHubWaitpointCondition, + baseline: GitHubPullRequestSnapshot, + current: GitHubPullRequestSnapshot, +): GitHubWaitpointEvaluation { + switch (condition) { + case "checks_settled": { + const settled = + current.checks.length > 0 && current.checks.every((check) => check.status === "completed"); + const failed = current.checks.filter( + (check) => + check.conclusion !== null && + !["success", "skipped", "neutral"].includes(check.conclusion), + ).length; + return { + satisfied: settled, + summary: settled + ? `${current.checks.length} checks settled (${failed} unsuccessful).` + : `${current.checks.filter((check) => check.status === "pending").length} checks remain pending.`, + }; + } + case "new_review_activity": { + const baselineIds = new Set(baseline.reviewActivity.map((activity) => activity.id)); + const added = current.reviewActivity.filter((activity) => !baselineIds.has(activity.id)); + return { + satisfied: added.length > 0, + summary: + added.length > 0 + ? `${added.length} new review or comment event${added.length === 1 ? "" : "s"}.` + : "No new review or comment activity.", + }; + } + case "pull_request_closed": + return { + satisfied: current.state !== "open", + summary: + current.state === "merged" + ? "Pull request merged." + : current.state === "closed" + ? "Pull request closed without merging." + : "Pull request remains open.", + }; + } +} + +export const make = Effect.gen(function* () { + const gitHubCli = yield* GitHubCli; + return GitHubPullRequestProbe.of({ + get: Effect.fn("GitHubPullRequestProbe.get")(function* (input) { + const output = yield* gitHubCli.execute({ + cwd: input.cwd, + args: [ + "pr", + "view", + String(input.pullRequestNumber), + "--repo", + input.repository, + "--json", + WATCH_FIELDS, + ], + }); + const raw = yield* decodeRawSnapshot(output.stdout).pipe( + Effect.mapError( + (cause) => + new GitHubPullRequestProbeDecodeError({ + repository: input.repository, + pullRequestNumber: input.pullRequestNumber, + cause, + }), + ), + ); + return normalizeSnapshot(raw); + }), + }); +}); + +export const layer = Layer.effect(GitHubPullRequestProbe, make); diff --git a/apps/server/src/github/GitHubWaitpointRegistration.test.ts b/apps/server/src/github/GitHubWaitpointRegistration.test.ts new file mode 100644 index 00000000000..81064e306cf --- /dev/null +++ b/apps/server/src/github/GitHubWaitpointRegistration.test.ts @@ -0,0 +1,102 @@ +import { assert, it } from "@effect/vitest"; +import { ThreadId } from "@t3tools/contracts"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as TestClock from "effect/testing/TestClock"; + +import { ProjectionSnapshotQuery } from "../orchestration/Services/ProjectionSnapshotQuery.ts"; +import { + GitHubWaitpointRepository, + layer as repositoryLayer, +} from "../persistence/GitHubWaitpoints.ts"; +import { SqlitePersistenceMemory } from "../persistence/Layers/Sqlite.ts"; +import { + GitHubPullRequestProbe, + type GitHubPullRequestSnapshot, +} from "./GitHubPullRequestProbe.ts"; +import { GitHubWaitpointRegistration, layer } from "./GitHubWaitpointRegistration.ts"; + +const baseline: GitHubPullRequestSnapshot = { + url: "https://github.com/pingdotgg/t3code/pull/4262", + state: "open", + headSha: "abc123", + mergedAt: null, + updatedAt: "2026-07-22T11:16:04.000Z", + checks: [{ name: "test", status: "pending", conclusion: null }], + reviewActivity: [], +}; + +const persistence = repositoryLayer.pipe(Layer.provideMerge(SqlitePersistenceMemory)); +let probeCalls = 0; +const probe = Layer.succeed( + GitHubPullRequestProbe, + GitHubPullRequestProbe.of({ + get: () => + Effect.sync(() => { + probeCalls += 1; + return baseline; + }), + }), +); +const snapshots = Layer.succeed(ProjectionSnapshotQuery, { + getThreadDetailById: () => + Effect.succeed( + Option.some({ + latestTurn: { turnId: "t3-turn-1" }, + }), + ), +} as unknown as ProjectionSnapshotQuery["Service"]); +const registrationLayer = layer.pipe( + Layer.provideMerge(persistence), + Layer.provideMerge(probe), + Layer.provideMerge(snapshots), +); +const test = it.layer(Layer.mergeAll(registrationLayer, persistence, TestClock.layer())); + +test("GitHubWaitpointRegistration", (it) => { + it.effect("captures an initial snapshot and persists a bounded durable wait", () => + Effect.gen(function* () { + const registration = yield* GitHubWaitpointRegistration; + const repository = yield* GitHubWaitpointRepository; + + const result = yield* registration.register({ + idempotencyKey: "call-1", + threadId: ThreadId.make("thread-1"), + cwd: "/tmp/project", + repository: "pingdotgg/t3code", + pullRequestNumber: 4262, + condition: "checks_settled", + timeoutMinutes: 60, + reason: "Address any failed checks.", + }); + const duplicate = yield* registration.register({ + idempotencyKey: "call-1", + threadId: ThreadId.make("thread-1"), + cwd: "/tmp/project", + repository: "pingdotgg/t3code", + pullRequestNumber: 4262, + condition: "checks_settled", + timeoutMinutes: 60, + reason: "This retry must not replace the original waitpoint.", + }); + + assert.deepStrictEqual(result, { id: "github:thread-1:call-1" }); + assert.deepStrictEqual(duplicate, result); + assert.equal(probeCalls, 1); + const stored = yield* repository.getById({ id: result.id }); + assert.isTrue(Option.isSome(stored)); + if (Option.isNone(stored)) return; + + assert.deepInclude(stored.value, { + baseline, + originatingTurnId: "t3-turn-1", + nextPollAt: "1970-01-01T00:00:30.000Z", + deadlineAt: "1970-01-01T01:00:00.000Z", + state: "pending", + }); + assert.match(stored.value.continuationPrompt, /Address any failed checks/); + assert.match(stored.value.continuationPrompt, /pingdotgg\/t3code#4262/); + }), + ); +}); diff --git a/apps/server/src/github/GitHubWaitpointRegistration.ts b/apps/server/src/github/GitHubWaitpointRegistration.ts new file mode 100644 index 00000000000..0b4d4dfd6f5 --- /dev/null +++ b/apps/server/src/github/GitHubWaitpointRegistration.ts @@ -0,0 +1,145 @@ +/** Registers a durable GitHub waitpoint from a provider dynamic-tool call. */ +import { ThreadId } from "@t3tools/contracts"; +import * as Context from "effect/Context"; +import * as DateTime from "effect/DateTime"; +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 type { ProjectionRepositoryError } from "../persistence/Errors.ts"; +import { + GitHubWaitpointRepository, + type GitHubWaitpointCondition, + type GitHubWaitpointRepositoryError, +} from "../persistence/GitHubWaitpoints.ts"; +import { ProjectionSnapshotQuery } from "../orchestration/Services/ProjectionSnapshotQuery.ts"; +import { + GitHubPullRequestProbe, + type GitHubPullRequestProbeError, +} from "./GitHubPullRequestProbe.ts"; + +const FIRST_POLL_DELAY_SECONDS = 30; + +export interface RegisterGitHubWaitpointInput { + readonly idempotencyKey: string; + readonly threadId: ThreadId; + readonly cwd: string; + readonly repository: string; + readonly pullRequestNumber: number; + readonly condition: GitHubWaitpointCondition; + readonly timeoutMinutes: number; + readonly reason?: string; +} + +export type GitHubWaitpointRegistrationError = + | GitHubPullRequestProbeError + | GitHubWaitpointRepositoryError + | ProjectionRepositoryError + | GitHubWaitpointRegistrationThreadUnavailableError + | GitHubWaitpointRegistrationUnavailableError; + +export class GitHubWaitpointRegistrationThreadUnavailableError extends Schema.TaggedErrorClass()( + "GitHubWaitpointRegistrationThreadUnavailableError", + { threadId: ThreadId }, +) { + override get message(): string { + return `Cannot register a GitHub wait because thread ${this.threadId} has no current T3 turn.`; + } +} + +export class GitHubWaitpointRegistrationUnavailableError extends Schema.TaggedErrorClass()( + "GitHubWaitpointRegistrationUnavailableError", + {}, +) { + override get message(): string { + return "Durable GitHub waitpoint registration is unavailable."; + } +} + +export interface GitHubWaitpointRegistrationService { + readonly register: ( + input: RegisterGitHubWaitpointInput, + ) => Effect.Effect<{ readonly id: string }, GitHubWaitpointRegistrationError>; +} + +export class GitHubWaitpointRegistration extends Context.Reference( + "t3/github/GitHubWaitpointRegistration/GitHubWaitpointRegistration", + { + defaultValue: () => ({ + register: () => Effect.fail(new GitHubWaitpointRegistrationUnavailableError()), + }), + }, +) {} + +function conditionLabel(condition: GitHubWaitpointCondition): string { + switch (condition) { + case "checks_settled": + return "all reported checks have settled"; + case "new_review_activity": + return "new review or comment activity is available"; + case "pull_request_closed": + return "the pull request has merged or closed"; + } +} + +function continuationPrompt(input: RegisterGitHubWaitpointInput): string { + const reason = input.reason?.trim(); + return [ + `T3 GitHub watcher observed that ${conditionLabel(input.condition)} for ${input.repository}#${input.pullRequestNumber}.`, + "Re-read the pull request and continue the task from the latest GitHub state.", + ...(reason ? [`Original reason for waiting: ${reason}`] : []), + ].join(" "); +} + +export const make = Effect.gen(function* () { + const repository = yield* GitHubWaitpointRepository; + const probe = yield* GitHubPullRequestProbe; + const snapshots = yield* ProjectionSnapshotQuery; + + return GitHubWaitpointRegistration.of({ + register: Effect.fn("GitHubWaitpointRegistration.register")(function* (input) { + const id = `github:${input.threadId}:${input.idempotencyKey}`; + const existing = yield* repository.getById({ id }); + if (Option.isSome(existing)) return { id }; + + const thread = yield* snapshots.getThreadDetailById(input.threadId); + if (Option.isNone(thread) || thread.value.latestTurn === null) { + return yield* new GitHubWaitpointRegistrationThreadUnavailableError({ + threadId: input.threadId, + }); + } + + const baseline = yield* probe.get({ + cwd: input.cwd, + repository: input.repository, + pullRequestNumber: input.pullRequestNumber, + }); + const now = yield* DateTime.now; + const createdAt = DateTime.formatIso(now); + yield* repository.register({ + id, + threadId: input.threadId, + originatingTurnId: thread.value.latestTurn.turnId, + repository: input.repository, + pullRequestNumber: input.pullRequestNumber, + condition: input.condition, + baseline, + continuationPrompt: continuationPrompt(input), + nextPollAt: DateTime.formatIso(DateTime.add(now, { seconds: FIRST_POLL_DELAY_SECONDS })), + deadlineAt: DateTime.formatIso(DateTime.add(now, { minutes: input.timeoutMinutes })), + createdAt, + }); + yield* Effect.logInfo("github.waitpoint.registered", { + waitpointId: id, + threadId: input.threadId, + repository: input.repository, + pullRequestNumber: input.pullRequestNumber, + condition: input.condition, + }); + return { id }; + }), + }); +}); + +export const layer = Layer.effect(GitHubWaitpointRegistration, make); diff --git a/apps/server/src/github/GitHubWaitpointRuntime.ts b/apps/server/src/github/GitHubWaitpointRuntime.ts new file mode 100644 index 00000000000..9693afbcc0d --- /dev/null +++ b/apps/server/src/github/GitHubWaitpointRuntime.ts @@ -0,0 +1,43 @@ +/** Live service graph and scoped polling loop for local GitHub waitpoints. */ +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Schedule from "effect/Schedule"; + +import * as GitHubWaitpoints from "../persistence/GitHubWaitpoints.ts"; +import * as GitHubCli from "../sourceControl/GitHubCli.ts"; +import * as GitHubPullRequestProbe from "./GitHubPullRequestProbe.ts"; +import * as GitHubWaitpointRegistration from "./GitHubWaitpointRegistration.ts"; +import { + GitHubWaitpointWorker, + layer as workerLayer, + threadGatewayLayer, +} from "./GitHubWaitpointWorker.ts"; + +const WORKER_TICK_INTERVAL = "5 seconds" as const; + +const probeLayer = GitHubPullRequestProbe.layer.pipe(Layer.provide(GitHubCli.layer)); + +const registrationLayer = GitHubWaitpointRegistration.layer.pipe( + Layer.provideMerge(GitHubWaitpoints.layer), + Layer.provideMerge(probeLayer), +); + +const workerServicesLayer = workerLayer.pipe( + Layer.provideMerge(GitHubWaitpoints.layer), + Layer.provideMerge(probeLayer), + Layer.provideMerge(threadGatewayLayer), +); + +const workerLoopLayer = Layer.effectDiscard( + Effect.gen(function* () { + const worker = yield* GitHubWaitpointWorker; + const tick = worker.processDue.pipe( + Effect.catchCause((cause) => + Effect.logWarning("github.waitpoint.worker.tick-failed", { cause }), + ), + ); + yield* Effect.forkScoped(tick.pipe(Effect.repeat(Schedule.spaced(WORKER_TICK_INTERVAL)))); + }), +).pipe(Layer.provideMerge(workerServicesLayer)); + +export const layer = Layer.merge(registrationLayer, workerLoopLayer); diff --git a/apps/server/src/github/GitHubWaitpointThreadGateway.test.ts b/apps/server/src/github/GitHubWaitpointThreadGateway.test.ts new file mode 100644 index 00000000000..563d591945d --- /dev/null +++ b/apps/server/src/github/GitHubWaitpointThreadGateway.test.ts @@ -0,0 +1,70 @@ +import { assert, it } from "@effect/vitest"; +import { ThreadId } from "@t3tools/contracts"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as Ref from "effect/Ref"; +import * as Stream from "effect/Stream"; + +import { OrchestrationEngineService } from "../orchestration/Services/OrchestrationEngine.ts"; +import { ProjectionSnapshotQuery } from "../orchestration/Services/ProjectionSnapshotQuery.ts"; +import { GitHubWaitpointThreadGateway, threadGatewayLayer } from "./GitHubWaitpointWorker.ts"; + +it.effect("dispatches a deterministic continuation command for a ready thread", () => + Effect.gen(function* () { + const commands = yield* Ref.make>([]); + const thread = { + id: ThreadId.make("thread-1"), + runtimeMode: "full-access" as const, + interactionMode: "default" as const, + latestTurn: { turnId: "turn-1", state: "completed" }, + session: { status: "ready", activeTurnId: null }, + }; + const snapshots = Layer.succeed(ProjectionSnapshotQuery, { + getThreadDetailById: () => Effect.succeed(Option.some(thread)), + } as unknown as ProjectionSnapshotQuery["Service"]); + const orchestration = Layer.succeed( + OrchestrationEngineService, + OrchestrationEngineService.of({ + readEvents: () => Stream.empty, + dispatch: (command) => + Ref.update(commands, (current) => [...current, command]).pipe(Effect.as({ sequence: 1 })), + streamDomainEvents: Stream.empty, + latestSequence: Effect.succeed(0), + }), + ); + const live = threadGatewayLayer.pipe(Layer.provide(Layer.merge(snapshots, orchestration))); + + yield* Effect.gen(function* () { + const gateway = yield* GitHubWaitpointThreadGateway; + const input = { + waitpointId: "github:thread-1:call-1", + threadId: ThreadId.make("thread-1"), + prompt: "GitHub checks settled. Continue.", + createdAt: "2026-07-22T12:00:00.000Z", + expectedLatestTurnId: "turn-1", + }; + yield* gateway.resume(input); + yield* gateway.resume(input); + const staleError = yield* gateway + .resume({ ...input, expectedLatestTurnId: "turn-older" }) + .pipe(Effect.flip); + assert.equal(staleError._tag, "GitHubWaitpointThreadUnavailableError"); + }).pipe(Effect.provide(live)); + + const dispatched = yield* Ref.get(commands); + assert.lengthOf(dispatched, 2); + assert.deepInclude(dispatched[0], { + type: "thread.turn.start", + commandId: "github-waitpoint:github:thread-1:call-1", + threadId: "thread-1", + message: { + messageId: "github-waitpoint:github:thread-1:call-1", + role: "user", + text: "GitHub checks settled. Continue.", + attachments: [], + }, + }); + assert.deepEqual(dispatched[0], dispatched[1]); + }), +); diff --git a/apps/server/src/github/GitHubWaitpointWorker.test.ts b/apps/server/src/github/GitHubWaitpointWorker.test.ts new file mode 100644 index 00000000000..1cc508b6308 --- /dev/null +++ b/apps/server/src/github/GitHubWaitpointWorker.test.ts @@ -0,0 +1,153 @@ +import { assert, it } from "@effect/vitest"; +import { ThreadId } from "@t3tools/contracts"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as Ref from "effect/Ref"; +import * as TestClock from "effect/testing/TestClock"; + +import { + GitHubWaitpointRepository, + layer as repositoryLayer, +} from "../persistence/GitHubWaitpoints.ts"; +import { SqlitePersistenceMemory } from "../persistence/Layers/Sqlite.ts"; +import { + GitHubPullRequestProbe, + type GitHubPullRequestSnapshot, +} from "./GitHubPullRequestProbe.ts"; +import { + GitHubWaitpointThreadGateway, + GitHubWaitpointWorker, + layer, +} from "./GitHubWaitpointWorker.ts"; + +const baseline: GitHubPullRequestSnapshot = { + url: "https://github.com/pingdotgg/t3code/pull/4262", + state: "open", + headSha: "abc123", + mergedAt: null, + updatedAt: "2026-07-22T11:16:04.000Z", + checks: [{ name: "test", status: "pending", conclusion: null }], + reviewActivity: [], +}; +const settled: GitHubPullRequestSnapshot = { + ...baseline, + checks: [{ name: "test", status: "completed", conclusion: "failure" }], +}; + +it.effect("resumes a ready thread exactly once when its GitHub condition is satisfied", () => + Effect.gen(function* () { + const resumed = yield* Ref.make< + ReadonlyArray<{ readonly id: string; readonly prompt: string }> + >([]); + const persistence = repositoryLayer.pipe(Layer.provideMerge(SqlitePersistenceMemory)); + const probe = Layer.succeed( + GitHubPullRequestProbe, + GitHubPullRequestProbe.of({ get: () => Effect.succeed(settled) }), + ); + const gateway = Layer.succeed( + GitHubWaitpointThreadGateway, + GitHubWaitpointThreadGateway.of({ + getStatus: () => Effect.succeed(Option.some({ ready: true, latestTurnId: "turn-1" })), + resume: (input) => + Ref.update(resumed, (current) => [ + ...current, + { id: input.waitpointId, prompt: input.prompt }, + ]), + }), + ); + const workerLayer = layer.pipe( + Layer.provideMerge(persistence), + Layer.provideMerge(probe), + Layer.provideMerge(gateway), + ); + const programLayer = Layer.mergeAll(workerLayer, persistence, TestClock.layer()); + + yield* Effect.gen(function* () { + const repository = yield* GitHubWaitpointRepository; + const worker = yield* GitHubWaitpointWorker; + yield* repository.register({ + id: "github:thread-1:call-1", + threadId: ThreadId.make("thread-1"), + originatingTurnId: "turn-1", + repository: "pingdotgg/t3code", + pullRequestNumber: 4262, + condition: "checks_settled", + baseline, + continuationPrompt: "Continue the pull request task.", + nextPollAt: "1970-01-01T00:00:00.000Z", + deadlineAt: "1970-01-01T01:00:00.000Z", + createdAt: "1970-01-01T00:00:00.000Z", + }); + + yield* worker.processDue; + yield* worker.processDue; + + const stored = yield* repository.getById({ id: "github:thread-1:call-1" }); + assert.isTrue(Option.isSome(stored)); + if (Option.isNone(stored)) return; + assert.equal(stored.value.state, "delivered"); + assert.deepStrictEqual(yield* Ref.get(resumed), [ + { + id: "github:thread-1:call-1", + prompt: + "Continue the pull request task. GitHub observation: 1 checks settled (1 unsuccessful).", + }, + ]); + }).pipe(Effect.provide(programLayer)); + }), +); + +it.effect("does not probe or resume until the originating provider turn has settled", () => + Effect.gen(function* () { + const probes = yield* Ref.make(0); + const persistence = repositoryLayer.pipe(Layer.provideMerge(SqlitePersistenceMemory)); + const probe = Layer.succeed( + GitHubPullRequestProbe, + GitHubPullRequestProbe.of({ + get: () => Ref.update(probes, (count) => count + 1).pipe(Effect.as(settled)), + }), + ); + const gateway = Layer.succeed( + GitHubWaitpointThreadGateway, + GitHubWaitpointThreadGateway.of({ + getStatus: () => Effect.succeed(Option.some({ ready: false, latestTurnId: "turn-1" })), + resume: () => Effect.die("resume must not run while the thread is active"), + }), + ); + const workerLayer = layer.pipe( + Layer.provideMerge(persistence), + Layer.provideMerge(probe), + Layer.provideMerge(gateway), + ); + const programLayer = Layer.mergeAll(workerLayer, persistence, TestClock.layer()); + + yield* Effect.gen(function* () { + const repository = yield* GitHubWaitpointRepository; + const worker = yield* GitHubWaitpointWorker; + yield* repository.register({ + id: "github:thread-1:call-active", + threadId: ThreadId.make("thread-1"), + originatingTurnId: "turn-1", + repository: "pingdotgg/t3code", + pullRequestNumber: 4262, + condition: "checks_settled", + baseline, + continuationPrompt: "Continue.", + nextPollAt: "1970-01-01T00:00:00.000Z", + deadlineAt: "1970-01-01T01:00:00.000Z", + createdAt: "1970-01-01T00:00:00.000Z", + }); + + yield* worker.processDue; + + assert.equal(yield* Ref.get(probes), 0); + const stored = yield* repository.getById({ id: "github:thread-1:call-active" }); + assert.isTrue(Option.isSome(stored)); + if (Option.isSome(stored)) { + assert.equal(stored.value.state, "pending"); + assert.equal(stored.value.nextPollAt, "1970-01-01T00:00:05.000Z"); + } + }).pipe(Effect.provide(programLayer)); + }), +); diff --git a/apps/server/src/github/GitHubWaitpointWorker.ts b/apps/server/src/github/GitHubWaitpointWorker.ts new file mode 100644 index 00000000000..b61859d98fa --- /dev/null +++ b/apps/server/src/github/GitHubWaitpointWorker.ts @@ -0,0 +1,303 @@ +/** + * Restartable local worker for durable GitHub waitpoints. + * + * Polling happens in the T3 host process. A model turn is started only after a + * condition is satisfied and an atomic delivery lease has been claimed. + */ +import { CommandId, MessageId, ThreadId } from "@t3tools/contracts"; +import * as Context from "effect/Context"; +import * as DateTime from "effect/DateTime"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as Result from "effect/Result"; +import * as Schema from "effect/Schema"; + +import type { OrchestrationDispatchError } from "../orchestration/Errors.ts"; +import { OrchestrationEngineService } from "../orchestration/Services/OrchestrationEngine.ts"; +import { ProjectionSnapshotQuery } from "../orchestration/Services/ProjectionSnapshotQuery.ts"; +import type { ProjectionRepositoryError } from "../persistence/Errors.ts"; +import { + GitHubWaitpointRepository, + type GitHubWaitpoint, + type GitHubWaitpointRepositoryError, +} from "../persistence/GitHubWaitpoints.ts"; +import { + evaluateGitHubWaitpoint, + GitHubPullRequestProbe, + GitHubPullRequestSnapshot, + type GitHubPullRequestProbeError, +} from "./GitHubPullRequestProbe.ts"; + +const DUE_BATCH_SIZE = 25; +const ACTIVE_TURN_RETRY_SECONDS = 5; +const POLL_INTERVAL_SECONDS = 30; +const DELIVERY_LEASE_SECONDS = 60; + +export class GitHubWaitpointThreadUnavailableError extends Schema.TaggedErrorClass()( + "GitHubWaitpointThreadUnavailableError", + { + threadId: ThreadId, + detail: Schema.String, + }, +) { + override get message(): string { + return `Cannot resume thread ${this.threadId}: ${this.detail}`; + } +} + +export interface GitHubWaitpointThreadStatus { + readonly ready: boolean; + readonly latestTurnId: string | null; +} + +export interface ResumeGitHubWaitpointInput { + readonly waitpointId: string; + readonly threadId: ThreadId; + readonly prompt: string; + readonly createdAt: string; + readonly expectedLatestTurnId: string; +} + +type GitHubWaitpointThreadGatewayError = + | ProjectionRepositoryError + | OrchestrationDispatchError + | GitHubWaitpointThreadUnavailableError; + +export class GitHubWaitpointThreadGateway extends Context.Service< + GitHubWaitpointThreadGateway, + { + readonly getStatus: ( + threadId: ThreadId, + ) => Effect.Effect, ProjectionRepositoryError>; + readonly resume: ( + input: ResumeGitHubWaitpointInput, + ) => Effect.Effect; + } +>()("t3/github/GitHubWaitpointWorker/GitHubWaitpointThreadGateway") {} + +function isThreadReady(thread: { + readonly latestTurn: { readonly state: string } | null; + readonly session: { + readonly status: string; + readonly activeTurnId: string | null; + } | null; +}): boolean { + return ( + thread.latestTurn?.state !== "running" && + thread.session?.activeTurnId == null && + thread.session?.status !== "starting" && + thread.session?.status !== "running" + ); +} + +export const makeThreadGateway = Effect.gen(function* () { + const snapshots = yield* ProjectionSnapshotQuery; + const orchestration = yield* OrchestrationEngineService; + + return GitHubWaitpointThreadGateway.of({ + getStatus: (threadId) => + snapshots.getThreadDetailById(threadId).pipe( + Effect.map( + Option.map((thread) => ({ + ready: isThreadReady(thread), + latestTurnId: thread.latestTurn?.turnId ?? null, + })), + ), + ), + resume: Effect.fn("GitHubWaitpointThreadGateway.resume")(function* (input) { + const threadOption = yield* snapshots.getThreadDetailById(input.threadId); + if (Option.isNone(threadOption)) { + return yield* new GitHubWaitpointThreadUnavailableError({ + threadId: input.threadId, + detail: "thread no longer exists or is archived", + }); + } + const thread = threadOption.value; + if (thread.latestTurn?.turnId !== input.expectedLatestTurnId) { + return yield* new GitHubWaitpointThreadUnavailableError({ + threadId: input.threadId, + detail: "thread advanced after the waitpoint was registered", + }); + } + if (!isThreadReady(thread)) { + return yield* new GitHubWaitpointThreadUnavailableError({ + threadId: input.threadId, + detail: "another turn is active", + }); + } + yield* orchestration.dispatch({ + type: "thread.turn.start", + commandId: CommandId.make(`github-waitpoint:${input.waitpointId}`), + threadId: input.threadId, + message: { + messageId: MessageId.make(`github-waitpoint:${input.waitpointId}`), + role: "user", + text: input.prompt, + attachments: [], + }, + runtimeMode: thread.runtimeMode, + interactionMode: thread.interactionMode, + createdAt: input.createdAt, + }); + }), + }); +}); + +export const threadGatewayLayer = Layer.effect(GitHubWaitpointThreadGateway, makeThreadGateway); + +type GitHubWaitpointWorkerError = + | GitHubWaitpointRepositoryError + | GitHubPullRequestProbeError + | GitHubWaitpointThreadGatewayError + | Schema.SchemaError; + +export class GitHubWaitpointWorker extends Context.Service< + GitHubWaitpointWorker, + { + readonly processDue: Effect.Effect; + } +>()("t3/github/GitHubWaitpointWorker") {} + +function errorMessage(error: unknown): string { + return error instanceof Error ? error.message : String(error); +} + +export const make = Effect.gen(function* () { + const repository = yield* GitHubWaitpointRepository; + const probe = yield* GitHubPullRequestProbe; + const threads = yield* GitHubWaitpointThreadGateway; + const decodeBaseline = Schema.decodeUnknownEffect(GitHubPullRequestSnapshot); + + const processWaitpoint = Effect.fn("GitHubWaitpointWorker.processWaitpoint")(function* ( + waitpoint: GitHubWaitpoint, + now: DateTime.Utc, + ) { + const nowIso = DateTime.formatIso(now); + if (waitpoint.deadlineAt <= nowIso) { + yield* repository.markExpired({ + id: waitpoint.id, + expiredAt: nowIso, + lastError: "Waitpoint deadline elapsed.", + }); + return; + } + + const threadStatus = yield* threads.getStatus(waitpoint.threadId); + if (Option.isNone(threadStatus)) { + yield* repository.markExpired({ + id: waitpoint.id, + expiredAt: nowIso, + lastError: "Thread no longer exists or is archived.", + }); + return; + } + if (!threadStatus.value.ready) { + yield* repository.reschedule({ + id: waitpoint.id, + nextPollAt: DateTime.formatIso(DateTime.add(now, { seconds: ACTIVE_TURN_RETRY_SECONDS })), + updatedAt: nowIso, + lastError: null, + }); + return; + } + if (threadStatus.value.latestTurnId !== waitpoint.originatingTurnId) { + yield* repository.markExpired({ + id: waitpoint.id, + expiredAt: nowIso, + lastError: "Thread advanced after this waitpoint was registered.", + }); + return; + } + + const baselineResult = yield* Effect.result(decodeBaseline(waitpoint.baseline)); + if (Result.isFailure(baselineResult)) { + yield* repository.markExpired({ + id: waitpoint.id, + expiredAt: nowIso, + lastError: "Stored GitHub baseline is incompatible with this T3 Code version.", + }); + return; + } + + const probeResult = yield* Effect.result( + probe.get({ + cwd: process.cwd(), + repository: waitpoint.repository, + pullRequestNumber: waitpoint.pullRequestNumber, + }), + ); + if (Result.isFailure(probeResult)) { + yield* repository.reschedule({ + id: waitpoint.id, + nextPollAt: DateTime.formatIso(DateTime.add(now, { seconds: POLL_INTERVAL_SECONDS })), + updatedAt: nowIso, + lastError: errorMessage(probeResult.failure), + }); + return; + } + + const evaluation = evaluateGitHubWaitpoint( + waitpoint.condition, + baselineResult.success, + probeResult.success, + ); + if (!evaluation.satisfied) { + yield* repository.reschedule({ + id: waitpoint.id, + nextPollAt: DateTime.formatIso(DateTime.add(now, { seconds: POLL_INTERVAL_SECONDS })), + updatedAt: nowIso, + lastError: null, + }); + return; + } + + const claim = yield* repository.claim({ + id: waitpoint.id, + now: nowIso, + leaseExpiresAt: DateTime.formatIso(DateTime.add(now, { seconds: DELIVERY_LEASE_SECONDS })), + }); + if (Option.isNone(claim)) return; + + const resumeResult = yield* Effect.result( + threads.resume({ + waitpointId: waitpoint.id, + threadId: waitpoint.threadId, + prompt: `${waitpoint.continuationPrompt} GitHub observation: ${evaluation.summary}`, + createdAt: nowIso, + expectedLatestTurnId: waitpoint.originatingTurnId, + }), + ); + if (Result.isFailure(resumeResult)) { + yield* repository.reschedule({ + id: waitpoint.id, + nextPollAt: DateTime.formatIso(DateTime.add(now, { seconds: ACTIVE_TURN_RETRY_SECONDS })), + updatedAt: nowIso, + lastError: errorMessage(resumeResult.failure), + }); + return; + } + + yield* repository.markDelivered({ id: waitpoint.id, deliveredAt: nowIso }); + yield* Effect.logInfo("github.waitpoint.delivered", { + waitpointId: waitpoint.id, + threadId: waitpoint.threadId, + repository: waitpoint.repository, + pullRequestNumber: waitpoint.pullRequestNumber, + condition: waitpoint.condition, + }); + }); + + const processDue = Effect.gen(function* () { + const now = yield* DateTime.now; + const due = yield* repository.listDue({ now: DateTime.formatIso(now), limit: DUE_BATCH_SIZE }); + yield* Effect.forEach(due, (waitpoint) => processWaitpoint(waitpoint, now), { + concurrency: 1, + discard: true, + }); + }); + + return GitHubWaitpointWorker.of({ processDue }); +}); + +export const layer = Layer.effect(GitHubWaitpointWorker, make); diff --git a/apps/server/src/persistence/Errors.ts b/apps/server/src/persistence/Errors.ts index 03edaec77d6..3c9c7be54b3 100644 --- a/apps/server/src/persistence/Errors.ts +++ b/apps/server/src/persistence/Errors.ts @@ -132,6 +132,7 @@ export type OrchestrationCommandReceiptRepositoryError = | PersistenceDecodeError; export type ProviderSessionRuntimeRepositoryError = PersistenceSqlError | PersistenceDecodeError; +export type GitHubWaitpointRepositoryError = PersistenceSqlError | PersistenceDecodeError; export type AuthPairingLinkRepositoryError = PersistenceSqlError | PersistenceDecodeError; export type AuthSessionRepositoryError = PersistenceSqlError | PersistenceDecodeError; diff --git a/apps/server/src/persistence/GitHubWaitpoints.test.ts b/apps/server/src/persistence/GitHubWaitpoints.test.ts new file mode 100644 index 00000000000..1579952470c --- /dev/null +++ b/apps/server/src/persistence/GitHubWaitpoints.test.ts @@ -0,0 +1,134 @@ +import { assert, it } from "@effect/vitest"; +import { ThreadId } from "@t3tools/contracts"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import { describe } from "vite-plus/test"; + +import { SqlitePersistenceMemory } from "./Layers/Sqlite.ts"; +import { GitHubWaitpointRepository, layer } from "./GitHubWaitpoints.ts"; + +const repositoryLayer = layer.pipe(Layer.provideMerge(SqlitePersistenceMemory)); + +const registeredWaitpoint = { + id: "github:thread-1:call-1", + threadId: ThreadId.make("thread-1"), + originatingTurnId: "turn-1", + repository: "pingdotgg/t3code", + pullRequestNumber: 4262, + condition: "checks_settled" as const, + baseline: { headSha: "abc", checksPending: 2 }, + continuationPrompt: "Checks settled. Continue the pull request task.", + nextPollAt: "2026-07-22T12:00:30.000Z", + deadlineAt: "2026-07-23T12:00:00.000Z", + createdAt: "2026-07-22T12:00:00.000Z", +}; + +describe("GitHubWaitpointRepository", () => { + it.effect("persists a waitpoint and treats its originating tool call as idempotent", () => + Effect.gen(function* () { + const repository = yield* GitHubWaitpointRepository; + + yield* repository.register(registeredWaitpoint); + yield* repository.register({ + ...registeredWaitpoint, + continuationPrompt: "A duplicate call must not replace the original.", + }); + + const result = yield* repository.getById({ id: registeredWaitpoint.id }); + assert.isTrue(Option.isSome(result)); + if (Option.isNone(result)) return; + + assert.deepStrictEqual(result.value, { + ...registeredWaitpoint, + state: "pending", + deliveryLeaseExpiresAt: null, + attemptCount: 0, + lastError: null, + updatedAt: registeredWaitpoint.createdAt, + deliveredAt: null, + }); + }).pipe(Effect.provide(repositoryLayer)), + ); + + it.effect("claims due work once and recovers a delivery after its lease expires", () => + Effect.gen(function* () { + const repository = yield* GitHubWaitpointRepository; + const claimWaitpoint = { ...registeredWaitpoint, id: "github:thread-1:call-claim" }; + yield* repository.register(claimWaitpoint); + + assert.deepStrictEqual( + yield* repository.listDue({ now: "2026-07-22T12:00:29.999Z", limit: 10 }), + [], + ); + assert.lengthOf(yield* repository.listDue({ now: "2026-07-22T12:00:30.000Z", limit: 10 }), 1); + + const firstClaim = yield* repository.claim({ + id: claimWaitpoint.id, + now: "2026-07-22T12:00:30.000Z", + leaseExpiresAt: "2026-07-22T12:01:30.000Z", + }); + const duplicateClaim = yield* repository.claim({ + id: claimWaitpoint.id, + now: "2026-07-22T12:00:31.000Z", + leaseExpiresAt: "2026-07-22T12:01:31.000Z", + }); + assert.isTrue(Option.isSome(firstClaim)); + assert.isTrue(Option.isNone(duplicateClaim)); + + assert.lengthOf(yield* repository.listDue({ now: "2026-07-22T12:01:30.000Z", limit: 10 }), 1); + const recoveredClaim = yield* repository.claim({ + id: claimWaitpoint.id, + now: "2026-07-22T12:01:30.000Z", + leaseExpiresAt: "2026-07-22T12:02:30.000Z", + }); + assert.isTrue(Option.isSome(recoveredClaim)); + + yield* repository.markDelivered({ + id: claimWaitpoint.id, + deliveredAt: "2026-07-22T12:01:31.000Z", + }); + assert.deepStrictEqual( + yield* repository.listDue({ now: "2026-07-22T12:03:00.000Z", limit: 10 }), + [], + ); + }).pipe(Effect.provide(repositoryLayer)), + ); + + it.effect("reschedules transient failures and expires waits that reach their deadline", () => + Effect.gen(function* () { + const repository = yield* GitHubWaitpointRepository; + const retryWaitpoint = { ...registeredWaitpoint, id: "github:thread-1:call-retry" }; + yield* repository.register(retryWaitpoint); + + yield* repository.reschedule({ + id: retryWaitpoint.id, + nextPollAt: "2026-07-22T12:02:00.000Z", + updatedAt: "2026-07-22T12:01:00.000Z", + lastError: "GitHub is temporarily unavailable.", + }); + const rescheduled = yield* repository.getById({ id: retryWaitpoint.id }); + assert.isTrue(Option.isSome(rescheduled)); + if (Option.isNone(rescheduled)) return; + assert.deepInclude(rescheduled.value, { + state: "pending", + nextPollAt: "2026-07-22T12:02:00.000Z", + lastError: "GitHub is temporarily unavailable.", + deliveryLeaseExpiresAt: null, + }); + + yield* repository.markExpired({ + id: retryWaitpoint.id, + expiredAt: "2026-07-22T12:02:01.000Z", + lastError: "Waitpoint deadline elapsed.", + }); + const expired = yield* repository.getById({ id: retryWaitpoint.id }); + assert.isTrue(Option.isSome(expired)); + if (Option.isNone(expired)) return; + assert.deepInclude(expired.value, { + state: "expired", + lastError: "Waitpoint deadline elapsed.", + }); + }).pipe(Effect.provide(repositoryLayer)), + ); +}); diff --git a/apps/server/src/persistence/GitHubWaitpoints.ts b/apps/server/src/persistence/GitHubWaitpoints.ts new file mode 100644 index 00000000000..67d6751a0b3 --- /dev/null +++ b/apps/server/src/persistence/GitHubWaitpoints.ts @@ -0,0 +1,398 @@ +/** + * Durable storage for GitHub-backed thread waitpoints. + * + * A waitpoint is keyed by the originating dynamic-tool call. Registration is + * therefore idempotent even if Codex retries the same request after a transport + * interruption. + */ +import { IsoDateTime, NonNegativeInt, ThreadId } from "@t3tools/contracts"; +import * as Context from "effect/Context"; +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 * as Struct from "effect/Struct"; +import * as SqlClient from "effect/unstable/sql/SqlClient"; +import * as SqlSchema from "effect/unstable/sql/SqlSchema"; + +import { + PersistenceDecodeError, + PersistenceSqlError, + type GitHubWaitpointRepositoryError, +} from "./Errors.ts"; +export type { GitHubWaitpointRepositoryError } from "./Errors.ts"; + +export const GitHubWaitpointCondition = Schema.Literals([ + "checks_settled", + "new_review_activity", + "pull_request_closed", +]); +export type GitHubWaitpointCondition = typeof GitHubWaitpointCondition.Type; + +export const GitHubWaitpointState = Schema.Literals([ + "pending", + "delivering", + "delivered", + "expired", +]); +export type GitHubWaitpointState = typeof GitHubWaitpointState.Type; + +export const GitHubWaitpoint = Schema.Struct({ + id: Schema.String, + threadId: ThreadId, + originatingTurnId: Schema.String, + repository: Schema.String, + pullRequestNumber: Schema.Int, + condition: GitHubWaitpointCondition, + baseline: Schema.Unknown, + continuationPrompt: Schema.String, + state: GitHubWaitpointState, + nextPollAt: IsoDateTime, + deadlineAt: IsoDateTime, + deliveryLeaseExpiresAt: Schema.NullOr(IsoDateTime), + attemptCount: NonNegativeInt, + lastError: Schema.NullOr(Schema.String), + createdAt: IsoDateTime, + updatedAt: IsoDateTime, + deliveredAt: Schema.NullOr(IsoDateTime), +}); +export type GitHubWaitpoint = typeof GitHubWaitpoint.Type; + +export const RegisterGitHubWaitpointInput = Schema.Struct({ + id: GitHubWaitpoint.fields.id, + threadId: GitHubWaitpoint.fields.threadId, + originatingTurnId: GitHubWaitpoint.fields.originatingTurnId, + repository: GitHubWaitpoint.fields.repository, + pullRequestNumber: GitHubWaitpoint.fields.pullRequestNumber, + condition: GitHubWaitpoint.fields.condition, + baseline: GitHubWaitpoint.fields.baseline, + continuationPrompt: GitHubWaitpoint.fields.continuationPrompt, + nextPollAt: GitHubWaitpoint.fields.nextPollAt, + deadlineAt: GitHubWaitpoint.fields.deadlineAt, + createdAt: GitHubWaitpoint.fields.createdAt, +}); +export type RegisterGitHubWaitpointInput = typeof RegisterGitHubWaitpointInput.Type; + +const GetGitHubWaitpointInput = Schema.Struct({ id: Schema.String }); +export type GetGitHubWaitpointInput = typeof GetGitHubWaitpointInput.Type; +const ListDueGitHubWaitpointsInput = Schema.Struct({ + now: IsoDateTime, + limit: Schema.Int.check(Schema.isGreaterThan(0)), +}); +const ClaimGitHubWaitpointInput = Schema.Struct({ + id: Schema.String, + now: IsoDateTime, + leaseExpiresAt: IsoDateTime, +}); +const MarkGitHubWaitpointDeliveredInput = Schema.Struct({ + id: Schema.String, + deliveredAt: IsoDateTime, +}); +const RescheduleGitHubWaitpointInput = Schema.Struct({ + id: Schema.String, + nextPollAt: IsoDateTime, + updatedAt: IsoDateTime, + lastError: Schema.NullOr(Schema.String), +}); +const MarkGitHubWaitpointExpiredInput = Schema.Struct({ + id: Schema.String, + expiredAt: IsoDateTime, + lastError: Schema.String, +}); + +export class GitHubWaitpointRepository extends Context.Service< + GitHubWaitpointRepository, + { + readonly register: ( + input: RegisterGitHubWaitpointInput, + ) => Effect.Effect; + readonly getById: ( + input: GetGitHubWaitpointInput, + ) => Effect.Effect, GitHubWaitpointRepositoryError>; + readonly listDue: ( + input: typeof ListDueGitHubWaitpointsInput.Type, + ) => Effect.Effect, GitHubWaitpointRepositoryError>; + readonly claim: ( + input: typeof ClaimGitHubWaitpointInput.Type, + ) => Effect.Effect, GitHubWaitpointRepositoryError>; + readonly markDelivered: ( + input: typeof MarkGitHubWaitpointDeliveredInput.Type, + ) => Effect.Effect; + readonly reschedule: ( + input: typeof RescheduleGitHubWaitpointInput.Type, + ) => Effect.Effect; + readonly markExpired: ( + input: typeof MarkGitHubWaitpointExpiredInput.Type, + ) => Effect.Effect; + } +>()("t3/persistence/GitHubWaitpoints/GitHubWaitpointRepository") {} + +const GitHubWaitpointDbRow = GitHubWaitpoint.mapFields( + Struct.assign({ baseline: Schema.fromJsonString(Schema.Unknown) }), +); +const GitHubWaitpointRawDbRow = Schema.Struct({ + id: Schema.Unknown, + threadId: Schema.Unknown, + originatingTurnId: Schema.Unknown, + repository: Schema.Unknown, + pullRequestNumber: Schema.Unknown, + condition: Schema.Unknown, + baseline: Schema.Unknown, + continuationPrompt: Schema.Unknown, + state: Schema.Unknown, + nextPollAt: Schema.Unknown, + deadlineAt: Schema.Unknown, + deliveryLeaseExpiresAt: Schema.Unknown, + attemptCount: Schema.Unknown, + lastError: Schema.Unknown, + createdAt: Schema.Unknown, + updatedAt: Schema.Unknown, + deliveredAt: Schema.Unknown, +}); +const RegisterGitHubWaitpointDbInput = RegisterGitHubWaitpointInput.mapFields( + Struct.assign({ baseline: Schema.fromJsonString(Schema.Unknown) }), +); + +function mapRepositoryError(operation: string, threadId?: ThreadId) { + return (cause: unknown): GitHubWaitpointRepositoryError => + Schema.isSchemaError(cause) + ? PersistenceDecodeError.fromSchemaError( + `${operation}:decode`, + cause, + threadId === undefined ? undefined : { threadId }, + ) + : new PersistenceSqlError({ + operation: `${operation}:query`, + ...(threadId === undefined ? {} : { correlation: { threadId } }), + cause, + }); +} + +export const make = Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + + const insertWaitpoint = SqlSchema.void({ + Request: RegisterGitHubWaitpointDbInput, + execute: (input) => sql` + INSERT INTO github_waitpoints ( + id, + thread_id, + originating_turn_id, + repository, + pull_request_number, + condition, + baseline_json, + continuation_prompt, + state, + next_poll_at, + deadline_at, + delivery_lease_expires_at, + attempt_count, + last_error, + created_at, + updated_at, + delivered_at + ) + VALUES ( + ${input.id}, + ${input.threadId}, + ${input.originatingTurnId}, + ${input.repository}, + ${input.pullRequestNumber}, + ${input.condition}, + ${input.baseline}, + ${input.continuationPrompt}, + 'pending', + ${input.nextPollAt}, + ${input.deadlineAt}, + NULL, + 0, + NULL, + ${input.createdAt}, + ${input.createdAt}, + NULL + ) + ON CONFLICT (id) DO NOTHING + `, + }); + + const selectWaitpoint = SqlSchema.findOneOption({ + Request: GetGitHubWaitpointInput, + Result: GitHubWaitpointRawDbRow, + execute: ({ id }) => sql` + SELECT + id, + thread_id AS "threadId", + originating_turn_id AS "originatingTurnId", + repository, + pull_request_number AS "pullRequestNumber", + condition, + baseline_json AS baseline, + continuation_prompt AS "continuationPrompt", + state, + next_poll_at AS "nextPollAt", + deadline_at AS "deadlineAt", + delivery_lease_expires_at AS "deliveryLeaseExpiresAt", + attempt_count AS "attemptCount", + last_error AS "lastError", + created_at AS "createdAt", + updated_at AS "updatedAt", + delivered_at AS "deliveredAt" + FROM github_waitpoints + WHERE id = ${id} + `, + }); + + const listDueWaitpoints = SqlSchema.findAll({ + Request: ListDueGitHubWaitpointsInput, + Result: GitHubWaitpointRawDbRow, + execute: ({ now, limit }) => sql` + SELECT + id, + thread_id AS "threadId", + originating_turn_id AS "originatingTurnId", + repository, + pull_request_number AS "pullRequestNumber", + condition, + baseline_json AS baseline, + continuation_prompt AS "continuationPrompt", + state, + next_poll_at AS "nextPollAt", + deadline_at AS "deadlineAt", + delivery_lease_expires_at AS "deliveryLeaseExpiresAt", + attempt_count AS "attemptCount", + last_error AS "lastError", + created_at AS "createdAt", + updated_at AS "updatedAt", + delivered_at AS "deliveredAt" + FROM github_waitpoints + WHERE (state = 'pending' AND next_poll_at <= ${now}) + OR (state = 'delivering' AND delivery_lease_expires_at <= ${now}) + ORDER BY next_poll_at ASC, created_at ASC, id ASC + LIMIT ${limit} + `, + }); + + const claimWaitpoint = SqlSchema.findOneOption({ + Request: ClaimGitHubWaitpointInput, + Result: GitHubWaitpointRawDbRow, + execute: ({ id, now, leaseExpiresAt }) => sql` + UPDATE github_waitpoints + SET + state = 'delivering', + delivery_lease_expires_at = ${leaseExpiresAt}, + attempt_count = attempt_count + 1, + updated_at = ${now} + WHERE id = ${id} + AND ( + (state = 'pending' AND next_poll_at <= ${now}) + OR (state = 'delivering' AND delivery_lease_expires_at <= ${now}) + ) + RETURNING + id, + thread_id AS "threadId", + originating_turn_id AS "originatingTurnId", + repository, + pull_request_number AS "pullRequestNumber", + condition, + baseline_json AS baseline, + continuation_prompt AS "continuationPrompt", + state, + next_poll_at AS "nextPollAt", + deadline_at AS "deadlineAt", + delivery_lease_expires_at AS "deliveryLeaseExpiresAt", + attempt_count AS "attemptCount", + last_error AS "lastError", + created_at AS "createdAt", + updated_at AS "updatedAt", + delivered_at AS "deliveredAt" + `, + }); + + const markWaitpointDelivered = SqlSchema.void({ + Request: MarkGitHubWaitpointDeliveredInput, + execute: ({ id, deliveredAt }) => sql` + UPDATE github_waitpoints + SET + state = 'delivered', + delivery_lease_expires_at = NULL, + last_error = NULL, + delivered_at = ${deliveredAt}, + updated_at = ${deliveredAt} + WHERE id = ${id} AND state = 'delivering' + `, + }); + + const rescheduleWaitpoint = SqlSchema.void({ + Request: RescheduleGitHubWaitpointInput, + execute: ({ id, nextPollAt, updatedAt, lastError }) => sql` + UPDATE github_waitpoints + SET + state = 'pending', + next_poll_at = ${nextPollAt}, + delivery_lease_expires_at = NULL, + last_error = ${lastError}, + updated_at = ${updatedAt} + WHERE id = ${id} AND state IN ('pending', 'delivering') + `, + }); + + const expireWaitpoint = SqlSchema.void({ + Request: MarkGitHubWaitpointExpiredInput, + execute: ({ id, expiredAt, lastError }) => sql` + UPDATE github_waitpoints + SET + state = 'expired', + delivery_lease_expires_at = NULL, + last_error = ${lastError}, + updated_at = ${expiredAt} + WHERE id = ${id} AND state IN ('pending', 'delivering') + `, + }); + + const decodeWaitpoint = Schema.decodeUnknownEffect(GitHubWaitpointDbRow); + const decodeOptionalWaitpoint = Option.match({ + onNone: () => Effect.succeed(Option.none()), + onSome: (row: typeof GitHubWaitpointRawDbRow.Type) => + decodeWaitpoint(row).pipe(Effect.map(Option.some)), + }); + + return GitHubWaitpointRepository.of({ + register: (input) => + insertWaitpoint(input).pipe( + Effect.mapError(mapRepositoryError("GitHubWaitpointRepository.register", input.threadId)), + ), + getById: (input) => + selectWaitpoint(input).pipe( + Effect.mapError(mapRepositoryError("GitHubWaitpointRepository.getById")), + Effect.flatMap(decodeOptionalWaitpoint), + Effect.mapError(mapRepositoryError("GitHubWaitpointRepository.getById")), + ), + listDue: (input) => + listDueWaitpoints(input).pipe( + Effect.mapError(mapRepositoryError("GitHubWaitpointRepository.listDue")), + Effect.flatMap((rows) => Effect.forEach(rows, (row) => decodeWaitpoint(row))), + Effect.mapError(mapRepositoryError("GitHubWaitpointRepository.listDue")), + ), + claim: (input) => + claimWaitpoint(input).pipe( + Effect.mapError(mapRepositoryError("GitHubWaitpointRepository.claim")), + Effect.flatMap(decodeOptionalWaitpoint), + Effect.mapError(mapRepositoryError("GitHubWaitpointRepository.claim")), + ), + markDelivered: (input) => + markWaitpointDelivered(input).pipe( + Effect.mapError(mapRepositoryError("GitHubWaitpointRepository.markDelivered")), + ), + reschedule: (input) => + rescheduleWaitpoint(input).pipe( + Effect.mapError(mapRepositoryError("GitHubWaitpointRepository.reschedule")), + ), + markExpired: (input) => + expireWaitpoint(input).pipe( + Effect.mapError(mapRepositoryError("GitHubWaitpointRepository.markExpired")), + ), + }); +}); + +export const layer = Layer.effect(GitHubWaitpointRepository, make); diff --git a/apps/server/src/persistence/Migrations.ts b/apps/server/src/persistence/Migrations.ts index cacb2c85b83..df941edb6bd 100644 --- a/apps/server/src/persistence/Migrations.ts +++ b/apps/server/src/persistence/Migrations.ts @@ -46,6 +46,7 @@ import Migration0030 from "./Migrations/030_ProjectionThreadShellArchiveIndexes. import Migration0031 from "./Migrations/031_AuthAuthorizationScopes.ts"; import Migration0032 from "./Migrations/032_AuthPairingProofKeyThumbprint.ts"; import Migration0033 from "./Migrations/033_ProjectionThreadsSettled.ts"; +import Migration0034 from "./Migrations/034_GitHubWaitpoints.ts"; /** * Migration loader with all migrations defined inline. @@ -91,6 +92,7 @@ export const migrationEntries = [ [31, "AuthAuthorizationScopes", Migration0031], [32, "AuthPairingProofKeyThumbprint", Migration0032], [33, "ProjectionThreadsSettled", Migration0033], + [34, "GitHubWaitpoints", Migration0034], ] as const; export const makeMigrationLoader = (throughId?: number) => diff --git a/apps/server/src/persistence/Migrations/034_GitHubWaitpoints.test.ts b/apps/server/src/persistence/Migrations/034_GitHubWaitpoints.test.ts new file mode 100644 index 00000000000..2ed734875ce --- /dev/null +++ b/apps/server/src/persistence/Migrations/034_GitHubWaitpoints.test.ts @@ -0,0 +1,51 @@ +import { assert, it } from "@effect/vitest"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as SqlClient from "effect/unstable/sql/SqlClient"; + +import { runMigrations } from "../Migrations.ts"; +import * as NodeSqliteClient from "../NodeSqliteClient.ts"; + +const layer = it.layer(Layer.mergeAll(NodeSqliteClient.layerMemory())); + +layer("034_GitHubWaitpoints", (it) => { + it.effect("installs durable GitHub waitpoint storage and its due-work index", () => + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + + yield* runMigrations({ toMigrationInclusive: 33 }); + yield* runMigrations({ toMigrationInclusive: 34 }); + + const columns = yield* sql<{ readonly name: string }>` + PRAGMA table_info(github_waitpoints) + `; + const indexes = yield* sql<{ readonly name: string }>` + PRAGMA index_list(github_waitpoints) + `; + + assert.deepStrictEqual( + columns.map((column) => column.name), + [ + "id", + "thread_id", + "originating_turn_id", + "repository", + "pull_request_number", + "condition", + "baseline_json", + "continuation_prompt", + "state", + "next_poll_at", + "deadline_at", + "delivery_lease_expires_at", + "attempt_count", + "last_error", + "created_at", + "updated_at", + "delivered_at", + ], + ); + assert.ok(indexes.some((index) => index.name === "idx_github_waitpoints_due")); + }), + ); +}); diff --git a/apps/server/src/persistence/Migrations/034_GitHubWaitpoints.ts b/apps/server/src/persistence/Migrations/034_GitHubWaitpoints.ts new file mode 100644 index 00000000000..56daf2d7544 --- /dev/null +++ b/apps/server/src/persistence/Migrations/034_GitHubWaitpoints.ts @@ -0,0 +1,35 @@ +import * as Effect from "effect/Effect"; +import * as SqlClient from "effect/unstable/sql/SqlClient"; + +export default Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + + yield* sql` + CREATE TABLE IF NOT EXISTS github_waitpoints ( + id TEXT PRIMARY KEY, + thread_id TEXT NOT NULL, + originating_turn_id TEXT NOT NULL, + repository TEXT NOT NULL, + pull_request_number INTEGER NOT NULL, + condition TEXT NOT NULL CHECK ( + condition IN ('checks_settled', 'new_review_activity', 'pull_request_closed') + ), + baseline_json TEXT NOT NULL, + continuation_prompt TEXT NOT NULL, + state TEXT NOT NULL CHECK (state IN ('pending', 'delivering', 'delivered', 'expired')), + next_poll_at TEXT NOT NULL, + deadline_at TEXT NOT NULL, + delivery_lease_expires_at TEXT, + attempt_count INTEGER NOT NULL DEFAULT 0, + last_error TEXT, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL, + delivered_at TEXT + ) + `; + + yield* sql` + CREATE INDEX IF NOT EXISTS idx_github_waitpoints_due + ON github_waitpoints (state, next_poll_at, delivery_lease_expires_at) + `; +}); diff --git a/apps/server/src/provider/Drivers/CodexDriver.ts b/apps/server/src/provider/Drivers/CodexDriver.ts index ffcc94ca77d..81792292a47 100644 --- a/apps/server/src/provider/Drivers/CodexDriver.ts +++ b/apps/server/src/provider/Drivers/CodexDriver.ts @@ -34,6 +34,7 @@ import { ChildProcessSpawner } from "effect/unstable/process"; import { makeCodexTextGeneration } from "../../textGeneration/CodexTextGeneration.ts"; import { ServerConfig } from "../../config.ts"; import { ServerSettingsService } from "../../serverSettings.ts"; +import { GitHubWaitpointRegistration } from "../../github/GitHubWaitpointRegistration.ts"; import { ProviderDriverError } from "../Errors.ts"; import { makeCodexAdapter } from "../Layers/CodexAdapter.ts"; import { checkCodexProviderStatus, makePendingCodexProvider } from "../Layers/CodexProvider.ts"; @@ -119,6 +120,7 @@ export const CodexDriver: ProviderDriver = { const httpClient = yield* HttpClient.HttpClient; const serverSettings = yield* ServerSettingsService; const eventLoggers = yield* ProviderEventLoggers; + const gitHubWaitpointRegistration = yield* GitHubWaitpointRegistration; const processEnv = mergeProviderInstanceEnvironment(environment); const homeLayout = yield* resolveCodexHomeLayout(config); const continuationIdentity = codexContinuationIdentity(homeLayout); @@ -158,6 +160,7 @@ export const CodexDriver: ProviderDriver = { const adapter = yield* makeCodexAdapter(effectiveConfig, { instanceId, environment: processEnv, + registerGitHubWaitpoint: gitHubWaitpointRegistration.register, ...(eventLoggers.native ? { nativeEventLogger: eventLoggers.native } : {}), }); const textGeneration = yield* makeCodexTextGeneration(effectiveConfig, processEnv); diff --git a/apps/server/src/provider/Layers/CodexAdapter.ts b/apps/server/src/provider/Layers/CodexAdapter.ts index 38a5887cdc3..d1e288ffbdf 100644 --- a/apps/server/src/provider/Layers/CodexAdapter.ts +++ b/apps/server/src/provider/Layers/CodexAdapter.ts @@ -40,6 +40,7 @@ import * as EffectCodexSchema from "effect-codex-app-server/schema"; import { getModelSelectionStringOptionValue } from "@t3tools/shared/model"; import { getCodexServiceTierOptionValue } from "../../codexModelOptions.ts"; import * as McpProviderSession from "../../mcp/McpProviderSession.ts"; +import type { GitHubWaitpointRegistrationService } from "../../github/GitHubWaitpointRegistration.ts"; import { ProviderAdapterRequestError, @@ -83,6 +84,7 @@ export interface CodexAdapterLiveOptions { >; readonly nativeEventLogPath?: string; readonly nativeEventLogger?: EventNdjsonLogger; + readonly registerGitHubWaitpoint?: GitHubWaitpointRegistrationService["register"]; } interface CodexAdapterSessionContext { @@ -1411,6 +1413,9 @@ export const makeCodexAdapter = Effect.fn("makeCodexAdapter")(function* ( ? { model: input.modelSelection.model } : {}), ...(serviceTier ? { serviceTier } : {}), + ...(options?.registerGitHubWaitpoint + ? { registerGitHubWaitpoint: options.registerGitHubWaitpoint } + : {}), ...(mcpSession ? { environment: { diff --git a/apps/server/src/provider/Layers/CodexDynamicTools.test.ts b/apps/server/src/provider/Layers/CodexDynamicTools.test.ts index 9a6d5a49222..2fc4498b033 100644 --- a/apps/server/src/provider/Layers/CodexDynamicTools.test.ts +++ b/apps/server/src/provider/Layers/CodexDynamicTools.test.ts @@ -1,6 +1,7 @@ import * as NodeAssert from "node:assert/strict"; import { it } from "@effect/vitest"; +import { ThreadId } from "@t3tools/contracts"; import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; import * as Fiber from "effect/Fiber"; @@ -13,6 +14,7 @@ import { makeT3CodexDynamicToolWaitRegistry, T3_CODEX_DYNAMIC_TOOLS, T3_CODEX_DYNAMIC_TOOL_NAMESPACE, + T3_CODEX_GITHUB_WAIT_TOOL_NAME, T3_CODEX_WAIT_MAX_DURATION_MS, T3_CODEX_WAIT_MIN_DURATION_MS, T3_CODEX_WAIT_TOOL_NAME, @@ -33,10 +35,17 @@ function waitCall( }; } +function githubWaitCall(arguments_: unknown): EffectCodexSchema.DynamicToolCallParams { + return waitCall(arguments_, { + callId: "github-wait-call-1", + tool: T3_CODEX_GITHUB_WAIT_TOOL_NAME, + }); +} + describe("T3 Codex dynamic tools", () => { it("publishes a bounded namespaced wait tool", () => { const namespace = T3_CODEX_DYNAMIC_TOOLS[0]; - const wait = namespace?.tools[0]; + const wait = namespace?.tools.find((tool) => tool.name === T3_CODEX_WAIT_TOOL_NAME); NodeAssert.equal(namespace?.type, "namespace"); NodeAssert.equal(namespace?.name, T3_CODEX_DYNAMIC_TOOL_NAMESPACE); @@ -62,6 +71,47 @@ describe("T3 Codex dynamic tools", () => { }); }); + it.effect("registers a durable GitHub waitpoint and returns without holding the turn open", () => + Effect.gen(function* () { + let registered: unknown; + const response = yield* handleT3CodexDynamicToolCall( + githubWaitCall({ + repository: "pingdotgg/t3code", + pullRequestNumber: 4262, + condition: "checks_settled", + timeoutMinutes: 60, + reason: "Wait for CI before addressing failures.", + }), + Effect.never, + { + threadId: ThreadId.make("thread-1"), + cwd: "/tmp/project", + registerGitHubWaitpoint: (input) => + Effect.sync(() => { + registered = input; + return { id: "github:thread-1:github-wait-call-1" }; + }), + }, + ); + + NodeAssert.equal(response.success, true); + NodeAssert.match( + response.contentItems[0]?.type === "inputText" ? response.contentItems[0].text : "", + /registered/i, + ); + NodeAssert.deepStrictEqual(registered, { + idempotencyKey: "github-wait-call-1", + threadId: ThreadId.make("thread-1"), + cwd: "/tmp/project", + repository: "pingdotgg/t3code", + pullRequestNumber: 4262, + condition: "checks_settled", + timeoutMinutes: 60, + reason: "Wait for CI before addressing failures.", + }); + }), + ); + it.effect("does not complete before the requested duration", () => Effect.gen(function* () { const fiber = yield* handleT3CodexDynamicToolCall( diff --git a/apps/server/src/provider/Layers/CodexDynamicTools.ts b/apps/server/src/provider/Layers/CodexDynamicTools.ts index 4ccc8ca8d96..9b841f6408c 100644 --- a/apps/server/src/provider/Layers/CodexDynamicTools.ts +++ b/apps/server/src/provider/Layers/CodexDynamicTools.ts @@ -5,11 +5,17 @@ import * as Ref from "effect/Ref"; import * as Result from "effect/Result"; import * as Schema from "effect/Schema"; import type * as EffectCodexSchema from "effect-codex-app-server/schema"; +import type { ThreadId } from "@t3tools/contracts"; + +import type { GitHubWaitpointCondition } from "../../persistence/GitHubWaitpoints.ts"; export const T3_CODEX_DYNAMIC_TOOL_NAMESPACE = "t3"; export const T3_CODEX_WAIT_TOOL_NAME = "wait"; +export const T3_CODEX_GITHUB_WAIT_TOOL_NAME = "await_github"; export const T3_CODEX_WAIT_MIN_DURATION_MS = 1_000; export const T3_CODEX_WAIT_MAX_DURATION_MS = 3_600_000; +export const T3_CODEX_GITHUB_WAIT_DEFAULT_TIMEOUT_MINUTES = 24 * 60; +export const T3_CODEX_GITHUB_WAIT_MAX_TIMEOUT_MINUTES = 7 * 24 * 60; const T3CodexWaitArguments = Schema.Struct({ durationMs: Schema.Int.check(Schema.isGreaterThanOrEqualTo(T3_CODEX_WAIT_MIN_DURATION_MS)).check( @@ -18,6 +24,39 @@ const T3CodexWaitArguments = Schema.Struct({ reason: Schema.optionalKey(Schema.String.check(Schema.isMaxLength(500))), }); const decodeT3CodexWaitArguments = Schema.decodeUnknownEffect(T3CodexWaitArguments); +const T3CodexGitHubWaitArguments = Schema.Struct({ + repository: Schema.String.check(Schema.isPattern(/^[^/\s]+\/[^/\s]+$/)).check( + Schema.isMaxLength(200), + ), + pullRequestNumber: Schema.Int.check(Schema.isGreaterThan(0)), + condition: Schema.Literals(["checks_settled", "new_review_activity", "pull_request_closed"]), + timeoutMinutes: Schema.optionalKey( + Schema.Int.check(Schema.isGreaterThan(0)).check( + Schema.isLessThanOrEqualTo(T3_CODEX_GITHUB_WAIT_MAX_TIMEOUT_MINUTES), + ), + ), + reason: Schema.optionalKey(Schema.String.check(Schema.isMaxLength(500))), +}); +const decodeT3CodexGitHubWaitArguments = Schema.decodeUnknownEffect(T3CodexGitHubWaitArguments); + +export interface RegisterGitHubWaitpointInput { + readonly idempotencyKey: string; + readonly threadId: ThreadId; + readonly cwd: string; + readonly repository: string; + readonly pullRequestNumber: number; + readonly condition: GitHubWaitpointCondition; + readonly timeoutMinutes: number; + readonly reason?: string; +} + +export interface T3CodexDynamicToolContext { + readonly threadId: ThreadId; + readonly cwd: string; + readonly registerGitHubWaitpoint: ( + input: RegisterGitHubWaitpointInput, + ) => Effect.Effect<{ readonly id: string }, { readonly message: string }>; +} export const T3_CODEX_DYNAMIC_TOOLS = [ { @@ -49,6 +88,47 @@ export const T3_CODEX_DYNAMIC_TOOLS = [ additionalProperties: false, }, }, + { + type: "function", + name: T3_CODEX_GITHUB_WAIT_TOOL_NAME, + description: + "Register a durable T3 Code waitpoint for a GitHub pull request. T3 polls GitHub locally through the authenticated gh CLI without model inference, survives a T3 restart, and starts one continuation turn when the condition is met. After registration, finish the current turn instead of polling.", + inputSchema: { + type: "object", + properties: { + repository: { + type: "string", + pattern: "^[^/\\s]+/[^/\\s]+$", + maxLength: 200, + description: "GitHub repository in owner/name form.", + }, + pullRequestNumber: { + type: "integer", + minimum: 1, + description: "Pull request number to watch.", + }, + condition: { + type: "string", + enum: ["checks_settled", "new_review_activity", "pull_request_closed"], + description: "GitHub condition that should resume this thread.", + }, + timeoutMinutes: { + type: "integer", + minimum: 1, + maximum: T3_CODEX_GITHUB_WAIT_MAX_TIMEOUT_MINUTES, + default: T3_CODEX_GITHUB_WAIT_DEFAULT_TIMEOUT_MINUTES, + description: "Stop watching after this many minutes (default 24 hours).", + }, + reason: { + type: "string", + maxLength: 500, + description: "Optional short explanation of the work to continue.", + }, + }, + required: ["repository", "pullRequestNumber", "condition"], + additionalProperties: false, + }, + }, ], }, ] as const satisfies ReadonlyArray; @@ -63,11 +143,49 @@ function textResponse(success: boolean, text: string): EffectCodexSchema.Dynamic export const handleT3CodexDynamicToolCall = Effect.fn("handleT3CodexDynamicToolCall")(function* ( payload: EffectCodexSchema.DynamicToolCallParams, cancelled: Effect.Effect = Effect.never, + context?: T3CodexDynamicToolContext, ): Effect.fn.Return { - if ( - payload.namespace !== T3_CODEX_DYNAMIC_TOOL_NAMESPACE || - payload.tool !== T3_CODEX_WAIT_TOOL_NAME - ) { + if (payload.namespace !== T3_CODEX_DYNAMIC_TOOL_NAMESPACE) { + return textResponse( + false, + `Unknown T3 Code dynamic tool: ${payload.namespace ?? ""}.${payload.tool}`, + ); + } + + if (payload.tool === T3_CODEX_GITHUB_WAIT_TOOL_NAME) { + if (context === undefined) { + return textResponse(false, "Durable GitHub waits are unavailable in this T3 Code runtime."); + } + const decoded = yield* Effect.result(decodeT3CodexGitHubWaitArguments(payload.arguments)); + if (Result.isFailure(decoded)) { + return textResponse( + false, + "Invalid GitHub wait arguments. Use owner/repository, a positive pull request number, and a supported condition.", + ); + } + const registration = yield* Effect.result( + context.registerGitHubWaitpoint({ + idempotencyKey: payload.callId, + threadId: context.threadId, + cwd: context.cwd, + repository: decoded.success.repository, + pullRequestNumber: decoded.success.pullRequestNumber, + condition: decoded.success.condition, + timeoutMinutes: + decoded.success.timeoutMinutes ?? T3_CODEX_GITHUB_WAIT_DEFAULT_TIMEOUT_MINUTES, + ...(decoded.success.reason === undefined ? {} : { reason: decoded.success.reason }), + }), + ); + if (Result.isFailure(registration)) { + return textResponse(false, `Could not register GitHub wait: ${registration.failure.message}`); + } + return textResponse( + true, + `GitHub wait registered as ${registration.success.id}. Finish this turn now; T3 Code will resume the thread when the condition is met.`, + ); + } + + if (payload.tool !== T3_CODEX_WAIT_TOOL_NAME) { return textResponse( false, `Unknown T3 Code dynamic tool: ${payload.namespace ?? ""}.${payload.tool}`, @@ -112,7 +230,9 @@ export interface T3CodexDynamicToolWaitRegistry { } export const makeT3CodexDynamicToolWaitRegistry = Effect.fn("makeT3CodexDynamicToolWaitRegistry")( - function* (): Effect.fn.Return { + function* ( + context?: T3CodexDynamicToolContext, + ): Effect.fn.Return { const stateRef = yield* Ref.make({ pending: new Map(), cancelledTurnIds: new Set(), @@ -142,6 +262,7 @@ export const makeT3CodexDynamicToolWaitRegistry = Effect.fn("makeT3CodexDynamicT return yield* handleT3CodexDynamicToolCall( payload, registered ? Deferred.await(cancelled) : Effect.void, + context, ).pipe( Effect.ensuring( Ref.update(stateRef, (current) => { diff --git a/apps/server/src/provider/Layers/CodexSessionRuntime.ts b/apps/server/src/provider/Layers/CodexSessionRuntime.ts index 4fd1cc847b7..bf49181acbb 100644 --- a/apps/server/src/provider/Layers/CodexSessionRuntime.ts +++ b/apps/server/src/provider/Layers/CodexSessionRuntime.ts @@ -40,6 +40,7 @@ import { makeT3CodexDynamicToolWaitRegistry, T3_CODEX_DYNAMIC_TOOLS } from "./Co import { codexSessionAppServerArgs } from "./codexLaunchArgs.ts"; import { expandHomePath } from "../../pathExpansion.ts"; import { buildCodexDeveloperInstructions } from "../CodexDeveloperInstructions.ts"; +import type { GitHubWaitpointRegistrationService } from "../../github/GitHubWaitpointRegistration.ts"; const decodeV2TurnStartResponse = Schema.decodeUnknownEffect(EffectCodexSchema.V2TurnStartResponse); const decodeV2ThreadStartResponse = Schema.decodeUnknownEffect( EffectCodexSchema.V2ThreadStartResponse, @@ -110,6 +111,7 @@ export interface CodexSessionRuntimeOptions { readonly serviceTier?: CodexServiceTier | undefined; readonly resumeCursor?: CodexResumeCursor; readonly appServerArgs?: ReadonlyArray; + readonly registerGitHubWaitpoint?: GitHubWaitpointRegistrationService["register"]; } export interface CodexSessionRuntimeSendTurnInput { @@ -743,7 +745,15 @@ export const makeCodexSessionRuntime = ( const pendingApprovalsRef = yield* Ref.make(new Map()); const approvalCorrelationsRef = yield* Ref.make(new Map()); const pendingUserInputsRef = yield* Ref.make(new Map()); - const dynamicToolWaitRegistry = yield* makeT3CodexDynamicToolWaitRegistry(); + const dynamicToolWaitRegistry = yield* makeT3CodexDynamicToolWaitRegistry( + options.registerGitHubWaitpoint === undefined + ? undefined + : { + threadId: options.threadId, + cwd: options.cwd, + registerGitHubWaitpoint: options.registerGitHubWaitpoint, + }, + ); const collabReceiverTurnsRef = yield* Ref.make(new Map()); const closedRef = yield* Ref.make(false); diff --git a/apps/server/src/server.ts b/apps/server/src/server.ts index 0e6db87b109..a6443628452 100644 --- a/apps/server/src/server.ts +++ b/apps/server/src/server.ts @@ -30,6 +30,7 @@ import * as CheckpointStore from "./checkpointing/CheckpointStore.ts"; import * as AzureDevOpsCli from "./sourceControl/AzureDevOpsCli.ts"; import * as BitbucketApi from "./sourceControl/BitbucketApi.ts"; import * as GitHubCli from "./sourceControl/GitHubCli.ts"; +import * as GitHubWaitpointRuntime from "./github/GitHubWaitpointRuntime.ts"; import * as GitLabCli from "./sourceControl/GitLabCli.ts"; import * as TextGeneration from "./textGeneration/TextGeneration.ts"; import { ProviderInstanceRegistryHydrationLive } from "./provider/Layers/ProviderInstanceRegistryHydration.ts"; @@ -183,6 +184,11 @@ const ProviderLayerLive = ProviderServiceLive.pipe( const PersistenceLayerLive = Layer.empty.pipe(Layer.provideMerge(SqlitePersistenceLayerLive)); +const OrchestrationAndGitHubWaitpointLayerLive = GitHubWaitpointRuntime.layer.pipe( + Layer.provideMerge(OrchestrationLayerLive), + Layer.provideMerge(PersistenceLayerLive), +); + const VcsDriverRegistryLayerLive = VcsDriverRegistry.layer.pipe( Layer.provide(VcsProjectConfig.layer), ); @@ -281,7 +287,7 @@ const CloudManagedEndpointRuntimeLive = Layer.mergeAll( const ProviderRuntimeLayerLive = ProviderSessionReaperLive.pipe( Layer.provideMerge(ProviderLayerLive), - Layer.provideMerge(OrchestrationLayerLive), + Layer.provideMerge(OrchestrationAndGitHubWaitpointLayerLive), ); const RuntimeCoreDependenciesLive = ReactorLayerLive.pipe( diff --git a/docs/project/deferred-thread-resume.md b/docs/project/deferred-thread-resume.md index f734f94a443..34010a60bda 100644 --- a/docs/project/deferred-thread-resume.md +++ b/docs/project/deferred-thread-resume.md @@ -13,8 +13,8 @@ This project is split into two deliberately different capabilities: implementation milestone and the scope of the initial PR. 2. **Durable resume** — the current turn ends, T3 persists a waitpoint, and a scheduler starts a new turn after a timer or external event. This is required for restarts, webhook triggers, and - long-lived watches such as pull-request CI. It is a follow-up because its lifecycle and delivery - semantics are materially different from a blocking tool call. + long-lived watches such as pull-request CI. The first durable adapter is a local GitHub watcher; + it does not depend on T3 Connect or public webhook infrastructure. ## Problem @@ -110,9 +110,9 @@ Suggested waitpoint fields: Delivery should be at-least-once with an atomic claim and idempotent continuation key. On startup, the scheduler reloads pending waitpoints and re-arms timers. External triggers should enter through authenticated, scoped adapters rather than exposing a generic unauthenticated callback. A GitHub PR -watcher can then translate check-suite, review, and pull-request webhooks into normalized waitpoint -signals. The sidebar/thread lifecycle should add an explicit `waiting` state only for this durable -mode; a live wait remains an active tool call. +watcher can translate polling results or future webhooks into normalized waitpoint signals. An +explicit sidebar `waiting` state remains a possible follow-up; the first durable adapter relies on +the registered tool activity and normal thread-running transition when delivery occurs. ## Alternatives considered @@ -127,6 +127,46 @@ mode; a live wait remains an active tool call. - **Keep a WebSocket open for every trigger:** useful as an adapter transport, but not sufficient as the source of truth because sockets do not survive process or network failure. +## Durable GitHub watcher + +The second implementation milestone adds `t3.await_github` for Codex. The tool takes an +`owner/repository`, pull-request number, condition, optional reason, and timeout. Registration reads +an initial snapshot through the authenticated `gh` CLI, persists the waitpoint, and returns +immediately. The agent is instructed to finish its current turn instead of polling. + +Supported conditions are deliberately narrow: + +- `checks_settled`: at least one reported check exists and every check is terminal. Failed checks + still satisfy the condition so the resumed agent can diagnose them. +- `new_review_activity`: a review or issue comment appears that was not in the registration snapshot. +- `pull_request_closed`: the pull request is merged or closed without merging. + +The local worker wakes every five seconds to find due database rows. It first verifies that the +originating T3 turn has settled, then queries GitHub no more than once per row every 30 seconds. A +busy thread is retried after five seconds without querying GitHub. The first poll occurs 30 seconds +after registration. If the user advances the thread with another turn, the older waitpoint expires +instead of injecting a stale continuation later. + +Delivery uses a 60-second database lease and a deterministic orchestration command id. If T3 stops +after dispatching the continuation but before marking the row delivered, the lease expires and the +worker retries the same command id; the orchestration command-receipt store deduplicates it. Pending +rows and expired leases are discovered from SQLite on startup, so no in-memory timer is the source +of truth. + +The continuation is a clearly identified T3 GitHub watcher message containing the observed result. +Normal orchestration events then move the thread back to its running state in connected clients. +No bespoke webhook route, GitHub App, T3 Connect link, or new frontend state is required for this +milestone. + +### Durable GitHub non-goals + +- Hosted monitoring while the local T3 server is offline. The worker catches up when T3 runs again. +- Generic arbitrary webhooks or user-supplied callback URLs. +- Durable timer waits; `t3.wait` remains the live timer primitive. +- Cross-provider exposure before those providers support an equivalent host-tool hook. +- Conditional HTTP requests or cross-waitpoint GitHub request coalescing. Those are optimizations if + observed watch volume approaches GitHub API limits. + ## Test plan Focused automated coverage: @@ -138,6 +178,11 @@ Focused automated coverage: - Duration bounds, malformed arguments, wrong namespaces, and unknown tool names fail without sleep. - Cancellation wins the race and returns a failed/cancelled tool result. - Existing provider adapter tests continue to verify canonical dynamic-tool activity mapping. +- Migration and repository tests cover idempotent registration, due ordering, exclusive claims, + expired-lease recovery, rescheduling, delivery, and expiry. +- GitHub adapter tests cover normalization and each supported condition. +- Worker tests cover active-turn deferral, satisfied-condition delivery, and duplicate suppression. +- The orchestration gateway test verifies deterministic continuation command and message ids. Targeted validation: @@ -154,6 +199,11 @@ omitting the tool rather than breaking the session. Existing provider event inge dynamic-tool lifecycle. A follow-up may add wait duration/cancellation metrics once the contract is used beyond Codex. +The GitHub milestone adds migration 034 and no feature flag. Removing the runtime layer stops new +polls without deleting rows; reverting the code leaves the additive table inert. Structured logs are +emitted when a waitpoint is registered, delivered, or a worker tick fails. Per-condition counters and +request coalescing can be added after real watch volume is observed. + ## Risks - Experimental app-server protocol changes: isolate the raw field construction and response decode @@ -163,6 +213,10 @@ used beyond Codex. it before sending the interrupt request. - Confusing live and durable behavior: name and document the initial tool as a live wait and do not claim restart or webhook guarantees. +- GitHub CLI authentication loss: retain the pending row, store the last error, and retry until its + deadline rather than waking the model with an infrastructure failure. +- Duplicate continuation after a crash: use a delivery lease plus deterministic orchestration + command id so the existing command receipt is the idempotency boundary. ## Definition of done for the initial PR @@ -172,3 +226,13 @@ used beyond Codex. - Focused tests and server typecheck pass. - The provider documentation explains the capability and its live-session limitation. - The PR explicitly links this design and leaves durable waitpoints as follow-up scope. + +## Definition of done for the GitHub milestone + +- `t3.await_github` validates and snapshots a pull request through authenticated `gh`. +- Registration returns immediately and the agent is told to finish its current turn. +- Pending rows and expired delivery leases survive a T3 restart. +- A busy thread is never probed or resumed; a satisfied condition starts one continuation turn. +- Missing threads and elapsed deadlines move the waitpoint to `expired`. +- Focused migration, repository, adapter, worker, gateway, and Codex runtime tests pass. +- The server package typechecks and the live `gh` payload shape is smoke-tested against a real PR. diff --git a/docs/providers/codex.md b/docs/providers/codex.md index e9e68d2c228..5decfc29af3 100644 --- a/docs/providers/codex.md +++ b/docs/providers/codex.md @@ -35,9 +35,18 @@ T3 Code and does not repeatedly wake the model, so the agent can perform one sta wait instead of spending inference on frequent polling. This is a live-session wait. It is cancelled if the current turn is interrupted, the provider -session closes, or T3 Code stops. It does not wake a completed thread after an app restart and does -not subscribe to external webhooks. Durable timers and event-triggered continuation are described in -the [deferred thread resume design](../project/deferred-thread-resume.md). +session closes, or T3 Code stops. + +For pull-request work, Codex also receives `t3.await_github`. It can wait for all reported checks to +settle, a new review or comment, or the pull request to merge or close. T3 stores this wait in its +local database, checks GitHub through your authenticated `gh` CLI, and starts one continuation turn +when the condition is met. This survives a T3 restart and does not require T3 Connect or a public +webhook endpoint. The local T3 server must eventually run again to observe the change and resume the +thread. + +GitHub waits expire after 24 hours by default and may be bounded from one minute through seven days. +See the [deferred thread resume design](../project/deferred-thread-resume.md) for delivery and recovery +semantics. ## I Want Work And Personal Codex Accounts