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 }, +) {}