authorgravatar for julius0216@outlook.comJulius Marminge <julius0216@outlook.com> 2026-03-27 18:43:35-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-03-27 18:43:35-07:00
log23b3f0c3d0a7de1ca860db03a9b8cacec6f2f7f6
tree2a557c31a9c86fbdf097e82521263fdffa123233
parente08cea3a233263660af3f5f83ab1e58d8b13ec5b
signature Signed by PGP key B5690EEEBB952194

Refactor Codex adapter lifecycle helpers (#1478)

1 files changed, 222 insertions(+), 218 deletions(-)

apps/server/src/provider/Layers/CodexAdapter.ts+222-218
...@@ -20,6 +20,7 @@ import {...@@ -20,6 +20,7 @@ import {
20 ProviderItemId,20 ProviderItemId,
21 ThreadId,21 ThreadId,
22 TurnId,22 TurnId,
23 ProviderSendTurnInput,
23} from "@t3tools/contracts";24} from "@t3tools/contracts";
24import { Effect, FileSystem, Layer, Queue, Schema, ServiceMap, Stream } from "effect";25import { Effect, FileSystem, Layer, Queue, Schema, ServiceMap, Stream } from "effect";
2526
...@@ -1327,38 +1328,40 @@ function mapToRuntimeEvents(...@@ -1327,38 +1328,40 @@ function mapToRuntimeEvents(
1327 return [];1328 return [];
1328}1329}
13291330
1330const makeCodexAdapter = (options?: CodexAdapterLiveOptions) =>1331const makeCodexAdapter = Effect.fn("makeCodexAdapter")(function* (
1331 Effect.gen(function* () {1332 options?: CodexAdapterLiveOptions,
1332 const fileSystem = yield* FileSystem.FileSystem;1333) {
1333 const serverConfig = yield* Effect.service(ServerConfig);1334 const fileSystem = yield* FileSystem.FileSystem;
1334 const nativeEventLogger =1335 const serverConfig = yield* Effect.service(ServerConfig);
1335 options?.nativeEventLogger ??1336 const nativeEventLogger =
1336 (options?.nativeEventLogPath !== undefined1337 options?.nativeEventLogger ??
1337 ? yield* makeEventNdjsonLogger(options.nativeEventLogPath, {1338 (options?.nativeEventLogPath !== undefined
1338 stream: "native",1339 ? yield* makeEventNdjsonLogger(options.nativeEventLogPath, {
1339 })1340 stream: "native",
1340 : undefined);1341 })
13411342 : undefined);
1342 const manager = yield* Effect.acquireRelease(
1343 Effect.gen(function* () {
1344 if (options?.manager) {
1345 return options.manager;
1346 }
1347 const services = yield* Effect.services<never>();
1348 return options?.makeManager?.(services) ?? new CodexAppServerManager(services);
1349 }),
1350 (manager) =>
1351 Effect.sync(() => {
1352 try {
1353 manager.stopAll();
1354 } catch {
1355 // Finalizers should never fail and block shutdown.
1356 }
1357 }),
1358 );
1359 const serverSettingsService = yield* ServerSettingsService;
13601343
1361 const startSession: CodexAdapterShape["startSession"] = Effect.fn(function* (input) {1344 const acquireManager = Effect.fn("acquireManager")(function* () {
1345 if (options?.manager) {
1346 return options.manager;
1347 }
1348 const services = yield* Effect.services<never>();
1349 return options?.makeManager?.(services) ?? new CodexAppServerManager(services);
1350 });
1351
1352 const manager = yield* Effect.acquireRelease(acquireManager(), (manager) =>
1353 Effect.sync(() => {
1354 try {
1355 manager.stopAll();
1356 } catch {
1357 // Finalizers should never fail and block shutdown.
1358 }
1359 }),
1360 );
1361 const serverSettingsService = yield* ServerSettingsService;
1362
1363 const startSession: CodexAdapterShape["startSession"] = Effect.fn("startSession")(
1364 function* (input) {
1362 if (input.provider !== undefined && input.provider !== PROVIDER) {1365 if (input.provider !== undefined && input.provider !== PROVIDER) {
1363 return yield* new ProviderAdapterValidationError({1366 return yield* new ProviderAdapterValidationError({
1364 provider: PROVIDER,1367 provider: PROVIDER,
...@@ -1407,211 +1410,212 @@ const makeCodexAdapter = (options?: CodexAdapterLiveOptions) =>...@@ -1407,211 +1410,212 @@ const makeCodexAdapter = (options?: CodexAdapterLiveOptions) =>
1407 cause,1410 cause,
1408 }),1411 }),
1409 });1412 });
1413 },
1414 );
1415
1416 const resolveAttachment = Effect.fn("resolveAttachment")(function* (
1417 input: ProviderSendTurnInput,
1418 attachment: NonNullable<ProviderSendTurnInput["attachments"]>[number],
1419 ) {
1420 const attachmentPath = resolveAttachmentPath({
1421 attachmentsDir: serverConfig.attachmentsDir,
1422 attachment,
1410 });1423 });
1424 if (!attachmentPath) {
1425 return yield* toRequestError(
1426 input.threadId,
1427 "turn/start",
1428 new Error(`Invalid attachment id '${attachment.id}'.`),
1429 );
1430 }
1431 const bytes = yield* fileSystem.readFile(attachmentPath).pipe(
1432 Effect.mapError(
1433 (cause) =>
1434 new ProviderAdapterRequestError({
1435 provider: PROVIDER,
1436 method: "turn/start",
1437 detail: toMessage(cause, "Failed to read attachment file."),
1438 cause,
1439 }),
1440 ),
1441 );
1442 return {
1443 type: "image" as const,
1444 url: `data:${attachment.mimeType};base64,${Buffer.from(bytes).toString("base64")}`,
1445 };
1446 });
14111447
1412 const sendTurn: CodexAdapterShape["sendTurn"] = (input) =>1448 const sendTurn: CodexAdapterShape["sendTurn"] = Effect.fn("sendTurn")(function* (input) {
1413 Effect.gen(function* () {1449 const codexAttachments = yield* Effect.forEach(
1414 const codexAttachments = yield* Effect.forEach(1450 input.attachments ?? [],
1415 input.attachments ?? [],1451 (attachment) => resolveAttachment(input, attachment),
1416 (attachment) =>1452 { concurrency: 1 },
1417 Effect.gen(function* () {1453 );
1418 const attachmentPath = resolveAttachmentPath({
1419 attachmentsDir: serverConfig.attachmentsDir,
1420 attachment,
1421 });
1422 if (!attachmentPath) {
1423 return yield* toRequestError(
1424 input.threadId,
1425 "turn/start",
1426 new Error(`Invalid attachment id '${attachment.id}'.`),
1427 );
1428 }
1429 const bytes = yield* fileSystem.readFile(attachmentPath).pipe(
1430 Effect.mapError(
1431 (cause) =>
1432 new ProviderAdapterRequestError({
1433 provider: PROVIDER,
1434 method: "turn/start",
1435 detail: toMessage(cause, "Failed to read attachment file."),
1436 cause,
1437 }),
1438 ),
1439 );
1440 return {
1441 type: "image" as const,
1442 url: `data:${attachment.mimeType};base64,${Buffer.from(bytes).toString("base64")}`,
1443 };
1444 }),
1445 { concurrency: 1 },
1446 );
14471454
1448 return yield* Effect.tryPromise({1455 return yield* Effect.tryPromise({
1449 try: () => {1456 try: () => {
1450 const managerInput = {1457 const managerInput = {
1451 threadId: input.threadId,1458 threadId: input.threadId,
1452 ...(input.input !== undefined ? { input: input.input } : {}),1459 ...(input.input !== undefined ? { input: input.input } : {}),
1453 ...(input.modelSelection?.provider === "codex"1460 ...(input.modelSelection?.provider === "codex"
1454 ? { model: input.modelSelection.model }1461 ? { model: input.modelSelection.model }
1455 : {}),1462 : {}),
1456 ...(input.modelSelection?.provider === "codex" &&1463 ...(input.modelSelection?.provider === "codex" &&
1457 input.modelSelection.options?.reasoningEffort !== undefined1464 input.modelSelection.options?.reasoningEffort !== undefined
1458 ? { effort: input.modelSelection.options.reasoningEffort }1465 ? { effort: input.modelSelection.options.reasoningEffort }
1459 : {}),1466 : {}),
1460 ...(input.modelSelection?.provider === "codex" &&1467 ...(input.modelSelection?.provider === "codex" && input.modelSelection.options?.fastMode
1461 input.modelSelection.options?.fastMode1468 ? { serviceTier: "fast" }
1462 ? { serviceTier: "fast" }1469 : {}),
1463 : {}),1470 ...(input.interactionMode !== undefined
1464 ...(input.interactionMode !== undefined1471 ? { interactionMode: input.interactionMode }
1465 ? { interactionMode: input.interactionMode }1472 : {}),
1466 : {}),1473 ...(codexAttachments.length > 0 ? { attachments: codexAttachments } : {}),
1467 ...(codexAttachments.length > 0 ? { attachments: codexAttachments } : {}),1474 };
1468 };1475 return manager.sendTurn(managerInput);
1469 return manager.sendTurn(managerInput);1476 },
1470 },1477 catch: (cause) => toRequestError(input.threadId, "turn/start", cause),
1471 catch: (cause) => toRequestError(input.threadId, "turn/start", cause),1478 }).pipe(
1472 }).pipe(1479 Effect.map((result) => ({
1473 Effect.map((result) => ({1480 ...result,
1474 ...result,1481 threadId: input.threadId,
1475 threadId: input.threadId,1482 })),
1476 })),1483 );
1477 );1484 });
1478 });
14791485
1480 const interruptTurn: CodexAdapterShape["interruptTurn"] = (threadId, turnId) =>1486 const interruptTurn: CodexAdapterShape["interruptTurn"] = (threadId, turnId) =>
1481 Effect.tryPromise({1487 Effect.tryPromise({
1482 try: () => manager.interruptTurn(threadId, turnId),1488 try: () => manager.interruptTurn(threadId, turnId),
1483 catch: (cause) => toRequestError(threadId, "turn/interrupt", cause),1489 catch: (cause) => toRequestError(threadId, "turn/interrupt", cause),
1484 });1490 });
14851491
1486 const readThread: CodexAdapterShape["readThread"] = (threadId) =>1492 const readThread: CodexAdapterShape["readThread"] = (threadId) =>
1487 Effect.tryPromise({1493 Effect.tryPromise({
1488 try: () => manager.readThread(threadId),1494 try: () => manager.readThread(threadId),
1489 catch: (cause) => toRequestError(threadId, "thread/read", cause),1495 catch: (cause) => toRequestError(threadId, "thread/read", cause),
1490 }).pipe(1496 }).pipe(
1491 Effect.map((snapshot) => ({1497 Effect.map((snapshot) => ({
1492 threadId,1498 threadId,
1493 turns: snapshot.turns,1499 turns: snapshot.turns,
1494 })),1500 })),
1501 );
1502
1503 const rollbackThread: CodexAdapterShape["rollbackThread"] = (threadId, numTurns) => {
1504 if (!Number.isInteger(numTurns) || numTurns < 1) {
1505 return Effect.fail(
1506 new ProviderAdapterValidationError({
1507 provider: PROVIDER,
1508 operation: "rollbackThread",
1509 issue: "numTurns must be an integer >= 1.",
1510 }),
1495 );1511 );
1512 }
14961513
1497 const rollbackThread: CodexAdapterShape["rollbackThread"] = (threadId, numTurns) => {1514 return Effect.tryPromise({
1498 if (!Number.isInteger(numTurns) || numTurns < 1) {1515 try: () => manager.rollbackThread(threadId, numTurns),
1499 return Effect.fail(1516 catch: (cause) => toRequestError(threadId, "thread/rollback", cause),
1500 new ProviderAdapterValidationError({1517 }).pipe(
1501 provider: PROVIDER,1518 Effect.map((snapshot) => ({
1502 operation: "rollbackThread",1519 threadId,
1503 issue: "numTurns must be an integer >= 1.",1520 turns: snapshot.turns,
1504 }),1521 })),
1505 );1522 );
1506 }1523 };
15071524
1508 return Effect.tryPromise({1525 const respondToRequest: CodexAdapterShape["respondToRequest"] = (threadId, requestId, decision) =>
1509 try: () => manager.rollbackThread(threadId, numTurns),1526 Effect.tryPromise({
1510 catch: (cause) => toRequestError(threadId, "thread/rollback", cause),1527 try: () => manager.respondToRequest(threadId, requestId, decision),
1511 }).pipe(1528 catch: (cause) => toRequestError(threadId, "item/requestApproval/decision", cause),
1512 Effect.map((snapshot) => ({1529 });
1513 threadId,
1514 turns: snapshot.turns,
1515 })),
1516 );
1517 };
15181530
1519 const respondToRequest: CodexAdapterShape["respondToRequest"] = (1531 const respondToUserInput: CodexAdapterShape["respondToUserInput"] = (
1520 threadId,1532 threadId,
1521 requestId,1533 requestId,
1522 decision,1534 answers,
1523 ) =>1535 ) =>
1524 Effect.tryPromise({1536 Effect.tryPromise({
1525 try: () => manager.respondToRequest(threadId, requestId, decision),1537 try: () => manager.respondToUserInput(threadId, requestId, answers),
1526 catch: (cause) => toRequestError(threadId, "item/requestApproval/decision", cause),1538 catch: (cause) => toRequestError(threadId, "item/tool/requestUserInput", cause),
1527 });1539 });
15281540
1529 const respondToUserInput: CodexAdapterShape["respondToUserInput"] = (1541 const stopSession: CodexAdapterShape["stopSession"] = (threadId) =>
1530 threadId,1542 Effect.sync(() => {
1531 requestId,1543 manager.stopSession(threadId);
1532 answers,1544 });
1533 ) =>
1534 Effect.tryPromise({
1535 try: () => manager.respondToUserInput(threadId, requestId, answers),
1536 catch: (cause) => toRequestError(threadId, "item/tool/requestUserInput", cause),
1537 });
15381545
1539 const stopSession: CodexAdapterShape["stopSession"] = (threadId) =>1546 const listSessions: CodexAdapterShape["listSessions"] = () =>
1540 Effect.sync(() => {1547 Effect.sync(() => manager.listSessions());
1541 manager.stopSession(threadId);
1542 });
15431548
1544 const listSessions: CodexAdapterShape["listSessions"] = () =>1549 const hasSession: CodexAdapterShape["hasSession"] = (threadId) =>
1545 Effect.sync(() => manager.listSessions());1550 Effect.sync(() => manager.hasSession(threadId));
15461551
1547 const hasSession: CodexAdapterShape["hasSession"] = (threadId) =>1552 const stopAll: CodexAdapterShape["stopAll"] = () =>
1548 Effect.sync(() => manager.hasSession(threadId));1553 Effect.sync(() => {
1554 manager.stopAll();
1555 });
15491556
1550 const stopAll: CodexAdapterShape["stopAll"] = () =>1557 const runtimeEventQueue = yield* Queue.unbounded<ProviderRuntimeEvent>();
1551 Effect.sync(() => {
1552 manager.stopAll();
1553 });
15541558
1555 const runtimeEventQueue = yield* Queue.unbounded<ProviderRuntimeEvent>();1559 const writeNativeEvent = Effect.fn("writeNativeEvent")(function* (event: ProviderEvent) {
15561560 if (!nativeEventLogger) {
1557 yield* Effect.acquireRelease(1561 return;
1558 Effect.gen(function* () {1562 }
1559 const writeNativeEvent = (event: ProviderEvent) =>1563 yield* nativeEventLogger.write(event, event.threadId);
1560 Effect.gen(function* () {1564 });
1561 if (!nativeEventLogger) {
1562 return;
1563 }
1564 yield* nativeEventLogger.write(event, event.threadId);
1565 });
1566
1567 const services = yield* Effect.services<never>();
1568 const listener = (event: ProviderEvent) =>
1569 Effect.gen(function* () {
1570 yield* writeNativeEvent(event);
1571 const runtimeEvents = mapToRuntimeEvents(event, event.threadId);
1572 if (runtimeEvents.length === 0) {
1573 yield* Effect.logDebug("ignoring unhandled Codex provider event", {
1574 method: event.method,
1575 threadId: event.threadId,
1576 turnId: event.turnId,
1577 itemId: event.itemId,
1578 });
1579 return;
1580 }
1581 yield* Queue.offerAll(runtimeEventQueue, runtimeEvents);
1582 }).pipe(Effect.runPromiseWith(services));
1583 manager.on("event", listener);
1584 return listener;
1585 }),
1586 (listener) =>
1587 Effect.gen(function* () {
1588 yield* Effect.sync(() => {
1589 manager.off("event", listener);
1590 });
1591 yield* Queue.shutdown(runtimeEventQueue);
1592 }),
1593 );
15941565
1595 return {1566 const registerListener = Effect.fn("registerListener")(function* () {
1596 provider: PROVIDER,1567 const services = yield* Effect.services<never>();
1597 capabilities: {1568 const listenerEffect = Effect.fn("listener")(function* (event: ProviderEvent) {
1598 sessionModelSwitch: "in-session",1569 yield* writeNativeEvent(event);
1599 },1570 const runtimeEvents = mapToRuntimeEvents(event, event.threadId);
1600 startSession,1571 if (runtimeEvents.length === 0) {
1601 sendTurn,1572 yield* Effect.logDebug("ignoring unhandled Codex provider event", {
1602 interruptTurn,1573 method: event.method,
1603 readThread,1574 threadId: event.threadId,
1604 rollbackThread,1575 turnId: event.turnId,
1605 respondToRequest,1576 itemId: event.itemId,
1606 respondToUserInput,1577 });
1607 stopSession,1578 return;
1608 listSessions,1579 }
1609 hasSession,1580 yield* Queue.offerAll(runtimeEventQueue, runtimeEvents);
1610 stopAll,1581 });
1611 streamEvents: Stream.fromQueue(runtimeEventQueue),1582 const listener = (event: ProviderEvent) =>
1612 } satisfies CodexAdapterShape;1583 listenerEffect(event).pipe(Effect.runPromiseWith(services));
1584 manager.on("event", listener);
1585 return listener;
1586 });
1587
1588 const unregisterListener = Effect.fn("unregisterListener")(function* (
1589 listener: (event: ProviderEvent) => Promise<void>,
1590 ) {
1591 yield* Effect.sync(() => {
1592 manager.off("event", listener);
1593 });
1594 yield* Queue.shutdown(runtimeEventQueue);
1613 });1595 });
16141596
1597 yield* Effect.acquireRelease(registerListener(), unregisterListener);
1598
1599 return {
1600 provider: PROVIDER,
1601 capabilities: {
1602 sessionModelSwitch: "in-session",
1603 },
1604 startSession,
1605 sendTurn,
1606 interruptTurn,
1607 readThread,
1608 rollbackThread,
1609 respondToRequest,
1610 respondToUserInput,
1611 stopSession,
1612 listSessions,
1613 hasSession,
1614 stopAll,
1615 streamEvents: Stream.fromQueue(runtimeEventQueue),
1616 } satisfies CodexAdapterShape;
1617});
1618
1615export const CodexAdapterLive = Layer.effect(CodexAdapter, makeCodexAdapter());1619export const CodexAdapterLive = Layer.effect(CodexAdapter, makeCodexAdapter());
16161620
1617export function makeCodexAdapterLive(options?: CodexAdapterLiveOptions) {1621export function makeCodexAdapterLive(options?: CodexAdapterLiveOptions) {