lambert/codex-tanstack-start
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: stream assistant output via Codex app-server

Browse Source
This commit is contained in:
lambert
2026-09-11 16:17:51 +08:00
parent 94812a4282
commit 6f4584d2f8
24 changed files with 839 additions and 158 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+12 -11
View File
@@ -1,13 +1,14 @@
# Codex + TanStack Start Demo
Minimal proof of concept for using `@openai/codex-sdk` as a **local personal-agent runtime** behind a TanStack Start web UI.
Minimal proof of concept for using the local Codex app-server as a **personal-agent runtime** behind a TanStack Start web UI.
The browser never receives Codex credentials or raw SDK objects. TanStack Start runs the Codex SDK on the server side, normalizes structured events into an application-owned `ChatEvent` contract, and streams those events back to the React UI.
The browser never receives Codex credentials or raw protocol objects. TanStack Start runs the Codex app-server on the server side, normalizes its structured notifications into an application-owned `ChatEvent` contract, and streams those events back to the React UI.
## What this demo validates
- TanStack Start server functions can stream an async generator to the browser.
- `@openai/codex-sdk` can start and resume Codex threads.
- Codex app-server can start and resume Codex threads.
- Codex app-server `item/agentMessage/delta` notifications reach the UI as real assistant deltas.
- The server-side SDK can reuse the Codex/ChatGPT authentication already available on the host.
- The browser persists only `{ threadId, messages }` in versioned localStorage.
- Agent activity is rendered without exposing raw reasoning, command output, MCP payloads, local paths, or authentication material.
@@ -26,13 +27,13 @@ TanStack Start Server Function
ChatEvent normalizer
|
v
CodexRuntime (server only)
CodexRuntime interface (server only)
|
v
@openai/codex-sdk
CodexAppServerRuntime
|
v
Codex CLI runtime + local ChatGPT/Codex authentication
codex app-server + local ChatGPT/Codex authentication
```
Client state follows one path:
@@ -87,14 +88,14 @@ If `CODEX_WORKSPACE_ROOT` is omitted, V0 uses the server process working directo
## Safety baseline
Every Codex thread is created/resumed with:
Every Codex app-server thread/turn is created or resumed with:
```text
model: luna
modelReasoningEffort: high
sandboxMode: read-only
effort: high
sandbox: read-only
sandboxPolicy.networkAccess: false
approvalPolicy: never
networkAccessEnabled: false
```
This demo therefore targets repository inspection, explanation, reasoning, and other read-only workflows. File mutation is deliberately not enabled.
@@ -122,7 +123,7 @@ Transient activities, errors, credentials, SDK events, and command output are no
## Known limitations
- V0 has one active browser conversation and one configured workspace root.
- Assistant text is emitted on completed Codex message items; this is structured event streaming, not token-by-token text rendering.
- Assistant text is emitted from app-server deltas while the turn is running, then calibrated with the completed item snapshot.
- The runtime is fixed to the `luna` model with `high` reasoning effort; there is no UI for changing the model, reasoning effort, workspace, or permissions.
- There is no write-mode approval flow; the runtime is intentionally read-only.
- New Chat ignores any remaining client-side events from the previous turn, but does not currently propagate an explicit cancellation signal through the TanStack Start RPC to terminate the underlying Codex process immediately.
docs/ARCHITECTURE.md
+10 -6
View File
@@ -2,17 +2,19 @@
## Boundary
`@openai/codex-sdk` is server-only. Browser code must never import the SDK or read local Codex authentication files.
The Codex app-server is server-only. Browser code must never import the SDK/protocol client or read local Codex authentication files.
## Runtime flow
```text
POST user turn
-> chat server function
-> CodexService
-> startThread() or resumeThread(threadId)
-> runStreamed()
-> normalize Codex ThreadEvent to app ChatEvent
-> CodexRuntime interface
-> CodexAppServerRuntime
-> thread/start or thread/resume
-> turn/start
-> item/agentMessage/delta notifications
-> normalize app-server notifications to app ChatEvent
-> async stream back to browser
-> reducer updates message/activity state
```
@@ -24,7 +26,9 @@ The UI depends on an application-owned event contract rather than Codex SDK even
Initial event families:
- `thread.started`
- `assistant.message`
- `assistant.started`
- `assistant.delta`
- `assistant.completed`
- `activity.started`
- `activity.updated`
- `activity.completed`
docs/INTEGRATION.md
+1 -1
View File
@@ -13,7 +13,7 @@ getCodexRuntime().streamTurn({
})
```
The SDK `ThreadEvent` is converted at the seam to Task 02's structural server-only event type. Browser-reachable modules do not import `@openai/codex-sdk`.
Codex app-server notifications are converted at the seam to Task 02's structural server-only event type. Browser-reachable modules do not import `@openai/codex-sdk` or app-server protocol objects.
### Task 02 -> Task 04
docs/VALIDATION.md
+2 -1
View File
@@ -8,9 +8,10 @@ Validation was rerun in GitHub Actions on Node.js 22 with a network-enabled npm
- dependency installation: PASS (`npm install --no-audit --no-fund`)
- `npm run typecheck`: PASS
- `npm test`: PASS — 9 test files, 34 tests
- `npm test`: PASS — 10 test files, 37 tests
- `npm run build`: PASS
- V0 runtime policy remains read-only (`sandboxMode: read-only`, `approvalPolicy: never`, network disabled)
- app-server smoke verification: PASS (`item/started` -> multiple `item/agentMessage/delta` -> `item/completed` -> `turn/completed`)
- documented authentication path reuses the host Codex/ChatGPT login; `OPENAI_API_KEY` is not required for that subscription-auth path
## Integration defects fixed during final validation
docs/tasks/01-codex-runtime.md
+4 -3
View File
@@ -3,9 +3,10 @@
Implement the server-only Codex runtime wrapper.
## Requirements
- Instantiate `Codex` only in server-only code.
- Support `startThread()` and `resumeThread(threadId)`.
- Expose a streaming turn method based on `runStreamed()`.
- Keep the Codex app-server process and protocol client in server-only code.
- Support `thread/start` and `thread/resume(threadId)`.
- Expose a streaming turn method based on `turn/start` notifications.
- Forward `item/agentMessage/delta` events without exposing raw app-server messages.
- Use the `luna` model with `high` reasoning effort for every thread.
- Use a read-only sandbox/workspace posture for V0.
- Validate/resolve workspace path server-side.
docs/tasks/04-state-persistence-tests.md
+1 -1
View File
@@ -21,7 +21,7 @@ Task 04 exports deterministic client-state primitives from `src/features/chat/ch
- `ChatStateEvent` + `applyChatEvent(state, event)`
- `initialChatState`
Task 05 should adapt Task 02's transport `ChatEvent` into `ChatStateEvent`, then dispatch it as `{ type: 'event.received', event }`. The state layer intentionally has no dependency on `@openai/codex-sdk`.
Task 05 should adapt Task 02's transport `ChatEvent` into `ChatStateEvent`, then dispatch it as `{ type: 'event.received', event }`. The state layer intentionally has no dependency on Codex SDK or app-server protocol details.
Persistence helpers live in `src/features/chat/chat.storage.ts`:
src/features/chat/chat-event.adapter.test.ts
+17 -9
View File
@@ -1,20 +1,28 @@
import { describe, expect, it } from 'vitest'
import { toChatStateEvent } from './chat-event.adapter'
it('converts an assistant transport event into a persisted reducer message', () => {
it('converts assistant lifecycle events into reducer events', () => {
expect(
toChatStateEvent(
{ type: 'assistant.message', id: 'm1', text: 'Hello' },
{ type: 'assistant.started', id: 'm1' },
123,
),
).toEqual({
type: 'assistant.message',
message: {
id: 'm1',
role: 'assistant',
content: 'Hello',
createdAt: 123,
},
type: 'assistant.started',
id: 'm1',
createdAt: 123,
})
expect(toChatStateEvent({ type: 'assistant.delta', id: 'm1', delta: 'Hello' }, 123)).toEqual({
type: 'assistant.delta',
id: 'm1',
delta: 'Hello',
})
expect(toChatStateEvent({ type: 'assistant.completed', id: 'm1', text: 'Hello' }, 123)).toEqual({
type: 'assistant.completed',
id: 'm1',
text: 'Hello',
})
})
src/features/chat/chat-event.adapter.ts
+10 -8
View File
@@ -10,17 +10,19 @@ export function toChatStateEvent(
case 'thread.started':
return event
case 'assistant.message':
case 'assistant.started':
return {
type: 'assistant.message',
message: {
id: event.id,
role: 'assistant',
content: event.text,
createdAt: receivedAt,
},
type: 'assistant.started',
id: event.id,
createdAt: receivedAt,
}
case 'assistant.delta':
return event
case 'assistant.completed':
return event
case 'activity.started':
case 'activity.updated':
return {
src/features/chat/chat.reducer.test.ts
+49 -9
View File
@@ -96,18 +96,58 @@ describe('applyChatEvent', () => {
expect(next.messages).toEqual([userMessage])
})
it('upserts assistant messages by id', () => {
const first = applyChatEvent(initialChatState, {
type: 'assistant.message',
message: assistantMessage,
it('creates one assistant message, appends deltas, and calibrates on completion', () => {
const running = chatReducer(initialChatState, { type: 'turn.started' })
const started = applyChatEvent(running, {
type: 'assistant.started',
id: assistantMessage.id,
createdAt: assistantMessage.createdAt,
})
const updated = applyChatEvent(first, {
type: 'assistant.message',
message: { ...assistantMessage, content: 'Hi again' },
const duplicateStarted = applyChatEvent(started, {
type: 'assistant.started',
id: assistantMessage.id,
createdAt: 99,
})
const firstDelta = applyChatEvent(duplicateStarted, {
type: 'assistant.delta',
id: assistantMessage.id,
delta: 'Hi',
})
const secondDelta = applyChatEvent(firstDelta, {
type: 'assistant.delta',
id: assistantMessage.id,
delta: ' there',
})
const completed = applyChatEvent(secondDelta, {
type: 'assistant.completed',
id: assistantMessage.id,
text: 'Hi again',
})
expect(updated.messages).toHaveLength(1)
expect(updated.messages[0]?.content).toBe('Hi again')
expect(duplicateStarted.messages).toHaveLength(1)
expect(secondDelta.messages[0]?.content).toBe('Hi there')
expect(completed.messages[0]).toEqual({ ...assistantMessage, content: 'Hi again' })
expect(completed.status).toBe('running')
})
it('does not mix deltas from different assistant ids', () => {
const first = applyChatEvent(initialChatState, {
type: 'assistant.started',
id: 'a1',
createdAt: 1,
})
const second = applyChatEvent(first, {
type: 'assistant.started',
id: 'a2',
createdAt: 2,
})
const updated = applyChatEvent(second, {
type: 'assistant.delta',
id: 'a2',
delta: 'second',
})
expect(updated.messages.map((message) => message.content)).toEqual(['', 'second'])
})
it('tracks activity lifecycle deterministically', () => {
src/features/chat/chat.reducer.ts
+55 -3
View File
@@ -39,7 +39,9 @@ export interface ChatState extends PersistedConversation {
*/
export type ChatStateEvent =
| { type: 'thread.started'; threadId: string }
| { type: 'assistant.message'; message: ChatMessage }
| { type: 'assistant.started'; id: string; createdAt: number }
| { type: 'assistant.delta'; id: string; delta: string }
| { type: 'assistant.completed'; id: string; text: string }
| { type: 'activity.started'; activity: AgentActivity }
| {
type: 'activity.updated'
@@ -99,12 +101,62 @@ export function applyChatEvent(state: ChatState, event: ChatStateEvent): ChatSta
threadId: event.threadId,
}
case 'assistant.message':
case 'assistant.started':
if (state.messages.some((message) => message.id === event.id)) {
return state
}
return {
...state,
messages: upsertMessage(state.messages, event.message),
messages: [
...state.messages,
{
id: event.id,
role: 'assistant',
content: '',
createdAt: event.createdAt,
},
],
}
case 'assistant.delta': {
const index = state.messages.findIndex((message) => message.id === event.id)
if (index === -1) {
return {
...state,
messages: [
...state.messages,
{
id: event.id,
role: 'assistant',
content: event.delta,
createdAt: Date.now(),
},
],
}
}
const message = state.messages[index]
if (!message || message.role !== 'assistant') return state
const messages = [...state.messages]
messages[index] = { ...message, content: message.content + event.delta }
return { ...state, messages }
}
case 'assistant.completed': {
const existing = state.messages.find((message) => message.id === event.id)
return {
...state,
messages: upsertMessage(state.messages, {
id: event.id,
role: 'assistant',
content: event.text,
createdAt: existing?.createdAt ?? Date.now(),
}),
}
}
case 'activity.started':
return {
...state,
src/features/chat/chat.types.ts
+10 -1
View File
@@ -35,7 +35,16 @@ export type ChatEvent =
threadId: string
}
| {
type: 'assistant.message'
type: 'assistant.started'
id: string
}
| {
type: 'assistant.delta'
id: string
delta: string
}
| {
type: 'assistant.completed'
id: string
text: string
}
src/features/chat/use-chat-controller.test.ts
+9
View File
@@ -0,0 +1,9 @@
import { describe, expect, it } from 'vitest'
import { isCurrentChatGeneration } from './use-chat-controller'
describe('isCurrentChatGeneration', () => {
it('rejects events from a stream superseded by New Chat', () => {
expect(isCurrentChatGeneration(3, 4)).toBe(false)
expect(isCurrentChatGeneration(4, 4)).toBe(true)
})
})
src/features/chat/use-chat-controller.ts
+9 -2
View File
@@ -21,6 +21,13 @@ export interface ChatController {
newChat: () => void
}
export function isCurrentChatGeneration(
streamGeneration: number,
currentGeneration: number,
): boolean {
return streamGeneration === currentGeneration
}
export function useChatController(): ChatController {
const [state, dispatch] = useReducer(chatReducer, initialChatState)
const [restored, setRestored] = useState(false)
@@ -69,14 +76,14 @@ export function useChatController(): ChatController {
})
for await (const event of events) {
if (generation !== generationRef.current) return
if (!isCurrentChatGeneration(generation, generationRef.current)) return
dispatch({
type: 'event.received',
event: toChatStateEvent(event, Date.now()),
})
}
} catch (error) {
if (generation !== generationRef.current) return
if (!isCurrentChatGeneration(generation, generationRef.current)) return
dispatch({
type: 'event.received',
event: {
src/server-functions/__tests__/chat-stream.test.ts
+8 -1
View File
@@ -7,6 +7,11 @@ describe('streamNormalizedChatEvents', () => {
const events: CodexThreadEvent[] = [
{ type: 'thread.started', thread_id: 'thread-1' },
{ type: 'turn.started' },
{
type: 'item.started',
item: { id: 'm1', type: 'agent_message', text: '' },
},
{ type: 'item.agent_message.delta', item_id: 'm1', delta: 'Done' },
{
type: 'item.completed',
item: { id: 'm1', type: 'agent_message', text: 'Done' },
@@ -37,7 +42,9 @@ describe('streamNormalizedChatEvents', () => {
expect(output.map((event) => event.type)).toEqual([
'thread.started',
'assistant.message',
'assistant.started',
'assistant.delta',
'assistant.completed',
'turn.completed',
])
})
src/server-functions/__tests__/codex-event-normalizer.test.ts
+34 -6
View File
@@ -1,5 +1,9 @@
import { describe, expect, it } from 'vitest'
import { normalizeCodexEvent, sanitizeErrorMessage } from '../codex-event-normalizer'
import {
createCodexEventNormalizer,
normalizeCodexEvent,
sanitizeErrorMessage,
} from '../codex-event-normalizer'
describe('normalizeCodexEvent', () => {
it('maps thread ids and usage without exposing SDK field names', () => {
@@ -31,13 +35,37 @@ describe('normalizeCodexEvent', () => {
])
})
it('emits assistant text only when the item completes', () => {
const item = { id: 'm1', type: 'agent_message' as const, text: 'Hello' }
it('converts app-server deltas into assistant deltas', () => {
expect(normalizeCodexEvent({
type: 'item.agent_message.delta',
item_id: 'm1',
delta: 'Hello',
})).toEqual([{ type: 'assistant.delta', id: 'm1', delta: 'Hello' }])
})
expect(normalizeCodexEvent({ type: 'item.updated', item })).toEqual([])
expect(normalizeCodexEvent({ type: 'item.completed', item })).toEqual([
{ type: 'assistant.message', id: 'm1', text: 'Hello' },
it('converts cumulative snapshots into safe deltas and final correction', () => {
const normalizer = createCodexEventNormalizer()
const started = { id: 'm1', type: 'agent_message' as const, text: '' }
expect(normalizer.normalize({ type: 'item.started', item: started })).toEqual([
{ type: 'assistant.started', id: 'm1' },
])
expect(normalizer.normalize({
type: 'item.updated',
item: { ...started, text: 'React' },
})).toEqual([{ type: 'assistant.delta', id: 'm1', delta: 'React' }])
expect(normalizer.normalize({
type: 'item.updated',
item: { ...started, text: 'React 是' },
})).toEqual([{ type: 'assistant.delta', id: 'm1', delta: ' 是' }])
expect(normalizer.normalize({
type: 'item.updated',
item: { ...started, text: 'A corrected answer' },
})).toEqual([{ type: 'assistant.completed', id: 'm1', text: 'A corrected answer' }])
expect(normalizer.normalize({
type: 'item.completed',
item: { ...started, text: 'A corrected answer' },
})).toEqual([{ type: 'assistant.completed', id: 'm1', text: 'A corrected answer' }])
})
it('does not expose reasoning text, command output, or MCP payloads', () => {
src/server-functions/chat-stream.ts
+3 -2
View File
@@ -1,6 +1,6 @@
import type { ChatEvent, ChatRequest } from '../features/chat/chat.types'
import type { CodexThreadEvent } from './codex-event.types'
import { normalizeCodexEvent } from './codex-event-normalizer'
import { createCodexEventNormalizer } from './codex-event-normalizer'
export type CodexTurnStreamFactory = (
request: ChatRequest,
@@ -12,9 +12,10 @@ export async function* streamNormalizedChatEvents(
createCodexTurnStream: CodexTurnStreamFactory,
): AsyncGenerator<ChatEvent> {
const source = await createCodexTurnStream(request)
const normalizer = createCodexEventNormalizer()
for await (const codexEvent of source) {
for (const chatEvent of normalizeCodexEvent(codexEvent)) {
for (const chatEvent of normalizer.normalize(codexEvent)) {
yield chatEvent
}
}
src/server-functions/chat.runtime.server.ts
+2 -2
View File
@@ -16,8 +16,8 @@ export async function* streamCodexTurn(
prompt: request.message,
threadId: request.threadId,
})) {
// Task 02 intentionally owns a small structural mirror of the SDK event
// contract so browser-reachable modules never import @openai/codex-sdk.
// Task 02 intentionally owns a small structural mirror of the server event
// contract so browser-reachable modules never import Codex runtime details.
yield event as unknown as CodexThreadEvent
}
}
src/server-functions/codex-event-normalizer.ts
+52 -5
View File
@@ -8,7 +8,7 @@ import type {
import type { CodexThreadEvent, CodexThreadItem, CodexUsage } from './codex-event.types'
/**
* Converts Codex SDK events into the stable, browser-safe application contract.
* Converts Codex runtime events into the stable, browser-safe application contract.
*
* Security boundary:
* - raw SDK objects are never forwarded
@@ -18,6 +18,25 @@ import type { CodexThreadEvent, CodexThreadItem, CodexUsage } from './codex-even
* - only an opaque thread identifier required for resume is exposed
*/
export function normalizeCodexEvent(event: CodexThreadEvent): ChatEvent[] {
return createCodexEventNormalizer().normalize(event)
}
export function createCodexEventNormalizer(): {
normalize: (event: CodexThreadEvent) => ChatEvent[]
} {
const assistantSnapshots = new Map<string, string>()
return {
normalize(event) {
return normalizeCodexEventWithSnapshots(event, assistantSnapshots)
},
}
}
function normalizeCodexEventWithSnapshots(
event: CodexThreadEvent,
assistantSnapshots: Map<string, string>,
): ChatEvent[] {
switch (event.type) {
case 'thread.started':
return [{ type: 'thread.started', threadId: event.thread_id }]
@@ -34,21 +53,49 @@ export function normalizeCodexEvent(event: CodexThreadEvent): ChatEvent[] {
case 'error':
return [{ type: 'error', message: sanitizeErrorMessage(event.message) }]
case 'item.agent_message.delta':
if (!event.delta) return []
return [{ type: 'assistant.delta', id: event.item_id, delta: event.delta }]
case 'item.started':
case 'item.updated':
case 'item.completed':
return normalizeItemEvent(event.type, event.item)
return normalizeItemEvent(event.type, event.item, assistantSnapshots)
}
}
function normalizeItemEvent(
eventType: 'item.started' | 'item.updated' | 'item.completed',
item: CodexThreadItem,
assistantSnapshots: Map<string, string>,
): ChatEvent[] {
if (item.type === 'agent_message') {
return eventType === 'item.completed'
? [{ type: 'assistant.message', id: item.id, text: item.text }]
: []
if (eventType === 'item.started') {
const previous = assistantSnapshots.get(item.id)
assistantSnapshots.set(item.id, item.text)
const events: ChatEvent[] = [{ type: 'assistant.started', id: item.id }]
if (!previous && item.text) {
events.push({ type: 'assistant.delta', id: item.id, delta: item.text })
}
return events
}
if (eventType === 'item.updated') {
const previous = assistantSnapshots.get(item.id) ?? ''
assistantSnapshots.set(item.id, item.text)
if (item.text === previous) return []
if (item.text.startsWith(previous)) {
const delta = item.text.slice(previous.length)
return delta ? [{ type: 'assistant.delta', id: item.id, delta }] : []
}
// A non-prefix snapshot cannot be represented as an append-only delta.
// Use a full correction event so the reducer never displays duplicated text.
return [{ type: 'assistant.completed', id: item.id, text: item.text }]
}
assistantSnapshots.delete(item.id)
return [{ type: 'assistant.completed', id: item.id, text: item.text }]
}
const normalizedActivity = toActivity(item)
src/server-functions/codex-event.types.ts
+4 -4
View File
@@ -1,9 +1,8 @@
/**
* Deliberately small structural mirror of the Codex SDK event contract.
* Deliberately small structural mirror of the server-side Codex event contract.
*
* Task 02 does not import `@openai/codex-sdk` into browser-reachable modules.
* Task 01/05 can pass the SDK's `ThreadEvent` values directly because these
* shapes are structurally compatible.
* Browser-reachable modules never import Codex SDK or app-server protocol types.
* The runtime adapter converts server notifications into these shapes first.
*/
export type CodexUsage = {
input_tokens: number
@@ -72,4 +71,5 @@ export type CodexThreadEvent =
| { type: 'item.started'; item: CodexThreadItem }
| { type: 'item.updated'; item: CodexThreadItem }
| { type: 'item.completed'; item: CodexThreadItem }
| { type: 'item.agent_message.delta'; item_id: string; delta: string }
| { type: 'error'; message: string }
src/server/codex/codex-app-server.server.ts
+445
View File
@@ -0,0 +1,445 @@
import { spawn, type ChildProcessWithoutNullStreams } from 'node:child_process'
import { createInterface } from 'node:readline'
import {
CodexRuntimeError,
normalizeCodexRuntimeError,
} from './codex.errors'
import { normalizeThreadId } from './thread-selection.server'
import { resolveWorkspace } from './workspace.server'
import type {
CodexRuntime,
CodexRuntimeOptions,
StreamCodexTurnInput,
} from './codex-runtime'
import type {
CodexThreadEvent,
CodexThreadItem,
CodexUsage,
} from '../../server-functions/codex-event.types'
const LUNA_HIGH_OPTIONS = {
model: 'luna',
effort: 'high',
approvalPolicy: 'never',
sandbox: 'read-only',
sandboxPolicy: { type: 'readOnly', networkAccess: false },
} as const
type JsonRecord = Record<string, unknown>
type AppServerMessage = JsonRecord & {
id?: number | string
method?: string
params?: unknown
result?: unknown
error?: unknown
}
type AppServerRequestResult = {
response: AppServerMessage
notifications: AppServerMessage[]
}
type UsageState = { value: CodexUsage }
const EMPTY_USAGE: CodexUsage = {
input_tokens: 0,
cached_input_tokens: 0,
cache_write_input_tokens: 0,
output_tokens: 0,
reasoning_output_tokens: 0,
}
function isRecord(value: unknown): value is JsonRecord {
return typeof value === 'object' && value !== null
}
function stringValue(value: unknown): string | null {
return typeof value === 'string' ? value : null
}
function normalizePrompt(prompt: string): string {
const normalized = prompt.trim()
if (!normalized) {
throw new CodexRuntimeError('INVALID_INPUT', 'Prompt must not be empty.')
}
return normalized
}
function rpcErrorMessage(error: unknown): string {
if (isRecord(error)) {
const message = stringValue(error.message)
if (message) return message
}
return 'Codex app-server request failed.'
}
function appServerErrorMessage(message: AppServerMessage): string {
const params = isRecord(message.params) ? message.params : null
const error = params && isRecord(params.error) ? params.error : null
return stringValue(error?.message) ?? 'Codex app-server reported an error.'
}
function numberValue(value: unknown): number {
return typeof value === 'number' && Number.isFinite(value) ? value : 0
}
function mapUsage(value: unknown): CodexUsage | null {
if (!isRecord(value)) return null
return {
input_tokens: numberValue(value.inputTokens),
cached_input_tokens: numberValue(value.cachedInputTokens),
cache_write_input_tokens: numberValue(value.cacheWriteInputTokens),
output_tokens: numberValue(value.outputTokens),
reasoning_output_tokens: numberValue(value.reasoningOutputTokens),
}
}
function mapRuntimeStatus(value: unknown): 'in_progress' | 'completed' | 'failed' {
if (value === 'inProgress') return 'in_progress'
if (value === 'failed' || value === 'declined') return 'failed'
return 'completed'
}
function mapFileChangeKind(value: unknown): 'add' | 'delete' | 'update' {
if (isRecord(value) && (value.type === 'add' || value.type === 'delete')) {
return value.type
}
return 'update'
}
function mapAppServerItem(value: unknown): CodexThreadItem | null {
if (!isRecord(value)) return null
const id = stringValue(value.id)
const type = stringValue(value.type)
if (!id || !type) return null
switch (type) {
case 'agentMessage': {
return { id, type: 'agent_message', text: stringValue(value.text) ?? '' }
}
case 'reasoning': {
const summary = Array.isArray(value.summary)
? value.summary.filter((item): item is string => typeof item === 'string')
: []
return { id, type: 'reasoning', text: summary.join('\n') }
}
case 'commandExecution': {
const command = stringValue(value.command)
if (!command) return null
return {
id,
type: 'command_execution',
command,
aggregated_output: stringValue(value.aggregatedOutput) ?? '',
exit_code: typeof value.exitCode === 'number' ? value.exitCode : undefined,
status: mapRuntimeStatus(value.status),
}
}
case 'fileChange': {
const changes = Array.isArray(value.changes)
? value.changes.flatMap((change) => {
if (!isRecord(change) || typeof change.path !== 'string') return []
return [{ path: change.path, kind: mapFileChangeKind(change.kind) }]
})
: []
return {
id,
type: 'file_change',
changes,
status: value.status === 'failed' || value.status === 'declined' ? 'failed' : 'completed',
}
}
case 'mcpToolCall': {
const server = stringValue(value.server)
const tool = stringValue(value.tool)
if (!server || !tool) return null
return {
id,
type: 'mcp_tool_call',
server,
tool,
arguments: undefined,
result: undefined,
error: undefined,
status: mapRuntimeStatus(value.status),
}
}
case 'webSearch':
return {
id,
type: 'web_search',
query: stringValue(value.query) ?? 'Web search',
}
case 'error': {
const message = stringValue(value.message)
return message === null ? null : { id, type: 'error', message }
}
default:
return null
}
}
class AppServerConnection {
private readonly child: ChildProcessWithoutNullStreams
private readonly lines
private readonly messages
private requestId = 0
private stderr = ''
constructor(codexPath: string) {
this.child = spawn(codexPath, ['app-server', '--stdio'], {
cwd: process.cwd(),
stdio: 'pipe',
})
this.child.stderr.setEncoding('utf8')
this.child.stderr.on('data', (chunk: string) => {
this.stderr += chunk
})
this.lines = createInterface({ input: this.child.stdout })
this.messages = this.lines[Symbol.asyncIterator]()
}
send(method: string, params: unknown): number {
if (this.child.stdin.destroyed) {
throw new Error('Codex app-server stdin is closed.')
}
const id = ++this.requestId
this.child.stdin.write(`${JSON.stringify({ id, method, params })}\n`)
return id
}
notify(method: string, params?: unknown): void {
if (this.child.stdin.destroyed) return
this.child.stdin.write(`${JSON.stringify({ method, ...(params === undefined ? {} : { params }) })}\n`)
}
async request(method: string, params: unknown): Promise<AppServerRequestResult> {
const id = this.send(method, params)
const notifications: AppServerMessage[] = []
while (true) {
const message = await this.nextMessage()
if (message.id === id) {
if (message.error !== undefined) {
throw new Error(rpcErrorMessage(message.error))
}
return { response: message, notifications }
}
if (this.isServerRequest(message)) {
this.rejectServerRequest(message)
} else if (message.method) {
notifications.push(message)
}
}
}
async nextNotification(): Promise<AppServerMessage> {
while (true) {
const message = await this.nextMessage()
if (this.isServerRequest(message)) {
this.rejectServerRequest(message)
continue
}
return message
}
}
async close(): Promise<void> {
this.lines.close()
if (this.child.exitCode !== null) return
await new Promise<void>((resolve) => {
const finish = () => resolve()
this.child.once('exit', finish)
this.child.kill()
})
}
private async nextMessage(): Promise<AppServerMessage> {
const next = await this.messages.next()
if (next.done) {
const detail = this.stderr.trim()
throw new Error(detail ? `Codex app-server exited: ${detail}` : 'Codex app-server exited unexpectedly.')
}
try {
const parsed: unknown = JSON.parse(next.value)
if (!isRecord(parsed)) throw new Error('Codex app-server returned a non-object message.')
return parsed as AppServerMessage
} catch (error) {
throw new Error(`Invalid Codex app-server message: ${error instanceof Error ? error.message : 'unknown error'}`)
}
}
private isServerRequest(message: AppServerMessage): boolean {
return message.id !== undefined && message.method !== undefined && message.result === undefined && message.error === undefined
}
private rejectServerRequest(message: AppServerMessage): void {
if (message.id === undefined || this.child.stdin.destroyed) return
this.child.stdin.write(`${JSON.stringify({
id: message.id,
error: { code: -32000, message: 'Interactive server requests are disabled.' },
})}\n`)
}
}
function threadIdFromResponse(response: AppServerMessage): string {
const result = isRecord(response.result) ? response.result : null
const thread = result && isRecord(result.thread) ? result.thread : null
const threadId = thread && stringValue(thread.id)
if (!threadId) throw new Error('Codex app-server did not return a thread id.')
return threadId
}
function notificationEvents(
message: AppServerMessage,
usage: UsageState,
): CodexThreadEvent[] {
const params = isRecord(message.params) ? message.params : null
if (!params || !message.method) return []
switch (message.method) {
case 'thread/started': {
const thread = isRecord(params.thread) ? params.thread : null
const threadId = thread && stringValue(thread.id)
return threadId ? [{ type: 'thread.started', thread_id: threadId }] : []
}
case 'turn/started':
return [{ type: 'turn.started' }]
case 'item/started': {
const item = mapAppServerItem(params.item)
return item ? [{ type: 'item.started', item }] : []
}
case 'item/completed': {
const item = mapAppServerItem(params.item)
return item ? [{ type: 'item.completed', item }] : []
}
case 'item/agentMessage/delta': {
const itemId = stringValue(params.itemId)
const delta = stringValue(params.delta)
return itemId && delta ? [{ type: 'item.agent_message.delta', item_id: itemId, delta }] : []
}
case 'thread/tokenUsage/updated': {
const tokenUsage = isRecord(params.tokenUsage) ? params.tokenUsage : null
const last = tokenUsage && mapUsage(tokenUsage.last)
if (last) usage.value = last
return []
}
case 'turn/completed': {
const turn = isRecord(params.turn) ? params.turn : null
const status = turn && stringValue(turn.status)
if (status === 'failed' || status === 'interrupted') {
const error = turn && isRecord(turn.error) ? turn.error : null
return [{
type: 'turn.failed',
error: { message: stringValue(error?.message) ?? `Codex turn ${status}.` },
}]
}
return [{ type: 'turn.completed', usage: usage.value }]
}
case 'error':
return [{ type: 'error', message: appServerErrorMessage(message) }]
default:
return []
}
}
export function normalizeAppServerNotification(
message: unknown,
usage: UsageState,
): CodexThreadEvent[] {
return isRecord(message) ? notificationEvents(message as AppServerMessage, usage) : []
}
export class CodexAppServerRuntime implements CodexRuntime {
private readonly workspaceRoot?: string
private readonly codexPath: string
constructor(options: CodexRuntimeOptions = {}) {
this.workspaceRoot = options.workspaceRoot
this.codexPath = options.codexPath ?? process.env.CODEX_APP_SERVER_PATH ?? 'codex'
}
async *streamTurn(input: StreamCodexTurnInput): AsyncGenerator<CodexThreadEvent> {
const prompt = normalizePrompt(input.prompt)
const threadId = normalizeThreadId(input.threadId)
const workingDirectory = await resolveWorkspace({
requestedPath: input.workspacePath,
allowedRoot: this.workspaceRoot,
})
if (input.signal?.aborted) throw new Error('Codex turn was cancelled.')
const connection = new AppServerConnection(this.codexPath)
const abortHandler = () => connection.close().catch(() => undefined)
input.signal?.addEventListener('abort', abortHandler, { once: true })
try {
await connection.request('initialize', {
clientInfo: {
name: 'codex-tanstack-demo',
title: 'Codex TanStack Start Demo',
version: '0.1.0',
},
capabilities: { experimentalApi: true, requestAttestation: false },
})
connection.notify('initialized')
const threadRequest = threadId ? 'thread/resume' : 'thread/start'
const threadResult = await connection.request(threadRequest, {
...(threadId ? { threadId } : {}),
cwd: workingDirectory,
model: LUNA_HIGH_OPTIONS.model,
approvalPolicy: LUNA_HIGH_OPTIONS.approvalPolicy,
sandbox: LUNA_HIGH_OPTIONS.sandbox,
})
const activeThreadId = threadIdFromResponse(threadResult.response)
yield { type: 'thread.started', thread_id: activeThreadId }
const turnResult = await connection.request('turn/start', {
threadId: activeThreadId,
input: [{ type: 'text', text: prompt }],
cwd: workingDirectory,
model: LUNA_HIGH_OPTIONS.model,
effort: LUNA_HIGH_OPTIONS.effort,
approvalPolicy: LUNA_HIGH_OPTIONS.approvalPolicy,
sandboxPolicy: LUNA_HIGH_OPTIONS.sandboxPolicy,
})
const usage: UsageState = { value: { ...EMPTY_USAGE } }
for (const message of turnResult.notifications) {
for (const event of normalizeAppServerNotification(message, usage)) yield event
if (message.method === 'turn/completed') return
}
while (true) {
const message = await connection.nextNotification()
for (const event of normalizeAppServerNotification(message, usage)) yield event
if (message.method === 'turn/completed') return
}
} catch (error) {
throw normalizeCodexRuntimeError(error)
} finally {
input.signal?.removeEventListener('abort', abortHandler)
await connection.close()
}
}
}
src/server/codex/codex-app-server.test.ts
+70
View File
@@ -0,0 +1,70 @@
import { describe, expect, it } from 'vitest'
import {
normalizeAppServerNotification,
} from './codex-app-server.server'
const usage = {
value: {
input_tokens: 1,
cached_input_tokens: 0,
cache_write_input_tokens: 0,
output_tokens: 2,
reasoning_output_tokens: 0,
},
}
describe('normalizeAppServerNotification', () => {
it('preserves the assistant lifecycle and delta sequence', () => {
expect(normalizeAppServerNotification({
method: 'item/started',
params: {
item: {
type: 'agentMessage',
id: 'm1',
text: '',
},
},
}, usage)).toEqual([
{ type: 'item.started', item: { id: 'm1', type: 'agent_message', text: '' } },
])
expect(normalizeAppServerNotification({
method: 'item/agentMessage/delta',
params: { itemId: 'm1', delta: 'Hello' },
}, usage)).toEqual([
{ type: 'item.agent_message.delta', item_id: 'm1', delta: 'Hello' },
])
expect(normalizeAppServerNotification({
method: 'item/completed',
params: {
item: {
type: 'agentMessage',
id: 'm1',
text: 'Hello',
},
},
}, usage)).toEqual([
{ type: 'item.completed', item: { id: 'm1', type: 'agent_message', text: 'Hello' } },
])
})
it('does not carry MCP arguments or results across the runtime boundary', () => {
const events = normalizeAppServerNotification({
method: 'item/completed',
params: {
item: {
type: 'mcpToolCall',
id: 'tool-1',
server: 'github',
tool: 'search',
arguments: { token: 'SECRET' },
result: { secret: 'SECRET' },
status: 'completed',
},
},
}, usage)
expect(JSON.stringify(events)).not.toContain('SECRET')
})
})
src/server/codex/codex-runtime.ts
+18
View File
@@ -0,0 +1,18 @@
import type { CodexThreadEvent } from '../../server-functions/codex-event.types'
export type StreamCodexTurnInput = {
prompt: string
threadId?: string | null
workspacePath?: string
signal?: AbortSignal
}
export type CodexRuntimeOptions = {
workspaceRoot?: string
codexPath?: string
}
/** Runtime boundary kept independent from the browser transport and UI. */
export interface CodexRuntime {
streamTurn(input: StreamCodexTurnInput): AsyncGenerator<CodexThreadEvent>
}
src/server/codex/codex.server.ts
+12 -81
View File
@@ -1,85 +1,16 @@
import { Codex, type ThreadEvent } from "@openai/codex-sdk";
import { CodexAppServerRuntime } from './codex-app-server.server'
import type {
CodexRuntime,
CodexRuntimeOptions,
StreamCodexTurnInput,
} from './codex-runtime'
import { CodexRuntimeError, normalizeCodexRuntimeError } from "./codex.errors";
import {
normalizeThreadId,
selectThread,
type CodexClientLike,
type ThreadOptionsLike,
} from "./thread-selection.server";
import { resolveWorkspace } from "./workspace.server";
export type { CodexRuntime, CodexRuntimeOptions, StreamCodexTurnInput } from './codex-runtime'
export { CodexAppServerRuntime } from './codex-app-server.server'
export type StreamCodexTurnInput = {
prompt: string;
threadId?: string | null;
workspacePath?: string;
signal?: AbortSignal;
};
let singleton: CodexRuntime | undefined
export type CodexRuntimeOptions = {
workspaceRoot?: string;
};
const LUNA_HIGH_THREAD_OPTIONS = {
model: "luna",
modelReasoningEffort: "high",
sandboxMode: "read-only",
approvalPolicy: "never",
networkAccessEnabled: false,
} as const;
function normalizePrompt(prompt: string): string {
const normalized = prompt.trim();
if (!normalized) {
throw new CodexRuntimeError("INVALID_INPUT", "Prompt must not be empty.");
}
return normalized;
}
/**
* Server-only wrapper around @openai/codex-sdk.
*
* Authentication is intentionally not accepted from the browser. The SDK reuses
* the Codex/ChatGPT authentication available to the server-side Codex runtime.
*/
export class CodexRuntime {
private readonly client: CodexClientLike<ThreadEvent>;
private readonly workspaceRoot?: string;
constructor(options: CodexRuntimeOptions = {}, client?: CodexClientLike<ThreadEvent>) {
this.workspaceRoot = options.workspaceRoot;
this.client = client ?? (new Codex() as CodexClientLike<ThreadEvent>);
}
async *streamTurn(input: StreamCodexTurnInput): AsyncGenerator<ThreadEvent> {
const prompt = normalizePrompt(input.prompt);
const threadId = normalizeThreadId(input.threadId);
const workingDirectory = await resolveWorkspace({
requestedPath: input.workspacePath,
allowedRoot: this.workspaceRoot,
});
const threadOptions: ThreadOptionsLike = {
...LUNA_HIGH_THREAD_OPTIONS,
workingDirectory,
};
const thread = selectThread(this.client, threadId, threadOptions);
try {
const { events } = await thread.runStreamed(prompt, { signal: input.signal });
for await (const event of events) {
yield event;
}
} catch (error) {
throw normalizeCodexRuntimeError(error);
}
}
}
let singleton: CodexRuntime | undefined;
export function getCodexRuntime(): CodexRuntime {
singleton ??= new CodexRuntime();
return singleton;
export function getCodexRuntime(options?: CodexRuntimeOptions): CodexRuntime {
singleton ??= new CodexAppServerRuntime(options)
return singleton
}
src/server/codex/index.server.ts
+2 -2
View File
@@ -1,4 +1,4 @@
export { CodexRuntime, getCodexRuntime } from "./codex.server";
export type { CodexRuntimeOptions, StreamCodexTurnInput } from "./codex.server";
export { CodexAppServerRuntime, getCodexRuntime } from './codex.server'
export type { CodexRuntime, CodexRuntimeOptions, StreamCodexTurnInput } from './codex.server'
export { CodexRuntimeError, normalizeCodexRuntimeError } from "./codex.errors";
export type { CodexRuntimeErrorCode } from "./codex.errors";