From 5b2d183254f64d13ef81ea445750df845b9c064a Mon Sep 17 00:00:00 2001 From: Rene Date: Wed, 9 Sep 2026 18:58:43 +0200 Subject: [PATCH] fix(server): Route explicit native goal control Bind goal reads and status updates to the persisted T3 provider instance and live native session. Preserve objective, budget and accounting, reject stale owners and repeated attempts, and leave asynchronous continuation with Codex. Cover HTTP scopes, routing, active-turn guards, uncertain RPC outcomes and a disposable native quota-denial round trip. --- .../src/orchestration/goalControlHttp.test.ts | 162 ++++++++++ apps/server/src/orchestration/http.ts | 21 ++ .../src/provider/Layers/CodexAdapter.test.ts | 8 + .../src/provider/Layers/CodexAdapter.ts | 12 + .../CodexGoalControl.integration.test.ts | 149 ++++++++++ .../provider/Layers/CodexGoalControl.test.ts | 276 ++++++++++++++++++ .../src/provider/Layers/CodexGoalControl.ts | 171 +++++++++++ .../provider/Layers/CodexSessionRuntime.ts | 18 +- .../Layers/ProviderGoalControl.test.ts | 124 ++++++++ .../provider/Layers/ProviderGoalControl.ts | 83 ++++++ .../src/provider/Layers/ProviderService.ts | 2 + .../src/provider/Services/ProviderAdapter.ts | 7 + .../src/provider/Services/ProviderService.ts | 7 + packages/contracts/src/environmentHttp.ts | 13 + packages/contracts/src/index.ts | 1 + packages/contracts/src/providerGoal.ts | 75 +++++ 16 files changed, 1128 insertions(+), 1 deletion(-) create mode 100644 apps/server/src/orchestration/goalControlHttp.test.ts create mode 100644 apps/server/src/provider/Layers/CodexGoalControl.integration.test.ts create mode 100644 apps/server/src/provider/Layers/CodexGoalControl.test.ts create mode 100644 apps/server/src/provider/Layers/CodexGoalControl.ts create mode 100644 apps/server/src/provider/Layers/ProviderGoalControl.test.ts create mode 100644 apps/server/src/provider/Layers/ProviderGoalControl.ts create mode 100644 packages/contracts/src/providerGoal.ts diff --git a/apps/server/src/orchestration/goalControlHttp.test.ts b/apps/server/src/orchestration/goalControlHttp.test.ts new file mode 100644 index 0000000000000..eed56b10c3237 --- /dev/null +++ b/apps/server/src/orchestration/goalControlHttp.test.ts @@ -0,0 +1,162 @@ +import * as NodeServices from "@effect/platform-node/NodeServices"; +import * as NodeHttpPlatform from "@effect/platform-node/NodeHttpPlatform"; +import * as Schema from "effect/Schema"; +import * as Etag from "effect/unstable/http/Etag"; +import * as ServerConfig from "../config.ts"; +import * as WorkspacePaths from "../workspace/WorkspacePaths.ts"; +import { assert, describe, it } from "vite-plus/test"; +import { + AuthSessionId, + AuthOrchestrationReadScope, + AuthOrchestrationOperateScope, + EnvironmentOrchestrationHttpApi, + EnvironmentAuthenticatedAuth, + EnvironmentAuthenticatedPrincipal, + EnvironmentAuthInvalidError, + ProviderGoalControlError, + ProviderInstanceId, + ThreadId, + type AuthEnvironmentScope, + type ProviderGoalControlInput, +} from "@t3tools/contracts"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as HttpApi from "effect/unstable/httpapi/HttpApi"; +import * as HttpApiBuilder from "effect/unstable/httpapi/HttpApiBuilder"; +import * as HttpRouter from "effect/unstable/http/HttpRouter"; +import { orchestrationHttpApiLayer } from "./http.ts"; +import { ProjectionSnapshotQuery } from "./Services/ProjectionSnapshotQuery.ts"; +import { OrchestrationEngineService } from "./Services/OrchestrationEngine.ts"; +import { ProviderService } from "../provider/Services/ProviderService.ts"; + +const decodeGoalError = Schema.decodeUnknownSync(ProviderGoalControlError); +class TestApi extends HttpApi.make("environment").add(EnvironmentOrchestrationHttpApi) {} +const get: ProviderGoalControlInput = { + action: "get", + threadId: ThreadId.make("disposable-http"), + expectedProviderInstanceId: ProviderInstanceId.make("codex-test"), + expectedNativeThreadId: "native-http", +}; +const resume: Extract = { + ...get, + action: "resume", + ownerId: "b220998f-3b30-4b02-a9c7-91696d58222f", + attemptId: "e208fae3-8a65-4f79-9fb9-340a81f5d08f", + expectedGoal: { + threadId: "native-http", + objective: "Fixture", + status: "usageLimited", + tokenBudget: null, + tokensUsed: 321, + timeUsedSeconds: 5, + createdAt: 1, + updatedAt: 2, + }, +}; +function fixture(scopes: ReadonlyArray | null, deny = false) { + let calls = 0; + const auth = Layer.succeed(EnvironmentAuthenticatedAuth, (httpEffect) => + scopes === null + ? Effect.fail( + new EnvironmentAuthInvalidError({ + code: "auth_invalid", + reason: "missing_credential", + traceId: "test", + }), + ) + : Effect.provideService(httpEffect, EnvironmentAuthenticatedPrincipal, { + sessionId: AuthSessionId.make("test"), + subject: "test-user", + method: "bearer-access-token", + scopes: new Set(scopes), + }), + ); + const service = Layer.succeed(ProviderService, { + goalControl: (input: ProviderGoalControlInput) => { + calls++; + return deny + ? Effect.fail( + new ProviderGoalControlError({ + reason: "native_request_failed", + message: "usage_limit_exceeded", + nativeCode: 429, + }), + ) + : Effect.succeed({ + threadId: input.threadId, + providerInstanceId: input.expectedProviderInstanceId, + ownerId: resume.ownerId, + goal: + input.action === "get" + ? resume.expectedGoal + : { ...input.expectedGoal, status: "active" as const }, + turnDispatched: false as const, + continuationObserved: false as const, + }); + }, + } as unknown as ProviderService["Service"]); + const routes = HttpApiBuilder.layer(TestApi).pipe( + Layer.provide(orchestrationHttpApiLayer), + Layer.provide(auth), + Layer.provide(Layer.succeed(ProjectionSnapshotQuery, {} as ProjectionSnapshotQuery["Service"])), + Layer.provide( + Layer.succeed(OrchestrationEngineService, {} as OrchestrationEngineService["Service"]), + ), + Layer.provideMerge(service), + Layer.provideMerge( + Layer.mergeAll( + NodeHttpPlatform.layer, + Etag.layer, + WorkspacePaths.layer, + ServerConfig.layerTest(process.cwd(), { prefix: "t3-goal-http-" }), + ).pipe(Layer.provideMerge(NodeServices.layer)), + ), + ); + const web = HttpRouter.toWebHandler(routes, { disableLogger: true }); + return { + ...web, + calls: () => calls, + request: (body: unknown) => + web.handler( + new Request("http://localhost/api/orchestration/goal-control", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify(body), + }), + ), + }; +} +describe("goal-control HTTP contract", () => { + it("requires authentication and operate scope for resume; read scope alone can inspect", async () => { + for (const [scopes, body, status, calls] of [ + [null, get, 401, 0], + [[AuthOrchestrationReadScope], get, 200, 1], + [[AuthOrchestrationReadScope], resume, 403, 0], + [[AuthOrchestrationOperateScope], resume, 200, 1], + ] as const) { + const f = fixture(scopes); + try { + assert.equal((await f.request(body)).status, status); + assert.equal(f.calls(), calls); + } finally { + await f.dispose(); + } + } + }); + it("rejects missing resume ownership and preserves native quota diagnostics", async () => { + const f = fixture([AuthOrchestrationOperateScope], true); + try { + assert.equal((await f.request({ ...get, action: "resume" })).status, 400); + assert.equal(f.calls(), 0); + const response = await f.request(resume); + assert.equal(response.status, 409); + const body = decodeGoalError(await response.json()); + assert.equal(body.reason, "native_request_failed"); + assert.equal(body.nativeCode, 429); + assert.equal(body.message, "usage_limit_exceeded"); + assert.equal(f.calls(), 1); + } finally { + await f.dispose(); + } + }); +}); diff --git a/apps/server/src/orchestration/http.ts b/apps/server/src/orchestration/http.ts index f7147106c7a9b..90738adf4da04 100644 --- a/apps/server/src/orchestration/http.ts +++ b/apps/server/src/orchestration/http.ts @@ -1,5 +1,6 @@ import { AuthOrchestrationOperateScope, + ProviderGoalControlError, AuthOrchestrationReadScope, EnvironmentHttpApi, } from "@t3tools/contracts"; @@ -7,6 +8,7 @@ import * as Effect from "effect/Effect"; import * as Option from "effect/Option"; import * as HttpApiBuilder from "effect/unstable/httpapi/HttpApiBuilder"; +import { ProviderService } from "../provider/Services/ProviderService.ts"; import { projectThreadDetailSnapshot } from "./ActivityPayloadProjection.ts"; import { cleanupFailedUploadedAttachments, normalizeDispatchCommand } from "./Normalizer.ts"; import { @@ -27,6 +29,25 @@ export const orchestrationHttpApiLayer = HttpApiBuilder.group( const orchestrationEngine = yield* OrchestrationEngineService; return handlers + .handle( + "goalControl", + Effect.fn("environment.orchestration.goalControl")(function* (args) { + yield* annotateEnvironmentRequest(args.endpoint.name); + yield* requireEnvironmentScope( + args.payload.action === "get" + ? AuthOrchestrationReadScope + : AuthOrchestrationOperateScope, + ); + const service = yield* Effect.serviceOption(ProviderService); + if (Option.isNone(service) || !service.value.goalControl) { + return yield* new ProviderGoalControlError({ + reason: "unsupported", + message: "This server does not expose native goal control.", + }); + } + return yield* service.value.goalControl(args.payload); + }), + ) .handle( "snapshot", Effect.fn("environment.orchestration.snapshot")(function* (args) { diff --git a/apps/server/src/provider/Layers/CodexAdapter.test.ts b/apps/server/src/provider/Layers/CodexAdapter.test.ts index f7c6036885d9d..0aa88618ab148 100644 --- a/apps/server/src/provider/Layers/CodexAdapter.test.ts +++ b/apps/server/src/provider/Layers/CodexAdapter.test.ts @@ -5,6 +5,7 @@ import * as NodeOS from "node:os"; import * as NodePath from "node:path"; import { ApprovalRequestId, + ProviderGoalControlError, CodexSettings, EventId, ProviderDriverKind, @@ -60,6 +61,13 @@ const asEventId = (value: string): EventId => EventId.make(value); const asItemId = (value: string): ProviderItemId => ProviderItemId.make(value); class FakeCodexRuntime implements CodexSessionRuntimeShape { + readonly goalControl: CodexSessionRuntimeShape["goalControl"] = () => + Effect.fail( + new ProviderGoalControlError({ + reason: "unsupported", + message: "Goal control is not configured in this fixture.", + }), + ); private readonly eventQueue = Effect.runSync(Queue.unbounded()); private readonly now = "2026-01-01T00:00:00.000Z"; diff --git a/apps/server/src/provider/Layers/CodexAdapter.ts b/apps/server/src/provider/Layers/CodexAdapter.ts index 2d88e58dc1fb4..92228c631b113 100644 --- a/apps/server/src/provider/Layers/CodexAdapter.ts +++ b/apps/server/src/provider/Layers/CodexAdapter.ts @@ -9,6 +9,7 @@ */ import { EventId, + ProviderGoalControlError, type CanonicalItemType, type CanonicalRequestType, type CodexSettings, @@ -2713,6 +2714,17 @@ export const makeCodexAdapter = Effect.fn("makeCodexAdapter")(function* ( promptlessTurnContinuation: true, }, startSession, + goalControl: (input) => + requireSession(input.threadId).pipe( + Effect.mapError( + () => + new ProviderGoalControlError({ + reason: "owner_unavailable", + message: "The owning Codex session is no longer loaded.", + }), + ), + Effect.flatMap((session) => session.runtime.goalControl(input)), + ), sendTurn, compaction: { type: "native", start: compactThread }, interruptTurn, diff --git a/apps/server/src/provider/Layers/CodexGoalControl.integration.test.ts b/apps/server/src/provider/Layers/CodexGoalControl.integration.test.ts new file mode 100644 index 0000000000000..ba9333b4c29a6 --- /dev/null +++ b/apps/server/src/provider/Layers/CodexGoalControl.integration.test.ts @@ -0,0 +1,149 @@ +// @effect-diagnostics nodeBuiltinImport:off - disposable loopback quota fixture. +import * as NodeHttp from "node:http"; +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { it } from "@effect/vitest"; +import { ProviderInstanceId, ThreadId } from "@t3tools/contracts"; +import * as Effect from "effect/Effect"; +import * as Fiber from "effect/Fiber"; +import * as Stream from "effect/Stream"; +import * as FileSystem from "effect/FileSystem"; +import * as Schema from "effect/Schema"; +import { ChildProcess, ChildProcessSpawner } from "effect/unstable/process"; +import * as CodexSchema from "effect-codex-app-server/schema"; +import * as CodexClient from "effect-codex-app-server/client"; +import { assert, describe } from "vite-plus/test"; +import { makeCodexSessionRuntime } from "./CodexSessionRuntime.ts"; +import { buildCodexInitializeParams } from "./CodexProvider.ts"; + +// Opt in with a native binary. The disposable provider points only to an offline loopback endpoint. +const binary = process.env.T3_CODEX_GOAL_TEST_BINARY; +const decodeCompleted = Schema.decodeUnknownEffect(CodexSchema.V2TurnCompletedNotification); +const decodeCursor = Schema.decodeUnknownEffect(Schema.Struct({ threadId: Schema.String })); +describe.runIf(binary)("native Codex goal-control round trip", () => { + it.effect( + "resumes a disposable goal through its owning runtime with one native continuation and no accounting reset", + () => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const home = yield* fs.makeTempDirectoryScoped({ prefix: "t3-goal-roundtrip-" }); + const denialServer = yield* Effect.acquireRelease( + Effect.promise(async () => { + const server = NodeHttp.createServer((request, response) => { + request.resume(); + response.writeHead(429, { "content-type": "application/json" }); + response.end( + '{"error":{"type":"usage_limit_reached","message":"Disposable quota denial","plan_type":"plus","resets_at":4102444800}}', + ); + }); + await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve)); + return server; + }), + (server) => + Effect.promise(() => new Promise((resolve) => server.close(() => resolve()))), + ); + const address = denialServer.address(); + if (address === null || typeof address === "string") + return yield* Effect.die("Missing quota fixture port"); + yield* fs.writeFileString( + `${home}/config.toml`, + `model_provider = "offline" +[model_providers.offline] +name = "offline" +base_url = "http://127.0.0.1:${address.port}/v1" +wire_api = "responses" +request_max_retries = 0 +stream_max_retries = 0 +`, + ); + const env = { HOME: home, CODEX_HOME: home, PATH: process.env.PATH ?? "" }; + const seeded = yield* Effect.gen(function* () { + const spawner = yield* ChildProcessSpawner.ChildProcessSpawner; + const child = yield* spawner.spawn( + ChildProcess.make(binary!, ["app-server"], { + cwd: home, + env, + extendEnv: false, + forceKillAfter: "2 seconds", + }), + ); + return yield* Effect.gen(function* () { + const client = yield* CodexClient.CodexAppServerClient; + yield* client.request("initialize", buildCodexInitializeParams()); + yield* client.notify("initialized", undefined); + const thread = yield* client.request("thread/start", { cwd: home }); + return (yield* client.request("thread/goal/set", { + threadId: thread.thread.id, + objective: "Disposable goal-control fixture", + status: "paused", + tokenBudget: 1024, + })).goal; + }).pipe(Effect.provide(CodexClient.layerChildProcess(child))); + }).pipe(Effect.scoped); + const threadId = ThreadId.make("disposable-t3-goal-thread"); + const providerInstanceId = ProviderInstanceId.make("offline-fixture"); + const runtime = yield* makeCodexSessionRuntime({ + threadId, + providerInstanceId, + binaryPath: binary!, + homePath: home, + environment: env, + cwd: home, + runtimeMode: "full-access", + resumeCursor: { threadId: seeded.threadId }, + }); + const session = yield* runtime.start(); + const cursor = yield* decodeCursor(session.resumeCursor); + assert.equal(cursor.threadId, seeded.threadId); + const binding = { + threadId, + expectedProviderInstanceId: providerInstanceId, + expectedNativeThreadId: seeded.threadId, + }; + const snapshot = yield* runtime.goalControl({ ...binding, action: "get" }); + assert.deepEqual(snapshot.goal, seeded); + const before = yield* runtime.readThread; + const continuation = yield* runtime.events.pipe( + Stream.takeUntil((event) => event.method === "turn/completed"), + Stream.runCollect, + Effect.forkScoped, + ); + const input = { + ...binding, + action: "resume" as const, + ownerId: snapshot.ownerId, + attemptId: "43d82415-87de-441e-8110-72223d33c8f5", + expectedGoal: seeded, + }; + const resumed = yield* runtime.goalControl(input); + assert.equal(resumed.goal?.status, "active"); + assert.equal(resumed.goal?.objective, seeded.objective); + assert.equal(resumed.goal?.tokenBudget, seeded.tokenBudget); + assert.equal(resumed.goal?.tokensUsed, seeded.tokensUsed); + assert.equal(resumed.goal?.timeUsedSeconds, seeded.timeUsedSeconds); + assert.equal(resumed.goal?.createdAt, seeded.createdAt); + assert.isFalse(resumed.turnDispatched); + assert.equal( + (yield* runtime.goalControl(input).pipe(Effect.flip)).reason, + "attempt_already_used", + ); + const observed = yield* Fiber.join(continuation); + const after = yield* runtime.readThread; + assert.equal(observed.filter((event) => event.method === "turn/started").length, 1); + assert.equal(after.turns.length, before.turns.length + 1); + const terminal = observed.find((event) => event.method === "turn/completed"); + assert.isDefined(terminal); + const completed = yield* decodeCompleted(terminal!.payload); + assert.equal(completed.turn.status, "failed"); + assert.equal(completed.turn.error?.codexErrorInfo, "usageLimitExceeded"); + const limited = yield* runtime.goalControl({ ...binding, action: "get" }); + assert.equal(limited.goal?.status, "usageLimited"); + assert.equal(limited.goal?.objective, seeded.objective); + assert.equal(limited.goal?.tokenBudget, seeded.tokenBudget); + assert.equal( + (yield* runtime.goalControl(input).pipe(Effect.flip)).reason, + "attempt_already_used", + ); + yield* runtime.close; + }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)), + ); +}); diff --git a/apps/server/src/provider/Layers/CodexGoalControl.test.ts b/apps/server/src/provider/Layers/CodexGoalControl.test.ts new file mode 100644 index 0000000000000..d6e86d180a421 --- /dev/null +++ b/apps/server/src/provider/Layers/CodexGoalControl.test.ts @@ -0,0 +1,276 @@ +import { it } from "@effect/vitest"; +import { assert, describe } from "vite-plus/test"; +import { + ProviderInstanceId, + ProviderDriverKind, + ThreadId, + type ProviderNativeGoal, + type ProviderSession, + type ProviderGoalControlInput, +} from "@t3tools/contracts"; +import * as Deferred from "effect/Deferred"; +import * as Fiber from "effect/Fiber"; +import * as TestClock from "effect/testing/TestClock"; +import * as Effect from "effect/Effect"; +import * as Semaphore from "effect/Semaphore"; +import * as CodexErrors from "effect-codex-app-server/errors"; +import type * as CodexClient from "effect-codex-app-server/client"; +import { makeCodexGoalControl } from "./CodexGoalControl.ts"; + +const ownerId = "b220998f-3b30-4b02-a9c7-91696d58222f"; +const attemptId = "e208fae3-8a65-4f79-9fb9-340a81f5d08f"; +const instance = ProviderInstanceId.make("codex-fixture"); +const t3 = ThreadId.make("t3-disposable"); +const get: ProviderGoalControlInput = { + action: "get", + threadId: t3, + expectedProviderInstanceId: instance, + expectedNativeThreadId: "native-disposable", +}; +const initial: ProviderNativeGoal = { + threadId: "native-disposable", + objective: "Keep the existing goal", + status: "usageLimited", + tokenBudget: 1000, + tokensUsed: 321, + timeUsedSeconds: 45, + createdAt: 10, + updatedAt: 20, +}; +const setup = Effect.fn("goalFixture")(function* () { + let goal: ProviderNativeGoal | null = { ...initial }; + let status = "idle"; + let session: ProviderSession = { + threadId: t3, + provider: ProviderDriverKind.make("codex"), + providerInstanceId: instance, + status: "ready", + runtimeMode: "full-access", + resumeCursor: { threadId: initial.threadId }, + createdAt: "2026-09-09T00:00:00Z", + updatedAt: "2026-09-09T00:00:00Z", + }; + let error: CodexErrors.CodexAppServerRequestError | undefined; + let corrupt = false; + let hang = false; + let denyStatus = false; + const entered = yield* Deferred.make(); + const calls: Array<{ method: string; params: unknown }> = []; + const request = (method: string, params: unknown) => + Effect.suspend((): Effect.Effect => { + calls.push({ method, params }); + if (method === "thread/goal/get") return Effect.succeed({ goal }); + if (method === "thread/read") + return Effect.succeed({ thread: { id: initial.threadId, status: { type: status } } }); + assert.equal(method, "thread/goal/set", "goal control must never wake or interrupt a turn"); + if (error) return Effect.fail(error); + if (hang) return Deferred.succeed(entered, undefined).pipe(Effect.andThen(Effect.never)); + const update = params as { status: ProviderNativeGoal["status"] }; + goal = { + ...goal!, + status: denyStatus ? "usageLimited" : update.status, + updatedAt: 21, + ...(corrupt ? { tokensUsed: 0 } : {}), + }; + return Effect.succeed({ goal }); + }); + const mutex = yield* Semaphore.make(1); + const control = makeCodexGoalControl({ + ownerId, + session: Effect.sync(() => session), + mutex, + client: { request: request as CodexClient.CodexAppServerClient["Service"]["request"] }, + }); + const resume = (): Extract => ({ + ...get, + action: "resume", + ownerId, + attemptId, + expectedGoal: { ...goal! }, + }); + return { + control, + calls, + resume, + mutex, + goal: () => goal, + setGoal: (value: ProviderNativeGoal | null) => { + goal = value; + }, + setStatus: (value: string) => { + status = value; + }, + setSession: (value: Partial) => { + session = { ...session, ...value }; + }, + deny: () => { + error = new CodexErrors.CodexAppServerRequestError({ + code: 429, + errorMessage: "usage_limit_exceeded: try after reset", + }); + }, + entered: Deferred.await(entered), + hang: () => { + hang = true; + }, + denyStatus: () => { + denyStatus = true; + }, + corrupt: () => { + corrupt = true; + }, + }; +}); + +describe("native goal control", () => { + it.effect("bounds a lost native response and refuses to replay its mutation", () => + Effect.gen(function* () { + const f = yield* setup(); + f.hang(); + const input = f.resume(); + const pending = yield* f.control(input).pipe(Effect.flip, Effect.forkScoped); + yield* f.entered; + yield* TestClock.adjust("10 seconds"); + const error = yield* Fiber.join(pending); + assert.equal(error.reason, "native_request_failed"); + assert.include(error.message, "timed out"); + assert.equal((yield* f.control(input).pipe(Effect.flip)).reason, "attempt_already_used"); + assert.equal(f.calls.filter((c) => c.method === "thread/goal/set").length, 1); + }).pipe(Effect.scoped), + ); + it.effect("returns a native usage-limited result without silently resetting it again", () => + Effect.gen(function* () { + const f = yield* setup(); + f.denyStatus(); + assert.equal((yield* f.control(f.resume())).goal?.status, "usageLimited"); + assert.equal(f.calls.filter((c) => c.method === "thread/goal/set").length, 1); + }), + ); + + it.effect( + "reads and resumes the exact native thread without replacing accounting or starting a turn", + () => + Effect.gen(function* () { + const f = yield* setup(); + assert.deepEqual((yield* f.control(get)).goal, initial); + const result = yield* f.control(f.resume()); + assert.equal(result.goal?.status, "active"); + assert.equal(result.goal?.tokensUsed, 321); + assert.equal(result.goal?.timeUsedSeconds, 45); + assert.equal(result.goal?.tokenBudget, 1000); + assert.equal(result.goal?.objective, initial.objective); + assert.isFalse(result.turnDispatched); + assert.deepEqual( + f.calls.filter((c) => c.method === "thread/goal/set"), + [{ method: "thread/goal/set", params: { threadId: initial.threadId, status: "active" } }], + ); + }), + ); + it.effect("allows explicit recovery from a previous failed turn without an active turn", () => + Effect.gen(function* () { + const f = yield* setup(); + f.setSession({ status: "error" }); + f.setStatus("systemError"); + assert.equal((yield* f.control(f.resume())).goal?.status, "active"); + }), + ); + it.effect( + "preserves an unbudgeted goal and supports pause through the same status-only API", + () => + Effect.gen(function* () { + const f = yield* setup(); + f.setGoal({ ...initial, tokenBudget: null }); + yield* f.control({ ...f.resume(), action: "pause" }); + assert.equal(f.goal()?.status, "paused"); + assert.isNull(f.goal()?.tokenBudget); + }), + ); + it.effect("rejects wrong or stale live owners before RPC", () => + Effect.gen(function* () { + for (const patch of [ + { ownerId: attemptId }, + { expectedNativeThreadId: "other" }, + { expectedProviderInstanceId: ProviderInstanceId.make("other") }, + { threadId: ThreadId.make("other") }, + ]) { + const f = yield* setup(); + assert.equal( + (yield* f.control({ ...f.resume(), ...patch }).pipe(Effect.flip)).reason, + "owner_mismatch", + ); + assert.isEmpty(f.calls); + } + }), + ); + it.effect("rejects a closed owner rather than recovering or starting another session", () => + Effect.gen(function* () { + const f = yield* setup(); + f.setSession({ status: "closed" }); + assert.equal((yield* f.control(get).pipe(Effect.flip)).reason, "owner_unavailable"); + assert.isEmpty(f.calls); + }), + ); + it.effect("does not change or interrupt active turns even when local session state lags", () => + Effect.gen(function* () { + for (const local of [true, false]) { + const f = yield* setup(); + if (local) f.setSession({ status: "running" }); + else f.setStatus("active"); + assert.equal((yield* f.control(f.resume()).pipe(Effect.flip)).reason, "active_turn"); + assert.isFalse(f.calls.some((c) => c.method === "thread/goal/set")); + } + }), + ); + it.effect("rejects stale snapshots, completed goals and exhausted budgets", () => + Effect.gen(function* () { + for (const [patch, reason] of [ + [{ status: "complete" }, "terminal_goal"], + [{ tokenBudget: 321 }, "budget_exhausted"], + ] as const) { + const f = yield* setup(); + f.setGoal({ ...initial, ...patch }); + assert.equal((yield* f.control(f.resume()).pipe(Effect.flip)).reason, reason); + } + const f = yield* setup(); + const input = f.resume(); + f.setGoal({ ...initial, tokensUsed: 322 }); + assert.equal((yield* f.control(input).pipe(Effect.flip)).reason, "goal_changed"); + f.setGoal(null); + assert.equal((yield* f.control(input).pipe(Effect.flip)).reason, "goal_missing"); + }), + ); + it.effect("propagates native quota denial and never replays the same attempt", () => + Effect.gen(function* () { + const f = yield* setup(); + f.deny(); + const input = f.resume(); + const error = yield* f.control(input).pipe(Effect.flip); + assert.equal(error.reason, "native_request_failed"); + assert.equal(error.nativeCode, 429); + assert.include(error.message, "usage_limit_exceeded"); + assert.equal((yield* f.control(input).pipe(Effect.flip)).reason, "attempt_already_used"); + assert.equal(f.calls.filter((c) => c.method === "thread/goal/set").length, 1); + assert.deepEqual(f.goal(), initial); + }), + ); + it.effect("serializes concurrent resumes and rejects duplicate mutation attempts", () => + Effect.gen(function* () { + const f = yield* setup(); + const input = f.resume(); + const outcomes = yield* Effect.all( + [f.control(input).pipe(Effect.result), f.control(input).pipe(Effect.result)], + { concurrency: "unbounded" }, + ); + assert.equal(outcomes.filter((r) => r._tag === "Success").length, 1); + assert.equal(f.calls.filter((c) => c.method === "thread/goal/set").length, 1); + }), + ); + it.effect("reports unexpected native accounting changes without a corrective reset", () => + Effect.gen(function* () { + const f = yield* setup(); + f.corrupt(); + assert.equal((yield* f.control(f.resume()).pipe(Effect.flip)).reason, "native_state_changed"); + assert.equal(f.calls.filter((c) => c.method === "thread/goal/set").length, 1); + }), + ); +}); diff --git a/apps/server/src/provider/Layers/CodexGoalControl.ts b/apps/server/src/provider/Layers/CodexGoalControl.ts new file mode 100644 index 0000000000000..c809e59248742 --- /dev/null +++ b/apps/server/src/provider/Layers/CodexGoalControl.ts @@ -0,0 +1,171 @@ +import { + ProviderGoalControlError, + type ProviderGoalControlInput, + type ProviderGoalControlResult, + ProviderNativeGoal, + type ProviderSession, +} from "@t3tools/contracts"; +import * as Effect from "effect/Effect"; +import * as Schema from "effect/Schema"; +import type * as Semaphore from "effect/Semaphore"; +import * as CodexErrors from "effect-codex-app-server/errors"; +import type * as CodexClient from "effect-codex-app-server/client"; + +const goalEquivalence = Schema.toEquivalence(ProviderNativeGoal); +const decodeCursor = Schema.decodeUnknownOption(Schema.Struct({ threadId: Schema.String })); +const isGoalControlError = Schema.is(ProviderGoalControlError); +const isNativeRequestError = Schema.is(CodexErrors.CodexAppServerRequestError); +const sameGoal = (a: ProviderNativeGoal, b: ProviderNativeGoal) => + goalEquivalence( + { ...a, tokenBudget: a.tokenBudget ?? null }, + { ...b, tokenBudget: b.tokenBudget ?? null }, + ); + +export const makeCodexGoalControl = (options: { + readonly ownerId: string; + readonly session: Effect.Effect; + readonly client: Pick; + readonly mutex: Semaphore.Semaphore; +}) => { + const attempts = new Set(); + return Effect.fn("CodexGoalControl.control")(function* ( + input: ProviderGoalControlInput, + ): Effect.fn.Return { + return yield* Effect.gen(function* () { + const session = yield* options.session; + const cursor = decodeCursor(session.resumeCursor); + if (session.status === "closed" || session.status === "connecting") { + return yield* new ProviderGoalControlError({ + reason: "owner_unavailable", + message: "The owning Codex session is not ready.", + }); + } + if ( + session.threadId !== input.threadId || + cursor._tag === "None" || + cursor.value.threadId !== input.expectedNativeThreadId || + session.providerInstanceId !== input.expectedProviderInstanceId || + (input.action !== "get" && input.ownerId !== options.ownerId) + ) { + return yield* new ProviderGoalControlError({ + reason: "owner_mismatch", + message: + "Goal control owner does not match the live Codex session. Read a fresh owner before retrying.", + }); + } + const threadId = cursor.value.threadId; + const response = yield* options.client.request("thread/goal/get", { threadId }); + let goal = response.goal ?? null; + if (goal && goal.threadId !== threadId) { + return yield* new ProviderGoalControlError({ + reason: "owner_mismatch", + message: "Codex returned a goal for another native thread.", + }); + } + if (input.action !== "get") { + if (attempts.has(input.attemptId)) { + return yield* new ProviderGoalControlError({ + reason: "attempt_already_used", + message: + "This goal-control attempt was already sent; read the goal to reconcile its outcome. It will not be replayed.", + }); + } + if (!goal) + return yield* new ProviderGoalControlError({ + reason: "goal_missing", + message: "No existing native goal to update.", + }); + if (!sameGoal(goal, input.expectedGoal)) + return yield* new ProviderGoalControlError({ + reason: "goal_changed", + message: "The native goal changed since it was read. No update was sent.", + }); + if (goal.status === "complete") + return yield* new ProviderGoalControlError({ + reason: "terminal_goal", + message: "A completed goal cannot be resumed or replaced by this operation.", + }); + if ( + input.action === "resume" && + goal.tokenBudget != null && + goal.tokensUsed >= goal.tokenBudget + ) { + return yield* new ProviderGoalControlError({ + reason: "budget_exhausted", + message: + "The existing goal token budget is exhausted. Goal control does not increase it.", + }); + } + const thread = yield* options.client.request("thread/read", { + threadId, + includeTurns: false, + }); + if (thread.thread.id !== threadId) + return yield* new ProviderGoalControlError({ + reason: "owner_mismatch", + message: "Codex returned another native thread.", + }); + if ( + session.activeTurnId || + session.status === "running" || + thread.thread.status.type === "active" || + thread.thread.status.type === "notLoaded" + ) { + return yield* new ProviderGoalControlError({ + reason: "active_turn", + message: + "The owning Codex thread is active or unavailable. Let its active turn finish; goal control never interrupts it.", + }); + } + if (attempts.size >= 128) + return yield* new ProviderGoalControlError({ + reason: "attempt_capacity", + message: + "This owner has exhausted its goal-control attempt capacity. No update was sent.", + }); + // Record before crossing the RPC boundary, including failures and lost responses. + attempts.add(input.attemptId); + const status = input.action === "resume" ? "active" : "paused"; + if (goal.status !== status) { + const before = goal; + goal = (yield* options.client.request("thread/goal/set", { threadId, status })).goal; + if (!sameGoal({ ...goal, status: before.status, updatedAt: before.updatedAt }, before)) { + return yield* new ProviderGoalControlError({ + reason: "native_state_changed", + message: + "Codex changed goal identity, budget, or accounting during a status-only update. Read the native state; no corrective write or retry was sent.", + }); + } + } + } + return { + threadId: input.threadId, + providerInstanceId: input.expectedProviderInstanceId, + ownerId: options.ownerId, + goal, + turnDispatched: false, + continuationObserved: false, + } satisfies ProviderGoalControlResult; + }).pipe( + options.mutex.withPermit, + Effect.timeout("10 seconds"), + Effect.mapError((error) => { + if (isGoalControlError(error)) return error; + if (isNativeRequestError(error)) { + return new ProviderGoalControlError({ + reason: error.code === -32601 ? "unsupported" : "native_request_failed", + message: error.errorMessage, + nativeCode: error.code, + }); + } + return new ProviderGoalControlError({ + reason: "native_request_failed", + message: + error._tag === "TimeoutError" + ? "Goal control timed out. An update may have been accepted; read native state before any new intent. The attempt will not be replayed." + : error.message, + }); + }), + ); + }); +}; diff --git a/apps/server/src/provider/Layers/CodexSessionRuntime.ts b/apps/server/src/provider/Layers/CodexSessionRuntime.ts index 78f4b9aa8e505..eff0bced2934c 100644 --- a/apps/server/src/provider/Layers/CodexSessionRuntime.ts +++ b/apps/server/src/provider/Layers/CodexSessionRuntime.ts @@ -1,5 +1,8 @@ import { ApprovalRequestId, + type ProviderGoalControlInput, + type ProviderGoalControlResult, + type ProviderGoalControlError, DEFAULT_MODEL, EventId, ProviderDriverKind, @@ -28,6 +31,7 @@ import * as Layer from "effect/Layer"; import * as Queue from "effect/Queue"; import * as Ref from "effect/Ref"; import * as Schema from "effect/Schema"; +import * as Semaphore from "effect/Semaphore"; import * as Scope from "effect/Scope"; import * as Stream from "effect/Stream"; import { ChildProcess, ChildProcessSpawner } from "effect/unstable/process"; @@ -36,6 +40,7 @@ import * as CodexErrors from "effect-codex-app-server/errors"; import * as CodexRpc from "effect-codex-app-server/rpc"; import * as EffectCodexSchema from "effect-codex-app-server/schema"; +import { makeCodexGoalControl } from "./CodexGoalControl.ts"; import { buildCodexInitializeParams } from "./CodexProvider.ts"; import { codexSessionAppServerArgs } from "./codexLaunchArgs.ts"; import { expandHomePath } from "../../pathExpansion.ts"; @@ -191,6 +196,9 @@ export interface CodexThreadSnapshot { } export interface CodexSessionRuntimeShape { + readonly goalControl: ( + input: ProviderGoalControlInput, + ) => Effect.Effect; readonly start: () => Effect.Effect; readonly getSession: Effect.Effect; readonly sendTurn: ( @@ -2318,8 +2326,16 @@ export const makeCodexSessionRuntime = ( yield* Queue.shutdown(events); }); + const goalMutex = yield* Semaphore.make(1); + const goalControl = makeCodexGoalControl({ + ownerId: yield* randomUUIDv4("provider-event"), + session: Ref.get(sessionRef), + client, + mutex: goalMutex, + }); return { start, + goalControl, getSession: Ref.get(sessionRef), compactThread: Effect.gen(function* () { const providerThreadId = yield* readProviderThreadId; @@ -2381,7 +2397,7 @@ export const makeCodexSessionRuntime = ( ? { resumeCursor: { threadId: resumedProviderThreadId } } : {}), } satisfies ProviderTurnStartResult; - }), + }).pipe(goalMutex.withPermit), interruptTurn: (turnId) => Effect.gen(function* () { const providerThreadId = yield* readProviderThreadId; diff --git a/apps/server/src/provider/Layers/ProviderGoalControl.test.ts b/apps/server/src/provider/Layers/ProviderGoalControl.test.ts new file mode 100644 index 0000000000000..c4aba61339d23 --- /dev/null +++ b/apps/server/src/provider/Layers/ProviderGoalControl.test.ts @@ -0,0 +1,124 @@ +import { it } from "@effect/vitest"; +import { assert, describe } from "vite-plus/test"; +import { + ProviderInstanceId, + ProviderDriverKind, + ThreadId, + type ProviderGoalControlInput, + type ProviderGoalControlResult, +} from "@t3tools/contracts"; +import * as Effect from "effect/Effect"; +import * as Option from "effect/Option"; +import type { ProviderAdapterRegistryShape } from "../Services/ProviderAdapterRegistry.ts"; +import type { + ProviderSessionDirectoryShape, + ProviderRuntimeBinding, +} from "../Services/ProviderSessionDirectory.ts"; +import { routeGoalControl } from "./ProviderGoalControl.ts"; + +const input: ProviderGoalControlInput = { + action: "get", + threadId: ThreadId.make("t3-thread"), + expectedProviderInstanceId: ProviderInstanceId.make("codex-two"), + expectedNativeThreadId: "native-thread", +}; +const binding: ProviderRuntimeBinding = { + threadId: input.threadId, + provider: ProviderDriverKind.make("codex"), + providerInstanceId: input.expectedProviderInstanceId, + resumeCursor: { threadId: input.expectedNativeThreadId }, +}; +function fixture(value: ProviderRuntimeBinding | null = binding, loaded = true, supported = true) { + const routed: string[] = []; + let controls = 0; + const directory = { + getBinding: () => Effect.succeed(Option.fromNullishOr(value)), + } as unknown as ProviderSessionDirectoryShape; + const registry = { + getByInstance: (id: string) => { + routed.push(id); + return Effect.succeed({ + provider: ProviderDriverKind.make("codex"), + hasSession: (threadId: string) => { + assert.equal(threadId, input.threadId); + return Effect.succeed(loaded); + }, + ...(supported + ? { + goalControl: (request: ProviderGoalControlInput) => { + controls++; + assert.deepEqual(request, input); + return Effect.succeed({ + threadId: request.threadId, + providerInstanceId: request.expectedProviderInstanceId, + ownerId: "b220998f-3b30-4b02-a9c7-91696d58222f", + goal: null, + turnDispatched: false, + continuationObserved: false, + } satisfies ProviderGoalControlResult); + }, + } + : {}), + }); + }, + } as unknown as ProviderAdapterRegistryShape; + return { control: routeGoalControl(directory, registry), routed, controls: () => controls }; +} +describe("T3 goal routing", () => { + it.effect("rejects a missing owner before adapter lookup", () => + Effect.gen(function* () { + const f = fixture(null); + assert.equal((yield* f.control(input).pipe(Effect.flip)).reason, "owner_unavailable"); + assert.isEmpty(f.routed); + }), + ); + it.effect( + "routes through the persisted instance rather than a default account or native ID as T3 ID", + () => + Effect.gen(function* () { + const f = fixture(); + yield* f.control(input); + assert.deepEqual(f.routed, ["codex-two"]); + assert.equal(f.controls(), 1); + }), + ); + it.effect("rejects conflicting persisted ownership before selecting an adapter", () => + Effect.gen(function* () { + for (const patch of [ + { providerInstanceId: ProviderInstanceId.make("codex-one") }, + { resumeCursor: { threadId: "wrong" } }, + { resumeCursor: null }, + ]) { + const f = fixture({ ...binding, ...patch }); + assert.equal((yield* f.control(input).pipe(Effect.flip)).reason, "owner_mismatch"); + assert.isEmpty(f.routed); + } + }), + ); + it.effect( + "rejects unloaded owners without session recovery and diagnoses unsupported adapters", + () => + Effect.gen(function* () { + assert.equal( + (yield* fixture(binding, false).control(input).pipe(Effect.flip)).reason, + "owner_unavailable", + ); + assert.equal( + (yield* fixture(binding, true, false).control(input).pipe(Effect.flip)).reason, + "unsupported", + ); + }), + ); + it.effect("rejects missing resume intent and owner fields", () => + Effect.gen(function* () { + const f = fixture(); + assert.equal( + (yield* f + .control({ ...input, action: "resume" } as ProviderGoalControlInput) + .pipe(Effect.flip)).reason, + "invalid_input", + ); + assert.isEmpty(f.routed); + }), + ); +}); diff --git a/apps/server/src/provider/Layers/ProviderGoalControl.ts b/apps/server/src/provider/Layers/ProviderGoalControl.ts new file mode 100644 index 0000000000000..7606aa1015ba9 --- /dev/null +++ b/apps/server/src/provider/Layers/ProviderGoalControl.ts @@ -0,0 +1,83 @@ +import { ProviderGoalControlError, ProviderGoalControlInput } from "@t3tools/contracts"; +import * as Effect from "effect/Effect"; +import * as Option from "effect/Option"; +import * as Schema from "effect/Schema"; +import type { ProviderAdapterRegistryShape } from "../Services/ProviderAdapterRegistry.ts"; +import type { ProviderSessionDirectoryShape } from "../Services/ProviderSessionDirectory.ts"; + +const decodeCursor = Schema.decodeUnknownOption(Schema.Struct({ threadId: Schema.String })); +const decodeInput = Schema.decodeUnknownEffect(ProviderGoalControlInput); + +export const routeGoalControl = ( + directory: ProviderSessionDirectoryShape, + registry: ProviderAdapterRegistryShape, +) => + Effect.fn("ProviderGoalControl.route")(function* (raw: ProviderGoalControlInput) { + const input = yield* decodeInput(raw).pipe( + Effect.mapError( + () => + new ProviderGoalControlError({ + reason: "invalid_input", + message: "Goal control requires an exact thread binding and explicit action.", + }), + ), + ); + const binding = yield* directory.getBinding(input.threadId).pipe( + Effect.mapError( + () => + new ProviderGoalControlError({ + reason: "owner_unavailable", + message: "Cannot read the provider owner for this T3 thread.", + }), + ), + ); + if (Option.isNone(binding)) { + return yield* new ProviderGoalControlError({ + reason: "owner_unavailable", + message: "This T3 thread has no provider session binding.", + }); + } + const cursor = decodeCursor(binding.value.resumeCursor); + if ( + binding.value.providerInstanceId !== input.expectedProviderInstanceId || + Option.isNone(cursor) || + cursor.value.threadId !== input.expectedNativeThreadId + ) { + return yield* new ProviderGoalControlError({ + reason: "owner_mismatch", + message: + "The T3 thread's provider instance or native thread has changed. Read its current binding before retrying.", + }); + } + const adapter = yield* registry.getByInstance(input.expectedProviderInstanceId).pipe( + Effect.mapError( + () => + new ProviderGoalControlError({ + reason: "owner_unavailable", + message: "The bound provider instance is not available.", + }), + ), + ); + if (binding.value.provider !== adapter.provider) { + return yield* new ProviderGoalControlError({ + reason: "owner_mismatch", + message: "The provider binding and live adapter disagree.", + }); + } + if (!adapter.goalControl) { + return yield* new ProviderGoalControlError({ + reason: "unsupported", + message: + "This provider adapter does not expose native goal control. A turn notification does not resume a limited goal.", + }); + } + // Never recover/start a session here: recovery can replace a missing native thread. + if (!(yield* adapter.hasSession(input.threadId))) { + return yield* new ProviderGoalControlError({ + reason: "owner_unavailable", + message: + "The owning provider session is not loaded. Resume it through T3 before reading a fresh goal-control owner.", + }); + } + return yield* adapter.goalControl(input); + }); diff --git a/apps/server/src/provider/Layers/ProviderService.ts b/apps/server/src/provider/Layers/ProviderService.ts index 13da014180b74..9767cbe6876b7 100644 --- a/apps/server/src/provider/Layers/ProviderService.ts +++ b/apps/server/src/provider/Layers/ProviderService.ts @@ -49,6 +49,7 @@ import * as Schema from "effect/Schema"; import * as SchemaIssue from "effect/SchemaIssue"; import * as Stream from "effect/Stream"; +import { routeGoalControl } from "./ProviderGoalControl.ts"; import { resolveAttachmentPath } from "../../attachmentStore.ts"; import * as ServerConfig from "../../config.ts"; import { @@ -2259,6 +2260,7 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( return { startSession, + goalControl: routeGoalControl(directory, registry), sendTurn, compactThread, interruptTurn, diff --git a/apps/server/src/provider/Services/ProviderAdapter.ts b/apps/server/src/provider/Services/ProviderAdapter.ts index c9b62fd795256..995e5bcef1e01 100644 --- a/apps/server/src/provider/Services/ProviderAdapter.ts +++ b/apps/server/src/provider/Services/ProviderAdapter.ts @@ -8,6 +8,9 @@ * @module ProviderAdapter */ import type { + ProviderGoalControlInput, + ProviderGoalControlResult, + ProviderGoalControlError, ApprovalRequestId, ProviderApprovalDecision, ProviderDriverKind, @@ -65,6 +68,10 @@ export interface ProviderThreadSnapshot { } export interface ProviderAdapterShape { + readonly goalControl?: ( + input: ProviderGoalControlInput, + ) => Effect.Effect; + /** * Provider kind implemented by this adapter. */ diff --git a/apps/server/src/provider/Services/ProviderService.ts b/apps/server/src/provider/Services/ProviderService.ts index c189e2916ff1e..f941f112a8c25 100644 --- a/apps/server/src/provider/Services/ProviderService.ts +++ b/apps/server/src/provider/Services/ProviderService.ts @@ -12,6 +12,9 @@ * @module ProviderService */ import type { + ProviderGoalControlInput, + ProviderGoalControlResult, + ProviderGoalControlError, ProviderInterruptTurnInput, ProviderInstanceId, ProviderRespondToRequestInput, @@ -39,6 +42,10 @@ import type { ProviderInstanceRoutingInfo } from "./ProviderAdapterRegistry.ts"; * ProviderServiceShape - Service API for provider session and turn orchestration. */ export interface ProviderServiceShape { + readonly goalControl?: ( + input: ProviderGoalControlInput, + ) => Effect.Effect; + /** * Start a provider session. */ diff --git a/packages/contracts/src/environmentHttp.ts b/packages/contracts/src/environmentHttp.ts index 9cdd21393fead..2631e678e5700 100644 --- a/packages/contracts/src/environmentHttp.ts +++ b/packages/contracts/src/environmentHttp.ts @@ -1,3 +1,8 @@ +import { + ProviderGoalControlInput, + ProviderGoalControlResult, + ProviderGoalControlError, +} from "./providerGoal.ts"; import * as Context from "effect/Context"; import type * as DateTime from "effect/DateTime"; import * as Schema from "effect/Schema"; @@ -505,6 +510,14 @@ const EnvironmentOrchestrationThreadSnapshotQuery = { }; export class EnvironmentOrchestrationHttpApi extends HttpApiGroup.make("orchestration") + .add( + HttpApiEndpoint.post("goalControl", "/api/orchestration/goal-control", { + headers: OptionalBearerHeaders, + payload: ProviderGoalControlInput, + success: ProviderGoalControlResult, + error: [...EnvironmentOrchestrationDispatchErrors, ProviderGoalControlError], + }).middleware(EnvironmentAuthenticatedAuth), + ) .add( HttpApiEndpoint.get("snapshot", "/api/orchestration/snapshot", { headers: OptionalBearerHeaders, diff --git a/packages/contracts/src/index.ts b/packages/contracts/src/index.ts index 74a1b4939f1af..3c47e313f2b44 100644 --- a/packages/contracts/src/index.ts +++ b/packages/contracts/src/index.ts @@ -11,6 +11,7 @@ export * from "./remoteAccess.ts"; export * from "./ipc.ts"; export * from "./terminal.ts"; export * from "./provider.ts"; +export * from "./providerGoal.ts"; export * from "./providerInstance.ts"; export * from "./providerSetup.ts"; export * from "./providerRuntime.ts"; diff --git a/packages/contracts/src/providerGoal.ts b/packages/contracts/src/providerGoal.ts new file mode 100644 index 0000000000000..bab0dfac3da70 --- /dev/null +++ b/packages/contracts/src/providerGoal.ts @@ -0,0 +1,75 @@ +import * as Schema from "effect/Schema"; +import { ThreadId, TrimmedNonEmptyString } from "./baseSchemas.ts"; +import { ProviderInstanceId } from "./providerInstance.ts"; + +export const ProviderNativeGoal = Schema.Struct({ + threadId: TrimmedNonEmptyString, + objective: Schema.String, + status: Schema.Literals([ + "active", + "paused", + "blocked", + "usageLimited", + "budgetLimited", + "complete", + ]), + tokenBudget: Schema.optionalKey(Schema.NullOr(Schema.Int)), + tokensUsed: Schema.Int, + timeUsedSeconds: Schema.Int, + createdAt: Schema.Int, + updatedAt: Schema.Int, +}); +export type ProviderNativeGoal = typeof ProviderNativeGoal.Type; + +const binding = { + threadId: ThreadId, + expectedProviderInstanceId: ProviderInstanceId, + expectedNativeThreadId: TrimmedNonEmptyString, +}; +export const ProviderGoalControlInput = Schema.Union([ + Schema.Struct({ ...binding, action: Schema.Literal("get") }), + Schema.Struct({ + ...binding, + action: Schema.Literals(["resume", "pause"]), + ownerId: Schema.String.check(Schema.isUUID()), + attemptId: Schema.String.check(Schema.isUUID()), + expectedGoal: ProviderNativeGoal, + }), +]); +export type ProviderGoalControlInput = typeof ProviderGoalControlInput.Type; + +export const ProviderGoalControlResult = Schema.Struct({ + threadId: ThreadId, + providerInstanceId: ProviderInstanceId, + ownerId: Schema.String.check(Schema.isUUID()), + goal: Schema.NullOr(ProviderNativeGoal), + // T3 sends no turn/start. Codex owns any asynchronous goal continuation. + turnDispatched: Schema.Literal(false), + // Acceptance is distinct from continuation; observe the existing native turn events. + continuationObserved: Schema.Literal(false), +}); +export type ProviderGoalControlResult = typeof ProviderGoalControlResult.Type; + +export class ProviderGoalControlError extends Schema.TaggedError()( + "ProviderGoalControlError", + { + reason: Schema.Literals([ + "invalid_input", + "owner_unavailable", + "owner_mismatch", + "unsupported", + "active_turn", + "goal_missing", + "goal_changed", + "terminal_goal", + "budget_exhausted", + "attempt_already_used", + "attempt_capacity", + "native_request_failed", + "native_state_changed", + ]), + message: Schema.String, + nativeCode: Schema.optionalKey(Schema.Number), + }, + { httpApiStatus: 409 }, +) {}