MARSOHS-440: doctl-codex proxy M3-M5

[julia-ye]

Implementation plan here (from Adil Hafeez): https://github.internal.digitalocean.com/digitalocean/cthulhu/pull/166147/files
Has milestones that this is based on (M3-M5)

Jira: https://do-internal.atlassian.net/browse/MARSOHS-440

[adilhafeez]

Can you explain this feature a bit more with some sample use?

[ElanHasson]

Review: M3–M5 codex proxy

Reviewed the full diff against feat/agents-subcommands, ran the suite under -race, and reproduced the concurrency findings rather than eyeballing them. The protocol translation work is genuinely strong — the shape-by-shape sourcing against codex-rs and the "why this field is omitted" comments are better than most of what I review, and commandExecutionRequestApprovalParams' note about AcceptWithExecpolicyAmendment silently reintroducing unhonorable approvals is a great catch.

The problems are concentrated in the merge resolution and in lifecycle/concurrency, not in the translation layer. Findings are ordered by severity.

I've opened #1895 stacked on this branch with fixes for #2 and #3 below, since both are mechanical and verified. Everything else is a design call and left to you.

---

🔴 1. The merge dropped ensureEventLoop-before-SendInput, so a turn can silently lose all of its events

This is the one I'd most want addressed before merge.

