feat(examples): add Worker-native Codex harness - #2208
Conversation
|
There was a problem hiding this comment.
Note
This report is out of date. Scroll down for Devin Review's latest report on this PR.
Devin Review found 8 potential issues.
4 flags not posted on this PR by your GitHub settings — view them in Devin Review. (Configure)
| const existing = this.#effect(operation.operation_id, action.effect_id); | ||
| if (existing?.status === "completed" && existing.result !== null) { | ||
| return parseEffectResult(existing.result); | ||
| } | ||
| if (!existing) { | ||
| this.lifecycle.storage.sql.exec( | ||
| `INSERT INTO cf_codex_effects | ||
| (operation_id, effect_id, kind, status, request, created_at) | ||
| VALUES (?, ?, ?, 'pending', ?, ?)`, | ||
| operation.operation_id, | ||
| action.effect_id, | ||
| action.type, | ||
| JSON.stringify(action), | ||
| Date.now() | ||
| ); | ||
| } | ||
|
|
||
| const result = | ||
| action.type === "model" | ||
| ? await completeCodexModel(this.model, action) | ||
| : await performWorkspaceTool(this.workspace, action); |
There was a problem hiding this comment.
| const [root, session] = url.pathname.split("/").filter(Boolean); | ||
| if (root !== "sessions" || session === undefined) { | ||
| return json({ error: "use /sessions/:session" }, { status: 404 }); | ||
| } | ||
| return env.Coder.getByName(session).fetch(request); |
⚪ agents import sizesMeasured 267 runtime imports as minified bundles. The primary size is gzip; raw minified size is included for diagnosis. An existing import growing by more than 10% is marked red. This report is informational.
Compared No import sizes changed. All 267 current runtime imports
Reported by agent-think[bot]. |
The example is the demo: remove smoke.mjs, the node:test adapter suite and its tsx tsconfig, the /health, results and abort routes, and the "static-wasm" mode/metadata leftovers. Remove the nx `build` script so CI does not need a Rust toolchain; `start` and `deploy` still build the kernel. Fix recovery after a Durable Object restart: a stored terminal action now settles the operation instead of failing it, a replayed terminal operation closes its stream, and an identical resubmit of a still-queued operation re-syncs its Tasks wake. Join text blocks before trimming so boundary whitespace survives, and stop gating UI completion on the demo file. Trim deployment URLs and smoke timing tables from the RFCs. Claude-Session: https://claude.ai/code/session_01KEFnjoMnZBMewsuD9qxGrL
Drop the harness's own cf_codex_effects table and run each model or Workspace effect as a named Tasks step keyed by its effect ID. Tasks replays settled results on later attempts and hands the step's AbortSignal to the model call, so cancellation reaches in-flight work. Document the interrupted step policy in the RFC: tools re-run idempotently, model rounds re-issue and accept a rare duplicate call. Claude-Session: https://claude.ai/code/session_01KEFnjoMnZBMewsuD9qxGrL
There was a problem hiding this comment.
Note
This report is out of date. Scroll down for Devin Review's latest report on this PR.
Devin Review found 2 new potential issues.
4 flags not posted on this PR by your GitHub settings — view them in Devin Review. (Configure)
| if (operation.status === "completed" || operation.status === "failed") { | ||
| // Replayed after settling: make sure the stream is closed and finish. | ||
| await this.#flushEvents(operation, true); | ||
| return projectTaskResult(operation); |
Replace the HTTP polling routes with a WebSocket protocol on the WebSockets capability: a session snapshot on connect, subscribe to replay-then-tail an operation's Streams log, and submit and restart commands. Operation state changes broadcast to every connection. The client connects with useAgent from agents/react and a small useCodexSession hook layers the protocol on that socket, so the whole transcript and every operation's events reload from durable state on reconnect. The worker routes with routeAgentRequest. Claude-Session: https://claude.ai/code/session_01KEFnjoMnZBMewsuD9qxGrL
There was a problem hiding this comment.
Note
This report is out of date. Scroll down for Devin Review's latest report on this PR.
Devin Review found 4 new potential issues.
4 flags not posted on this PR by your GitHub settings — view them in Devin Review. (Configure)
| if (block.type === "text") continue; | ||
| if (!block.complete) continue; | ||
| frames.push({ | ||
| type: "response.output_item.done", | ||
| item: { | ||
| type: "function_call", | ||
| call_id: block.id, | ||
| name: block.name, | ||
| arguments: block.input | ||
| } | ||
| }); | ||
| } | ||
| if (text.length > 0) { | ||
| frames.push( | ||
| { type: "response.output_text.delta", delta: text }, | ||
| { | ||
| type: "response.output_item.done", | ||
| item: { | ||
| type: "message", | ||
| role: "assistant", | ||
| content: [{ type: "output_text", text }] | ||
| } | ||
| } | ||
| ); | ||
| } |
| function isClientMessage(value: unknown): value is CodexClientMessage { | ||
| return ( | ||
| typeof value === "object" && | ||
| value !== null && | ||
| "type" in value && | ||
| typeof value.type === "string" | ||
| ); | ||
| } |
There was a problem hiding this comment.
| const path = action.arguments.path; | ||
| if (typeof path !== "string") { | ||
| return { type: "error", message: `${action.name} requires a path` }; | ||
| } | ||
| if (action.name === "workspace_write") { | ||
| const content = action.arguments.content; | ||
| if (typeof content !== "string") { | ||
| return { type: "error", message: "workspace_write requires content" }; | ||
| } | ||
| await workspace.writeFile(path, content); |
agents
@cloudflare/ai-chat
@cloudflare/codemode
hono-agents
@cloudflare/shell
@cloudflare/think
@cloudflare/voice
@cloudflare/worker-bundler
commit: |
Deploy as codex-harness-example, give the Vite dev server its own inspector port so it can run beside the other harness examples, and document setting CLOUDFLARE_ACCOUNT_ID for logins with several accounts.
agents/react resolved a second React copy from the agents package, which broke every hook call in the browser.
A closed stream ends as soon as it is replayed, and every stream end asked for a snapshot that resubscribed to every operation. Subscribe to each operation once per connection and refresh only after a live tail ends, and scroll to the bottom only when content is added.
Add a synthetic-model stress worker (src/stress) with a driver and a CDP heap profiler (scripts/) so the kernel, Tasks, Streams, and SQLite paths can be pushed without Workers AI. Fix what the runs broke: - the 16-transition cap failed any turn with more than six tool calls; cap model rounds at 24 and transitions at 256 instead - the stored model action repeated the checkpoint's input, halving the transcript a turn could hold before SQLITE_TOOBIG; store it without the input and rehydrate on read - a 1 MB tool argument could not be journaled as a Tasks step result; bound prompts, tool arguments, and tool outputs at 256 KB, replacing oversized arguments with an error the model sees - session listings carried every checkpoint; omit kernel state from listings and let the UI read one checkpoint on demand Record the results and limits in the README and RFC.
There was a problem hiding this comment.
Note
This report is out of date. Scroll down for Devin Review's latest report on this PR.
Devin Review found 1 new potential issue.
4 flags not posted on this PR by your GitHub settings — view them in Devin Review. (Configure)
| await this.streams.open(streamId, { tag: operationId }); | ||
| this.lifecycle.storage.sql.exec( | ||
| `INSERT INTO cf_codex_operations | ||
| (operation_id, stream_id, status, prompt, started_at) | ||
| VALUES (?, ?, 'queued', ?, ?)`, | ||
| operationId, | ||
| streamId, | ||
| prompt, | ||
| Date.now() | ||
| ); |
There was a problem hiding this comment.
…ample # Conflicts: # design/AGENTS.md # examples/next/README.md
…e limits The kernel checkpoint no longer carries the transcript. It is a cursor over one turn: phase, round, and pending tool calls, a few hundred bytes however long the conversation is. Everything with size moves to the SDK's durable primitives: - prompts, assistant messages, and tool outputs are Sessions messages, so large ones chunk across rows; each round hydrates a byte-budgeted window and Sessions compacts the branch past a token threshold - tool calls reach the kernel as a pointer to the stored assistant message and tool outputs are stored as tool messages, so Tasks journals effects by id instead of by value - events append to the operation's Streams log in the same transaction as the checkpoint; the journal table is gone - workspace_read takes offset and max_bytes, and the Workspace spills large files to an R2 bucket - the prompt shows a marker for any tool input or output over 64 KB, with a ranged read to page it back, so one large write cannot evict the rest of the turn from the context window Remove the 256 KB payload caps and the transition cap; the round cap is a configurable option. The stress suite now completes 8 MB prompts, 8 MB tool payloads, 60-round turns, 200 calls in a round, 200 turns on one object, and 32 concurrent objects with a 0.5 KB checkpoint throughout.
A transient Workers AI capacity error failed the whole turn. A round that returns no usable response now throws inside its Tasks step, which retries with backoff before the turn fails. Nothing is stored for a failed round, so a retry starts clean.
There was a problem hiding this comment.
Note
This report is out of date. Scroll down for Devin Review's latest report on this PR.
Devin Review found 8 new potential issues.
4 flags not posted on this PR by your GitHub settings — view them in Devin Review. (Configure)
| await this.session.appendMessage({ | ||
| id: userMessageId(operationId), | ||
| role: "user", | ||
| parts: [{ type: "text", text: prompt }], | ||
| metadata: { operationId } | ||
| }); |
| if (action.name === "workspace_read") { | ||
| const content = await workspace.readFile(path); | ||
| if (content === null) | ||
| return { success: false, output: { path, found: false } }; | ||
| // Files have no size limit; the model pages through big ones by range. | ||
| const bytes = new TextEncoder().encode(content); | ||
| const offset = clampInteger(input.offset, 0, bytes.byteLength); | ||
| const maxBytes = clampInteger(input.max_bytes, 1, DEFAULT_READ_BYTES); | ||
| const end = Math.min(bytes.byteLength, offset + maxBytes); |
| const existing = this.#operation(operationId); | ||
| if (existing) { | ||
| if (existing.prompt !== promptPreview(prompt)) { | ||
| throw new Error( | ||
| `Codex operation ${operationId} already exists with different input` | ||
| ); | ||
| } |
| const end = Math.min(bytes.byteLength, offset + maxBytes); | ||
| return { | ||
| success: true, | ||
| output: { | ||
| path, | ||
| content: new TextDecoder().decode(bytes.subarray(offset, end)), | ||
| offset, | ||
| end, | ||
| total_bytes: bytes.byteLength, | ||
| ...(end < bytes.byteLength ? { next_offset: end } : {}) |
| const text = blocks | ||
| .filter((block) => block.type === "text") | ||
| .map((block) => block.value) | ||
| .join("") | ||
| .trim(); |
There was a problem hiding this comment.
🟡 Model output loses whitespace
trim removes intentional edge whitespace before events and final output are generated. Code blocks and whitespace-sensitive answers change.
| const text = blocks | |
| .filter((block) => block.type === "text") | |
| .map((block) => block.value) | |
| .join("") | |
| .trim(); | |
| const text = blocks | |
| .filter((block) => block.type === "text") | |
| .map((block) => block.value) | |
| .join(""); |
Was this helpful? React with 👍 or 👎 to provide feedback.
| this.lifecycle.storage.transactionSync(() => { | ||
| for (const event of transition.events) { | ||
| if (event.seq < writer.cursor) continue; | ||
| if (event.seq !== writer.cursor) { | ||
| throw new Error( | ||
| `Codex event gap for ${operation.operation_id}: expected ${writer.cursor}, got ${event.seq}` | ||
| ); | ||
| } | ||
| writer.append(event); | ||
| } |
| return step.do( | ||
| `effect:${action.effect_id}`, | ||
| { retries: MODEL_ROUND_RETRIES, timeout: "10 minutes" }, | ||
| ({ signal }) => this.#runEffect(operation, checkpoint, action, signal) |
| submit: (input) => this.submit(input), | ||
| operation: (operationId) => this.snapshot(operationId), | ||
| message: (id) => this.message(id), | ||
| readFile: (path) => this.readFile(path), |
There was a problem hiding this comment.
Devin Review found 3 new potential issues.
5 flags not posted on this PR by your GitHub settings — view them in Devin Review. (Configure)
| if ( | ||
| !(error instanceof Error) || | ||
| error.name === "AttemptSupersededError" | ||
| ) { | ||
| throw error; | ||
| } | ||
| this.#settleFailed(input.operationId, errorMessage(error)); |
There was a problem hiding this comment.
🔴 Platform failures become terminal turns
When a step exhausts transient platform retries, #drive converts the propagated error into a terminal turn failure. Tasks cannot defer recovery to a fresh invocation.
Prompt for agents
Preserve Tasks control-flow and platform-failure semantics in CodexHarness.#drive. The catch in examples/next/harnesses/codex/src/codex-harness.ts currently rethrows non-Error controls and AttemptSupersededError only. ReplayStep also intentionally propagates code-update resets, memory-limit resets, and exhausted transient platform failures so Tasks can defer or apply breaker policy. Use the Tasks/retries predicates or a framework-supported error classifier to rethrow every control/platform failure, and settle the Codex operation only for terminal application errors.
Was this helpful? React with 👍 or 👎 to provide feedback.
| case "stream_end": { | ||
| // Only an operation we were tailing live needs its settled state | ||
| // and file; a replayed closed stream ends immediately. | ||
| const operation = stateRef.current.operations.find( | ||
| (candidate) => candidate.operationId === message.operationId | ||
| ); | ||
| if (operation && isActive(operation)) { | ||
| sendRef.current({ type: "snapshot", id: crypto.randomUUID() }); | ||
| } |
There was a problem hiding this comment.
🟡 Completed turns leave workspace view stale
After completion, isActive suppresses the snapshot request at stream end. The sidebar keeps the file state captured before the turn.
| case "stream_end": { | |
| // Only an operation we were tailing live needs its settled state | |
| // and file; a replayed closed stream ends immediately. | |
| const operation = stateRef.current.operations.find( | |
| (candidate) => candidate.operationId === message.operationId | |
| ); | |
| if (operation && isActive(operation)) { | |
| sendRef.current({ type: "snapshot", id: crypto.randomUUID() }); | |
| } | |
| case "stream_end": { | |
| sendRef.current({ type: "snapshot", id: crypto.randomUUID() }); | |
| return; | |
| } |
Was this helpful? React with 👍 or 👎 to provide feedback.
| const result = await this.#performEffect( | ||
| operation, | ||
| transition.checkpoint, | ||
| transition.action, | ||
| step | ||
| ); | ||
| command = { | ||
| type: "resolve_effect", | ||
| checkpoint: transition.checkpoint, | ||
| effect_id: transition.action.effect_id, | ||
| result | ||
| }; |
There was a problem hiding this comment.
🟡 Round limit permits extra call
When the kernel emits the first disallowed round, #performEffect calls the model before checking the limit. The operation pays for an unused response.
Prompt for agents
Enforce maxRounds before dispatching each model action, including the model action emitted after the last tool in a batch. Do not call the provider or append an assistant message for a round that the next loop iteration will reject. Clarify whether maxRounds counts model calls or completed continuations and test small limits such as 1.
Was this helpful? React with 👍 or 👎 to provide feedback.
Summary
Adds
examples/next/harnesses/codex, a Worker-native Codex harness composed asCodexHarness extends LifecycleCapabilityon a plainDurableObject. Example only: no package export, no changeset.step.do, Streams records the kernel events, and a Shell Workspace holds the files.LanguageModelV4directly and runs Kimi K2.7 Code throughworkers-ai-provider@4, Workers AI, and AI Gateway. Every model-emitted tool call is preserved and settled sequentially before the next round.WebSocketscapability. The client connects withuseAgentfromagents/react; auseCodexSessionhook replays-then-tails each operation's Streams log, so the transcript and events reload from durable state on reconnect. A "Restart and verify" action aborts the Durable Object and shows the operation and file recovered.Known gaps
Run
Rust 1.95 with the
wasm32-unknown-unknowntarget is required locally;pnpm run startbuilds the kernel first. CI does not build the kernel.