feat/agents-subcommands contains this, from the M0–M2 PR (#1878):

// Open (or confirm) the event-loop stream before SendInput, not after:
// SendInput is what makes the harness start the run, so a reader
// attached only afterward can race a fast run to completion and miss
// every event it emits. See ensureEventLoop.
if err := f.ensureEventLoop(ctx); err != nil {
    return nil, &agentproxy.RPCError{Code: -32000, Message: "opening event stream failed: " + err.Error()}
}

Merge a422044 resolved facade.go entirely in favor of this branch, and ensureEventLoop, lookupTurn, and stopEventLoop are all gone at HEAD. turn/start now does the opposite of what that comment warns against:

resp, err := f.Sessions.SendInput(...)   // harness starts the run here
...
f.trackTurn(ctx, resp.RunID)             // ...and only now do we go open a stream, asynchronously

trackTurn is fire-and-forget: it spawns runEventLoop, which then calls StreamSession, which does an HTTP round trip. Meanwhile godo is explicit that this is unrecoverable:

Live (the default) reads GET .../sessions/{id}/events, served by the data plane. Delivery is forward-only from the moment of attach, so a live stream never carries events that predate it.

So anything the run emits before the attach completes is gone for good. Because runEventLoop opens with cursor == "", there's no Last-Event-ID to recover it either. If run.started is the event lost, the damage compounds: ts.startedAt stays 0 and ts.itemStarted stays false, so the client never gets turn/started or item/started, subsequent item/agentMessage/deltas reference an itemId the client was never told about, and finishTurn skips item/completed entirely. A fast turn renders as nothing at all.

Removing lookupTurn reopens the mirror-image race for second-and-later turns. It existed precisely for this:

// lookupTurn returns the tracked turnState for runID, retrying briefly if
// it's not there yet. turn/start calls SendInput, which is what makes the
// harness create the run and start emitting its events, and only registers
// runID in f.turns once SendInput returns — so this goroutine can observe
// that run's first event fractionally before trackTurn's map write becomes
// visible to it.

At HEAD, drainStream does a single lookup and silently drops on miss:

ts, ok := f.turns[ev.RunID]
if !ok {
    continue // event for a run this facade isn't tracking
}

For turn 2+, the shared loop is already draining while Dispatch is still between SendInput and trackTurn, so that continue discards real events for a real turn. The first turn is structurally safe here (trackTurn inserts before spawning), but that's the only case that is.

Worth checking whether this was a deliberate revert or a conflict resolved on autopilot — if the latter, it's worth re-diffing the whole merge against the base, since these were the two functions I happened to look for.

🔴 2. All four --replay tests fail on this branch

--- FAIL: TestFacade_Replay_ThreadResume (2.00s)
--- FAIL: TestFacade_Replay_FiresOnlyOnce (2.00s)
--- FAIL: TestFacade_Replay_UnaffectedByConcurrentLiveTurnsReset (2.00s)
--- FAIL: TestFacade_Replay_RetriesAfterAbortedAttempt (2.11s)

codex facade: replay history fetch failed, will retry on next connect:
GET .../v2/agents/sessions/sess_test123/stream?replay_only=true: 404 404 page not found

agentproxytest registers only .../events, but godo routes replay-only reads to .../stream?replay_only=true. Another merge seam: the harness kept the base's route list while this branch's new replay code needs the control-plane route. handleStream already handles replay_only and QueueReplayHistory already documents serving it, so it's purely the missing registration — one line makes all four pass. Fixed in #1895.

Two knock-on notes. TestFacade_Replay_Disabled passes for the wrong reason right now (it asserts expectNone, which a 404 also satisfies), as does the first half of TestFacade_Replay_RetriesAfterAbortedAttempt. And the checks reported on this PR are only GitGuardian and Semgrep — if Go tests aren't gating, that's how four red tests on the headline feature got this far.

🔴 3. failAllTrackedTurns can crash the process

f.mu.Lock()
turns := f.turns
f.mu.Unlock()

for runID, ts := range turns { // ranges the *live* map
    f.finishTurn(runID, ts, "failed", ...)
}

turns aliases f.turns. finishTurn deletes from it, a concurrent turn/start inserts via trackTurn, and a --replay goroutine's own finishTurn deletes from it — all under f.mu, but a lock held only on the writer's side doesn't protect an open iterator. This is fatal error: concurrent map iteration and map write, an unrecoverable crash, and it fires on the connection-loss path — the moment users can least afford one.

Reproduced under -race:

WARNING: DATA RACE
Write at 0x00c0003c78c0 by goroutine 10:
  runtime.mapassign_faststr()
Previous read at 0x00c0003c78c0 by goroutine 9:
  runtime.mapIterStart()
  codex.(*Facade).failAllTrackedTurns() facade.go:1173

Fixed in #1895 by copying under the lock, with a -race regression test. I used a copy rather than detaching via f.turns = nil because detaching widens the window in finding #7.

🔴 4. One shared Facade across connections races on f.notifier

This PR changes Serve/ServeListener from func() Facade to a single instance, and deletes the comment that explained why the factory existed:

newFacade is called once per accepted connection, not once for the whole listener: a facade like codex's carries per-connection state... reusing one Facade instance across a disconnect/reconnect would leak that state into the new connection — notifications from a still-unwinding previous connection's background goroutine could even land on the new socket via a shared notifier.

That hazard is now live. SetNotifier writes f.notifier from the HTTP handler goroutine on every connection, while background goroutines read it via notify, with no synchronization on the field:

WARNING: DATA RACE
Read at 0x00c000238020 by goroutine 14:
  codex.(*Facade).notify() facade.go:902
Previous write at 0x00c000238020 by goroutine 13:
  codex.(*Facade).SetNotifier() facade.go:165

(Synthetic repro, but the access pattern is the real one.) And maybeReplay's own comment confirms the overlap isn't theoretical:

proxy.go's "one slot" accept loop only frees up once the old connection's handleConn returns, but that old connection's own replaySessionHistory goroutine can still be blocked reading the replay-only stream after the WS itself is gone

So an old connection's replay goroutine outlives its socket, and when it resumes it reads whatever f.notifier now points at — the new client's connection. Beyond the race itself, that's stale history from a dead connection being written onto a live one's timeline.

Guarding the field with a mutex (or atomic.Value) fixes the race but not the delivery leak. Cleanly separating the two probably means either passing the notifier down through the call chain instead of stashing it, or having each goroutine capture its own notifier at start and drop writes once its connection's ctx is done. Worth asking whether the factory needed to be removed at all — the only state that genuinely wants to outlive a connection is replayDone, which could live outside the Facade.

🔴 5. --replay re-prompts for historical approvals and re-resolves settled HITLs

replaySessionHistory feeds every historical event through the same translateEvent used for live events, and translateEvent has no replay awareness. So hitl.requested records in durable history get re-driven as if they were happening now. Confirmed against the harness:

codex facade: auto-rejecting file_change HITL hitl-old-1 (pushing codex toward shell commands for file edits)
harness received a resolve for an already-historical HITL:
  {RequestID:hitl-old-1 Outcome:HITL_OUTCOME_REJECT Source:RESOLUTION_SOURCE_OUT_OF_BAND}
client was re-prompted to approve a historical command:
  method=item/commandExecution/requestApproval Command:rm -rf /tmp/x Cwd:/workspace

Two distinct problems. file_change and unknown kinds fire real ResolveHITL POSTs against hitl_ids the harness settled long ago — spurious writes whose failure mode depends on how the harness treats a re-resolve. command_execution is worse: the user is shown a modal asking them to approve a command that already ran, with no indication it's history, and a goroutine blocks on that answer. Approving or declining then races a resolve against an already-closed request.

Since run.hitl_resolved is in the history too, the outcome is already known at replay time. Simplest fix is a replay bool on translateEvent that renders past approvals as completed items and never sends requests or calls ResolveHITL.

---

🟡 6. replayDone outlives the connection, so a reconnecting client gets a blank thread

replayDone is per-Facade, and the Facade is now per-process. When codex disconnects and a fresh codex --remote connects, that's a new TUI with empty scrollback — but maybeReplay short-circuits and it sees no history at all, despite --replay. The doc comment's reasoning ("must not repeat and re-show history a second time to a client that already saw it") holds within one connection but not across them. Under the old per-connection factory this worked correctly. Consider keying on the connection rather than the process, which the notifier question in #4 also points toward.

🟡 7. The reconnect cursor resets whenever the loop exits and restarts

cursor is a local in runEventLoop. When the loop exits via the noTurnsLeft path, the next turn/start starts a fresh loop with cursor == "" — so that attach is forward-only with no Last-Event-ID, reproducing finding #1's window on every turn that follows an idle exit, not just the first. Hoisting the cursor onto the Facade would let a restarted loop resume where the last one stopped.

Related: this is why I fixed #3 with a copy rather than f.turns = nil. Detaching lets a turn/start racing runEventLoop's exit register a turn into a fresh map while streamStarted is still true, so no loop is started for it, and the deferred cleanup then discards it. That window exists today; I didn't want to widen it.

🟡 8. Replayed usage inflates the live thread total

translateEvent's RunUsageRecorded case accumulates into f.totalUsage, and replay goes through the same path — so historical token usage is added to the running total the TUI displays for the current thread. Given total is thread-scoped this may be intentional, but combined with #6 it's inconsistent: whether your total includes history depends on whether you're the first client to connect since the process started.

🟡 9. A failed item/started doesn't propagate clientDead

if !f.notify("turn/started", ...) {
    return true
}
ts.itemStarted = f.notify("item/started", ...)   // failure swallowed

The turn/started failure correctly reports a dead client; the item/started failure immediately below it only records itemStarted = false. If the connection dies between the two, the loop keeps draining and notifying against a socket that's gone, instead of unwinding. Looks unintentional given the line above it.

🟡 10. CheckOrigin always returns true (pre-existing)

Not introduced here, but M4 raises the stakes enough to flag. The upgrader accepts any Origin:

CheckOrigin: func(r *http.Request) bool { return true },

with the rationale that "no browser reaches this." Browsers can reach loopback WebSockets, and WebSockets aren't subject to CORS — so any page the user visits while the proxy is listening can connect to ws://127.0.0.1:1144 and drive the session: send turn/start, read the full response stream, and now answer item/commandExecution/requestApproval requests. The default port is fixed and documented, and there's no auth. The one-client slot is the only mitigation, and it only helps while codex is actually attached. Rejecting requests whose Origin header is present and non-loopback would close this without affecting websocat or codex, neither of which sends one.

---

🟢 11. A duplicate reply can wedge the read loop

deliverReply does a blocking send on a buffer-1 channel:

ch, ok := w.pending[key]
...
ch <- msg

If two frames arrive with the same id before Request wakes and clears the pending entry, the second send blocks forever — inside handleConn's read loop, which stops reading the connection entirely. A non-blocking send with a default: that logs the duplicate keeps a misbehaving client from wedging the pump.

🟢 12. gofmt

harness.go's godo import landed inside the stdlib group; gofmt -l flags the file. Fixed in #1895.

---

Verification

go test -race ./internal/agentproxy/...   # 4 failures, all replay
gofmt -l internal/agentproxy/             # harness.go
GOOS=windows go build ./commands/         # ok (syscall.SIGTERM is fine on Windows)

With #1895 applied, go test -race ./internal/agentproxy/... is green and gofmt is clean.

Nice work on the protocol layer — the substance of my concerns is really about what the merge dropped and about Facade lifetime, not about the translation itself.

[adilhafeez]

whats the use case for this?

[adilhafeez]

why max 8?

[SSharma-10]

Looks like we are back to a single shared Facade across connections, with only the notifier getting swapped on connect.

On #1878 we switched to a new Facade per connection so reconnect (codex --remote) wouldn't keep sticky streamStarted / turn state. Was that intentional here for --replay, or can we restore the per-connection factory?

[julia-ye]

Restored in this branch — ServeListener takes newFacade func() Facade again (same as #1878). Each codex --remote connection gets a fresh Facade so streamStarted / turns / notifier can’t leak across reconnect. --replay is per-connection too (finding #6), so a reconnecting TUI with empty scrollback gets history again.

[SSharma-10]

Looks like SendInput is still happening before the SSE stream is attached (trackTurn → async runEventLoop → StreamSession).

We ran into this in #1878 and changed the flow to call ensureEventLoop before SendInput so we wouldn't miss early events against a live harness. It looks like that ordering may have been lost during the merge. Would you mind restoring the attach-before-SendInput flow?

[julia-ye]

Restored — ensureEventLoop runs before SendInput again (same ordering as #1878). trackTurn only registers the run.

[logwolvy]

This will be wired to OHP under the hood, no breaking changes hopefully (I have pushed my commit to switch to OHP for streaming; under review currenly), just FYI.

[ElanHasson]

Review: M3–M5 codex proxy (round 2, 6aebc49)

Re-reviewed against feat/agents-subcommands at 6aebc49. All twelve findings from my previous round are genuinely fixed, not just papered over — I verified each one at HEAD rather than trusting the commit message:

#	Finding	Status at HEAD
1	ensureEventLoop before SendInput	Restored (facade.go:885), lookupTurn back (facade.go:1006)
2	Four failing --replay tests	.../stream route registered; suite green
3	failAllTrackedTurns map-iteration crash	Copies under f.mu (facade.go:1260)
4	Shared Facade racing on notifier	newFacade func() Facade restored
5	--replay re-prompting settled HITLs	translateEvent(..., replay bool) gates the side effects
6	replayDone outliving the connection	Per-Facade, so per-connection
7	Reconnect cursor reset per loop	Hoisted to Facade.streamCursor
8	Replayed usage inflating the live total	Replay path returns early
9	item/started not propagating clientDead	Both notifies propagate now
10	CheckOrigin accepting any origin	allowedProxyOrigin + test
11	Duplicate reply wedging the read loop	Non-blocking send with default:
12	gofmt	Clean

go test -race -count=2 ./internal/agentproxy/... is green, go vet and gofmt are clean, and GOOS=windows/GOOS=darwin builds fine.

There is one new blocking finding, and it came in with the M5 commit (065c45bc) rather than with the fixes. It's serious enough that I'd hold the merge on it alone.

---

🔴 1. The reconnect cursor drops most of the event stream, because event ids are not ordered

drainStream treats the event id as a monotonically increasing position and discards anything that doesn't sort above the running cursor:

if ev.EventID != "" {
    if *cursor != "" && ev.EventID <= *cursor {
        continue
    }
    *cursor = ev.EventID
}

The doc comment justifies this by "relying on canonical event ids being ULIDs (lexicographically ordered by time)," and cites doctl agents attach's tokenDeduper as precedent. Neither holds up.

Attach is not precedent. attach's drainStream calls cursor.set(ev.EventID) unconditionally (commands/agents.go:1545) and never compares two ids. Its tokenDeduper dedupes on the text of a run.token_delta (dedup.allow(p.Text), reset on any non-token event) — content, not position. The ordinal comparison here is new, not a generalisation of something already proven in production.

The ids aren't ordered. The producer of every run.* event this facade consumes is the guest harness, and it mints them randomly:

• harness-api/public/events.proto:25 — string event_id = 1; // Random UUIDv4; unique event identity (not monotonic)
• plano crates/agent_harness/src/spi/events.rs — event_id: Uuid::new_v4()

And critically, the ULID path is not ordered either. harness-api's own events.NewEventID() is ulid.MustNew(ulid.Now(), rand.Reader) — that's 48 bits of millisecond timestamp plus 80 bits of fresh randomness per call, not ulid.Monotonic. Two ids minted in the same millisecond sort randomly against each other, and token deltas arrive many per millisecond. So the assumption fails under both candidate id schemes.

Because the cursor only ratchets upward, it converges on the maximum id seen so far and the drop rate climbs toward 100% as a turn goes on. I drove one ordinary turn (run.started, 12 × run.token_delta, run.completed) through a fake harness that mints ids the way the real one does. Both schemes, several runs each:

uuidv4 (events.proto / plano CanonicalEvent::new)
    token deltas: sent 12, delivered 3
    token deltas: sent 12, delivered 2
    token deltas: sent 12, delivered 6

ulid minted within one millisecond (harness-api events.NewEventID)
    token deltas: sent 12, delivered 6
    token deltas: sent 12, delivered 4
    token deltas: sent 12, delivered 5

run.completed was dropped in every single run, and that's the part that turns a rendering glitch into a broken session. With the turn never finished, noTurnsLeft stays false, so runEventLoop reconnects until the budget is gone and then synthesizes a failure:

codex facade: stream ended (err=<nil>), reconnecting     (×4)
codex facade: reconnected 5 times without staying healthy, giving up
turn/completed params: {Turn:{ID:run-uuid-1 Status:failed Error:"lost connection to hosted session"}}

So the user experience is: a truncated, garbled answer, a ~15s stall while the backoff runs down, and then a bogus "lost connection to hosted session" on a session that never lost anything. I also saw duplicate turn/started/item/started for one turn across those reconnects.

The reason CI didn't catch this is that agentproxytest's fixture mints fmt.Sprintf("evt-%d", i) (harness.go:323) — synthetic, monotonic, and nothing like what the harness emits. (Worth noting the fixture is itself only correct below ten events: "evt-10" < "evt-2" lexicographically, so a test queueing eleven would start silently dropping too.)

On the fix: I'd drop the ordinal comparison and follow attach — set the cursor unconditionally for its Last-Event-ID job, and if redelivery needs handling, do it with an explicit set of ids seen since the last attach, or with attach's content-based deduper. ev.Seq is the field that actually carries a monotonic position, but per the SSE contract it's 0 for everything until task D1 ships. That same note is the reason there's little to guard against right now: while positions are 0-0, Last-Event-ID "becomes a real resume point only once D1/History land," so reconnects are forward-only and can't redeliver a prefix in the first place. The dedup is currently costing the common path everything and buying nothing.

---

🟡 2. Replayed history is stamped with the current time

Every timestamp translateEvent produces comes from the clock, not the event: ts.startedAt = time.Now().Unix() (translate.go:49), StartedAtMs: time.Now().UnixMilli() (:132, :166), CompletedAtMs (:226, :265), and finishTurn's completedAt. That's fine live, but under --replay a conversation from last week renders as having happened in the last few milliseconds, with every turn duration ≈ 0. godo.HostedAgentEvent carries the real event time in At; using it (at least on the replay path) would make history look like history.

🟡 3. The thread token total silently resets after an idle drop

runEventLoop's deferred cleanup does f.totalUsage = tokenUsageBreakdown{} (facade.go:1149). The loop exits via noTurnsLeft whenever the stream drops with no turn in flight — a normal idle timeout between two user messages. The field's own doc says it "accumulates every live run.usage_recorded event seen across this connection," and threadTokenUsageUpdatedNotification's total is thread-scoped, so the TUI's running total jumping backwards after an idle pause contradicts both. turns and streamStarted genuinely are per-loop; totalUsage is per-connection and probably shouldn't be reset alongside them.

🟡 4. replaySessionHistory doesn't skip run-less events

drainStream skips events with no run id (facade.go:1306) precisely because control frames belong to no turn. The replay loop has no equivalent guard — turns[ev.RunID] on a stream.state frame synthesizes a phantom turnState keyed "" with itemID: "-msg". Harmless today because no run-less event kind matches a translateEvent case, but it's a guard that exists on one path and not its twin, and the fake harness emits stream.state on the replay stream too.

🟡 5. clientDead propagation is inconsistent across event kinds

RunStarted and TokenChunk both return true when f.notify fails, but the ToolCallStarted/ToolCallCompleted (translate.go:120, :157, :211, :256) and RunUsageRecorded (:403) notifies discard the result. A client that dies during a tool-call-heavy stretch keeps the loop draining and notifying at a closed socket until something else happens to notice — the same thing finding #9 fixed one case of last round.

---

🟢 6. lookupTurn costs 200 ms per permanently-untracked event

lookupTurn polls for eventClaimWait (200 ms) before giving up, which is right for the SendInput/trackTurn race it was written for, but it's applied to every miss including ones that can never resolve — an event for a run already deleted by finishTurn, say. Each one stalls the shared drain loop for a fifth of a second while every other in-flight turn waits. Bounding the retry to runs that could plausibly still be registering (e.g. only within a short window of a SendInput return) would keep the fix without the tax.

🟢 7. Replay notifications can race the thread/start reply onto the wire

maybeReplay spawns its goroutine before Dispatch returns (facade.go:810, :831), and handleConn only writes the thread/start result afterward. Nothing orders the two, so a replayed turn/started can in principle reach the client before the thread it refers to has been acknowledged. In practice replaySessionHistory does an HTTP round trip first and will essentially always lose that race — but it's ordering by luck, and gating the goroutine on the reply having been written would make it free.

---

Verification

go build ./...                                  # ok
go vet ./internal/agentproxy/... ./commands/     # clean
gofmt -l internal/agentproxy/ commands/agents.go # clean
go test -race -count=2 ./internal/agentproxy/... # ok
GOOS=windows go build ./commands/                # ok
GOOS=darwin  go build ./commands/                # ok

Finding #1 was reproduced with a throwaway fake harness minting ids both ways; everything else is from reading, and I've said so where I wasn't able to confirm behaviour directly.

The response to the last round was thorough — the ensureEventLoop ordering, the per-connection facade, and the replay HITL gating are all done properly, and the origin check went further than I asked for. Finding #1 is the one thing standing between this and a merge, and it's a small diff.