mirror of
https://gh-proxy.org/https://github.com/klinsmaya/multica.git
synced 2026-11-08 01:16:45 +08:00
Andrei
a47a1ced0c
* fix(agent): keep Codex commentary out of output on empty final answer * test: pin empty-output completion and single result delivery
4392 lines
156 KiB
Go
4392 lines
156 KiB
Go
package agent
|
|
|
|
import (
|
|
"bufio"
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"io"
|
|
"log/slog"
|
|
"os"
|
|
"os/exec"
|
|
"path/filepath"
|
|
"regexp"
|
|
"sort"
|
|
"strconv"
|
|
"strings"
|
|
"sync"
|
|
"sync/atomic"
|
|
"syscall"
|
|
"time"
|
|
|
|
"github.com/multica-ai/multica/server/pkg/redact"
|
|
)
|
|
|
|
// codexBlockedArgs are flags hardcoded by the daemon that must not be
|
|
// overridden by user-configured custom_args. The mcp_servers config keys
|
|
// live in the per-task `$CODEX_HOME/config.toml` (written by
|
|
// ensureCodexMcpConfig); user-supplied `-c mcp_servers.…` overrides are
|
|
// stripped separately by filterCodexCustomConfigOverrides because they
|
|
// share the `-c` flag with legitimate non-MCP overrides like `-c model=…`.
|
|
var codexBlockedArgs = map[string]blockedArgMode{
|
|
"--listen": blockedWithValue, // stdio:// transport for daemon communication
|
|
}
|
|
|
|
const (
|
|
codexFastServiceTier = "priority"
|
|
codexStandardServiceTier = "default"
|
|
codexFastModeFeature = "fast_mode"
|
|
)
|
|
|
|
// codexStderrTailBytes bounds the stderr tail captured for inclusion in
|
|
// error messages when codex exits before the JSON-RPC handshake (e.g. the
|
|
// user supplied a custom_args flag that the `app-server` subcommand
|
|
// rejects). Kept as its own constant so bumping codex independently of
|
|
// other agents stays easy if codex starts shipping longer failure traces.
|
|
const (
|
|
codexStderrTailBytes = 2048
|
|
defaultCodexSemanticInactivityTimeout = 10 * time.Minute
|
|
// defaultCodexFirstTurnNoProgressTimeout caps how long the first turn may
|
|
// stay completely silent after the app-server reports turn/started. Its only
|
|
// job is to fail fast instead of waiting out
|
|
// defaultCodexSemanticInactivityTimeout, so it only has to stay well under
|
|
// that 10 minute backstop.
|
|
//
|
|
// 30s turned out to sit inside the normal range rather than above it: two
|
|
// independent field reports measured the first progress event landing just
|
|
// past 30s on gpt-5.5, and ~39s for a WSL app-server (GH #5959). Both were
|
|
// healthy turns the watchdog killed. 60s clears that evidence with margin
|
|
// while keeping the fast-fail value (MUL-5542).
|
|
defaultCodexFirstTurnNoProgressTimeout = 60 * time.Second
|
|
defaultCodexHandshakeTimeout = 30 * time.Second
|
|
// defaultCodexTurnInterruptTimeout gives app-server a short, bounded window to
|
|
// finish an interrupted turn and report its final token usage before the
|
|
// process is torn down. This mirrors the native Codex stop flow without
|
|
// letting a wedged interrupt delay replacement executions indefinitely. The
|
|
// value is independently configurable per execution; lifecycle logs record
|
|
// the observed acknowledgement/completion latency so deployments can tune it
|
|
// from their own app-server evidence. On 2026-09-01, the reproducible real-
|
|
// agent probe in codex_interrupt_latency_integration_test.go measured 20
|
|
// interrupted generations and 20 interrupted active tools on WSL/Linux with
|
|
// codex-cli 0.147.0: turn/completed P99/max were 15ms/15ms and 13ms/13ms.
|
|
// Two seconds deliberately keeps more than 130x headroom over the observed
|
|
// worst protocol completion for host load and cross-platform scheduling.
|
|
defaultCodexTurnInterruptTimeout = 2 * time.Second
|
|
// thread/start and thread/resume may refresh the model catalog, initialize
|
|
// MCP integrations, and restore persisted history. Field evidence shows
|
|
// healthy calls crossing the 30s budget used for local initialize, so keep
|
|
// their default separate without slowing failures for lightweight RPCs.
|
|
defaultCodexThreadHandshakeTimeout = 60 * time.Second
|
|
codexVersionDiagnosticTimeout = 2 * time.Second
|
|
// Keep a raw agent-message tail small enough that a slow daemon does not
|
|
// make the stdout reader retain token-sized channel entries indefinitely.
|
|
// At one sixty-fourth of the scanner's maximum completed-snapshot line, any
|
|
// reconcilable item occupies at most 65 of the 256 message slots (the leading
|
|
// delta plus 64 aggregates), reserving the rest for status and tool events.
|
|
// Since this threshold is 512 KiB, a normal message will not reach it: only
|
|
// its leading delta is usually handed off during generation, while the
|
|
// remaining deltas stay pending until item/completed reconciles the item.
|
|
// A single provider event may be larger, in which case that event is flushed
|
|
// as one chunk rather than split at an arbitrary byte boundary.
|
|
codexAgentMessageAggregateBytes = agentStreamMaxLineBytes / 64
|
|
// codexGracefulShutdownTimeout bounds how long the lifecycle goroutine
|
|
// waits for codex to exit on its own after stdin is closed, before forcing
|
|
// a context-cancel kill. A clean exit lets codex run its shutdown path and
|
|
// flush buffered telemetry — OTEL batch exporters only force-flush on
|
|
// graceful shutdown, so killing it immediately (the prior behavior) drops
|
|
// the task's spans/metrics/logs.
|
|
codexGracefulShutdownTimeout = 10 * time.Second
|
|
)
|
|
|
|
// codexGracefulShutdownTimeoutNanos optionally overrides
|
|
// codexGracefulShutdownTimeout for tests, in nanoseconds. Zero or negative
|
|
// values keep the production default. Tests for the cleanup-on-scanner-
|
|
// overflow path (#4520) use it to shrink the grace window from 10 s to a
|
|
// few hundred ms so the regression runs in a normal `go test` budget
|
|
// instead of burning two full grace windows per cleanup phase. Mirrors
|
|
// the opencodeTerminateGraceNanos hook.
|
|
var codexGracefulShutdownTimeoutNanos atomic.Int64
|
|
var codexProcessWaitDelayNanos atomic.Int64
|
|
var activeCodexLaunches atomic.Int64
|
|
var maxActiveCodexLaunchesObserved atomic.Int64
|
|
var codexCleanupConfirmationOverride atomic.Int32
|
|
|
|
// codexCatalogRetryBackoff is the floor of the delay before retrying a model
|
|
// catalog refresh failure. Jitter below twice the floor is added on top, in
|
|
// steps of floor/500 (1ms at the production value). Package tests shorten it;
|
|
// production never reassigns it.
|
|
var codexCatalogRetryBackoff = 500 * time.Millisecond
|
|
|
|
func sanitizeCodexDiagnostic(value string) string {
|
|
return sanitizeAgentDiagnostic(value)
|
|
}
|
|
|
|
func codexProcessExitStatus(state *os.ProcessState) any {
|
|
if state == nil {
|
|
return nil
|
|
}
|
|
return state.String()
|
|
}
|
|
|
|
func codexGracefulShutdown() time.Duration {
|
|
if n := codexGracefulShutdownTimeoutNanos.Load(); n > 0 {
|
|
return time.Duration(n)
|
|
}
|
|
return codexGracefulShutdownTimeout
|
|
}
|
|
|
|
func codexProcessWaitDelay() time.Duration {
|
|
if n := codexProcessWaitDelayNanos.Load(); n > 0 {
|
|
return time.Duration(n)
|
|
}
|
|
return 10 * time.Second
|
|
}
|
|
|
|
type codexStderrClassification struct {
|
|
// modelRefreshFailure is the broad catalog failure bucket. It includes the
|
|
// narrower modelRefreshTimeout bucket, so the two counts are not additive.
|
|
modelRefreshFailure int
|
|
modelRefreshTimeout int
|
|
mcpInitTransport int
|
|
bareTimeout int
|
|
}
|
|
|
|
// classifyCodexStartupStderr emits bounded counters only. It deliberately does
|
|
// not make retry decisions: stderr is sibling diagnostic evidence, not proof
|
|
// that thread/start did or did not create a provider-side thread.
|
|
func classifyCodexStartupStderr(stderr string, timedOut bool) codexStderrClassification {
|
|
lower := strings.ToLower(sanitizeCodexDiagnostic(stderr))
|
|
classification := codexStderrClassification{
|
|
modelRefreshFailure: strings.Count(lower, codexModelCatalogRefreshFailureSignal),
|
|
modelRefreshTimeout: strings.Count(lower, codexModelCatalogRefreshTimeoutSignal),
|
|
}
|
|
for _, line := range strings.Split(lower, "\n") {
|
|
if strings.Contains(line, "mcp") && strings.Contains(line, "transport") &&
|
|
(strings.Contains(line, "error") || strings.Contains(line, "failed") || strings.Contains(line, "closed")) {
|
|
classification.mcpInitTransport++
|
|
}
|
|
}
|
|
// Any catalog refresh failure now owns the timeout classification, not only
|
|
// the narrower child-process timeout signal. This intentionally makes the
|
|
// existing bare-timeout bucket stricter across daemon versions.
|
|
if timedOut && classification.modelRefreshFailure == 0 && classification.mcpInitTransport == 0 {
|
|
classification.bareTimeout = 1
|
|
}
|
|
return classification
|
|
}
|
|
|
|
// CodexSemanticInactivityMarker prefixes timeout errors emitted when Codex
|
|
// stops making semantic progress while the process is still alive.
|
|
const CodexSemanticInactivityMarker = "codex semantic inactivity timeout"
|
|
|
|
// CodexFirstTurnNoProgressMarker identifies the app-server failure mode where
|
|
// Codex accepts a turn and then never emits any item, completion, or error.
|
|
const CodexFirstTurnNoProgressMarker = "codex app-server no progress timeout"
|
|
|
|
// CodexHandshakeTimeoutMarker identifies a Codex app-server startup RPC that
|
|
// did not answer within the bounded handshake window.
|
|
const CodexHandshakeTimeoutMarker = "codex app-server handshake timeout"
|
|
|
|
// CodexGuardianRateLimitMarker identifies an in-turn approval retry loop that
|
|
// the adapter stopped after repeated reviewer failures.
|
|
const CodexGuardianRateLimitMarker = "codex Guardian approval rate-limit circuit open"
|
|
|
|
const codexGuardianRateLimitThreshold = 2
|
|
const codexGuardianReviewDedupWindow = 32
|
|
const codexGuardianRateLimitError = CodexGuardianRateLimitMarker + ": two consecutive Guardian approval reviews failed with 429 Too Many Requests; check completed work before continuing after the provider limit clears"
|
|
|
|
type codexGuardianRateLimitCircuit struct {
|
|
consecutive int
|
|
recentReviewIDs [codexGuardianReviewDedupWindow]string
|
|
nextReviewIDSlot int
|
|
tripped bool
|
|
}
|
|
|
|
// observeReview consumes the Guardian-specific app-server notification. A
|
|
// synthesized commandExecution item may have no output when approval fails,
|
|
// so inspecting tool output would miss the failure we need to stop.
|
|
func (c *codexGuardianRateLimitCircuit) observeReview(params map[string]any) bool {
|
|
reviewID, _ := params["reviewId"].(string)
|
|
review, _ := params["review"].(map[string]any)
|
|
if reviewID == "" || review == nil {
|
|
return false
|
|
}
|
|
status, _ := review["status"].(string)
|
|
if status == "" || status == "inProgress" {
|
|
return false
|
|
}
|
|
for _, recentID := range c.recentReviewIDs {
|
|
if reviewID == recentID {
|
|
return false
|
|
}
|
|
}
|
|
c.recentReviewIDs[c.nextReviewIDSlot] = reviewID
|
|
c.nextReviewIDSlot = (c.nextReviewIDSlot + 1) % len(c.recentReviewIDs)
|
|
rationale, _ := review["rationale"].(string)
|
|
lower := strings.ToLower(rationale)
|
|
// A policy denial carries a risk level; a reviewer failure does not. Keep
|
|
// the 429 signal as a narrow gate, without depending on Codex's English
|
|
// failure prefix (the review notification schema is unstable).
|
|
if status != "denied" || review["riskLevel"] != nil ||
|
|
(!strings.Contains(lower, "429") && !strings.Contains(lower, "too many requests")) {
|
|
c.consecutive = 0
|
|
return false
|
|
}
|
|
c.consecutive++
|
|
if c.consecutive < codexGuardianRateLimitThreshold || c.tripped {
|
|
return false
|
|
}
|
|
c.tripped = true
|
|
return true
|
|
}
|
|
|
|
// codexResumeMarker and codexLineOverflowMarker are the two halves of the
|
|
// error text a resume-overflow produces: the method that failed (written by
|
|
// startOrResumeThread) and bufio's own ErrTooLong wording, which reaches the
|
|
// string through the reader goroutine.
|
|
const (
|
|
codexResumeMarker = "thread/resume failed"
|
|
codexLineOverflowMarker = "token too long"
|
|
)
|
|
|
|
// CodexResumeOverflowError reports whether an agent error string is the
|
|
// resume-overflow failure — the thread/resume response did not fit in our
|
|
// stdout line buffer, so the thread cannot be handed to us at all.
|
|
//
|
|
// This lives next to the code that writes both halves of the text so the two
|
|
// cannot drift apart, and it is exported because the daemon has only the error
|
|
// STRING at report time: by then the typed bufio.ErrTooLong that
|
|
// isCodexResumeOverflow matches in-process is long gone. Both markers are
|
|
// required — "thread/resume failed" alone covers ordinary rejections that a
|
|
// plain retry handles, and "token too long" alone would also match an overflow
|
|
// on some other RPC, where dropping the session pointer cures nothing.
|
|
//
|
|
// The session behind such a failure is unusable for resume until it shrinks,
|
|
// which it never does — codex rollouts are append-only. Callers use this to
|
|
// stop handing the same oversized thread to the next task (MUL-5722).
|
|
func CodexResumeOverflowError(errText string) bool {
|
|
if errText == "" {
|
|
return false
|
|
}
|
|
lower := strings.ToLower(errText)
|
|
return strings.Contains(lower, codexResumeMarker) && strings.Contains(lower, codexLineOverflowMarker)
|
|
}
|
|
|
|
// codexRetiredCompactPath is the request path only the legacy remote-compaction
|
|
// route ever calls. The v2 route compacts on the ordinary /responses stream
|
|
// instead, so this segment appears only when the legacy route was selected.
|
|
const codexRetiredCompactPath = "responses/compact"
|
|
|
|
// codexRemoteCompactMarker opens the error Codex writes when a remote
|
|
// compaction task fails. Both routes share it — compact_remote_v2.rs labels its
|
|
// own failures with the same phrase — so it establishes only that compaction is
|
|
// what failed, never which route ran.
|
|
const codexRemoteCompactMarker = "remote compact task"
|
|
|
|
// CodexRetiredCompactionError identifies failures on the retired
|
|
// /responses/compact route so callers can suggest correcting remote_compaction_v2.
|
|
// Require the path itself: the current route can emit the same compaction prefix
|
|
// and HTTP status, so those alone would produce the opposite advice.
|
|
//
|
|
// This is diagnostic text, not a resume-safety verdict or a config override.
|
|
// Correcting the setting can recover the existing conversation.
|
|
func CodexRetiredCompactionError(errText string) bool {
|
|
if errText == "" {
|
|
return false
|
|
}
|
|
lower := strings.ToLower(errText)
|
|
return strings.Contains(lower, codexRemoteCompactMarker) &&
|
|
strings.Contains(lower, codexRetiredCompactPath)
|
|
}
|
|
|
|
// codexModelCatalogRefreshFailureSignal matches the Codex models-manager error
|
|
// emitted when the model catalog could not be refreshed. Codex reports several
|
|
// distinct causes under this prefix ("timeout waiting for child process to
|
|
// exit", "stream disconnected before completion", ...), so match the shared
|
|
// prefix rather than one variant: they are all the same startup-blocking
|
|
// failure from the daemon's point of view.
|
|
const codexModelCatalogRefreshFailureSignal = "failed to refresh available models"
|
|
const codexModelCatalogRefreshTimeoutSignal = "failed to refresh available models: timeout waiting for child process to exit"
|
|
|
|
var errCodexProcessExited = errors.New("codex process exited")
|
|
|
|
type codexTimeoutKind int
|
|
|
|
const (
|
|
codexTimeoutNone codexTimeoutKind = iota
|
|
codexTimeoutSemanticInactivity
|
|
codexTimeoutFirstTurnNoProgress
|
|
)
|
|
|
|
type codexTimeoutDiagnostic struct {
|
|
Kind codexTimeoutKind
|
|
Timeout time.Duration
|
|
LastActivity string
|
|
ThreadID string
|
|
TurnID string
|
|
Model string
|
|
CodexVersion string
|
|
}
|
|
|
|
// codexFirstItemWaitObservation records the interval from turn/started to the
|
|
// first semantic progress or terminal outcome. start is called by the stdout
|
|
// reader before it publishes status:running, while finish is called by the
|
|
// lifecycle goroutine; the mutex keeps the cross-goroutine timestamp precise
|
|
// without changing the watchdog's existing channel-driven semantics.
|
|
type codexFirstItemWaitObservation struct {
|
|
mu sync.Mutex
|
|
startedAt time.Time
|
|
finishedAt time.Time
|
|
turnID string
|
|
outcome string
|
|
stderr codexStderrClassification
|
|
}
|
|
|
|
func (o *codexFirstItemWaitObservation) start(now time.Time, turnID string) {
|
|
o.mu.Lock()
|
|
defer o.mu.Unlock()
|
|
if o.startedAt.IsZero() {
|
|
o.startedAt = now
|
|
o.turnID = turnID
|
|
}
|
|
}
|
|
|
|
func (o *codexFirstItemWaitObservation) finish(now time.Time, outcome string, stderr codexStderrClassification) {
|
|
o.mu.Lock()
|
|
defer o.mu.Unlock()
|
|
if o.startedAt.IsZero() || o.outcome != "" {
|
|
return
|
|
}
|
|
o.finishedAt = now
|
|
o.outcome = outcome
|
|
o.stderr = stderr
|
|
}
|
|
|
|
func (o *codexFirstItemWaitObservation) snapshot() (time.Duration, string, string, codexStderrClassification, bool) {
|
|
o.mu.Lock()
|
|
defer o.mu.Unlock()
|
|
if o.startedAt.IsZero() || o.finishedAt.IsZero() || o.outcome == "" {
|
|
return 0, "", "", codexStderrClassification{}, false
|
|
}
|
|
return o.finishedAt.Sub(o.startedAt), o.turnID, o.outcome, o.stderr, true
|
|
}
|
|
|
|
// codexBackend implements Backend by spawning `codex app-server --listen stdio://`
|
|
// and communicating via JSON-RPC 2.0 over stdin/stdout.
|
|
type codexBackend struct {
|
|
cfg Config
|
|
}
|
|
|
|
func buildCodexArgs(opts ExecOptions, logger *slog.Logger) []string {
|
|
args := []string{"app-server", "--listen", "stdio://"}
|
|
launchArgs := NormalizeCodexLaunchArgs(opts.ExtraArgs, opts.CustomArgs, opts.McpConfig, logger)
|
|
if opts.ServiceTier == codexFastServiceTier {
|
|
launchArgs = enforceCodexFastMode(launchArgs, logger)
|
|
}
|
|
return append(args, launchArgs...)
|
|
}
|
|
|
|
// NormalizeCodexLaunchArgs returns the user-supplied Codex args (extra then
|
|
// custom) after lower-priority normalization: shell quoting stripped,
|
|
// protocol-critical flags removed, and — when a managed
|
|
// mcp_config owns the mcp_servers namespace — stray `-c mcp_servers.*`
|
|
// overrides dropped. buildCodexArgs prepends the fixed
|
|
// `app-server --listen stdio://` transport flags and may append an
|
|
// agent-owned runtime override (currently `--enable fast_mode`) after this
|
|
// result.
|
|
//
|
|
// It is exported so the daemon can reconstruct the *effective* launch args when
|
|
// deciding the Windows sandbox mode. A `-c windows.sandbox=…` opt-in may arrive
|
|
// shell-quoted (users commonly type custom_args with shell syntax, e.g.
|
|
// `'-c' 'windows.sandbox=unelevated'`), and only after this same normalization
|
|
// does it match the `-c windows.sandbox=…` shape the sandbox detector looks
|
|
// for. Reconstructing the args any other way lets the two drift, silently
|
|
// downgrading a user's isolation opt-in (MUL-4957).
|
|
func NormalizeCodexLaunchArgs(extraArgs, customArgs []string, mcpConfig json.RawMessage, logger *slog.Logger) []string {
|
|
extra := filterCustomArgs(extraArgs, codexBlockedArgs, logger)
|
|
custom := filterCustomArgs(customArgs, codexBlockedArgs, logger)
|
|
// Only claim ownership of the `mcp_servers` namespace when the agent
|
|
// actually has a managed mcp_config in the MCP Tab. Otherwise existing
|
|
// users who configure MCP via `custom_args: ["-c", "mcp_servers.…"]`
|
|
// would silently lose those entries. With managed mcp_config present,
|
|
// daemon-written `$CODEX_HOME/config.toml` is the authoritative source and
|
|
// stray `-c mcp_servers.*` overrides are dropped to keep last-wins from
|
|
// re-shadowing it.
|
|
if hasManagedCodexMcpConfig(mcpConfig) {
|
|
extra = filterCodexCustomConfigOverrides(extra, logger)
|
|
custom = filterCodexCustomConfigOverrides(custom, logger)
|
|
}
|
|
out := make([]string, 0, len(extra)+len(custom))
|
|
out = append(out, extra...)
|
|
out = append(out, custom...)
|
|
return out
|
|
}
|
|
|
|
// enforceCodexFastMode makes the explicit agent service-tier selection
|
|
// authoritative over daemon ExtraArgs, per-agent CustomArgs, and inherited
|
|
// config.toml. Codex's `--disable fast_mode` wins over `--enable fast_mode`
|
|
// regardless of argv order, so conflicting lower-priority disable flags must
|
|
// be removed before the managed enable is appended. Conflicting `-c` /
|
|
// `--config features.fast_mode=...` overrides are removed for the same
|
|
// precedence reason, even though current Codex versions let `--enable` win:
|
|
// the final argv should encode one unambiguous owner for this setting.
|
|
//
|
|
// Other disabled features are preserved, and this helper is used only for the
|
|
// catalog-owned `priority` tier. Future service tiers must not accidentally
|
|
// inherit Fast mode semantics.
|
|
func enforceCodexFastMode(args []string, logger *slog.Logger) []string {
|
|
return append(stripCodexFastModeConflicts(args, logger), "--enable", codexFastModeFeature)
|
|
}
|
|
|
|
// stripCodexFastModeConflicts removes the lower-priority overrides that would
|
|
// defeat an explicit priority tier, without appending the managed enable.
|
|
//
|
|
// Split out of enforceCodexFastMode because the enable belongs exactly once, on
|
|
// the final managed args, while the removal has to reach every argv region the
|
|
// user can write into — including a custom runtime profile's launch prefix.
|
|
// Prefix-first does not protect this setting: `--disable fast_mode` beats
|
|
// `--enable fast_mode` regardless of argv order, and a `-c
|
|
// features.fast_mode=false` beats config.toml from any position. Leaving the
|
|
// prefix unfiltered would let a profile override the tier the agent explicitly
|
|
// selected, inverting the precedence this package promises (GH #7046).
|
|
func stripCodexFastModeConflicts(args []string, logger *slog.Logger) []string {
|
|
args = filterCodexConfigOverrides(
|
|
args,
|
|
codexManagedFastModeConfigKeyRe,
|
|
"features.fast_mode",
|
|
logger,
|
|
)
|
|
filtered := make([]string, 0, len(args)+2)
|
|
for i := 0; i < len(args); i++ {
|
|
arg := args[i]
|
|
flag := arg
|
|
value := ""
|
|
hasInlineValue := false
|
|
if idx := strings.Index(arg, "="); idx > 0 {
|
|
flag = arg[:idx]
|
|
value = arg[idx+1:]
|
|
hasInlineValue = true
|
|
}
|
|
if flag == "--disable" {
|
|
if !hasInlineValue && i+1 < len(args) {
|
|
value = args[i+1]
|
|
}
|
|
if value == codexFastModeFeature {
|
|
if logger != nil {
|
|
logger.Warn("codex: ignored lower-priority feature disable",
|
|
"feature", codexFastModeFeature)
|
|
}
|
|
if !hasInlineValue {
|
|
i++
|
|
}
|
|
continue
|
|
}
|
|
}
|
|
filtered = append(filtered, arg)
|
|
}
|
|
return filtered
|
|
}
|
|
|
|
// hasManagedCodexMcpConfig reports whether the agent's mcp_config field is
|
|
// "present" in the API three-state sense: a non-null JSON value. Both
|
|
// `{}` and `{"mcpServers":{}}` count as present (the admin saved an empty
|
|
// managed set — strict mode, no global fallback); only SQL NULL or the
|
|
// literal JSON `null` count as absent (CLI default).
|
|
func hasManagedCodexMcpConfig(raw json.RawMessage) bool {
|
|
return hasManagedMcpConfig(raw)
|
|
}
|
|
|
|
// codexManagedMcpConfigKeyRe matches the daemon-managed config namespace
|
|
// (`mcp_servers.…`) when it appears as the value of a Codex `-c` /
|
|
// `--config` flag. Used by filterCodexCustomConfigOverrides to drop user
|
|
// overrides that would otherwise shadow what the MCP Tab writes into
|
|
// `$CODEX_HOME/config.toml`.
|
|
var codexManagedMcpConfigKeyRe = regexp.MustCompile(`^\s*mcp_servers(?:\s*\.|\s*=|\s*$)`)
|
|
|
|
var codexManagedFastModeConfigKeyRe = regexp.MustCompile(
|
|
`^\s*features\s*\.\s*fast_mode\s*(?:=|$)`)
|
|
|
|
// A daemon-managed shell_environment_policy must also win over profile and
|
|
// custom-arg overrides. Match root and profile policy keys without catching an
|
|
// unrelated table field that happens to use the same final key name.
|
|
const (
|
|
codexShellEnvPolicyKeyPattern = `(?:shell_environment_policy|"shell_environment_policy"|'shell_environment_policy')`
|
|
codexProfileNameKeyPattern = `(?:[A-Za-z0-9_-]+|"[^"]+"|'[^']+')`
|
|
)
|
|
|
|
var codexManagedShellEnvConfigKeyRe = regexp.MustCompile(
|
|
`^\s*(?:` + codexShellEnvPolicyKeyPattern + `|profiles\s*\.\s*` + codexProfileNameKeyPattern + `\s*\.\s*` + codexShellEnvPolicyKeyPattern + `)\s*(?:\.|=|$)`)
|
|
|
|
// filterCodexCustomConfigOverrides drops `-c mcp_servers.…=` and
|
|
// `--config mcp_servers.…=` entries from custom args. Codex's `-c` is
|
|
// last-wins (verified against codex-cli 0.132.0), so without this filter a
|
|
// user-written `-c mcp_servers.fetch=…` in custom_args would silently
|
|
// override whatever the MCP Tab saved into the per-task config.toml. We
|
|
// own the `mcp_servers` namespace via the managed block, so user attempts
|
|
// to write into it are dropped with a warning rather than allowed to win.
|
|
// Other `-c`/`--config` keys (e.g. `-c model="o3"`) pass through unchanged.
|
|
func filterCodexCustomConfigOverrides(args []string, logger *slog.Logger) []string {
|
|
return filterCodexConfigOverrides(args, codexManagedMcpConfigKeyRe, "mcp_servers", logger)
|
|
}
|
|
|
|
func filterCodexShellEnvConfigOverrides(args []string, logger *slog.Logger) []string {
|
|
return filterCodexConfigOverrides(args, codexManagedShellEnvConfigKeyRe, shellEnvironmentPolicyConfigNamespace, logger)
|
|
}
|
|
|
|
const shellEnvironmentPolicyConfigNamespace = "shell_environment_policy"
|
|
|
|
func filterCodexConfigOverrides(args []string, managedKeyRe *regexp.Regexp, namespace string, logger *slog.Logger) []string {
|
|
if len(args) == 0 {
|
|
return args
|
|
}
|
|
filtered := make([]string, 0, len(args))
|
|
for i := 0; i < len(args); i++ {
|
|
arg := args[i]
|
|
flag := arg
|
|
inlineValue := ""
|
|
hasInlineValue := false
|
|
if idx := strings.Index(arg, "="); idx > 0 {
|
|
flag = arg[:idx]
|
|
inlineValue = arg[idx+1:]
|
|
hasInlineValue = true
|
|
}
|
|
if flag == "-c" || flag == "--config" {
|
|
value := inlineValue
|
|
if !hasInlineValue && i+1 < len(args) {
|
|
value = args[i+1]
|
|
}
|
|
if managedKeyRe.MatchString(value) {
|
|
if logger != nil {
|
|
// Log the key only, never the value. Managed config values
|
|
// may contain secrets and must stay out of logs/argv.
|
|
key := value
|
|
if eqIdx := strings.Index(value, "="); eqIdx >= 0 {
|
|
key = value[:eqIdx]
|
|
}
|
|
logger.Warn("custom_args: blocked managed Codex config override",
|
|
"namespace", namespace, "flag", flag, "key", strings.TrimSpace(key))
|
|
}
|
|
if !hasInlineValue && i+1 < len(args) {
|
|
i++ // skip the value arg
|
|
}
|
|
continue
|
|
}
|
|
}
|
|
filtered = append(filtered, arg)
|
|
}
|
|
return filtered
|
|
}
|
|
|
|
// Markers delimiting the daemon-managed `[mcp_servers.*]` block in
|
|
// `$CODEX_HOME/config.toml`. Match the existing sandbox / multi-agent /
|
|
// memory marker pattern so ops can grep all managed blocks consistently.
|
|
const (
|
|
multicaCodexMcpBeginMarker = "# BEGIN multica-managed mcp_servers (do not edit; regenerated by daemon)"
|
|
multicaCodexMcpEndMarker = "# END multica-managed mcp_servers"
|
|
)
|
|
|
|
var codexMcpBlockRe = regexp.MustCompile(
|
|
`(?ms)^` + regexp.QuoteMeta(multicaCodexMcpBeginMarker) +
|
|
`.*?^` + regexp.QuoteMeta(multicaCodexMcpEndMarker) + `\n*`)
|
|
|
|
// userCodexMcpServersTableHeaderRe matches `[mcp_servers.<name>]` (and its
|
|
// quoted-key form `[mcp_servers."<name>"]`) at the start of a line. Used
|
|
// to strip user-provided mcp_servers tables from the per-task config when
|
|
// the agent has its own mcp_config — mirrors Claude's `--strict-mcp-config`
|
|
// model where the daemon's set is authoritative.
|
|
var userCodexMcpServersTableHeaderRe = regexp.MustCompile(
|
|
`^\s*\[\s*mcp_servers\s*\.\s*(?:"[^"]*"|[^\]\s]+)\s*\]\s*(?:#.*)?$`)
|
|
|
|
// ensureCodexMcpConfig writes (or clears) the daemon-managed
|
|
// `[mcp_servers.*]` block in `$CODEX_HOME/config.toml`. The block is the
|
|
// authoritative source of MCP servers for this run: with mcp_config set
|
|
// in the agent UI the daemon also strips any inherited
|
|
// `[mcp_servers.*]` tables from the per-task config so the user's global
|
|
// `~/.codex/config.toml` doesn't shadow or collide with the managed set.
|
|
//
|
|
// The file mode is 0o600 because `mcp_servers.<id>.env` values may carry
|
|
// secrets (API keys, bearer tokens); the per-task home is owned by the
|
|
// daemon's user, so 0o600 keeps secrets out of any world-readable copy
|
|
// while still letting the codex child read them.
|
|
//
|
|
// A malformed mcp_config is returned as an error and the caller decides
|
|
// whether to surface or warn — same fail-soft contract the prior argv
|
|
// path had.
|
|
func ensureCodexMcpConfig(configPath string, mcpConfig json.RawMessage, logger *slog.Logger) error {
|
|
data, err := os.ReadFile(configPath)
|
|
if err != nil && !os.IsNotExist(err) {
|
|
return fmt.Errorf("read config.toml: %w", err)
|
|
}
|
|
existing := string(data)
|
|
|
|
// Always strip a prior managed block so reruns and clear-config flows
|
|
// converge on a clean state.
|
|
stripped := codexMcpBlockRe.ReplaceAllString(existing, "")
|
|
|
|
managed := hasManagedCodexMcpConfig(mcpConfig)
|
|
block, _, renderErr := renderCodexMcpServersBlock(mcpConfig)
|
|
if renderErr != nil {
|
|
return renderErr
|
|
}
|
|
|
|
var updated string
|
|
if managed {
|
|
// Agent has a managed MCP set (possibly empty — `{}` /
|
|
// `{"mcpServers":{}}` count as "saved an empty set" in the API's
|
|
// three-state semantics, distinct from nil/null which means
|
|
// "fall back to CLI default"). Strip any user-defined
|
|
// `[mcp_servers.*]` tables inherited from `~/.codex/config.toml`
|
|
// so the managed set is strict — mirrors Claude's
|
|
// `--strict-mcp-config`. Two reasons we cannot mix:
|
|
// 1. TOML rejects redefining the same table; a user table
|
|
// named `[mcp_servers.fetch]` would crash codex if the
|
|
// agent also defined `fetch`.
|
|
// 2. An admin saving an explicit list in the MCP Tab would
|
|
// otherwise see user-global servers silently joined in,
|
|
// which contradicts the UI affordance.
|
|
stripped = stripCodexUserMcpServerTables(stripped)
|
|
stripped = strings.TrimRight(stripped, "\n")
|
|
// When the managed set is empty we still write the marker
|
|
// block (with no tables between). This pins "managed but
|
|
// empty" on disk so the next run can find and strip the
|
|
// markers, and so the file's intent is grep-able by ops.
|
|
if block == "" {
|
|
block = multicaCodexMcpBeginMarker + "\n" + multicaCodexMcpEndMarker + "\n"
|
|
}
|
|
if stripped == "" {
|
|
updated = block
|
|
} else {
|
|
updated = stripped + "\n\n" + block
|
|
}
|
|
} else {
|
|
// No managed config: just remove any prior managed block and
|
|
// leave inherited user tables alone (CLI default fallback).
|
|
updated = stripped
|
|
}
|
|
|
|
if updated == existing {
|
|
return nil
|
|
}
|
|
if err := os.WriteFile(configPath, []byte(updated), 0o600); err != nil {
|
|
return fmt.Errorf("write config.toml: %w", err)
|
|
}
|
|
// os.WriteFile applies the mode only when creating a new file; if the
|
|
// per-task config.toml was already on disk at 0o644 (the default mode
|
|
// used by execenv.copyFile when seeding from ~/.codex/config.toml),
|
|
// the secret-bearing values we just wrote would inherit that wider
|
|
// mode. Chmod unconditionally to keep the secret in the daemon
|
|
// owner's lane regardless of the prior mode.
|
|
if err := os.Chmod(configPath, 0o600); err != nil {
|
|
return fmt.Errorf("chmod config.toml to 0600: %w", err)
|
|
}
|
|
if logger != nil {
|
|
logger.Debug("codex: wrote managed mcp_servers block to config.toml",
|
|
"config_path", configPath, "managed", managed)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// renderCodexMcpServersBlock renders the agent's mcp_config JSON
|
|
// (Claude-style `{"mcpServers": {...}}`) as a TOML block of
|
|
// `[mcp_servers.<name>]` tables wrapped in BEGIN/END markers. Returns
|
|
// (block, hasServers, err); hasServers=false means the input had no
|
|
// servers to render (empty/null mcp_config) and the caller should only
|
|
// strip the prior managed block.
|
|
//
|
|
// Stdio server keys (`args`, `env`, `command`) pass through verbatim —
|
|
// Codex's config schema happens to use the same names today. Remote HTTP
|
|
// servers use Codex-specific keys, so they are normalised here rather than
|
|
// leaking provider details into the UI/dispatch layer.
|
|
func renderCodexMcpServersBlock(raw json.RawMessage) (string, bool, error) {
|
|
if len(raw) == 0 {
|
|
return "", false, nil
|
|
}
|
|
var parsed struct {
|
|
McpServers map[string]json.RawMessage `json:"mcpServers"`
|
|
}
|
|
if err := json.Unmarshal(raw, &parsed); err != nil {
|
|
return "", false, fmt.Errorf("parse mcp_config json: %w", err)
|
|
}
|
|
if len(parsed.McpServers) == 0 {
|
|
return "", false, nil
|
|
}
|
|
|
|
names := make([]string, 0, len(parsed.McpServers))
|
|
for name := range parsed.McpServers {
|
|
names = append(names, name)
|
|
}
|
|
sort.Strings(names)
|
|
|
|
var sb strings.Builder
|
|
sb.WriteString(multicaCodexMcpBeginMarker)
|
|
sb.WriteString("\n")
|
|
for i, name := range names {
|
|
if !isCodexBareTomlKey(name) {
|
|
return "", false, fmt.Errorf("mcp server name %q must be ASCII alphanumeric / _ / - to fit Codex's bare-key requirement", name)
|
|
}
|
|
var serverVal map[string]any
|
|
if err := json.Unmarshal(parsed.McpServers[name], &serverVal); err != nil {
|
|
return "", false, fmt.Errorf("mcp_servers.%s: %w", name, err)
|
|
}
|
|
if serverVal == nil {
|
|
return "", false, fmt.Errorf("mcp_servers.%s must be a JSON object", name)
|
|
}
|
|
serverVal = normalizeCodexMcpServerConfig(serverVal)
|
|
if i > 0 {
|
|
sb.WriteString("\n")
|
|
}
|
|
sb.WriteString("[mcp_servers.")
|
|
sb.WriteString(name)
|
|
sb.WriteString("]\n")
|
|
keys := make([]string, 0, len(serverVal))
|
|
for k := range serverVal {
|
|
keys = append(keys, k)
|
|
}
|
|
sort.Strings(keys)
|
|
for _, k := range keys {
|
|
tomlValue, err := jsonValueToCodexTOMLInline(serverVal[k])
|
|
if err != nil {
|
|
return "", false, fmt.Errorf("mcp_servers.%s.%s: %w", name, k, err)
|
|
}
|
|
sb.WriteString(codexTOMLKey(k))
|
|
sb.WriteString(" = ")
|
|
sb.WriteString(tomlValue)
|
|
sb.WriteString("\n")
|
|
}
|
|
}
|
|
sb.WriteString(multicaCodexMcpEndMarker)
|
|
sb.WriteString("\n")
|
|
return sb.String(), true, nil
|
|
}
|
|
|
|
func normalizeCodexMcpServerConfig(server map[string]any) map[string]any {
|
|
if !isCodexRemoteMcpServer(server) {
|
|
normalized := make(map[string]any, len(server))
|
|
for k, v := range server {
|
|
if isMulticaMcpSelectorKey(k) {
|
|
continue
|
|
}
|
|
normalized[k] = v
|
|
}
|
|
return normalized
|
|
}
|
|
|
|
normalized := make(map[string]any, len(server)+1)
|
|
for k, v := range server {
|
|
switch {
|
|
case isMulticaMcpSelectorKey(k):
|
|
continue
|
|
case k == "type":
|
|
continue
|
|
case k == "headers":
|
|
if _, ok := server["http_headers"]; !ok {
|
|
normalized["http_headers"] = v
|
|
}
|
|
default:
|
|
normalized[k] = v
|
|
}
|
|
}
|
|
normalized["experimental_use_rmcp_client"] = true
|
|
return normalized
|
|
}
|
|
|
|
func isMulticaMcpSelectorKey(k string) bool {
|
|
switch k {
|
|
case "tools", "prompts", "resources":
|
|
return true
|
|
default:
|
|
return false
|
|
}
|
|
}
|
|
|
|
func isCodexRemoteMcpServer(server map[string]any) bool {
|
|
if typ, ok := server["type"].(string); ok && strings.EqualFold(typ, "http") {
|
|
return true
|
|
}
|
|
_, hasURL := server["url"]
|
|
_, hasCommand := server["command"]
|
|
return hasURL && !hasCommand
|
|
}
|
|
|
|
// stripCodexUserMcpServerTables removes every `[mcp_servers.*]` table
|
|
// (header + body lines until the next top-level table header or EOF) from
|
|
// a TOML config string. Sub-tables like `[mcp_servers.fetch.env]` count
|
|
// as part of the parent table and are dropped along with it.
|
|
func stripCodexUserMcpServerTables(content string) string {
|
|
lines := strings.Split(content, "\n")
|
|
out := make([]string, 0, len(lines))
|
|
skipping := false
|
|
for _, line := range lines {
|
|
if userCodexMcpServersTableHeaderRe.MatchString(line) {
|
|
skipping = true
|
|
continue
|
|
}
|
|
if skipping {
|
|
trimmed := strings.TrimSpace(line)
|
|
if strings.HasPrefix(trimmed, "[") {
|
|
// Next table header. If it's still an `mcp_servers.*`
|
|
// table (including a sub-table), keep skipping; otherwise
|
|
// stop and emit this line.
|
|
if userCodexMcpServersTableHeaderRe.MatchString(line) ||
|
|
strings.HasPrefix(trimmed, "[mcp_servers.") ||
|
|
strings.HasPrefix(trimmed, "[ mcp_servers.") {
|
|
continue
|
|
}
|
|
skipping = false
|
|
out = append(out, line)
|
|
continue
|
|
}
|
|
continue
|
|
}
|
|
out = append(out, line)
|
|
}
|
|
return strings.Join(out, "\n")
|
|
}
|
|
|
|
// jsonValueToCodexTOMLInline serialises a JSON value as a TOML inline
|
|
// value. Only the subset Codex's `-c` accepts is supported: strings,
|
|
// numbers, booleans, arrays, and inline tables. JSON nulls are rejected
|
|
// because TOML has no null and silently dropping them would be confusing.
|
|
func jsonValueToCodexTOMLInline(v any) (string, error) {
|
|
switch x := v.(type) {
|
|
case nil:
|
|
return "", fmt.Errorf("null is not a valid TOML value")
|
|
case bool:
|
|
if x {
|
|
return "true", nil
|
|
}
|
|
return "false", nil
|
|
case float64:
|
|
if x == float64(int64(x)) {
|
|
return strconv.FormatInt(int64(x), 10), nil
|
|
}
|
|
return strconv.FormatFloat(x, 'f', -1, 64), nil
|
|
case string:
|
|
return codexTOMLBasicString(x), nil
|
|
case []any:
|
|
parts := make([]string, len(x))
|
|
for i, e := range x {
|
|
p, err := jsonValueToCodexTOMLInline(e)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
parts[i] = p
|
|
}
|
|
return "[" + strings.Join(parts, ", ") + "]", nil
|
|
case map[string]any:
|
|
keys := make([]string, 0, len(x))
|
|
for k := range x {
|
|
keys = append(keys, k)
|
|
}
|
|
sort.Strings(keys)
|
|
parts := make([]string, len(keys))
|
|
for i, k := range keys {
|
|
p, err := jsonValueToCodexTOMLInline(x[k])
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
parts[i] = codexTOMLKey(k) + " = " + p
|
|
}
|
|
return "{ " + strings.Join(parts, ", ") + " }", nil
|
|
default:
|
|
return "", fmt.Errorf("unsupported value type %T", v)
|
|
}
|
|
}
|
|
|
|
func codexTOMLBasicString(s string) string {
|
|
var sb strings.Builder
|
|
sb.Grow(len(s) + 2)
|
|
sb.WriteByte('"')
|
|
for _, r := range s {
|
|
switch r {
|
|
case '\\':
|
|
sb.WriteString(`\\`)
|
|
case '"':
|
|
sb.WriteString(`\"`)
|
|
case '\b':
|
|
sb.WriteString(`\b`)
|
|
case '\t':
|
|
sb.WriteString(`\t`)
|
|
case '\n':
|
|
sb.WriteString(`\n`)
|
|
case '\f':
|
|
sb.WriteString(`\f`)
|
|
case '\r':
|
|
sb.WriteString(`\r`)
|
|
default:
|
|
if r < 0x20 || r == 0x7f {
|
|
sb.WriteString(fmt.Sprintf(`\u%04x`, r))
|
|
} else {
|
|
sb.WriteRune(r)
|
|
}
|
|
}
|
|
}
|
|
sb.WriteByte('"')
|
|
return sb.String()
|
|
}
|
|
|
|
func codexTOMLKey(s string) string {
|
|
if isCodexBareTomlKey(s) {
|
|
return s
|
|
}
|
|
return codexTOMLBasicString(s)
|
|
}
|
|
|
|
func isCodexBareTomlKey(s string) bool {
|
|
if s == "" {
|
|
return false
|
|
}
|
|
for _, r := range s {
|
|
switch {
|
|
case r >= 'a' && r <= 'z':
|
|
case r >= 'A' && r <= 'Z':
|
|
case r >= '0' && r <= '9':
|
|
case r == '_' || r == '-':
|
|
default:
|
|
return false
|
|
}
|
|
}
|
|
return true
|
|
}
|
|
|
|
func (b *codexBackend) Execute(ctx context.Context, prompt string, opts ExecOptions) (*Session, error) {
|
|
firstSession, err := b.executeOnce(ctx, prompt, opts, 1)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
msgCh := make(chan Message, 256)
|
|
resCh := make(chan Result, 1)
|
|
var sessionMu sync.RWMutex
|
|
currentSession := firstSession
|
|
supplement := func(supplementCtx context.Context, instruction string) error {
|
|
sessionMu.RLock()
|
|
session := currentSession
|
|
sessionMu.RUnlock()
|
|
if session == nil || session.Supplement == nil {
|
|
return errors.New("codex turn does not accept additional messages")
|
|
}
|
|
return session.Supplement(supplementCtx, instruction)
|
|
}
|
|
supplementReady := func() bool {
|
|
sessionMu.RLock()
|
|
session := currentSession
|
|
sessionMu.RUnlock()
|
|
return session != nil && session.SupplementReady != nil && session.SupplementReady()
|
|
}
|
|
|
|
go func() {
|
|
defer func() {
|
|
sessionMu.Lock()
|
|
currentSession = nil
|
|
sessionMu.Unlock()
|
|
}()
|
|
defer close(msgCh)
|
|
defer close(resCh)
|
|
session := firstSession
|
|
attemptOpts := opts
|
|
for attempt := 1; attempt <= 2; attempt++ {
|
|
if attempt > 1 {
|
|
var err error
|
|
session, err = b.executeOnce(ctx, prompt, attemptOpts, attempt)
|
|
if err != nil {
|
|
resCh <- Result{Status: "failed", Error: err.Error()}
|
|
return
|
|
}
|
|
sessionMu.Lock()
|
|
currentSession = session
|
|
sessionMu.Unlock()
|
|
}
|
|
// Hold back the leading session-pin status messages until this
|
|
// attempt proves it made real progress. A retry never continues the
|
|
// discarded attempt's thread (initialize retries fail before any
|
|
// thread exists; catalog retries clear ResumeSessionID below), so
|
|
// forwarding its pin would leave the resume pointer aimed at a
|
|
// thread that never produced a turn (MUL-5110). The first non-pin
|
|
// message means the attempt is live: flush and stream from then on.
|
|
var heldPins []Message
|
|
holdingPins := true
|
|
flushHeldPins := func() {
|
|
for _, held := range heldPins {
|
|
msgCh <- held
|
|
}
|
|
heldPins = nil
|
|
holdingPins = false
|
|
}
|
|
for msg := range session.Messages {
|
|
if holdingPins && msg.Type == MessageStatus && msg.Status == "running" {
|
|
heldPins = append(heldPins, msg)
|
|
continue
|
|
}
|
|
if holdingPins {
|
|
flushHeldPins()
|
|
}
|
|
msgCh <- msg
|
|
}
|
|
result, ok := <-session.Result
|
|
if !ok {
|
|
flushHeldPins()
|
|
resCh <- Result{Status: "failed", Error: "codex attempt closed without result"}
|
|
return
|
|
}
|
|
retryReason := ""
|
|
switch {
|
|
case result.codexInitializeRetrySafe:
|
|
retryReason = "initialize"
|
|
case result.codexStartupRefreshRetrySafe:
|
|
retryReason = "model_catalog_refresh"
|
|
}
|
|
if retryReason == "" || attempt == 2 {
|
|
flushHeldPins()
|
|
resCh <- result
|
|
return
|
|
}
|
|
// The model catalog refresh reaches the network, so give the
|
|
// transient failure a little more room than the local initialize
|
|
// handshake gets. Both stay well inside the task timeout, and
|
|
// ctx.Done() below keeps the retry from extending it.
|
|
backoff := 75*time.Millisecond + time.Duration(time.Now().UnixNano()%50)*time.Millisecond
|
|
if retryReason == "model_catalog_refresh" {
|
|
backoff = codexCatalogRetryBackoff + time.Duration(time.Now().UnixNano()%1000)*(2*codexCatalogRetryBackoff/1000)
|
|
// The stalled attempt already reached turn/started, so the prior
|
|
// thread may hold the submitted input or an unfinished turn.
|
|
// Resuming it again could duplicate that input; start a fresh
|
|
// thread instead and keep ResumeExpected so codexTurnInput
|
|
// prepends the caller's continuity notice about the lost
|
|
// context.
|
|
if attemptOpts.ResumeSessionID != "" {
|
|
b.cfg.Logger.Warn("codex retry dropping resume pointer after model catalog refresh failure",
|
|
"prior_thread_id", attemptOpts.ResumeSessionID,
|
|
)
|
|
attemptOpts.ResumeSessionID = ""
|
|
attemptOpts.ResumeExpected = true
|
|
}
|
|
}
|
|
b.cfg.Logger.Warn("codex retry scheduled", "reason", retryReason, "attempt", attempt, "next_attempt", attempt+1, "backoff", backoff.String())
|
|
select {
|
|
case <-ctx.Done():
|
|
flushHeldPins()
|
|
resCh <- result
|
|
return
|
|
case <-time.After(backoff):
|
|
}
|
|
}
|
|
}()
|
|
|
|
return &Session{Supplement: supplement, SupplementReady: supplementReady, Messages: msgCh, Result: resCh}, nil
|
|
}
|
|
|
|
func (b *codexBackend) executeOnce(ctx context.Context, prompt string, opts ExecOptions, attempt int) (*Session, error) {
|
|
execPath := b.cfg.ExecutablePath
|
|
if execPath == "" {
|
|
execPath = "codex"
|
|
}
|
|
if _, err := exec.LookPath(execPath); err != nil {
|
|
return nil, fmt.Errorf("codex executable not found at %q: %w", execPath, err)
|
|
}
|
|
|
|
timeout := opts.Timeout
|
|
semanticInactivityTimeout := opts.SemanticInactivityTimeout
|
|
if semanticInactivityTimeout == 0 {
|
|
semanticInactivityTimeout = defaultCodexSemanticInactivityTimeout
|
|
}
|
|
handshakeTimeout, threadHandshakeTimeout := resolveCodexHandshakeTimeouts(opts)
|
|
runCtx, cancel := runContext(ctx, timeout)
|
|
|
|
// Materialise the agent's MCP config into the per-task
|
|
// `$CODEX_HOME/config.toml`. Argv would be the simpler path, but
|
|
// `mcp_servers.<id>.env` is allowed to carry secrets (Codex docs:
|
|
// https://developers.openai.com/codex/mcp#configure-with-configtoml)
|
|
// and our UI already treats mcp_config as a redacted-for-non-admins
|
|
// field. Process argv ends up in OS-level `ps` listings; daemon command
|
|
// logs redact values, but log redaction cannot protect the process list.
|
|
// Writing through config.toml at 0o600 keeps the secret values out of argv
|
|
// entirely.
|
|
codexHome := strings.TrimSpace(b.cfg.Env["CODEX_HOME"])
|
|
if codexHome != "" {
|
|
if err := ensureCodexMcpConfig(filepath.Join(codexHome, "config.toml"), opts.McpConfig, b.cfg.Logger); err != nil {
|
|
// Fail closed when we can't materialise the managed config.
|
|
// Warning-and-launching would silently fall back to the
|
|
// user's global `~/.codex/config.toml` MCP servers and
|
|
// look indistinguishable from "the saved config was
|
|
// applied", which is exactly the surprise the MCP Tab is
|
|
// supposed to remove.
|
|
cancel()
|
|
return nil, fmt.Errorf("apply codex mcp_config: %w", err)
|
|
}
|
|
} else if hasManagedCodexMcpConfig(opts.McpConfig) {
|
|
// Managed mcp_config saved but no CODEX_HOME to anchor it.
|
|
// Same reasoning as above: silently launching would inherit
|
|
// whatever MCP setup the host user has, which is the wrong
|
|
// shape of failure.
|
|
cancel()
|
|
return nil, fmt.Errorf("codex: mcp_config is set but CODEX_HOME env var is not configured; cannot apply managed MCP")
|
|
}
|
|
|
|
// A custom runtime profile's fixed_args reach codex as the launch prefix.
|
|
// Being first does not protect the daemon's managed config from them: a
|
|
// `-c key=value` override wins over the task-local config.toml from any
|
|
// argv position, so the prefix needs the same two removals ExtraArgs and
|
|
// CustomArgs get below.
|
|
runtimeCmd := b.cfg.commandAt(execPath)
|
|
if codexHome != "" {
|
|
// The daemon owns shell_environment_policy in the task-local config.
|
|
// Codex -c/--config overrides are last-wins, so remove user-provided
|
|
// root or profile policy overrides before building the final argv.
|
|
opts.ExtraArgs = filterCodexShellEnvConfigOverrides(opts.ExtraArgs, b.cfg.Logger)
|
|
opts.CustomArgs = filterCodexShellEnvConfigOverrides(opts.CustomArgs, b.cfg.Logger)
|
|
runtimeCmd = runtimeCmd.withFilteredPrefix(func(prefix []string) []string {
|
|
return filterCodexShellEnvConfigOverrides(prefix, b.cfg.Logger)
|
|
})
|
|
}
|
|
if hasManagedCodexMcpConfig(opts.McpConfig) {
|
|
// Mirrors NormalizeCodexLaunchArgs, which applies this to ExtraArgs and
|
|
// CustomArgs once an agent has a managed mcp_config.
|
|
runtimeCmd = runtimeCmd.withFilteredPrefix(func(prefix []string) []string {
|
|
return filterCodexCustomConfigOverrides(prefix, b.cfg.Logger)
|
|
})
|
|
}
|
|
if opts.ServiceTier == codexFastServiceTier {
|
|
// Mirrors enforceCodexFastMode, which buildCodexArgs applies to the
|
|
// managed args. The enable itself is appended there, once; the prefix
|
|
// only needs the conflicting disable/config overrides removed.
|
|
runtimeCmd = runtimeCmd.withFilteredPrefix(func(prefix []string) []string {
|
|
return stripCodexFastModeConflicts(prefix, b.cfg.Logger)
|
|
})
|
|
}
|
|
codexArgs := buildCodexArgs(opts, b.cfg.Logger)
|
|
// Keep the app-server alive long enough to handle turn/interrupt when the
|
|
// task context is cancelled. The lifecycle goroutine remains governed by
|
|
// runCtx and explicitly stops this process context during bounded cleanup.
|
|
processCtx, stopProcess := context.WithCancel(context.WithoutCancel(runCtx))
|
|
cmd := runtimeCmd.exec(processCtx, codexArgs...)
|
|
hideAgentWindow(cmd)
|
|
// Run codex in its own process group so a cancel-on-stuck cleanup
|
|
// reaches the whole tree — the codex Node wrapper plus the native
|
|
// Rust app-server it spawns — not just the direct child. Without
|
|
// this, killing the leader leaves grandchildren as orphans that
|
|
// keep consuming memory until the OS reaps them; see #4520, where a
|
|
// scanner overflow during thread/resume otherwise leaked Codex
|
|
// processes indefinitely. configureProcessGroup is a no-op on
|
|
// Windows.
|
|
// Override the default exec.CommandContext cancel behaviour. The
|
|
// default sends SIGKILL only to cmd.Process (the leader); we instead
|
|
// signal the whole process group so descendants die too. Returning
|
|
// nil keeps exec from logging a spurious error; cmd.WaitDelay below
|
|
// still backstops cmd.Wait() if the kill leaves an open pipe.
|
|
cmd.Cancel = func() error {
|
|
if cmd.Process != nil {
|
|
signalProcessGroup(cmd, syscall.SIGKILL)
|
|
}
|
|
return nil
|
|
}
|
|
// Bound the wait after the context is cancelled so a stuck child (or an
|
|
// open pipe held by a grandchild) can't hang cmd.Wait() forever. Matches
|
|
// the other long-lived backends (claude, copilot, cursor, …).
|
|
cmd.WaitDelay = codexProcessWaitDelay()
|
|
b.cfg.logAgentCommand(cmd, newAgentCommandLogArgs(codexArgs, trustAgentCommandPositional(0, "app-server")))
|
|
if opts.Cwd != "" {
|
|
cmd.Dir = opts.Cwd
|
|
}
|
|
cmd.Env = buildEnv(b.cfg.Env)
|
|
|
|
stdout, err := cmd.StdoutPipe()
|
|
if err != nil {
|
|
cancel()
|
|
stopProcess()
|
|
return nil, fmt.Errorf("codex stdout pipe: %w", err)
|
|
}
|
|
stdin, err := cmd.StdinPipe()
|
|
if err != nil {
|
|
cancel()
|
|
stopProcess()
|
|
return nil, fmt.Errorf("codex stdin pipe: %w", err)
|
|
}
|
|
// Codex stderr can contain auth/provider diagnostics. Capture a bounded
|
|
// tail and emit it only through the sanitizer in the cleanup event.
|
|
stderrBuf := newStderrTail(io.Discard, codexStderrTailBytes)
|
|
cmd.Stderr = stderrBuf
|
|
|
|
// Start and take ownership of the process tree in one step. On Windows the
|
|
// child is created suspended and placed in a Job Object before it runs, so
|
|
// the cleanup below reaches the Node wrapper, the native app-server, and the
|
|
// sandbox helpers underneath them rather than just the direct child. On Unix
|
|
// the process group configured above already covers that and this is a plain
|
|
// Start. Ownership that cannot be taken is logged, not fatal; a child that
|
|
// cannot be resumed is killed and reported here.
|
|
if err := startOwnedProcessTree(cmd, b.cfg.Logger); err != nil {
|
|
cancel()
|
|
stopProcess()
|
|
return nil, fmt.Errorf("start codex: %w", err)
|
|
}
|
|
activeLaunches := activeCodexLaunches.Add(1)
|
|
for {
|
|
maxSeen := maxActiveCodexLaunchesObserved.Load()
|
|
if activeLaunches <= maxSeen || maxActiveCodexLaunchesObserved.CompareAndSwap(maxSeen, activeLaunches) {
|
|
break
|
|
}
|
|
}
|
|
launchStarted := time.Now()
|
|
codexVersion := strings.TrimSpace(b.cfg.CodexVersion)
|
|
if codexVersion == "" {
|
|
codexVersion = "unknown"
|
|
}
|
|
|
|
b.cfg.Logger.Info("codex lifecycle", "phase", "spawn", "task_id", b.cfg.TaskID, "runtime_id", b.cfg.RuntimeID, "pid", cmd.Process.Pid, "process_group", cmd.Process.Pid, "cwd", opts.Cwd, "attempt", attempt, "active_launches", activeLaunches, "codex_version", codexVersion, "daemon_version", b.cfg.DaemonVersion)
|
|
|
|
msgCh := make(chan Message, 256)
|
|
resCh := make(chan Result, 1)
|
|
semanticActivityCh := make(chan string, 256)
|
|
guardianRateLimitTrip := make(chan struct{}, 1)
|
|
var guardianRateLimitTripped atomic.Bool
|
|
|
|
var outputMu sync.Mutex
|
|
// Result.Output is "final user-facing output selected by the backend"
|
|
// (agent.go), so it holds the deliverable only. output tracks the text the
|
|
// app-server labelled `phase: "final_answer"` and, for the legacy
|
|
// `agent_message` protocol which carries no phase, the last agent message.
|
|
// Every agent message still flows to msgCh, so the transcript is unchanged —
|
|
// only what the daemon forwards to a chat/channel reply narrows (GH #6006).
|
|
var output codexOutputSelection
|
|
var semanticObserved atomic.Bool
|
|
turnNotificationGate := &codexTurnNotificationGate{}
|
|
firstItemWait := &codexFirstItemWaitObservation{}
|
|
|
|
// turnDone is set before starting the reader goroutine so there is no
|
|
// race between the lifecycle goroutine writing and the reader reading.
|
|
turnDone := make(chan bool, 1) // true = aborted
|
|
|
|
var c *codexClient
|
|
observeMessage := func(msg Message) {
|
|
logCodexAgentMessage(b.cfg.Logger, msg)
|
|
activity := describeCodexSemanticActivity(msg)
|
|
if activity == "status:running" {
|
|
firstItemWait.start(time.Now(), c.activeTurnID())
|
|
}
|
|
trySendString(semanticActivityCh, activity)
|
|
if activity != "" {
|
|
semanticObserved.Store(true)
|
|
}
|
|
}
|
|
|
|
c = &codexClient{
|
|
cfg: b.cfg,
|
|
stdin: stdin,
|
|
pending: make(map[int]*pendingRPC),
|
|
processDone: make(chan struct{}),
|
|
handshakeTimeout: handshakeTimeout,
|
|
threadHandshakeTimeout: threadHandshakeTimeout,
|
|
pid: cmd.Process.Pid,
|
|
attempt: attempt,
|
|
activeLaunches: activeLaunches,
|
|
notificationProtocol: "unknown",
|
|
acceptNotification: turnNotificationGate.accept,
|
|
guardianRateLimitCircuit: &codexGuardianRateLimitCircuit{},
|
|
onGuardianRateLimit: func() {
|
|
guardianRateLimitTripped.Store(true)
|
|
b.cfg.Logger.Warn("codex Guardian approval rate-limit circuit opened",
|
|
"task_id", b.cfg.TaskID, "runtime_id", b.cfg.RuntimeID,
|
|
"failure_count", codexGuardianRateLimitThreshold)
|
|
guardianRateLimitTrip <- struct{}{}
|
|
},
|
|
onDiscardedNotification: func(string, map[string]any) {
|
|
// Any app-server notification proves the process made semantic
|
|
// progress, even when it is intentionally excluded from the active
|
|
// turn. Preserve initialize-retry safety without replaying content.
|
|
semanticObserved.Store(true)
|
|
},
|
|
onMessage: func(msg Message) {
|
|
observeMessage(msg)
|
|
trySend(msgCh, msg)
|
|
},
|
|
onAgentMessageChunk: func(text string) bool {
|
|
msg := Message{Type: MessageText, Content: text}
|
|
observeMessage(msg)
|
|
// Agent text is the user-visible transcript, so unlike auxiliary
|
|
// progress events it participates in reconciliation. A false return
|
|
// leaves the text pending for completed/terminal/EOF retry. Keep the
|
|
// send non-blocking because Session.Messages is optional; bounded
|
|
// coalescing prevents a token burst from filling the channel while
|
|
// preserving the reader's EOF and cancellation liveness.
|
|
return trySend(msgCh, msg)
|
|
},
|
|
onAgentMessage: func(text, phase string) {
|
|
// Delta events make MessageText incremental. Keep Result.Output's
|
|
// legacy fallback authoritative by updating it from the completed
|
|
// agent-message snapshot, not whichever delta happened to arrive last.
|
|
outputMu.Lock()
|
|
output.observeAgentMessage(text, phase)
|
|
outputMu.Unlock()
|
|
},
|
|
onFinalAnswer: func(text string) {
|
|
outputMu.Lock()
|
|
output.observeFinalAnswer(text)
|
|
outputMu.Unlock()
|
|
},
|
|
onSemanticActivity: func(description string) {
|
|
semanticObserved.Store(true)
|
|
b.cfg.Logger.Debug("codex semantic activity observed", "activity", description)
|
|
trySendString(semanticActivityCh, description)
|
|
},
|
|
onTurnDone: func(aborted bool) {
|
|
select {
|
|
case turnDone <- aborted:
|
|
default:
|
|
}
|
|
},
|
|
}
|
|
|
|
// Start reading stdout in background
|
|
readerDone := make(chan struct{})
|
|
go func() {
|
|
defer close(readerDone)
|
|
scanner := newAgentStreamScanner(stdout)
|
|
for scanner.Scan() {
|
|
line := strings.TrimSpace(scanner.Text())
|
|
if line == "" {
|
|
continue
|
|
}
|
|
c.handleLine(line)
|
|
}
|
|
// A cancelled or crashed app-server may close stdout without an
|
|
// item/completed snapshot. Preserve every complete delta JSON event
|
|
// already parsed; an incomplete final JSON token is ignored by
|
|
// handleLine and can never enter this buffer.
|
|
c.flushAgentMessageDeltas()
|
|
if err := scanner.Err(); err != nil {
|
|
// %w on BOTH: callers match errCodexProcessExited to decide the
|
|
// process is gone, and bufio.ErrTooLong to tell "we could not read
|
|
// the response" apart from "codex died". startOrResumeThread needs
|
|
// that distinction to report an oversized resume as a rejected
|
|
// resume rather than a crash (MUL-5722).
|
|
c.markProcessExited(fmt.Errorf("%w: %w", errCodexProcessExited, err))
|
|
return
|
|
}
|
|
c.markProcessExited(errCodexProcessExited)
|
|
}()
|
|
|
|
// drainAndWait closes stdin and joins the process exactly once. Read stderrBuf
|
|
// only after it returns, when os/exec's copy goroutines have finished.
|
|
//
|
|
// Wait for readerDone before cmd.Wait so Wait cannot close a pipe under an active
|
|
// read. Bound both reader shutdown and process reaping. Scanner overflow can leave
|
|
// the child blocked on unread stdout, so cancellation and WaitDelay must also
|
|
// cover that path instead of relying on stdin EOF to stop the child.
|
|
var waitOnce sync.Once
|
|
var cleanupConfirmed bool
|
|
var waitReturned bool
|
|
var cleanupWaitErr error
|
|
drainAndWait := func() {
|
|
waitOnce.Do(func() {
|
|
stdin.Close()
|
|
|
|
grace := codexGracefulShutdown()
|
|
waitCh := make(chan struct{})
|
|
var startWait sync.Once
|
|
startProcessWait := func() {
|
|
startWait.Do(func() {
|
|
go func() {
|
|
cleanupWaitErr = cmd.Wait()
|
|
close(waitCh)
|
|
}()
|
|
})
|
|
}
|
|
|
|
// Phase 1: let the reader finish before invoking cmd.Wait().
|
|
select {
|
|
case <-readerDone:
|
|
// reader drained cleanly (codex shutdown closed stdout)
|
|
// or aborted early (e.g. scanner overflow). Either way it
|
|
// is now safe to call cmd.Wait().
|
|
case <-time.After(grace):
|
|
// codex did not close stdout within the grace window. Force
|
|
// the shutdown via context cancellation — cmd.Cancel
|
|
// group-kills the tree, the reader unblocks when stdout
|
|
// EOFs, and we proceed to phase 2.
|
|
b.cfg.Logger.Warn("codex did not close stdout after stdin EOF; forcing shutdown",
|
|
"pid", cmd.Process.Pid,
|
|
"grace", grace.String(),
|
|
)
|
|
stopProcess()
|
|
// On Windows, Cancel terminates only the direct child. A
|
|
// descendant may keep inherited stdout open indefinitely. Start
|
|
// Wait now so os/exec's WaitDelay closes the pipe after its
|
|
// bounded deadline and lets the reader finish.
|
|
startProcessWait()
|
|
<-waitCh
|
|
select {
|
|
case <-readerDone:
|
|
case <-time.After(grace):
|
|
b.cfg.Logger.Warn("codex stdout reader remained open after bounded process wait",
|
|
"pid", cmd.Process.Pid,
|
|
"grace", grace.String(),
|
|
)
|
|
}
|
|
}
|
|
|
|
// Phase 2: bound cmd.Wait() in case the process is still alive
|
|
// (scanner-overflow case: reader exited early on its own while
|
|
// codex stayed blocked writing into a full stdout pipe).
|
|
startProcessWait()
|
|
select {
|
|
case <-waitCh:
|
|
waitReturned = true
|
|
// reaped cleanly.
|
|
case <-time.After(grace):
|
|
b.cfg.Logger.Warn("codex process still alive after reader exited; forcing shutdown",
|
|
"pid", cmd.Process.Pid,
|
|
"grace", grace.String(),
|
|
)
|
|
stopProcess()
|
|
// WaitDelay (10s) is the final backstop: even if the
|
|
// group-kill races with an open pipe held by a
|
|
// descendant, cmd.Wait() returns within WaitDelay of the
|
|
// cancel.
|
|
<-waitCh
|
|
waitReturned = true
|
|
}
|
|
// Wait returning with a ProcessState is the os/exec reap boundary.
|
|
// On Unix, ProcessState.Exited reports false for a process terminated
|
|
// by SIGKILL even though Wait successfully reaped it.
|
|
cleanupConfirmed = waitReturned && cmd.ProcessState != nil && waitProcessGroupGone(cmd, grace)
|
|
if codexCleanupConfirmationOverride.Load() < 0 {
|
|
cleanupConfirmed = false
|
|
}
|
|
b.cfg.Logger.Info("codex lifecycle",
|
|
"phase", "cleanup",
|
|
"task_id", b.cfg.TaskID,
|
|
"runtime_id", b.cfg.RuntimeID,
|
|
"pid", cmd.Process.Pid,
|
|
"process_group", cmd.Process.Pid,
|
|
"attempt", attempt,
|
|
"latency", time.Since(launchStarted).Round(time.Millisecond).String(),
|
|
"reaped", cleanupConfirmed,
|
|
"exit_status", codexProcessExitStatus(cmd.ProcessState),
|
|
"wait_error", cleanupWaitErr,
|
|
"stderr_bytes", stderrBuf.TotalBytes(),
|
|
"stderr_truncated", stderrBuf.TotalBytes() > codexStderrTailBytes,
|
|
)
|
|
// The tree has been reaped and observed; drop ownership. This is
|
|
// the only safe point to do so on Windows, where releasing kills
|
|
// whatever is still inside the job — which is precisely what should
|
|
// happen to anything that outlived the reap above. waitOnce makes it
|
|
// exactly-once per launch attempt.
|
|
releaseProcessGroup(cmd)
|
|
})
|
|
}
|
|
|
|
// Drive the session lifecycle in a goroutine.
|
|
// Shutdown sequence: lifecycle goroutine closes stdin + cancels context →
|
|
// codex process exits → reader goroutine's scanner.Scan() returns false →
|
|
// readerDone closes → lifecycle goroutine collects final output and sends Result.
|
|
go func() {
|
|
defer activeCodexLaunches.Add(-1)
|
|
defer cancel()
|
|
defer stopProcess()
|
|
defer close(msgCh)
|
|
defer close(resCh)
|
|
defer drainAndWait()
|
|
|
|
startTime := time.Now()
|
|
finalStatus := "completed"
|
|
var finalError string
|
|
|
|
// 1. Initialize handshake
|
|
initializeStarted := time.Now()
|
|
b.cfg.Logger.Info("codex lifecycle", "phase", "initialize_sent", "task_id", b.cfg.TaskID, "runtime_id", b.cfg.RuntimeID, "pid", cmd.Process.Pid, "attempt", attempt, "active_launches", activeLaunches)
|
|
_, err := c.request(runCtx, "initialize", map[string]any{
|
|
"clientInfo": map[string]any{
|
|
"name": "multica-agent-sdk",
|
|
"title": "Multica Agent SDK",
|
|
"version": "0.2.0",
|
|
},
|
|
"capabilities": map[string]any{
|
|
"experimentalApi": true,
|
|
},
|
|
})
|
|
if err != nil {
|
|
initializeLatency := time.Since(initializeStarted)
|
|
var handshakeErr *codexHandshakeTimeoutError
|
|
timedOut := errors.As(err, &handshakeErr) && handshakeErr.Method == "initialize"
|
|
if timedOut {
|
|
// A timed-out initialize may still complete after the host gives up.
|
|
// Kill the whole process group before waiting so a leader that exits
|
|
// on stdin EOF cannot leave detached-stdio descendants behind.
|
|
signalProcessGroup(cmd, syscall.SIGKILL)
|
|
} else if runCtx.Err() != nil {
|
|
// processCtx deliberately outlives runCtx so active turns can use
|
|
// turn/interrupt. Initialize has no interruptible turn, so a parent
|
|
// cancellation must take the fast process-tree cleanup path.
|
|
stopProcess()
|
|
}
|
|
drainAndWait() // flush os/exec stderr goroutine before sampling Tail
|
|
finalStatus = "failed"
|
|
finalError = fmt.Sprintf("codex initialize failed: %v", err)
|
|
contextEnded := errors.Is(err, context.DeadlineExceeded) || errors.Is(err, context.Canceled)
|
|
if !timedOut && !contextEnded {
|
|
// Timeout stderr is untrusted provider output and may echo opaque
|
|
// Config.Env/auth values that pattern sanitization cannot identify.
|
|
// The same applies when the parent task deadline/cancellation wins
|
|
// the race against the per-RPC handshake timeout.
|
|
// Keep it out of persisted/user-visible Results; cleanup lifecycle
|
|
// still records bounded byte/truncation metadata.
|
|
finalError = withAgentStderr(finalError, "codex", sanitizeCodexDiagnostic(stderrBuf.Tail()))
|
|
}
|
|
retrySafe := timedOut && !semanticObserved.Load() && cleanupConfirmed && codexInitializeRetrySupported()
|
|
if timedOut && !cleanupConfirmed {
|
|
finalError += "; retry suppressed: process cleanup/reap not confirmed"
|
|
} else if timedOut && cleanupConfirmed && !codexInitializeRetrySupported() {
|
|
finalError += "; retry suppressed: process-tree cleanup cannot be confirmed on this platform"
|
|
}
|
|
b.cfg.Logger.Warn("codex lifecycle", "phase", "initialize_failure", "task_id", b.cfg.TaskID, "runtime_id", b.cfg.RuntimeID, "pid", cmd.Process.Pid, "attempt", attempt, "latency", initializeLatency.Round(time.Millisecond).String(), "semantic_activity", semanticObserved.Load(), "cleanup_confirmed", cleanupConfirmed, "retry_safe", retrySafe)
|
|
resCh <- Result{Status: finalStatus, Error: finalError, DurationMs: time.Since(startTime).Milliseconds(), codexInitializeRetrySafe: retrySafe}
|
|
return
|
|
}
|
|
b.cfg.Logger.Info("codex lifecycle", "phase", "initialize_response", "task_id", b.cfg.TaskID, "runtime_id", b.cfg.RuntimeID, "pid", cmd.Process.Pid, "attempt", attempt, "latency", time.Since(initializeStarted).Round(time.Millisecond).String())
|
|
c.notify("initialized")
|
|
|
|
// 2. Start a new thread, or resume the prior one for this issue. When
|
|
// resume fails (thread GCed on the server, schema drift, etc.) we fall
|
|
// back to a fresh thread so the task still makes progress.
|
|
threadID, resumed, err := c.startOrResumeThread(runCtx, opts, b.cfg.Logger)
|
|
if err != nil {
|
|
var handshakeErr *codexHandshakeTimeoutError
|
|
timedOut := errors.As(err, &handshakeErr) && isCodexThreadSetupRPC(handshakeErr.Method)
|
|
if timedOut {
|
|
// A timed-out thread setup has an uncertain provider outcome. Kill
|
|
// the whole process group before waiting so a leader that exits on
|
|
// EOF cannot leave detached-stdio descendants behind.
|
|
signalProcessGroup(cmd, syscall.SIGKILL)
|
|
} else if runCtx.Err() != nil {
|
|
// No turn exists during thread start/resume, so there is nothing to
|
|
// interrupt and no reason to spend the graceful shutdown budget.
|
|
stopProcess()
|
|
}
|
|
drainAndWait() // flush os/exec stderr goroutine before sampling Tail
|
|
finalStatus = "failed"
|
|
stderrTail := sanitizeCodexDiagnostic(stderrBuf.Tail())
|
|
finalError = err.Error()
|
|
if c.threadSetupMethod != "" {
|
|
classification := classifyCodexStartupStderr(stderrTail, timedOut)
|
|
b.cfg.Logger.Warn("codex lifecycle",
|
|
"phase", strings.ReplaceAll(c.threadSetupMethod, "/", "_")+"_failure",
|
|
"task_id", b.cfg.TaskID,
|
|
"runtime_id", b.cfg.RuntimeID,
|
|
"pid", cmd.Process.Pid,
|
|
"attempt", attempt,
|
|
"active_launches", activeLaunches,
|
|
"method", c.threadSetupMethod,
|
|
"latency", time.Since(c.threadSetupStarted).Round(time.Millisecond).String(),
|
|
"latency_ms", time.Since(c.threadSetupStarted).Milliseconds(),
|
|
"cleanup_confirmed", cleanupConfirmed,
|
|
"reaped", cleanupConfirmed,
|
|
"retry_safe", false,
|
|
"retry_attempted", false,
|
|
"stderr_model_refresh_failure_count", classification.modelRefreshFailure,
|
|
"stderr_model_refresh_timeout_count", classification.modelRefreshTimeout,
|
|
"stderr_mcp_init_transport_count", classification.mcpInitTransport,
|
|
"stderr_bare_timeout_count", classification.bareTimeout,
|
|
)
|
|
}
|
|
resCh <- Result{
|
|
Status: finalStatus,
|
|
Error: finalError,
|
|
DurationMs: time.Since(startTime).Milliseconds(),
|
|
ResumeRejected: isCodexResumeOverflow(opts, err),
|
|
}
|
|
return
|
|
}
|
|
c.setThreadID(threadID)
|
|
if resumed {
|
|
b.cfg.Logger.Info("codex thread resumed", "thread_id", threadID)
|
|
} else {
|
|
b.cfg.Logger.Info("codex thread started", "thread_id", threadID)
|
|
}
|
|
|
|
// 3. Send turn and wait for completion. When a resume was expected but we
|
|
// ended up on a fresh thread (the live thread/resume RPC was rejected — a
|
|
// corrupt/incompatible rollout, server-side thread GC, schema drift — or a
|
|
// transport failure forced a fresh retry), prepend the caller's continuity
|
|
// notice so the agent does not assume continuity it no longer has. The
|
|
// daemon's pre-flight gates only catch cases detectable before launch;
|
|
// this covers the ones only the live resume reveals (MUL-4424).
|
|
//
|
|
// Whether that notice asks the agent to tell the USER is the caller's
|
|
// call, not ours: it depends on whether this surface's conversation is
|
|
// still readable, which this package cannot see (MUL-5722).
|
|
turnParams := map[string]any{
|
|
"threadId": threadID,
|
|
"input": codexTurnInput(prompt, opts.ResumeExpected, resumed, opts.ResumeContinuityNotice),
|
|
}
|
|
// Per-turn reasoning override. Mirrors the per-thread injection in
|
|
// startOrResumeThread; keeping both in sync is enforced by the
|
|
// shared `codexReasoningInjection` fixture in codex_test.go (see
|
|
// MUL-2339 — Trump's constraint that the three injection points
|
|
// must not drift independently).
|
|
applyCodexReasoningEffort(turnParams, opts.ThinkingLevel)
|
|
applyCodexServiceTier(turnParams, opts.ServiceTier)
|
|
waitingForTurn := true
|
|
var timeoutDiagnostic codexTimeoutDiagnostic
|
|
var processExitErr error
|
|
finishFirstItemWait := func(outcome string) {
|
|
firstItemWait.finish(
|
|
time.Now(),
|
|
outcome,
|
|
classifyCodexStartupStderr(stderrBuf.Tail(), strings.HasSuffix(outcome, "_timeout")),
|
|
)
|
|
}
|
|
finishTurn := func(aborted bool) {
|
|
waitingForTurn = false
|
|
switch {
|
|
case aborted:
|
|
finishFirstItemWait("turn_aborted")
|
|
finalStatus = "aborted"
|
|
if errMsg := c.getTurnError(); errMsg != "" {
|
|
finalError = errMsg
|
|
} else {
|
|
finalError = "turn was aborted"
|
|
}
|
|
default:
|
|
if errMsg := c.getTurnError(); errMsg != "" {
|
|
finishFirstItemWait("turn_failed")
|
|
finalStatus = "failed"
|
|
finalError = errMsg
|
|
} else {
|
|
finishFirstItemWait("turn_completed")
|
|
}
|
|
}
|
|
}
|
|
finishRunContextDone := func() {
|
|
waitingForTurn = false
|
|
if runCtx.Err() == context.DeadlineExceeded {
|
|
finishFirstItemWait("execution_timeout")
|
|
finalStatus = "timeout"
|
|
finalError = fmt.Sprintf("codex timed out after %s", timeout)
|
|
} else {
|
|
finishFirstItemWait("cancelled")
|
|
finalStatus = "aborted"
|
|
finalError = "execution cancelled"
|
|
}
|
|
}
|
|
turnNotificationGate.arm()
|
|
_, err = c.request(runCtx, "turn/start", turnParams)
|
|
if err != nil {
|
|
if runCtx.Err() != nil {
|
|
finishRunContextDone()
|
|
if !interruptCodexTurn(c, threadID, turnDone, opts.TurnInterruptTimeout, b.cfg.Logger) {
|
|
stopProcess()
|
|
}
|
|
} else {
|
|
select {
|
|
case aborted := <-turnDone:
|
|
finishTurn(aborted)
|
|
default:
|
|
drainAndWait() // flush os/exec stderr goroutine before sampling Tail
|
|
finalStatus = "failed"
|
|
finalError = withAgentStderr(fmt.Sprintf("codex turn/start failed: %v", err), "codex", sanitizeCodexDiagnostic(stderrBuf.Tail()))
|
|
resCh <- Result{Status: finalStatus, Error: finalError, DurationMs: time.Since(startTime).Milliseconds()}
|
|
return
|
|
}
|
|
}
|
|
}
|
|
|
|
lastSemanticActivity := time.Now()
|
|
lastSemanticActivityDescription := "turn/start"
|
|
semanticTimer := time.NewTimer(semanticInactivityTimeout)
|
|
defer semanticTimer.Stop()
|
|
|
|
firstTurnNoProgressTimeout := codexFirstTurnNoProgressTimeout(semanticInactivityTimeout, opts.FirstTurnNoProgressTimeout)
|
|
var firstTurnNoProgressTimer *time.Timer
|
|
var firstTurnNoProgressTimerC <-chan time.Time
|
|
firstTurnStarted := false
|
|
firstTurnProgressObserved := false
|
|
stopFirstTurnNoProgressTimer := func() {
|
|
if firstTurnNoProgressTimer == nil {
|
|
return
|
|
}
|
|
stopTimer(firstTurnNoProgressTimer)
|
|
firstTurnNoProgressTimerC = nil
|
|
}
|
|
defer stopFirstTurnNoProgressTimer()
|
|
|
|
for waitingForTurn {
|
|
select {
|
|
case aborted := <-turnDone:
|
|
finishTurn(aborted)
|
|
case <-guardianRateLimitTrip:
|
|
waitingForTurn = false
|
|
finishFirstItemWait("guardian_rate_limit")
|
|
finalStatus = "failed"
|
|
finalError = codexGuardianRateLimitError
|
|
if !interruptCodexTurn(c, threadID, turnDone, opts.TurnInterruptTimeout, b.cfg.Logger) {
|
|
stopProcess()
|
|
}
|
|
case activity := <-semanticActivityCh:
|
|
lastSemanticActivity = time.Now()
|
|
lastSemanticActivityDescription = activity
|
|
resetTimer(semanticTimer, semanticInactivityTimeout)
|
|
if activity == "status:running" && !firstTurnStarted {
|
|
firstTurnStarted = true
|
|
firstItemWait.start(time.Now(), c.activeTurnID())
|
|
firstTurnNoProgressTimer = time.NewTimer(firstTurnNoProgressTimeout)
|
|
firstTurnNoProgressTimerC = firstTurnNoProgressTimer.C
|
|
} else if firstTurnStarted && !firstTurnProgressObserved && isCodexFirstTurnProgressActivity(activity) {
|
|
firstTurnProgressObserved = true
|
|
if activity == "error:terminal" {
|
|
finishFirstItemWait("turn_failed")
|
|
} else {
|
|
finishFirstItemWait("progress")
|
|
}
|
|
stopFirstTurnNoProgressTimer()
|
|
}
|
|
case <-firstTurnNoProgressTimerC:
|
|
waitingForTurn = false
|
|
finishFirstItemWait("no_progress_timeout")
|
|
finalStatus = "timeout"
|
|
turnID := c.activeTurnID()
|
|
timeoutDiagnostic = codexTimeoutDiagnostic{
|
|
Kind: codexTimeoutFirstTurnNoProgress,
|
|
Timeout: firstTurnNoProgressTimeout,
|
|
LastActivity: lastSemanticActivityDescription,
|
|
ThreadID: threadID,
|
|
TurnID: turnID,
|
|
Model: opts.Model,
|
|
}
|
|
b.cfg.Logger.Warn(CodexFirstTurnNoProgressMarker,
|
|
"pid", cmd.Process.Pid,
|
|
"thread_id", threadID,
|
|
"turn_id", turnID,
|
|
"timeout", firstTurnNoProgressTimeout.String(),
|
|
"last_activity", lastSemanticActivityDescription,
|
|
)
|
|
case <-semanticTimer.C:
|
|
waitingForTurn = false
|
|
finishFirstItemWait("semantic_inactivity_timeout")
|
|
finalStatus = "timeout"
|
|
turnID := c.activeTurnID()
|
|
timeoutDiagnostic = codexTimeoutDiagnostic{
|
|
Kind: codexTimeoutSemanticInactivity,
|
|
Timeout: semanticInactivityTimeout,
|
|
LastActivity: lastSemanticActivityDescription,
|
|
ThreadID: threadID,
|
|
TurnID: turnID,
|
|
Model: opts.Model,
|
|
}
|
|
b.cfg.Logger.Warn(CodexSemanticInactivityMarker,
|
|
"pid", cmd.Process.Pid,
|
|
"thread_id", threadID,
|
|
"turn_id", turnID,
|
|
"timeout", semanticInactivityTimeout.String(),
|
|
"last_activity", lastSemanticActivityDescription,
|
|
"idle_for", time.Since(lastSemanticActivity).Round(time.Millisecond).String(),
|
|
)
|
|
case <-runCtx.Done():
|
|
finishRunContextDone()
|
|
if !interruptCodexTurn(c, threadID, turnDone, opts.TurnInterruptTimeout, b.cfg.Logger) {
|
|
stopProcess()
|
|
}
|
|
case <-c.processDone:
|
|
select {
|
|
case aborted := <-turnDone:
|
|
finishTurn(aborted)
|
|
default:
|
|
if runCtx.Err() != nil {
|
|
finishRunContextDone()
|
|
} else {
|
|
waitingForTurn = false
|
|
finishFirstItemWait("process_exit")
|
|
finalStatus = "failed"
|
|
processExitErr = c.getProcessErr()
|
|
if processExitErr == nil {
|
|
processExitErr = errCodexProcessExited
|
|
}
|
|
finalError = processExitErr.Error()
|
|
}
|
|
}
|
|
}
|
|
}
|
|
// A turn/completed notification can race the circuit signal in the
|
|
// select above. The already-observed repeated failures still win.
|
|
if guardianRateLimitTripped.Load() {
|
|
finalStatus = "failed"
|
|
finalError = codexGuardianRateLimitError
|
|
}
|
|
|
|
duration := time.Since(startTime)
|
|
b.cfg.Logger.Info("codex finished", "pid", cmd.Process.Pid, "status", finalStatus, "duration", duration.Round(time.Millisecond).String())
|
|
|
|
// Run cleanup. drainAndWait handles the graceful-then-cancel pattern
|
|
// in two bounded phases (see its declaration): wait for the reader,
|
|
// then wait for cmd.Wait(), force-cancelling either if the grace
|
|
// window expires. A clean shutdown lets codex flush OTEL telemetry;
|
|
// a stuck process is killed via the process-group SIGKILL.
|
|
drainAndWait()
|
|
|
|
if processExitErr != nil {
|
|
finalError = withAgentStderr(processExitErr.Error(), "codex", sanitizeCodexDiagnostic(stderrBuf.Tail()))
|
|
}
|
|
stderrTail := sanitizeCodexDiagnostic(stderrBuf.Tail())
|
|
if timeoutDiagnostic.Kind != codexTimeoutNone {
|
|
timeoutDiagnostic.CodexVersion = detectCodexVersionForDiagnostics(context.Background(), b.cfg.commandAt(execPath), cmd.Env, b.cfg.Logger)
|
|
finalError = buildCodexTimeoutDiagnosticError(timeoutDiagnostic, stderrTail)
|
|
}
|
|
|
|
// A first turn that produced no semantic progress because Codex could
|
|
// not load its model catalog is a startup-only failure: no tool ran and
|
|
// no content reached the user, so replaying the prompt cannot duplicate
|
|
// side effects. Reuse the same process-tree evidence initialize retries
|
|
// require (cleanupConfirmed plus platform support) rather than a bare
|
|
// ProcessState check: on Windows the daemon cannot prove the whole tree
|
|
// is gone, and a surviving app-server would race the retry.
|
|
startupRefreshRetrySafe := timeoutDiagnostic.Kind == codexTimeoutFirstTurnNoProgress &&
|
|
!firstTurnProgressObserved &&
|
|
strings.Contains(stderrTail, codexModelCatalogRefreshFailureSignal) &&
|
|
cleanupConfirmed && codexInitializeRetrySupported()
|
|
if startupRefreshRetrySafe {
|
|
b.cfg.Logger.Warn("codex startup model catalog refresh failure is retry safe",
|
|
"pid", cmd.Process.Pid,
|
|
"thread_id", threadID,
|
|
"attempt", attempt,
|
|
)
|
|
}
|
|
|
|
if waitLatency, firstTurnID, outcome, classification, ok := firstItemWait.snapshot(); ok {
|
|
if waitLatency < 0 {
|
|
waitLatency = 0
|
|
}
|
|
// On timeout, cmd.Wait is the synchronization point that guarantees
|
|
// the stderr copy goroutine has drained. Reclassify from the complete
|
|
// bounded tail so a last-moment catalog/MCP signal is not reported as
|
|
// a bare timeout. Successful waits keep the snapshot taken at first
|
|
// progress so later turn stderr cannot pollute first-item telemetry.
|
|
if strings.HasSuffix(outcome, "_timeout") {
|
|
classification = classifyCodexStartupStderr(stderrTail, true)
|
|
}
|
|
fields := []any{
|
|
"phase", "first_item_wait",
|
|
"task_id", b.cfg.TaskID,
|
|
"runtime_id", b.cfg.RuntimeID,
|
|
"pid", cmd.Process.Pid,
|
|
"attempt", attempt,
|
|
"active_launches", activeLaunches,
|
|
"method", "turn/start",
|
|
"thread_id", threadID,
|
|
"turn_id", firstTurnID,
|
|
"outcome", outcome,
|
|
"latency", waitLatency.Round(time.Millisecond).String(),
|
|
"latency_ms", waitLatency.Milliseconds(),
|
|
"timeout", firstTurnNoProgressTimeout.String(),
|
|
"semantic_inactivity_timeout", semanticInactivityTimeout.String(),
|
|
"codex_version", codexVersion,
|
|
"daemon_version", b.cfg.DaemonVersion,
|
|
"cleanup_confirmed", cleanupConfirmed,
|
|
"reaped", cleanupConfirmed,
|
|
// retry_safe describes the terminal attempt, not the measured wait
|
|
// interval. Successful samples therefore report false by design.
|
|
"retry_safe", startupRefreshRetrySafe,
|
|
"stderr_model_refresh_failure_count", classification.modelRefreshFailure,
|
|
"stderr_model_refresh_timeout_count", classification.modelRefreshTimeout,
|
|
"stderr_mcp_init_transport_count", classification.mcpInitTransport,
|
|
"stderr_bare_timeout_count", classification.bareTimeout,
|
|
}
|
|
switch outcome {
|
|
case "progress", "turn_completed":
|
|
b.cfg.Logger.Info("codex lifecycle", fields...)
|
|
default:
|
|
b.cfg.Logger.Warn("codex lifecycle", fields...)
|
|
}
|
|
}
|
|
|
|
outputMu.Lock()
|
|
finalOutput := codexDeliverableOutput(output)
|
|
outputMu.Unlock()
|
|
|
|
// Build usage map from accumulated codex usage.
|
|
// First check JSON-RPC notifications (often empty for Codex).
|
|
var usageMap map[string]TokenUsage
|
|
c.usageMu.Lock()
|
|
u := c.usage
|
|
c.usageMu.Unlock()
|
|
|
|
// Codex sometimes omits usage from JSON-RPC notifications, and can also
|
|
// omit the configured model when the agent relies on Codex's default.
|
|
// Scan the per-task session JSONL for either missing value. Session usage
|
|
// is only a fallback, so it must never replace a JSON-RPC usage sample.
|
|
taskCodexHome := strings.TrimSpace(b.cfg.Env["CODEX_HOME"])
|
|
usageMap = codexResultUsageMap(startTime, taskCodexHome, threadID, resumed, opts.Model, u)
|
|
|
|
resCh <- Result{
|
|
Status: finalStatus,
|
|
Output: finalOutput,
|
|
Error: finalError,
|
|
SessionID: threadID,
|
|
DurationMs: duration.Milliseconds(),
|
|
Usage: usageMap,
|
|
codexStartupRefreshRetrySafe: startupRefreshRetrySafe,
|
|
}
|
|
}()
|
|
|
|
supplement := func(supplementCtx context.Context, instruction string) error {
|
|
return supplementCodexTurn(supplementCtx, c, instruction)
|
|
}
|
|
supplementReady := func() bool {
|
|
return c.getThreadID() != "" && c.activeTurnID() != ""
|
|
}
|
|
return &Session{Supplement: supplement, SupplementReady: supplementReady, Messages: msgCh, Result: resCh}, nil
|
|
}
|
|
|
|
func supplementCodexTurn(ctx context.Context, c *codexClient, instruction string) error {
|
|
ctx, cancel := context.WithTimeout(ctx, 8*time.Second)
|
|
defer cancel()
|
|
threadID := c.getThreadID()
|
|
turnID := c.activeTurnID()
|
|
if threadID == "" || turnID == "" {
|
|
return errors.New("codex turn has not started")
|
|
}
|
|
_, err := c.request(ctx, "turn/steer", map[string]any{
|
|
"threadId": threadID,
|
|
"expectedTurnId": turnID,
|
|
"input": []map[string]any{{"type": "text", "text": instruction}},
|
|
})
|
|
return err
|
|
}
|
|
|
|
func resolveCodexHandshakeTimeouts(opts ExecOptions) (time.Duration, time.Duration) {
|
|
handshakeTimeout := opts.HandshakeTimeout
|
|
if handshakeTimeout <= 0 {
|
|
handshakeTimeout = defaultCodexHandshakeTimeout
|
|
}
|
|
threadHandshakeTimeout := opts.ThreadHandshakeTimeout
|
|
if threadHandshakeTimeout <= 0 {
|
|
if opts.HandshakeTimeout > 0 {
|
|
// Backward compatibility: an explicit legacy override historically
|
|
// bounded every startup RPC, including thread setup.
|
|
threadHandshakeTimeout = opts.HandshakeTimeout
|
|
} else {
|
|
threadHandshakeTimeout = defaultCodexThreadHandshakeTimeout
|
|
}
|
|
}
|
|
return handshakeTimeout, threadHandshakeTimeout
|
|
}
|
|
|
|
// The continuity notice this backend prepends is supplied by the caller via
|
|
// ExecOptions.ResumeContinuityNotice rather than written here. It used to be a
|
|
// constant in this file that mirrored the daemon's, which meant two hand-kept
|
|
// copies of the same paragraph and no way for this package to know which
|
|
// surface it was running on — the wording is only correct if you know whether
|
|
// the conversation is still readable (MUL-5722).
|
|
|
|
// codexTurnInput builds the input content for the first turn/start. When a
|
|
// resume was expected (resumeExpected) but the backend landed on a fresh thread
|
|
// (!resumed), it prepends the caller's continuity notice so the run does not
|
|
// silently continue as new. The notice is folded into the same text block as
|
|
// the prompt to stay within the single-text-block turn input Codex accepts.
|
|
//
|
|
// An empty notice means the caller has already disclosed the loss in the prompt
|
|
// itself, so this must add nothing — that is the whole guard against a turn
|
|
// carrying the same paragraph twice.
|
|
func codexTurnInput(prompt string, resumeExpected, resumed bool, notice string) []map[string]any {
|
|
text := prompt
|
|
if resumeExpected && !resumed {
|
|
text = notice + prompt
|
|
}
|
|
return []map[string]any{{"type": "text", "text": text}}
|
|
}
|
|
|
|
// startOrResumeThread picks between Codex's thread/resume and thread/start
|
|
// based on opts.ResumeSessionID. When a prior thread ID is provided it first
|
|
// tries thread/resume; recoverable protocol errors (unknown thread, schema
|
|
// mismatch) fall back to thread/start so the task still executes, while
|
|
// transport/process failures fail fast because the app-server can no longer
|
|
// answer a fresh start request. The returned threadID is what subsequent
|
|
// turn/start calls must reference, and resumed indicates whether the prior
|
|
// thread was picked up (only useful for logging).
|
|
func (c *codexClient) startOrResumeThread(ctx context.Context, opts ExecOptions, logger *slog.Logger) (string, bool, error) {
|
|
if priorThreadID := opts.ResumeSessionID; priorThreadID != "" {
|
|
// thread/resume reuses the thread's persisted model and reasoning
|
|
// effort; only override fields the daemon actually cares about.
|
|
// developerInstructions stays nil for the reason given on thread/start
|
|
// below.
|
|
resumeParams := map[string]any{
|
|
"threadId": priorThreadID,
|
|
"cwd": opts.Cwd,
|
|
"model": nilIfEmpty(opts.Model),
|
|
"developerInstructions": nil,
|
|
// Only omit history from the response, not the resumed model context.
|
|
// We need the thread ID, not an unbounded single-line turns payload.
|
|
"excludeTurns": true,
|
|
}
|
|
// Explicit override of the persisted reasoning effort: without
|
|
// this, a Codex resume silently reuses whatever level the prior
|
|
// session was created with, even when the user has flipped the
|
|
// agent's thinking_level since. See MUL-2339 — Elon flagged that
|
|
// resume must honour the live config, not the stored one.
|
|
applyCodexReasoningEffort(resumeParams, opts.ThinkingLevel)
|
|
applyCodexServiceTier(resumeParams, opts.ServiceTier)
|
|
c.threadSetupMethod = "thread/resume"
|
|
c.threadSetupStarted = time.Now()
|
|
logger.Info("codex lifecycle",
|
|
"phase", "thread_resume_sent",
|
|
"task_id", c.cfg.TaskID,
|
|
"runtime_id", c.cfg.RuntimeID,
|
|
"pid", c.pid,
|
|
"attempt", c.attempt,
|
|
"active_launches", c.activeLaunches,
|
|
"method", c.threadSetupMethod,
|
|
)
|
|
resumeResult, err := c.request(ctx, "thread/resume", resumeParams)
|
|
if err == nil {
|
|
if threadID := extractThreadID(resumeResult); threadID != "" {
|
|
logger.Info("codex lifecycle",
|
|
"phase", "thread_resume_response",
|
|
"task_id", c.cfg.TaskID,
|
|
"runtime_id", c.cfg.RuntimeID,
|
|
"pid", c.pid,
|
|
"attempt", c.attempt,
|
|
"active_launches", c.activeLaunches,
|
|
"method", c.threadSetupMethod,
|
|
"latency", time.Since(c.threadSetupStarted).Round(time.Millisecond).String(),
|
|
"latency_ms", time.Since(c.threadSetupStarted).Milliseconds(),
|
|
)
|
|
return threadID, true, nil
|
|
}
|
|
logger.Warn("codex thread/resume returned no thread ID; falling back to thread/start", "prior_thread_id", priorThreadID)
|
|
} else {
|
|
if isCodexTransportError(err) {
|
|
logger.Warn("codex thread/resume failed due to transport error; not falling back to thread/start", "prior_thread_id", priorThreadID, "error", err)
|
|
return "", false, fmt.Errorf("codex thread/resume failed: %w", err)
|
|
}
|
|
logger.Warn("codex thread/resume failed; falling back to thread/start", "prior_thread_id", priorThreadID, "error", err)
|
|
}
|
|
}
|
|
|
|
// developerInstructions is always nil: a thread started with this cwd loads
|
|
// the per-task AGENTS.md the daemon wrote there, so the runtime brief is
|
|
// already in context and inlining it would duplicate it on every turn.
|
|
// Confirmed end-to-end against codex-cli 0.144.6 driving the real
|
|
// app-server (thread/start -> turn/start) with developerInstructions unset
|
|
// (MUL-5392).
|
|
startParams := map[string]any{
|
|
"model": nilIfEmpty(opts.Model),
|
|
"modelProvider": nil,
|
|
"profile": nil,
|
|
"cwd": opts.Cwd,
|
|
"approvalPolicy": nil,
|
|
"sandbox": nil,
|
|
"config": nil,
|
|
"baseInstructions": nil,
|
|
"developerInstructions": nil,
|
|
"compactPrompt": nil,
|
|
"includeApplyPatchTool": nil,
|
|
"experimentalRawEvents": false,
|
|
"persistExtendedHistory": true,
|
|
}
|
|
applyCodexReasoningEffort(startParams, opts.ThinkingLevel)
|
|
applyCodexServiceTier(startParams, opts.ServiceTier)
|
|
c.threadSetupMethod = "thread/start"
|
|
c.threadSetupStarted = time.Now()
|
|
logger.Info("codex lifecycle",
|
|
"phase", "thread_start_sent",
|
|
"task_id", c.cfg.TaskID,
|
|
"runtime_id", c.cfg.RuntimeID,
|
|
"pid", c.pid,
|
|
"attempt", c.attempt,
|
|
"active_launches", c.activeLaunches,
|
|
"method", c.threadSetupMethod,
|
|
)
|
|
startResult, err := c.request(ctx, "thread/start", startParams)
|
|
if err != nil {
|
|
return "", false, fmt.Errorf("codex thread/start failed: %w", err)
|
|
}
|
|
threadID := extractThreadID(startResult)
|
|
if threadID == "" {
|
|
return "", false, fmt.Errorf("codex thread/start returned no thread ID")
|
|
}
|
|
logger.Info("codex lifecycle",
|
|
"phase", "thread_start_response",
|
|
"task_id", c.cfg.TaskID,
|
|
"runtime_id", c.cfg.RuntimeID,
|
|
"pid", c.pid,
|
|
"attempt", c.attempt,
|
|
"active_launches", c.activeLaunches,
|
|
"method", "thread/start",
|
|
"latency", time.Since(c.threadSetupStarted).Round(time.Millisecond).String(),
|
|
"latency_ms", time.Since(c.threadSetupStarted).Milliseconds(),
|
|
)
|
|
c.trySetThreadName(ctx, threadID, opts.ThreadName, logger)
|
|
return threadID, false, nil
|
|
}
|
|
|
|
func (c *codexClient) trySetThreadName(ctx context.Context, threadID, name string, logger *slog.Logger) {
|
|
name = strings.TrimSpace(name)
|
|
if name == "" {
|
|
return
|
|
}
|
|
if err := c.setThreadName(ctx, threadID, name); err != nil {
|
|
logger.Warn("codex thread/name/set failed; continuing without provider-native thread title",
|
|
"thread_id", threadID, "error", err)
|
|
}
|
|
}
|
|
|
|
func (c *codexClient) setThreadName(ctx context.Context, threadID, name string) error {
|
|
_, err := c.request(ctx, "thread/name/set", map[string]any{
|
|
"threadId": threadID,
|
|
"name": name,
|
|
})
|
|
return err
|
|
}
|
|
|
|
// applyCodexReasoningEffort writes the per-agent thinking_level into a
|
|
// Codex app-server request. The three points — thread/start.config,
|
|
// thread/resume.config, turn/start.effort — all flow through this helper
|
|
// so any future protocol/key change touches one site rather than three
|
|
// (per Trump's MUL-2339 review constraint).
|
|
//
|
|
// The shape is detected from the params keys:
|
|
// - turn/start always carries `input`, and the schema exposes the
|
|
// reasoning override as the top-level `effort` field.
|
|
// - thread/start and thread/resume nest it under
|
|
// `config.model_reasoning_effort`.
|
|
//
|
|
// Empty `level` is a no-op: we deliberately do NOT emit a key when the
|
|
// caller didn't request an override, so the upstream defaults (config
|
|
// file, account-scoped model preference) stay in charge. This also
|
|
// guarantees `effort: ""` never reaches the CLI — Codex rejects empty
|
|
// strings on this field.
|
|
func applyCodexReasoningEffort(params map[string]any, level string) {
|
|
if params == nil || level == "" {
|
|
return
|
|
}
|
|
if _, isTurnStart := params["input"]; isTurnStart {
|
|
params["effort"] = level
|
|
return
|
|
}
|
|
cfg, _ := params["config"].(map[string]any)
|
|
if cfg == nil {
|
|
cfg = map[string]any{}
|
|
}
|
|
cfg["model_reasoning_effort"] = level
|
|
params["config"] = cfg
|
|
}
|
|
|
|
// applyCodexServiceTier writes the catalog-owned service tier to the
|
|
// app-server's top-level serviceTier field. All three execution requests
|
|
// accept the same shape. Empty is a no-op so config.toml remains authoritative.
|
|
func applyCodexServiceTier(params map[string]any, tier string) {
|
|
if params == nil || tier == "" {
|
|
return
|
|
}
|
|
params["serviceTier"] = tier
|
|
}
|
|
|
|
func resetTimer(timer *time.Timer, d time.Duration) {
|
|
if !timer.Stop() {
|
|
select {
|
|
case <-timer.C:
|
|
default:
|
|
}
|
|
}
|
|
timer.Reset(d)
|
|
}
|
|
|
|
func stopTimer(timer *time.Timer) {
|
|
if timer == nil {
|
|
return
|
|
}
|
|
if !timer.Stop() {
|
|
select {
|
|
case <-timer.C:
|
|
default:
|
|
}
|
|
}
|
|
}
|
|
|
|
// codexFirstTurnNoProgressTimeout resolves the first-turn watchdog ceiling:
|
|
// how long the first turn may stay silent after turn/started before that one
|
|
// watchdog fails it. This is only the first-turn timer's own duration — NOT
|
|
// the effective first-item wait. The semantic-inactivity timer is armed
|
|
// concurrently (the same first status:running resets it and starts this one),
|
|
// so the real wait is min(this, semanticInactivityTimeout, execution timeout).
|
|
// Raising the override above the semantic timeout therefore does not extend the
|
|
// wait unless the semantic timeout is raised too; LoadConfig warns when it does
|
|
// not. See the run loop's competing-timer select for the interaction.
|
|
//
|
|
// An explicit configured override (MULTICA_CODEX_FIRST_TURN_TIMEOUT) is honored
|
|
// as-is, upward included: an operator whose app-server is legitimately slow to
|
|
// its first event (a heavy MCP boot, a cold model catalog) can lift this ceiling
|
|
// past the default that the semantic-inactivity timeout alone can never raise
|
|
// (GH #3262 / #5959).
|
|
//
|
|
// With no override the behaviour is byte-identical to before: the default
|
|
// ceiling, which the configured semantic-inactivity timeout can only ever
|
|
// shrink, never raise. The `* 4 / 5` cannot overflow here because that branch
|
|
// only runs when 0 < semanticInactivityTimeout <= defaultCodexFirstTurnNoProgressTimeout.
|
|
func codexFirstTurnNoProgressTimeout(semanticInactivityTimeout, configured time.Duration) time.Duration {
|
|
if configured > 0 {
|
|
return configured
|
|
}
|
|
if semanticInactivityTimeout <= 0 || semanticInactivityTimeout > defaultCodexFirstTurnNoProgressTimeout {
|
|
return defaultCodexFirstTurnNoProgressTimeout
|
|
}
|
|
scaled := semanticInactivityTimeout * 4 / 5
|
|
if scaled <= 0 {
|
|
return semanticInactivityTimeout
|
|
}
|
|
return scaled
|
|
}
|
|
|
|
func isCodexFirstTurnProgressActivity(activity string) bool {
|
|
return activity != "" && activity != "status:running" && activity != "error:retry"
|
|
}
|
|
|
|
func buildCodexTimeoutDiagnosticError(diag codexTimeoutDiagnostic, stderrTail string) string {
|
|
stderrTail = sanitizeCodexDiagnostic(stderrTail)
|
|
var msg string
|
|
switch diag.Kind {
|
|
case codexTimeoutFirstTurnNoProgress:
|
|
msg = fmt.Sprintf("%s after %s: received turn start but no item, message, tool, turn/completed, or error event (%s)",
|
|
CodexFirstTurnNoProgressMarker,
|
|
diag.Timeout,
|
|
formatCodexDiagnosticFields(diag),
|
|
)
|
|
case codexTimeoutSemanticInactivity:
|
|
msg = fmt.Sprintf("%s after %s without agent progress (last activity: %s; %s)",
|
|
CodexSemanticInactivityMarker,
|
|
diag.Timeout,
|
|
nonEmptyCodexDiagnosticValue(diag.LastActivity),
|
|
formatCodexDiagnosticFields(diag),
|
|
)
|
|
default:
|
|
msg = "codex timed out"
|
|
}
|
|
msg = appendCodexKnownStderrHint(msg, stderrTail)
|
|
return withAgentStderr(msg, "codex", stderrTail)
|
|
}
|
|
|
|
func formatCodexDiagnosticFields(diag codexTimeoutDiagnostic) string {
|
|
return fmt.Sprintf("codex_version=%q thread_id=%q turn_id=%q model=%q",
|
|
nonEmptyCodexDiagnosticValue(diag.CodexVersion),
|
|
nonEmptyCodexDiagnosticValue(diag.ThreadID),
|
|
nonEmptyCodexDiagnosticValue(diag.TurnID),
|
|
formatCodexDiagnosticModel(diag.Model),
|
|
)
|
|
}
|
|
|
|
func nonEmptyCodexDiagnosticValue(value string) string {
|
|
if strings.TrimSpace(value) == "" {
|
|
return "unknown"
|
|
}
|
|
return value
|
|
}
|
|
|
|
func formatCodexDiagnosticModel(model string) string {
|
|
if strings.TrimSpace(model) == "" {
|
|
return "default(empty)"
|
|
}
|
|
return model
|
|
}
|
|
|
|
func appendCodexKnownStderrHint(msg, stderrTail string) string {
|
|
if strings.Contains(stderrTail, codexModelCatalogRefreshFailureSignal) {
|
|
return msg + "; diagnosis: Codex could not load its model catalog, which blocks the first turn. This is usually a transient network failure reaching the Codex service. Check network/proxy connectivity and retry the task, or switch to another runtime while the Codex service is unreachable"
|
|
}
|
|
return msg
|
|
}
|
|
|
|
func detectCodexVersionForDiagnostics(ctx context.Context, runtimeCmd Command, env []string, logger *slog.Logger) string {
|
|
versionCtx, cancel := context.WithTimeout(ctx, codexVersionDiagnosticTimeout)
|
|
defer cancel()
|
|
|
|
cmd := runtimeCmd.exec(versionCtx, "--version")
|
|
cmd.Env = env
|
|
data, err := outputOwned(cmd, logger)
|
|
if err != nil {
|
|
if logger != nil {
|
|
logger.Debug("codex version diagnostic failed", "error", err)
|
|
}
|
|
return "unknown"
|
|
}
|
|
version, _ := extractVersionLine(string(data))
|
|
if strings.TrimSpace(version) == "" {
|
|
return "unknown"
|
|
}
|
|
return version
|
|
}
|
|
|
|
func trySendString(ch chan<- string, value string) {
|
|
select {
|
|
case ch <- value:
|
|
default:
|
|
}
|
|
}
|
|
|
|
// codexOutputSelection records what a turn produced that could become
|
|
// Result.Output. The zero value is an empty selection.
|
|
type codexOutputSelection struct {
|
|
// finalAnswer is the last non-blank message labelled `phase: "final_answer"`.
|
|
finalAnswer string
|
|
// finalAnswerObserved is set by any final_answer item, including one with
|
|
// empty or whitespace-only text: the app-server explicitly ended the turn
|
|
// without a deliverable, which is different from never labelling one.
|
|
finalAnswerObserved bool
|
|
lastAgentMessage string
|
|
// lastAgentMessagePhase is the phase label of lastAgentMessage; empty for
|
|
// the legacy protocol and for app-servers that do not label phases.
|
|
lastAgentMessagePhase string
|
|
}
|
|
|
|
func (s *codexOutputSelection) observeAgentMessage(text, phase string) {
|
|
s.lastAgentMessage = text
|
|
s.lastAgentMessagePhase = phase
|
|
}
|
|
|
|
func (s *codexOutputSelection) observeFinalAnswer(text string) {
|
|
s.finalAnswerObserved = true
|
|
if strings.TrimSpace(text) != "" {
|
|
s.finalAnswer = text
|
|
}
|
|
}
|
|
|
|
// codexDeliverableOutput picks Result.Output from what the turn produced: the
|
|
// message the app-server itself labelled `phase: "final_answer"`, or — for the
|
|
// legacy `agent_message` protocol, which carries no phase — the most recent
|
|
// agent message. Never every message joined: the intermediate ones narrate the
|
|
// work between tool calls, which belongs to the transcript and not to a chat or
|
|
// IM reply (GH #6006). A message explicitly labelled with another phase (for
|
|
// example `commentary`) is narration too, so an empty or missing final answer
|
|
// after it yields empty output instead of promoting the narration to the
|
|
// deliverable.
|
|
func codexDeliverableOutput(s codexOutputSelection) string {
|
|
if s.finalAnswer != "" {
|
|
return s.finalAnswer
|
|
}
|
|
if s.finalAnswerObserved || s.lastAgentMessagePhase != "" {
|
|
return ""
|
|
}
|
|
return s.lastAgentMessage
|
|
}
|
|
|
|
func logCodexAgentMessage(logger *slog.Logger, msg Message) {
|
|
if logger == nil {
|
|
return
|
|
}
|
|
attrs := []any{
|
|
"type", string(msg.Type),
|
|
"tool", msg.Tool,
|
|
"call_id", msg.CallID,
|
|
"status", msg.Status,
|
|
"content_len", len(msg.Content),
|
|
"output_len", len(msg.Output),
|
|
}
|
|
logger.Info("codex agent message received", attrs...)
|
|
if msg.Type == MessageToolResult {
|
|
logger.Info("codex tool_result observed", "tool", msg.Tool, "call_id", msg.CallID, "output_len", len(msg.Output))
|
|
}
|
|
}
|
|
|
|
func describeCodexSemanticActivity(msg Message) string {
|
|
switch msg.Type {
|
|
case MessageToolUse, MessageToolResult:
|
|
if msg.Tool != "" {
|
|
return fmt.Sprintf("%s:%s", msg.Type, msg.Tool)
|
|
}
|
|
case MessageStatus:
|
|
if msg.Status != "" {
|
|
return fmt.Sprintf("%s:%s", msg.Type, msg.Status)
|
|
}
|
|
}
|
|
return string(msg.Type)
|
|
}
|
|
|
|
// ── codexClient: JSON-RPC 2.0 transport ──
|
|
|
|
type codexClient struct {
|
|
cfg Config
|
|
stdin interface{ Write([]byte) (int, error) }
|
|
mu sync.Mutex
|
|
nextID int
|
|
pending map[int]*pendingRPC
|
|
processDone chan struct{}
|
|
processErr error
|
|
handshakeTimeout time.Duration
|
|
threadHandshakeTimeout time.Duration
|
|
pid int
|
|
attempt int
|
|
activeLaunches int64
|
|
threadSetupMethod string
|
|
threadSetupStarted time.Time
|
|
threadIDMu sync.RWMutex
|
|
threadID string
|
|
turnIDMu sync.RWMutex
|
|
lastTurnID string
|
|
turnID string
|
|
onMessage func(Message)
|
|
// onAgentMessageChunk reports whether a text chunk was handed to the
|
|
// daemon-facing message channel. Raw delta reconciliation advances only on
|
|
// true, so channel pressure cannot turn observed provider bytes into a false
|
|
// delivered prefix. Unit clients may leave this nil and use onMessage.
|
|
onAgentMessageChunk func(text string) bool
|
|
// onAgentMessage receives the authoritative completed text for one agent
|
|
// message and its phase label ("" when the protocol carries none).
|
|
// onAgentMessageChunk may deliver that text incrementally, so
|
|
// Result.Output fallbacks must use this callback rather than the last chunk.
|
|
onAgentMessage func(text, phase string)
|
|
onSemanticActivity func(description string)
|
|
onTurnDone func(aborted bool)
|
|
guardianRateLimitCircuit *codexGuardianRateLimitCircuit
|
|
onGuardianRateLimit func()
|
|
// onFinalAnswer fires only for an agent message the app-server itself
|
|
// labelled `phase: "final_answer"` — the turn's deliverable, as opposed to
|
|
// the intermediate agent messages that narrate work between tool calls.
|
|
// It fires for an empty final answer too, so the caller can tell "ended
|
|
// without a deliverable" from "never labelled one".
|
|
// Both still reach onMessage: the transcript keeps the whole timeline, only
|
|
// Result.Output is narrowed to the deliverable (GH #6006).
|
|
onFinalAnswer func(text string)
|
|
// acceptNotification isolates the active turn from same-thread history
|
|
// replay emitted while thread/resume is restoring prior conversation.
|
|
// Unit-level protocol tests leave it nil and exercise dispatch directly.
|
|
acceptNotification func(method string, params map[string]any) bool
|
|
// onDiscardedNotification preserves out-of-band safety signals (such as
|
|
// suppressing an initialize retry after observed activity) without letting
|
|
// filtered history mutate current-turn output or lifecycle state.
|
|
onDiscardedNotification func(method string, params map[string]any)
|
|
// agentMessageStreams tracks raw-v2 agent text on the single stdout reader.
|
|
// The first delta is handed off immediately; subsequent deltas are aggregated
|
|
// into bounded chunks and reconciled against item/completed. delivered is
|
|
// advanced only after onAgentMessageChunk confirms daemon handoff.
|
|
agentMessageStreams map[string]*codexAgentMessageStream
|
|
agentMessageOrder []string
|
|
|
|
notificationProtocol string // "unknown", "legacy", "raw"
|
|
turnCompleted bool
|
|
|
|
usageMu sync.Mutex
|
|
usage TokenUsage // accumulated from turn events
|
|
// usageTotal is the last accepted cumulative v2 usage snapshot for the
|
|
// active turn. Comparing later snapshots against it makes replay and
|
|
// duplicate delivery idempotent while retaining multi-response turns.
|
|
usageTotal codexRawTokenUsage
|
|
usageTotalSet bool
|
|
|
|
turnErrorMu sync.Mutex
|
|
turnError string // captured from turn/completed status=failed or terminal error notifications
|
|
}
|
|
|
|
type codexAgentMessageStream struct {
|
|
delivered strings.Builder
|
|
pending strings.Builder
|
|
}
|
|
|
|
// codexTurnNotificationGate keeps resume-time history replay from mutating the
|
|
// output or ending the new turn. Codex app-server can emit notifications before
|
|
// the turn/start RPC response, so the gate is armed before that request and uses
|
|
// turn/started (or legacy task_started) as the actual current-turn boundary.
|
|
// Protocols that omit those start events also omit a reliable current-turn ID;
|
|
// for compatibility, their post-arm events remain accepted. We therefore rely
|
|
// on the single stdout reader's ordering guarantee that resume history emitted
|
|
// before the thread/resume response is processed while the gate is still closed.
|
|
// A replay emitted after arm but before a start event cannot be distinguished
|
|
// from a valid legacy current-turn event without breaking those older streams.
|
|
// Its mutable lifecycle fields are only touched by the stdout reader goroutine;
|
|
// armed is atomic because the lifecycle goroutine flips it.
|
|
type codexTurnNotificationGate struct {
|
|
armed atomic.Bool
|
|
started bool
|
|
turnID string
|
|
}
|
|
|
|
func (g *codexTurnNotificationGate) arm() {
|
|
g.armed.Store(true)
|
|
}
|
|
|
|
func (g *codexTurnNotificationGate) accept(method string, params map[string]any) bool {
|
|
if !g.armed.Load() {
|
|
return false
|
|
}
|
|
|
|
if method == "codex/event" || strings.HasPrefix(method, "codex/event/") {
|
|
msg, _ := params["msg"].(map[string]any)
|
|
msgType, _ := msg["type"].(string)
|
|
if msgType == "task_started" {
|
|
g.started = true
|
|
return true
|
|
}
|
|
// Older Codex event streams can omit task_started. Once turn/start is
|
|
// armed, keep that compatibility; pre-arm replay is still excluded.
|
|
return true
|
|
}
|
|
|
|
switch {
|
|
case method == "turn/started":
|
|
g.started = true
|
|
g.turnID = extractNestedString(params, "turn", "id")
|
|
return true
|
|
case method == "turn/completed":
|
|
if !g.started {
|
|
// Older app-server versions can complete a turn without first
|
|
// emitting turn/started. The pre-arm boundary still rejects resume
|
|
// replay, while this keeps those versions functional.
|
|
return true
|
|
}
|
|
turnID := extractNestedString(params, "turn", "id")
|
|
return g.turnID == "" || turnID == "" || turnID == g.turnID
|
|
case method == "thread/status/changed" || strings.HasPrefix(method, "item/"):
|
|
if !g.started {
|
|
return true
|
|
}
|
|
turnID, _ := params["turnId"].(string)
|
|
return g.turnID == "" || turnID == "" || turnID == g.turnID
|
|
default:
|
|
// A terminal error may be the first notification produced by a failed
|
|
// turn/start, so it must remain observable even without turn/started.
|
|
return true
|
|
}
|
|
}
|
|
|
|
func (c *codexClient) setTurnError(msg string) {
|
|
if msg == "" {
|
|
return
|
|
}
|
|
c.turnErrorMu.Lock()
|
|
defer c.turnErrorMu.Unlock()
|
|
if c.turnError == "" {
|
|
c.turnError = msg
|
|
}
|
|
}
|
|
|
|
func (c *codexClient) getTurnError() string {
|
|
c.turnErrorMu.Lock()
|
|
defer c.turnErrorMu.Unlock()
|
|
return c.turnError
|
|
}
|
|
|
|
// setThreadID publishes the thread ID resolved by thread/start or
|
|
// thread/resume. The task goroutine writes it while the stdout goroutine is
|
|
// already dispatching notifications for that same thread, so the field needs
|
|
// the same synchronization as turnID (GH #8422).
|
|
func (c *codexClient) setThreadID(threadID string) {
|
|
c.threadIDMu.Lock()
|
|
c.threadID = threadID
|
|
c.threadIDMu.Unlock()
|
|
}
|
|
|
|
func (c *codexClient) getThreadID() string {
|
|
c.threadIDMu.RLock()
|
|
defer c.threadIDMu.RUnlock()
|
|
return c.threadID
|
|
}
|
|
|
|
func (c *codexClient) setActiveTurnID(turnID string) {
|
|
c.turnIDMu.Lock()
|
|
if turnID != "" {
|
|
c.lastTurnID = turnID
|
|
}
|
|
c.turnID = turnID
|
|
c.turnIDMu.Unlock()
|
|
}
|
|
|
|
// usageTurnID retains attribution after the turn closes its steering window.
|
|
func (c *codexClient) usageTurnID() string {
|
|
c.turnIDMu.RLock()
|
|
defer c.turnIDMu.RUnlock()
|
|
return c.lastTurnID
|
|
}
|
|
|
|
func (c *codexClient) activeTurnID() string {
|
|
c.turnIDMu.RLock()
|
|
defer c.turnIDMu.RUnlock()
|
|
return c.turnID
|
|
}
|
|
|
|
type pendingRPC struct {
|
|
ch chan rpcResult
|
|
method string
|
|
}
|
|
|
|
type rpcResult struct {
|
|
result json.RawMessage
|
|
err error
|
|
}
|
|
|
|
type codexHandshakeTimeoutError struct {
|
|
Method string
|
|
Timeout time.Duration
|
|
}
|
|
|
|
func (e *codexHandshakeTimeoutError) Error() string {
|
|
return fmt.Sprintf("%s: %s did not respond after %s", CodexHandshakeTimeoutMarker, e.Method, e.Timeout)
|
|
}
|
|
|
|
func (e *codexHandshakeTimeoutError) Unwrap() error {
|
|
return context.DeadlineExceeded
|
|
}
|
|
|
|
func isCodexHandshakeRPC(method string) bool {
|
|
switch method {
|
|
case "initialize", "thread/start", "thread/resume", "thread/name/set", "turn/start":
|
|
return true
|
|
default:
|
|
return false
|
|
}
|
|
}
|
|
|
|
func isCodexThreadSetupRPC(method string) bool {
|
|
return method == "thread/start" || method == "thread/resume"
|
|
}
|
|
|
|
func (c *codexClient) handshakeTimeoutFor(method string) time.Duration {
|
|
switch method {
|
|
case "thread/start", "thread/resume":
|
|
if c.threadHandshakeTimeout > 0 {
|
|
return c.threadHandshakeTimeout
|
|
}
|
|
return c.handshakeTimeout
|
|
default:
|
|
return c.handshakeTimeout
|
|
}
|
|
}
|
|
|
|
func codexRequestContextError(ctx context.Context) error {
|
|
var handshakeErr *codexHandshakeTimeoutError
|
|
if errors.As(context.Cause(ctx), &handshakeErr) {
|
|
return handshakeErr
|
|
}
|
|
return ctx.Err()
|
|
}
|
|
|
|
// interruptCodexTurn follows the app-server's native stop protocol before the
|
|
// cancelled execution tears down its process. Current app-server versions emit
|
|
// final usage in thread/tokenUsage/updated immediately before turn/completed,
|
|
// so an immediate process-group SIGKILL loses otherwise valid local token
|
|
// accounting for cancelled/replaced tasks. The bool reports whether the turn
|
|
// reached that authoritative completion boundary; callers must force-stop the
|
|
// process tree when it is false.
|
|
func interruptCodexTurn(c *codexClient, threadID string, turnDone <-chan bool, configuredTimeout time.Duration, logger *slog.Logger) bool {
|
|
select {
|
|
case <-turnDone:
|
|
// Completion won the cancellation race; usage has already been captured.
|
|
return true
|
|
default:
|
|
}
|
|
|
|
turnID := c.activeTurnID()
|
|
if threadID == "" || turnID == "" {
|
|
return false
|
|
}
|
|
if logger == nil {
|
|
logger = slog.Default()
|
|
}
|
|
interruptTimeout := configuredTimeout
|
|
if interruptTimeout <= 0 {
|
|
interruptTimeout = defaultCodexTurnInterruptTimeout
|
|
}
|
|
started := time.Now()
|
|
|
|
interruptCtx, cancelInterrupt := context.WithTimeout(context.Background(), interruptTimeout)
|
|
defer cancelInterrupt()
|
|
_, err := c.request(interruptCtx, "turn/interrupt", map[string]any{
|
|
"threadId": threadID,
|
|
"turnId": turnID,
|
|
})
|
|
if err != nil {
|
|
logger.Warn("codex turn interrupt failed", "thread_id", threadID, "turn_id", turnID, "error", err, "latency", time.Since(started).Round(time.Millisecond).String(), "timeout", interruptTimeout.String())
|
|
return false
|
|
}
|
|
|
|
select {
|
|
case <-turnDone:
|
|
logger.Info("codex turn interrupted", "thread_id", threadID, "turn_id", turnID, "latency", time.Since(started).Round(time.Millisecond).String(), "timeout", interruptTimeout.String())
|
|
return true
|
|
case <-interruptCtx.Done():
|
|
logger.Warn("codex turn interrupt completion timed out", "thread_id", threadID, "turn_id", turnID, "latency", time.Since(started).Round(time.Millisecond).String(), "timeout", interruptTimeout.String())
|
|
return false
|
|
}
|
|
}
|
|
|
|
func (c *codexClient) request(ctx context.Context, method string, params any) (json.RawMessage, error) {
|
|
if err := ctx.Err(); err != nil {
|
|
return nil, err
|
|
}
|
|
requestCtx := ctx
|
|
cancelRequest := func() {}
|
|
if timeout := c.handshakeTimeoutFor(method); timeout > 0 && isCodexHandshakeRPC(method) {
|
|
timeoutErr := &codexHandshakeTimeoutError{Method: method, Timeout: timeout}
|
|
requestCtx, cancelRequest = context.WithTimeoutCause(ctx, timeout, timeoutErr)
|
|
}
|
|
defer cancelRequest()
|
|
|
|
c.mu.Lock()
|
|
if c.processErr != nil {
|
|
err := c.processErr
|
|
c.mu.Unlock()
|
|
return nil, err
|
|
}
|
|
if c.processDone == nil {
|
|
c.processDone = make(chan struct{})
|
|
}
|
|
processDone := c.processDone
|
|
c.nextID++
|
|
id := c.nextID
|
|
pr := &pendingRPC{ch: make(chan rpcResult, 1), method: method}
|
|
c.pending[id] = pr
|
|
c.mu.Unlock()
|
|
|
|
msg := map[string]any{
|
|
"jsonrpc": "2.0",
|
|
"id": id,
|
|
"method": method,
|
|
"params": params,
|
|
}
|
|
data, err := json.Marshal(msg)
|
|
if err != nil {
|
|
c.mu.Lock()
|
|
delete(c.pending, id)
|
|
c.mu.Unlock()
|
|
return nil, err
|
|
}
|
|
data = append(data, '\n')
|
|
if _, err := c.stdin.Write(data); err != nil {
|
|
c.mu.Lock()
|
|
delete(c.pending, id)
|
|
c.mu.Unlock()
|
|
return nil, fmt.Errorf("write %s: %w", method, err)
|
|
}
|
|
if method == "turn/start" {
|
|
threadID := ""
|
|
if paramMap, ok := params.(map[string]any); ok {
|
|
threadID, _ = paramMap["threadId"].(string)
|
|
}
|
|
c.cfg.Logger.Info("codex turn/start sent", "request_id", id, "thread_id", threadID)
|
|
}
|
|
|
|
select {
|
|
case res := <-pr.ch:
|
|
return res.result, res.err
|
|
case <-processDone:
|
|
select {
|
|
case res := <-pr.ch:
|
|
return res.result, res.err
|
|
default:
|
|
}
|
|
c.mu.Lock()
|
|
delete(c.pending, id)
|
|
err := c.processErr
|
|
c.mu.Unlock()
|
|
if requestCtx.Err() != nil {
|
|
return nil, codexRequestContextError(requestCtx)
|
|
}
|
|
if err == nil {
|
|
err = errCodexProcessExited
|
|
}
|
|
return nil, err
|
|
case <-requestCtx.Done():
|
|
c.mu.Lock()
|
|
delete(c.pending, id)
|
|
c.mu.Unlock()
|
|
return nil, codexRequestContextError(requestCtx)
|
|
}
|
|
}
|
|
|
|
func (c *codexClient) notify(method string) {
|
|
msg := map[string]any{
|
|
"jsonrpc": "2.0",
|
|
"method": method,
|
|
}
|
|
data, _ := json.Marshal(msg)
|
|
data = append(data, '\n')
|
|
_, _ = c.stdin.Write(data)
|
|
}
|
|
|
|
func (c *codexClient) respond(id int, result any) {
|
|
msg := map[string]any{
|
|
"jsonrpc": "2.0",
|
|
"id": id,
|
|
"result": result,
|
|
}
|
|
data, _ := json.Marshal(msg)
|
|
data = append(data, '\n')
|
|
_, _ = c.stdin.Write(data)
|
|
}
|
|
|
|
func (c *codexClient) respondError(id int, code int, message string) {
|
|
msg := map[string]any{
|
|
"jsonrpc": "2.0",
|
|
"id": id,
|
|
"error": map[string]any{
|
|
"code": code,
|
|
"message": message,
|
|
},
|
|
}
|
|
data, _ := json.Marshal(msg)
|
|
data = append(data, '\n')
|
|
_, _ = c.stdin.Write(data)
|
|
}
|
|
|
|
func (c *codexClient) closeAllPending(err error) {
|
|
c.mu.Lock()
|
|
defer c.mu.Unlock()
|
|
for id, pr := range c.pending {
|
|
pr.ch <- rpcResult{err: err}
|
|
delete(c.pending, id)
|
|
}
|
|
}
|
|
|
|
func (c *codexClient) markProcessExited(err error) {
|
|
if err == nil {
|
|
err = errCodexProcessExited
|
|
}
|
|
c.mu.Lock()
|
|
defer c.mu.Unlock()
|
|
if c.processErr == nil {
|
|
c.processErr = err
|
|
if c.processDone != nil {
|
|
close(c.processDone)
|
|
}
|
|
}
|
|
for id, pr := range c.pending {
|
|
pr.ch <- rpcResult{err: err}
|
|
delete(c.pending, id)
|
|
}
|
|
}
|
|
|
|
func (c *codexClient) getProcessErr() error {
|
|
c.mu.Lock()
|
|
defer c.mu.Unlock()
|
|
return c.processErr
|
|
}
|
|
|
|
// isCodexResumeOverflow reports whether a failed startOrResumeThread failed
|
|
// because the thread/resume RESPONSE did not fit in our stdout line buffer.
|
|
//
|
|
// Codex serializes the entire thread into that single response, so a long
|
|
// enough thread overflows agentStreamMaxLineBytes and the reader goroutine
|
|
// dies with bufio.ErrTooLong. Every other transport error means codex is
|
|
// gone; this one means codex is fine and we cannot read it. The distinction
|
|
// is what makes it a resume REJECTION: the thread cannot be handed to us at
|
|
// all, so only starting over can cure it — exactly what Result.ResumeRejected
|
|
// documents, and the evidence shouldRetryWithFreshSession looks for first.
|
|
//
|
|
// Recovery has to go through the daemon rather than a local thread/start
|
|
// fallback. By the time we get here the reader goroutine has exited and codex
|
|
// is blocked writing the rest of the oversized line into a pipe nobody drains,
|
|
// so this process can no longer answer any RPC. The daemon's fresh-session
|
|
// retry re-execs codex, which is the only way back (MUL-5722).
|
|
//
|
|
// Requires ResumeSessionID: an overflow on a thread/start response is a
|
|
// different failure and nothing about the session pointer would fix it.
|
|
func isCodexResumeOverflow(opts ExecOptions, err error) bool {
|
|
return opts.ResumeSessionID != "" && errors.Is(err, bufio.ErrTooLong)
|
|
}
|
|
|
|
func isCodexTransportError(err error) bool {
|
|
if err == nil {
|
|
return false
|
|
}
|
|
if errors.Is(err, errCodexProcessExited) {
|
|
return true
|
|
}
|
|
var handshakeErr *codexHandshakeTimeoutError
|
|
if errors.As(err, &handshakeErr) {
|
|
return true
|
|
}
|
|
return strings.HasPrefix(err.Error(), "write ")
|
|
}
|
|
|
|
func (c *codexClient) handleLine(line string) {
|
|
var raw map[string]json.RawMessage
|
|
if err := json.Unmarshal([]byte(line), &raw); err != nil {
|
|
return
|
|
}
|
|
|
|
// Check if it's a response to our request
|
|
if _, hasID := raw["id"]; hasID {
|
|
if _, hasResult := raw["result"]; hasResult {
|
|
c.handleResponse(raw)
|
|
return
|
|
}
|
|
if _, hasError := raw["error"]; hasError {
|
|
c.handleResponse(raw)
|
|
return
|
|
}
|
|
// Server request (has id + method)
|
|
if _, hasMethod := raw["method"]; hasMethod {
|
|
c.handleServerRequest(raw)
|
|
return
|
|
}
|
|
}
|
|
|
|
// Notification (no id, has method)
|
|
if _, hasMethod := raw["method"]; hasMethod {
|
|
c.handleNotification(raw)
|
|
}
|
|
}
|
|
|
|
func (c *codexClient) handleResponse(raw map[string]json.RawMessage) {
|
|
var id int
|
|
if err := json.Unmarshal(raw["id"], &id); err != nil {
|
|
return
|
|
}
|
|
|
|
c.mu.Lock()
|
|
pr, ok := c.pending[id]
|
|
if ok {
|
|
delete(c.pending, id)
|
|
}
|
|
c.mu.Unlock()
|
|
|
|
if !ok {
|
|
return
|
|
}
|
|
|
|
if errData, hasErr := raw["error"]; hasErr {
|
|
var rpcErr struct {
|
|
Code int `json:"code"`
|
|
Message string `json:"message"`
|
|
}
|
|
_ = json.Unmarshal(errData, &rpcErr)
|
|
pr.ch <- rpcResult{err: fmt.Errorf("%s: %s (code=%d)", pr.method, rpcErr.Message, rpcErr.Code)}
|
|
} else {
|
|
pr.ch <- rpcResult{result: raw["result"]}
|
|
}
|
|
}
|
|
|
|
func (c *codexClient) handleServerRequest(raw map[string]json.RawMessage) {
|
|
var id int
|
|
_ = json.Unmarshal(raw["id"], &id)
|
|
|
|
var method string
|
|
_ = json.Unmarshal(raw["method"], &method)
|
|
|
|
// Auto-approve all exec/patch requests in daemon mode
|
|
switch method {
|
|
case "item/commandExecution/requestApproval", "execCommandApproval":
|
|
c.respond(id, map[string]any{"decision": "accept"})
|
|
case "item/fileChange/requestApproval", "applyPatchApproval":
|
|
c.respond(id, map[string]any{"decision": "accept"})
|
|
case "item/permissions/requestApproval":
|
|
c.respond(id, codexPermissionsApprovalResponse(raw["params"], c.cfg.Logger))
|
|
case "mcpServer/elicitation/request":
|
|
c.respond(id, map[string]any{"action": "accept", "content": nil, "_meta": nil})
|
|
default:
|
|
msg := fmt.Sprintf("unsupported codex app-server request: %s", method)
|
|
c.cfg.Logger.Warn("codex: unhandled server request", "method", method, "id", id)
|
|
c.setTurnError(msg)
|
|
c.respondError(id, -32601, msg)
|
|
}
|
|
}
|
|
|
|
// codexPermissionsApprovalResponse builds the auto-grant reply for a Codex
|
|
// item/permissions/requestApproval server request. In daemon mode there is no
|
|
// human to approve, so we echo back the requested network / fileSystem profile
|
|
// and scope it to the current turn, mirroring the other auto-accept branches in
|
|
// handleServerRequest.
|
|
//
|
|
// The grant is intentionally limited to the network / fileSystem keys we
|
|
// understand. A parse failure and any dropped key are logged so that a future
|
|
// app-server protocol that adds a new permission shape is visible in daemon
|
|
// logs instead of being silently narrowed away.
|
|
func codexPermissionsApprovalResponse(params json.RawMessage, logger *slog.Logger) map[string]any {
|
|
var payload struct {
|
|
Permissions map[string]any `json:"permissions"`
|
|
}
|
|
if err := json.Unmarshal(params, &payload); err != nil && logger != nil {
|
|
logger.Warn("codex: failed to parse permission approval request; granting empty turn-scoped profile", "error", err)
|
|
}
|
|
|
|
granted := map[string]any{}
|
|
var dropped []string
|
|
for key, value := range payload.Permissions {
|
|
switch key {
|
|
case "network", "fileSystem":
|
|
if value != nil {
|
|
granted[key] = value
|
|
}
|
|
default:
|
|
dropped = append(dropped, key)
|
|
}
|
|
}
|
|
if len(dropped) > 0 && logger != nil {
|
|
sort.Strings(dropped)
|
|
logger.Warn("codex: dropping unrecognized permission keys from approval request; add explicit handling if the app-server protocol expanded", "keys", dropped)
|
|
}
|
|
|
|
return map[string]any{
|
|
"permissions": granted,
|
|
"scope": "turn",
|
|
}
|
|
}
|
|
|
|
func (c *codexClient) handleNotification(raw map[string]json.RawMessage) {
|
|
var method string
|
|
_ = json.Unmarshal(raw["method"], &method)
|
|
|
|
var params map[string]any
|
|
if p, ok := raw["params"]; ok {
|
|
_ = json.Unmarshal(p, ¶ms)
|
|
}
|
|
// Filter multiplexed subagent threads before the current-turn gate. The
|
|
// gate mutates started/turnID on turn/started; letting another thread reach
|
|
// it first can replace the main turn ID and make subsequent main-thread
|
|
// items/completion look stale.
|
|
if c.isNotificationFromOtherThread(params) {
|
|
return
|
|
}
|
|
if c.acceptNotification != nil && !c.acceptNotification(method, params) {
|
|
if c.onDiscardedNotification != nil {
|
|
c.onDiscardedNotification(method, params)
|
|
}
|
|
return
|
|
}
|
|
|
|
// Legacy codex/event notifications
|
|
if method == "codex/event" || strings.HasPrefix(method, "codex/event/") {
|
|
c.notificationProtocol = "legacy"
|
|
msgData, ok := params["msg"]
|
|
if !ok {
|
|
return
|
|
}
|
|
msgMap, ok := msgData.(map[string]any)
|
|
if !ok {
|
|
return
|
|
}
|
|
c.handleEvent(msgMap)
|
|
return
|
|
}
|
|
|
|
// Raw v2 notifications
|
|
if c.notificationProtocol != "legacy" {
|
|
if c.notificationProtocol == "unknown" &&
|
|
(method == "turn/started" || method == "turn/completed" ||
|
|
method == "thread/started" || method == "error" || strings.HasPrefix(method, "item/")) {
|
|
c.notificationProtocol = "raw"
|
|
}
|
|
|
|
if c.notificationProtocol == "raw" {
|
|
c.handleRawNotification(method, params)
|
|
}
|
|
}
|
|
}
|
|
|
|
// codexPatchInputMaxBytes bounds the total diff/content bytes recorded for a
|
|
// single patch_apply event.
|
|
//
|
|
// The bound exists because the legacy protocol reports `add` and `delete` as
|
|
// whole-file contents rather than diffs, so a single generated file can be
|
|
// arbitrarily large. It is deliberately applied only to this new Codex payload:
|
|
// other providers already stream tool inputs through unbounded, and clamping
|
|
// them here would silently truncate transcripts that render correctly today.
|
|
// Unifying the limit at the persistence boundary is a separate change.
|
|
const codexPatchInputMaxBytes = 64 * 1024
|
|
|
|
// codexNormalizeLegacyChanges converts a legacy `patch_apply_*` changes map
|
|
// into the normalized change list.
|
|
//
|
|
// The legacy wire shape is map[path]FileChange, where FileChange is internally
|
|
// tagged on `type` and carries *different* payload fields per variant:
|
|
// add/delete carry whole-file `content`, while update carries `unified_diff`
|
|
// plus an optional `move_path`. Nothing in that shape is a plain unified diff
|
|
// for every case, which is why the normalized form keeps `diff` and `content`
|
|
// as alternatives instead of one field.
|
|
func codexNormalizeLegacyChanges(raw any) []any {
|
|
changes, ok := raw.(map[string]any)
|
|
if !ok || len(changes) == 0 {
|
|
return nil
|
|
}
|
|
|
|
// Go randomizes map iteration; sort so a replayed event yields a stable
|
|
// transcript instead of reshuffling the file list on every run.
|
|
paths := make([]string, 0, len(changes))
|
|
for path := range changes {
|
|
paths = append(paths, path)
|
|
}
|
|
sort.Strings(paths)
|
|
|
|
out := make([]any, 0, len(paths))
|
|
for _, path := range paths {
|
|
entry, ok := changes[path].(map[string]any)
|
|
if !ok {
|
|
continue
|
|
}
|
|
norm := map[string]any{"path": path}
|
|
if kind, _ := entry["type"].(string); kind != "" {
|
|
norm["kind"] = kind
|
|
}
|
|
if diff, _ := entry["unified_diff"].(string); diff != "" {
|
|
norm["diff"] = diff
|
|
}
|
|
// Presence, not non-emptiness: an added empty file has content "" and
|
|
// should still render as an empty body rather than a missing one.
|
|
if content, ok := entry["content"].(string); ok {
|
|
norm["content"] = content
|
|
}
|
|
if movePath, _ := entry["move_path"].(string); movePath != "" {
|
|
norm["move_path"] = movePath
|
|
}
|
|
out = append(out, norm)
|
|
}
|
|
if len(out) == 0 {
|
|
return nil
|
|
}
|
|
return out
|
|
}
|
|
|
|
// codexNormalizeRawChanges converts a v2 app-server `fileChange` item's
|
|
// changes array into the normalized change list.
|
|
//
|
|
// Unlike the legacy shape this is an ordered array of
|
|
// FileUpdateChange{path, kind, diff}, and `kind` is an *object*
|
|
// ({type, move_path?}) rather than a string — reading it as a string yields ""
|
|
// and silently loses the add/delete/update distinction.
|
|
//
|
|
// The `diff` field is misleadingly named: upstream's format_file_change_diff
|
|
// only produces a unified diff for `update`. For `add` and `delete` it returns
|
|
// the whole file's contents verbatim, and for a moved `update` it appends a
|
|
// trailing "\n\nMoved to: <path>" line. Feeding an added file's contents to a
|
|
// diff parser mislabels every line — and actively inverts the meaning of any
|
|
// line that happens to begin with '+' or '-' — so the payload is routed by
|
|
// `kind`, not by field name.
|
|
func codexNormalizeRawChanges(raw any) []any {
|
|
changes, ok := raw.([]any)
|
|
if !ok || len(changes) == 0 {
|
|
return nil
|
|
}
|
|
out := make([]any, 0, len(changes))
|
|
for _, change := range changes {
|
|
entry, ok := change.(map[string]any)
|
|
if !ok {
|
|
continue
|
|
}
|
|
norm := map[string]any{}
|
|
if path, _ := entry["path"].(string); path != "" {
|
|
norm["path"] = path
|
|
}
|
|
kind := ""
|
|
movePath := ""
|
|
switch k := entry["kind"].(type) {
|
|
case map[string]any:
|
|
kind, _ = k["type"].(string)
|
|
movePath, _ = k["move_path"].(string)
|
|
case string:
|
|
// Tolerate a flattened form so a future protocol tweak degrades
|
|
// to a labelled change instead of an unlabelled one.
|
|
kind = k
|
|
}
|
|
if kind != "" {
|
|
norm["kind"] = kind
|
|
}
|
|
if movePath != "" {
|
|
norm["move_path"] = movePath
|
|
}
|
|
|
|
body, bodyPresent := entry["diff"].(string)
|
|
switch kind {
|
|
case "add", "delete":
|
|
// Whole-file contents, despite arriving under `diff`. Keyed on the
|
|
// field being present rather than non-empty so an empty added file
|
|
// still renders as an empty body instead of a missing one.
|
|
if bodyPresent {
|
|
norm["content"] = body
|
|
} else if content, ok := entry["content"].(string); ok {
|
|
norm["content"] = content
|
|
}
|
|
default:
|
|
if body != "" {
|
|
norm["diff"] = stripCodexMovedToSuffix(body, movePath)
|
|
} else if content, ok := entry["content"].(string); ok {
|
|
norm["content"] = content
|
|
}
|
|
}
|
|
if len(norm) == 0 {
|
|
continue
|
|
}
|
|
out = append(out, norm)
|
|
}
|
|
if len(out) == 0 {
|
|
return nil
|
|
}
|
|
return out
|
|
}
|
|
|
|
// stripCodexMovedToSuffix removes the trailing "Moved to: <path>" line that
|
|
// upstream appends to a moved file's unified diff. The destination is already
|
|
// carried as move_path, and leaving the sentence in would render as two stray
|
|
// context lines at the end of the diff.
|
|
func stripCodexMovedToSuffix(diff, movePath string) string {
|
|
if movePath == "" {
|
|
return diff
|
|
}
|
|
suffix := "\n\nMoved to: " + movePath
|
|
return strings.TrimSuffix(diff, suffix)
|
|
}
|
|
|
|
// codexPatchInput wraps normalized changes into the tool_use input, applying
|
|
// the size bound. It returns nil when nothing could be normalized so a
|
|
// malformed or unrecognised event degrades to the previous payload-less
|
|
// behaviour rather than breaking the transcript.
|
|
func codexPatchInput(changes []any) map[string]any {
|
|
if len(changes) == 0 {
|
|
return nil
|
|
}
|
|
// Measure before anything rewrites the bodies, so the figure reported to the
|
|
// reader is the size of the patch they actually produced.
|
|
originalBytes := codexPatchBodyBytes(changes)
|
|
|
|
// Redaction must run BEFORE the size budget, never after it. Several rules
|
|
// only match a credential as a whole — the PEM rule needs both the BEGIN
|
|
// and the END marker — so cutting a body at the budget can strand the
|
|
// opening half of a private key in text that no pattern will match again.
|
|
// The daemon and the server each redact further down the path, and neither
|
|
// can recover a secret that truncation has already made unrecognisable.
|
|
safe, ok := redact.InputMap(map[string]any{"changes": changes})["changes"].([]any)
|
|
if !ok {
|
|
safe = changes
|
|
}
|
|
|
|
input := map[string]any{"changes": safe}
|
|
// Budget the redacted bodies, since those are what gets stored. Redaction
|
|
// usually shrinks a body — a whole key collapses to a short placeholder —
|
|
// so this is also the measurement that decides whether trimming is needed
|
|
// at all.
|
|
if codexPatchBodyBytes(safe) > codexPatchInputMaxBytes {
|
|
applyCodexPatchBudget(safe)
|
|
input["truncated"] = true
|
|
input["original_bytes"] = originalBytes
|
|
}
|
|
return input
|
|
}
|
|
|
|
// codexPatchBodyBytes totals the diff/content payload carried by a change list.
|
|
func codexPatchBodyBytes(changes []any) int {
|
|
total := 0
|
|
for _, change := range changes {
|
|
entry, ok := change.(map[string]any)
|
|
if !ok {
|
|
continue
|
|
}
|
|
for _, key := range []string{"diff", "content"} {
|
|
if body, _ := entry[key].(string); body != "" {
|
|
total += len(body)
|
|
}
|
|
}
|
|
}
|
|
return total
|
|
}
|
|
|
|
// applyCodexPatchBudget trims diff/content bodies in place until the payload
|
|
// fits codexPatchInputMaxBytes. Paths and kinds are never dropped: they are
|
|
// small and are the part a reviewer needs even when the body is gone.
|
|
func applyCodexPatchBudget(changes []any) {
|
|
remaining := codexPatchInputMaxBytes
|
|
for _, change := range changes {
|
|
entry, ok := change.(map[string]any)
|
|
if !ok {
|
|
continue
|
|
}
|
|
for _, key := range []string{"diff", "content"} {
|
|
body, _ := entry[key].(string)
|
|
if body == "" {
|
|
continue
|
|
}
|
|
if len(body) <= remaining {
|
|
remaining -= len(body)
|
|
continue
|
|
}
|
|
if kept := truncateUTF8(body, remaining); kept != "" {
|
|
entry[key] = kept
|
|
} else {
|
|
delete(entry, key)
|
|
}
|
|
entry["truncated"] = true
|
|
remaining = 0
|
|
}
|
|
}
|
|
}
|
|
|
|
// truncateUTF8 cuts s to at most max bytes without leaving a split rune, so the
|
|
// result still marshals as valid JSON.
|
|
func truncateUTF8(s string, max int) string {
|
|
if max <= 0 {
|
|
return ""
|
|
}
|
|
if len(s) <= max {
|
|
return s
|
|
}
|
|
return strings.ToValidUTF8(s[:max], "")
|
|
}
|
|
|
|
// codexNormalizePatchStatus maps both protocols' status spellings onto one
|
|
// snake_case vocabulary: the legacy enum is already snake_case
|
|
// (completed/failed/declined) while the v2 enum is camelCase and adds
|
|
// inProgress.
|
|
func codexNormalizePatchStatus(status string) string {
|
|
switch status {
|
|
case "":
|
|
return ""
|
|
case "inProgress", "in_progress":
|
|
return "in_progress"
|
|
default:
|
|
return status
|
|
}
|
|
}
|
|
|
|
// codexPatchResultOutput renders the tool_result line for a finished patch.
|
|
// It always produces a non-empty string when anything is known, because an
|
|
// empty output renders as an unexpandable blank row in the transcript.
|
|
func codexPatchResultOutput(status string, changes []any, stdout, stderr string) string {
|
|
segments := make([]string, 0, 3)
|
|
|
|
if headline := codexPatchHeadline(status, len(changes)); headline != "" {
|
|
segments = append(segments, headline)
|
|
}
|
|
if out := strings.TrimSpace(stdout); out != "" {
|
|
segments = append(segments, out)
|
|
}
|
|
if err := strings.TrimSpace(stderr); err != "" {
|
|
segments = append(segments, err)
|
|
}
|
|
return strings.Join(segments, "\n")
|
|
}
|
|
|
|
// codexMCPToolInput keeps the remote server visible in the transcript while
|
|
// preserving the provider-native argument shape under a stable key. Redact at
|
|
// the adapter boundary as well as in the daemon: other Message consumers must
|
|
// not have to know that an MCP invocation can carry credentials or private
|
|
// query text in deeply nested arguments.
|
|
func codexMCPToolInput(item map[string]any) map[string]any {
|
|
input := make(map[string]any, 2)
|
|
if server, _ := item["server"].(string); strings.TrimSpace(server) != "" {
|
|
input["server"] = server
|
|
}
|
|
if arguments, ok := item["arguments"]; ok {
|
|
input["arguments"] = arguments
|
|
}
|
|
return redact.InputMap(input)
|
|
}
|
|
|
|
func codexMCPToolName(item map[string]any) string {
|
|
if tool, _ := item["tool"].(string); strings.TrimSpace(tool) != "" {
|
|
return tool
|
|
}
|
|
return "mcp_tool"
|
|
}
|
|
|
|
// codexMCPToolResultOutput deliberately excludes result.content and
|
|
// structuredContent. Those values can be large and may contain private data;
|
|
// the Agent Log needs an auditable outcome, not a second copy of the provider
|
|
// payload that Codex already consumed.
|
|
func codexMCPToolResultOutput(item map[string]any) string {
|
|
status, _ := item["status"].(string)
|
|
status = codexNormalizePatchStatus(status)
|
|
if status == "" {
|
|
status = "completed"
|
|
}
|
|
|
|
segments := []string{status}
|
|
if durationMS := codexInt64(item, "durationMs", "duration_ms"); durationMS > 0 {
|
|
segments = append(segments, fmt.Sprintf("duration: %d ms", durationMS))
|
|
}
|
|
if errMessage := extractNestedString(item, "error", "message"); strings.TrimSpace(errMessage) != "" {
|
|
segments = append(segments, "error: "+sanitizeCodexDiagnostic(errMessage))
|
|
}
|
|
return strings.Join(segments, "\n")
|
|
}
|
|
|
|
func codexPatchHeadline(status string, fileCount int) string {
|
|
switch {
|
|
case status == "" && fileCount == 0:
|
|
return ""
|
|
case status == "":
|
|
return fmt.Sprintf("%s changed", pluralizeFiles(fileCount))
|
|
case fileCount == 0:
|
|
return status
|
|
default:
|
|
return fmt.Sprintf("%s (%s)", status, pluralizeFiles(fileCount))
|
|
}
|
|
}
|
|
|
|
func pluralizeFiles(n int) string {
|
|
if n == 1 {
|
|
return "1 file"
|
|
}
|
|
return fmt.Sprintf("%d files", n)
|
|
}
|
|
|
|
func (c *codexClient) handleEvent(msg map[string]any) {
|
|
msgType, _ := msg["type"].(string)
|
|
|
|
switch msgType {
|
|
case "task_started":
|
|
if c.onMessage != nil {
|
|
c.onMessage(Message{Type: MessageStatus, Status: "running", SessionID: c.getThreadID()})
|
|
}
|
|
case "agent_message":
|
|
text, _ := msg["message"].(string)
|
|
if text != "" && c.onMessage != nil {
|
|
c.onMessage(Message{Type: MessageText, Content: text})
|
|
}
|
|
if text != "" && c.onAgentMessage != nil {
|
|
c.onAgentMessage(text, "")
|
|
}
|
|
case "exec_command_begin":
|
|
callID, _ := msg["call_id"].(string)
|
|
command, _ := msg["command"].(string)
|
|
if c.onMessage != nil {
|
|
c.onMessage(Message{
|
|
Type: MessageToolUse,
|
|
Tool: "exec_command",
|
|
CallID: callID,
|
|
Input: map[string]any{"command": command},
|
|
})
|
|
}
|
|
case "exec_command_end":
|
|
callID, _ := msg["call_id"].(string)
|
|
output, _ := msg["output"].(string)
|
|
if c.onMessage != nil {
|
|
c.onMessage(Message{
|
|
Type: MessageToolResult,
|
|
Tool: "exec_command",
|
|
CallID: callID,
|
|
Output: output,
|
|
})
|
|
}
|
|
case "patch_apply_begin":
|
|
callID, _ := msg["call_id"].(string)
|
|
if c.onMessage != nil {
|
|
c.onMessage(Message{
|
|
Type: MessageToolUse,
|
|
Tool: "patch_apply",
|
|
CallID: callID,
|
|
Input: codexPatchInput(codexNormalizeLegacyChanges(msg["changes"])),
|
|
})
|
|
}
|
|
case "patch_apply_end":
|
|
callID, _ := msg["call_id"].(string)
|
|
stdout, _ := msg["stdout"].(string)
|
|
stderr, _ := msg["stderr"].(string)
|
|
status, _ := msg["status"].(string)
|
|
if status == "" {
|
|
// `status` postdates `success`; fall back so older builds still
|
|
// report an outcome rather than a bare file count.
|
|
if success, ok := msg["success"].(bool); ok {
|
|
if success {
|
|
status = "completed"
|
|
} else {
|
|
status = "failed"
|
|
}
|
|
}
|
|
}
|
|
changes := codexNormalizeLegacyChanges(msg["changes"])
|
|
if c.onMessage != nil {
|
|
c.onMessage(Message{
|
|
Type: MessageToolResult,
|
|
Tool: "patch_apply",
|
|
CallID: callID,
|
|
Output: codexPatchResultOutput(codexNormalizePatchStatus(status), changes, stdout, stderr),
|
|
})
|
|
}
|
|
case "task_complete":
|
|
// Extract usage from legacy task_complete if present.
|
|
c.extractUsageFromMap(msg)
|
|
if c.onTurnDone != nil {
|
|
c.onTurnDone(false)
|
|
}
|
|
case "turn_aborted":
|
|
if c.onTurnDone != nil {
|
|
c.onTurnDone(true)
|
|
}
|
|
}
|
|
}
|
|
|
|
func (c *codexClient) handleRawNotification(method string, params map[string]any) {
|
|
// Ignore notifications from threads other than the one we are tracking.
|
|
// Codex multiplexes subagent threads (e.g. memory consolidation) on the
|
|
// same stdio pipe; only our thread should drive turn lifecycle and output.
|
|
//
|
|
// The v2 app-server-protocol schema guarantees a top-level threadId on
|
|
// every notification. handleNotification performs this guard before the
|
|
// stateful current-turn gate; retain it here as defense in depth for direct
|
|
// callers. If a future codex revision introduces notifications without
|
|
// threadId, they fall through — re-audit this guard when bumping codex.
|
|
if c.isNotificationFromOtherThread(params) {
|
|
return
|
|
}
|
|
|
|
switch method {
|
|
case "thread/tokenUsage/updated":
|
|
c.updateThreadTokenUsage(params)
|
|
|
|
case "turn/started":
|
|
if turnID := extractNestedString(params, "turn", "id"); turnID != "" {
|
|
c.setActiveTurnID(turnID)
|
|
}
|
|
if c.onMessage != nil {
|
|
c.onMessage(Message{Type: MessageStatus, Status: "running", SessionID: c.getThreadID()})
|
|
}
|
|
|
|
case "turn/completed":
|
|
turnID := extractNestedString(params, "turn", "id")
|
|
status := extractNestedString(params, "turn", "status")
|
|
threadID, _ := params["threadId"].(string)
|
|
c.cfg.Logger.Info("codex turn/completed received", "thread_id", threadID, "turn_id", turnID, "status", status)
|
|
if c.turnCompleted {
|
|
return
|
|
}
|
|
c.turnCompleted = true
|
|
c.setActiveTurnID("")
|
|
aborted := status == "cancelled" || status == "canceled" ||
|
|
status == "aborted" || status == "interrupted"
|
|
|
|
// Capture the error message from failed turns so callers can surface
|
|
// a real reason instead of falling back to "empty output".
|
|
if status == "failed" {
|
|
errMsg := extractNestedString(params, "turn", "error", "message")
|
|
if errMsg == "" {
|
|
errMsg = "codex turn failed"
|
|
}
|
|
c.setTurnError(errMsg)
|
|
}
|
|
|
|
// Extract usage from turn/completed if present (e.g. params.turn.usage).
|
|
if turn, ok := params["turn"].(map[string]any); ok {
|
|
c.extractUsageFromMap(turn)
|
|
}
|
|
|
|
// Some cancellation and transport-failure paths do not emit an
|
|
// item/completed snapshot. Flush the aggregate of every complete delta
|
|
// notification before publishing the terminal boundary.
|
|
c.flushAgentMessageDeltas()
|
|
|
|
if c.onTurnDone != nil {
|
|
c.onTurnDone(aborted)
|
|
}
|
|
|
|
case "error":
|
|
// Top-level protocol error. Retrying notifications (willRetry=true) are
|
|
// transient reconnect attempts; only capture terminal errors so we
|
|
// don't stomp on a real failure later with a retry placeholder.
|
|
willRetry, _ := params["willRetry"].(bool)
|
|
errMsg := extractNestedString(params, "error", "message")
|
|
if errMsg == "" {
|
|
errMsg = extractNestedString(params, "message")
|
|
}
|
|
if errMsg != "" {
|
|
c.cfg.Logger.Warn("codex error notification", "message", errMsg, "will_retry", willRetry)
|
|
if c.onSemanticActivity != nil {
|
|
if willRetry {
|
|
c.onSemanticActivity("error:retry")
|
|
} else {
|
|
c.onSemanticActivity("error:terminal")
|
|
}
|
|
}
|
|
if !willRetry {
|
|
c.setTurnError(errMsg)
|
|
}
|
|
}
|
|
|
|
case "thread/status/changed":
|
|
// Status changes are informational. Only turn/completed carries the
|
|
// authoritative terminal state for a raw-protocol turn.
|
|
|
|
case "item/autoApprovalReview/completed":
|
|
// This item notification also signals semantic progress. Preserve the
|
|
// existing watchdog behavior when handling it specially below.
|
|
c.handleItemNotification(method, params)
|
|
if c.guardianRateLimitCircuit != nil &&
|
|
c.guardianRateLimitCircuit.observeReview(params) &&
|
|
c.onGuardianRateLimit != nil {
|
|
c.onGuardianRateLimit()
|
|
}
|
|
|
|
default:
|
|
if strings.HasPrefix(method, "item/") {
|
|
c.handleItemNotification(method, params)
|
|
}
|
|
}
|
|
}
|
|
|
|
func (c *codexClient) isNotificationFromOtherThread(params map[string]any) bool {
|
|
threadID, ok := params["threadId"].(string)
|
|
if !ok {
|
|
return false
|
|
}
|
|
// Compare against one snapshot: this runs on the stdout goroutine, ahead of
|
|
// the current-turn gate, so the gate's state cannot serialize the read.
|
|
currentThreadID := c.getThreadID()
|
|
return currentThreadID != "" && threadID != currentThreadID
|
|
}
|
|
|
|
func (c *codexClient) emitAgentMessageChunk(text string) bool {
|
|
if text == "" {
|
|
return true
|
|
}
|
|
if c.onAgentMessageChunk != nil {
|
|
return c.onAgentMessageChunk(text)
|
|
}
|
|
if c.onMessage != nil {
|
|
c.onMessage(Message{Type: MessageText, Content: text})
|
|
return true
|
|
}
|
|
return false
|
|
}
|
|
|
|
func (c *codexClient) agentMessageStream(itemID string) *codexAgentMessageStream {
|
|
if c.agentMessageStreams == nil {
|
|
c.agentMessageStreams = make(map[string]*codexAgentMessageStream)
|
|
}
|
|
stream := c.agentMessageStreams[itemID]
|
|
if stream == nil {
|
|
stream = &codexAgentMessageStream{}
|
|
c.agentMessageStreams[itemID] = stream
|
|
c.agentMessageOrder = append(c.agentMessageOrder, itemID)
|
|
}
|
|
return stream
|
|
}
|
|
|
|
func (c *codexClient) handleAgentMessageDelta(itemID, delta string) {
|
|
if itemID == "" || delta == "" {
|
|
return
|
|
}
|
|
|
|
stream := c.agentMessageStream(itemID)
|
|
// The leading delta is the latency-critical event. Deliver it immediately;
|
|
// coalesce the rest so a burst of token notifications cannot fill the
|
|
// daemon channel before its consumer is scheduled.
|
|
if stream.delivered.Len() == 0 && stream.pending.Len() == 0 {
|
|
if c.emitAgentMessageChunk(delta) {
|
|
stream.delivered.WriteString(delta)
|
|
return
|
|
}
|
|
}
|
|
|
|
stream.pending.WriteString(delta)
|
|
if stream.pending.Len() >= codexAgentMessageAggregateBytes {
|
|
c.flushAgentMessageStream(stream)
|
|
}
|
|
}
|
|
|
|
func (c *codexClient) flushAgentMessageStream(stream *codexAgentMessageStream) bool {
|
|
if stream == nil || stream.pending.Len() == 0 {
|
|
return true
|
|
}
|
|
text := stream.pending.String()
|
|
if !c.emitAgentMessageChunk(text) {
|
|
return false
|
|
}
|
|
stream.delivered.WriteString(text)
|
|
stream.pending.Reset()
|
|
return true
|
|
}
|
|
|
|
// flushAgentMessageDeltas preserves complete delta events when a turn is
|
|
// cancelled or stdout ends before item/completed. agentMessageOrder retains
|
|
// provider order; iterating the map directly would make two unfinished items
|
|
// nondeterministic.
|
|
func (c *codexClient) flushAgentMessageDeltas() {
|
|
for _, itemID := range c.agentMessageOrder {
|
|
stream := c.agentMessageStreams[itemID]
|
|
if stream == nil {
|
|
continue
|
|
}
|
|
if c.flushAgentMessageStream(stream) {
|
|
delete(c.agentMessageStreams, itemID)
|
|
}
|
|
}
|
|
}
|
|
|
|
func (c *codexClient) completeAgentMessage(itemID, text string) {
|
|
stream := c.agentMessageStreams[itemID]
|
|
if text == "" {
|
|
if c.flushAgentMessageStream(stream) {
|
|
delete(c.agentMessageStreams, itemID)
|
|
}
|
|
return
|
|
}
|
|
|
|
delivered := ""
|
|
if stream != nil {
|
|
delivered = stream.delivered.String()
|
|
} else {
|
|
stream = c.agentMessageStream(itemID)
|
|
}
|
|
|
|
if !strings.HasPrefix(text, delivered) {
|
|
// Already-persisted text cannot be retracted without a new event/schema.
|
|
// Appending the full snapshot would duplicate it, but every complete delta
|
|
// event still belongs in the append-only transcript. Flush pending normally;
|
|
// if the channel refuses it, leave the stream intact for terminal/EOF retry.
|
|
// onAgentMessage remains the authoritative Result.Output fallback.
|
|
if c.cfg.Logger != nil {
|
|
c.cfg.Logger.Warn("codex agent-message delta did not match completed text",
|
|
"item_id", itemID, "delivered_bytes", len(delivered), "completed_bytes", len(text))
|
|
}
|
|
if c.flushAgentMessageStream(stream) {
|
|
delete(c.agentMessageStreams, itemID)
|
|
}
|
|
return
|
|
}
|
|
|
|
// With an acknowledged prefix match, the completed snapshot supersedes
|
|
// observed-but-unacknowledged bytes and supplies their authoritative suffix.
|
|
stream.pending.Reset()
|
|
suffix := strings.TrimPrefix(text, delivered)
|
|
if suffix == "" || c.emitAgentMessageChunk(suffix) {
|
|
delete(c.agentMessageStreams, itemID)
|
|
return
|
|
}
|
|
// A process cancellation can race the final handoff. Keep the authoritative
|
|
// suffix so the stdout-EOF flush gets one more chance without duplicating the
|
|
// prefix that was already accepted.
|
|
stream.pending.WriteString(suffix)
|
|
}
|
|
|
|
func (c *codexClient) handleItemNotification(method string, params map[string]any) {
|
|
item, _ := params["item"].(map[string]any)
|
|
itemType, _ := item["type"].(string)
|
|
itemID, _ := item["id"].(string)
|
|
if method == "item/agentMessage/delta" {
|
|
// Unlike item/started and item/completed, the real app-server delta
|
|
// schema is flat: {threadId, turnId, itemId, delta}.
|
|
itemType = "agentMessage"
|
|
itemID, _ = params["itemId"].(string)
|
|
}
|
|
if isCodexItemProgressActivity(method) && c.onSemanticActivity != nil {
|
|
c.onSemanticActivity(describeCodexItemProgressActivity(method, itemType, itemID))
|
|
}
|
|
if item == nil && method != "item/agentMessage/delta" {
|
|
return
|
|
}
|
|
|
|
switch {
|
|
case method == "item/agentMessage/delta" && itemType == "agentMessage":
|
|
// JSON-RPC framing remains authoritative: handleLine invokes us only
|
|
// after a whole newline-delimited JSON event parsed successfully. Stream
|
|
// the event's text payload, not arbitrary stdout bytes.
|
|
delta, _ := params["delta"].(string)
|
|
c.handleAgentMessageDelta(itemID, delta)
|
|
|
|
case method == "item/started" && itemType == "commandExecution":
|
|
command, _ := item["command"].(string)
|
|
if c.onMessage != nil {
|
|
c.onMessage(Message{
|
|
Type: MessageToolUse,
|
|
Tool: "exec_command",
|
|
CallID: itemID,
|
|
Input: map[string]any{"command": command},
|
|
})
|
|
}
|
|
|
|
case method == "item/completed" && itemType == "commandExecution":
|
|
output, _ := item["aggregatedOutput"].(string)
|
|
if c.onMessage != nil {
|
|
c.onMessage(Message{
|
|
Type: MessageToolResult,
|
|
Tool: "exec_command",
|
|
CallID: itemID,
|
|
Output: output,
|
|
})
|
|
}
|
|
|
|
case method == "item/started" && itemType == "fileChange":
|
|
if c.onMessage != nil {
|
|
c.onMessage(Message{
|
|
Type: MessageToolUse,
|
|
Tool: "patch_apply",
|
|
CallID: itemID,
|
|
Input: codexPatchInput(codexNormalizeRawChanges(item["changes"])),
|
|
})
|
|
}
|
|
|
|
case method == "item/completed" && itemType == "fileChange":
|
|
status, _ := item["status"].(string)
|
|
changes := codexNormalizeRawChanges(item["changes"])
|
|
if c.onMessage != nil {
|
|
c.onMessage(Message{
|
|
Type: MessageToolResult,
|
|
Tool: "patch_apply",
|
|
CallID: itemID,
|
|
Output: codexPatchResultOutput(codexNormalizePatchStatus(status), changes, "", ""),
|
|
})
|
|
}
|
|
|
|
case method == "item/started" && itemType == "mcpToolCall":
|
|
if c.onMessage != nil {
|
|
c.onMessage(Message{
|
|
Type: MessageToolUse,
|
|
Tool: codexMCPToolName(item),
|
|
CallID: itemID,
|
|
Input: codexMCPToolInput(item),
|
|
})
|
|
}
|
|
|
|
case method == "item/completed" && itemType == "mcpToolCall":
|
|
status, _ := item["status"].(string)
|
|
if c.onMessage != nil {
|
|
c.onMessage(Message{
|
|
Type: MessageToolResult,
|
|
Tool: codexMCPToolName(item),
|
|
CallID: itemID,
|
|
Output: codexMCPToolResultOutput(item),
|
|
Status: codexNormalizePatchStatus(status),
|
|
})
|
|
}
|
|
|
|
case method == "item/completed" && itemType == "agentMessage":
|
|
text, _ := item["text"].(string)
|
|
phase, _ := item["phase"].(string)
|
|
if text != "" && c.onAgentMessage != nil {
|
|
c.onAgentMessage(text, phase)
|
|
}
|
|
c.completeAgentMessage(itemID, text)
|
|
if phase == "final_answer" {
|
|
// The gate exists so a subagent or a replayed history turn cannot
|
|
// end OUR turn early, and the thread guard at the top of this
|
|
// function already keeps foreign threads out. A final answer that
|
|
// arrives before we observed turn/started is still this thread's
|
|
// deliverable, and dropping it would fall back to narration. An
|
|
// empty final answer is reported too: it means the turn ended
|
|
// without a deliverable, not that the last narration is one.
|
|
if c.onFinalAnswer != nil {
|
|
c.onFinalAnswer(text)
|
|
}
|
|
// Keep the stream open until turn/completed. A final answer is the
|
|
// deliverable, not the authoritative lifecycle boundary.
|
|
}
|
|
}
|
|
}
|
|
|
|
// updateThreadTokenUsage consumes the v2 app-server's real usage notification.
|
|
// Resume can replay a historical snapshot after thread/resume returns, so only
|
|
// notifications attributed to this run's active or just-completed turn are eligible.
|
|
// A late final snapshot must survive closing supplement admission. The first current-
|
|
// turn snapshot contributes `last`; later snapshots contribute the monotonic
|
|
// delta from `total`. This retains multi-response tool loops without charging
|
|
// replayed history or duplicate snapshots twice.
|
|
func (c *codexClient) updateThreadTokenUsage(params map[string]any) {
|
|
turnID, _ := params["turnId"].(string)
|
|
if turnID == "" || turnID != c.usageTurnID() {
|
|
return
|
|
}
|
|
|
|
tokenUsage, ok := params["tokenUsage"].(map[string]any)
|
|
if !ok {
|
|
return
|
|
}
|
|
totalMap, ok := tokenUsage["total"].(map[string]any)
|
|
if !ok {
|
|
return
|
|
}
|
|
last, ok := tokenUsage["last"].(map[string]any)
|
|
if !ok {
|
|
return
|
|
}
|
|
|
|
total := codexRawTokenUsageFromV2(totalMap)
|
|
delta := codexRawTokenUsageFromV2(last)
|
|
c.usageMu.Lock()
|
|
if c.usageTotalSet {
|
|
delta = subtractCodexRawTokenUsage(total, c.usageTotal)
|
|
}
|
|
c.usageTotal = total
|
|
c.usageTotalSet = true
|
|
u := codexTokenUsage(delta)
|
|
c.usage.InputTokens += u.InputTokens
|
|
c.usage.OutputTokens += u.OutputTokens
|
|
c.usage.CacheReadTokens += u.CacheReadTokens
|
|
c.usage.CacheWriteTokens += u.CacheWriteTokens
|
|
c.usageMu.Unlock()
|
|
}
|
|
|
|
func codexRawTokenUsageFromV2(usage map[string]any) codexRawTokenUsage {
|
|
return codexRawTokenUsage{
|
|
InputTokens: codexInt64(usage, "inputTokens"),
|
|
OutputTokens: codexInt64(usage, "outputTokens"),
|
|
CachedInputTokens: codexInt64(usage, "cachedInputTokens"),
|
|
CacheWriteInputTokens: codexInt64(usage, "cacheWriteInputTokens"),
|
|
ReasoningOutputTokens: codexInt64(usage, "reasoningOutputTokens"),
|
|
}
|
|
}
|
|
|
|
func isCodexItemProgressActivity(method string) bool {
|
|
return strings.HasPrefix(method, "item/")
|
|
}
|
|
|
|
func describeCodexItemProgressActivity(method, itemType, itemID string) string {
|
|
if itemType == "" {
|
|
itemType = "unknown"
|
|
}
|
|
if itemID == "" {
|
|
return fmt.Sprintf("%s:%s", method, itemType)
|
|
}
|
|
return fmt.Sprintf("%s:%s:%s", method, itemType, itemID)
|
|
}
|
|
|
|
// extractUsageFromMap extracts token usage from a map that may contain
|
|
// "usage", "token_usage", or "tokens" fields. Handles various Codex formats.
|
|
func (c *codexClient) extractUsageFromMap(data map[string]any) {
|
|
// Try common field names for usage data.
|
|
var usageMap map[string]any
|
|
for _, key := range []string{"usage", "token_usage", "tokens"} {
|
|
if v, ok := data[key].(map[string]any); ok {
|
|
usageMap = v
|
|
break
|
|
}
|
|
}
|
|
if usageMap == nil {
|
|
return
|
|
}
|
|
|
|
c.usageMu.Lock()
|
|
defer c.usageMu.Unlock()
|
|
|
|
// Codex reports cached input as a prompt-token detail: cached_input_tokens
|
|
// are included in input_tokens. Persist mutually-exclusive buckets so
|
|
// dashboard cost math does not charge cached input twice.
|
|
inputTokens := codexInt64(usageMap, "input_tokens", "input", "prompt_tokens")
|
|
cacheReadTokens := codexInt64(usageMap, "cached_input_tokens", "cache_read_tokens", "cache_read_input_tokens")
|
|
c.usage.InputTokens += codexUncachedInputTokens(inputTokens, cacheReadTokens)
|
|
c.usage.OutputTokens += codexInt64(usageMap, "output_tokens", "output", "completion_tokens")
|
|
c.usage.CacheReadTokens += cacheReadTokens
|
|
c.usage.CacheWriteTokens += codexInt64(usageMap, "cache_write_tokens", "cache_creation_input_tokens")
|
|
}
|
|
|
|
func codexUncachedInputTokens(inputTokens, cachedInputTokens int64) int64 {
|
|
uncached := inputTokens - cachedInputTokens
|
|
if uncached < 0 {
|
|
return 0
|
|
}
|
|
return uncached
|
|
}
|
|
|
|
// codexPlainInputTokens removes both cache details from Codex's raw input
|
|
// total, which includes cache-read and cache-write input in v2 and JSONL usage.
|
|
func codexPlainInputTokens(inputTokens, cacheReadTokens, cacheWriteTokens int64) int64 {
|
|
return codexUncachedInputTokens(
|
|
codexUncachedInputTokens(inputTokens, cacheReadTokens),
|
|
cacheWriteTokens,
|
|
)
|
|
}
|
|
|
|
// codexInt64 returns the first non-zero int64 value from the map for the given keys.
|
|
func codexInt64(m map[string]any, keys ...string) int64 {
|
|
for _, key := range keys {
|
|
switch v := m[key].(type) {
|
|
case float64:
|
|
if v != 0 {
|
|
return int64(v)
|
|
}
|
|
case int64:
|
|
if v != 0 {
|
|
return v
|
|
}
|
|
}
|
|
}
|
|
return 0
|
|
}
|
|
|
|
// ── Codex session log scanner ──
|
|
|
|
// codexSessionUsage holds usage extracted from a Codex session JSONL file.
|
|
type codexSessionUsage struct {
|
|
usage TokenUsage
|
|
model string
|
|
}
|
|
|
|
// scanCodexSessionUsage extracts usage for threadID from its Codex rollout.
|
|
// Codex 0.144.4 embeds the app-server thread ID in both the rollout filename and
|
|
// the first session_meta item. Binding the scan to that ID prevents a
|
|
// concurrently-created rollout (for example a Codex subagent) from being billed
|
|
// to this task. A resumed rollout keeps its original date directory, so both flat
|
|
// and YYYY/MM/DD layouts are searched.
|
|
func scanCodexSessionUsage(startTime time.Time, codexHome, threadID string, resumed bool) *codexSessionUsage {
|
|
root := codexSessionRoot(codexHome)
|
|
if root == "" || strings.TrimSpace(threadID) == "" {
|
|
return nil
|
|
}
|
|
|
|
type candidate struct {
|
|
path string
|
|
modTime time.Time
|
|
}
|
|
var files []candidate
|
|
for _, path := range findCodexSessionRollouts(root, threadID) {
|
|
info, err := os.Stat(path)
|
|
if err != nil || info.ModTime().Before(startTime) {
|
|
continue
|
|
}
|
|
files = append(files, candidate{path: path, modTime: info.ModTime()})
|
|
}
|
|
if len(files) == 0 {
|
|
return nil
|
|
}
|
|
sort.Slice(files, func(i, j int) bool {
|
|
if files[i].modTime.Equal(files[j].modTime) {
|
|
return files[i].path < files[j].path
|
|
}
|
|
return files[i].modTime.Before(files[j].modTime)
|
|
})
|
|
|
|
// Multiple paths for one thread can transiently exist during layout migration.
|
|
// They have the same owner, so prefer the latest deterministically without ever
|
|
// crossing into a different thread's rollout.
|
|
result := parseCodexSessionFileSince(files[len(files)-1].path, startTime, resumed)
|
|
if result == nil || (result.usage.InputTokens == 0 && result.usage.OutputTokens == 0 &&
|
|
result.usage.CacheReadTokens == 0 && result.usage.CacheWriteTokens == 0 && result.model == "") {
|
|
return nil
|
|
}
|
|
return result
|
|
}
|
|
|
|
// codexResultUsageMap resolves a Codex result's model and usage without trading
|
|
// one provider signal for another. JSON-RPC usage is authoritative when present;
|
|
// the session log fills only a missing model. Session usage is used only when the
|
|
// JSON-RPC path reported no tokens. A default-model run therefore remains
|
|
// attributable instead of being stored as "unknown".
|
|
func codexResultUsageMap(startTime time.Time, codexHome, threadID string, resumed bool, model string, usage TokenUsage) map[string]TokenUsage {
|
|
usageEmpty := usage.InputTokens == 0 && usage.OutputTokens == 0 &&
|
|
usage.CacheReadTokens == 0 && usage.CacheWriteTokens == 0
|
|
var scanned *codexSessionUsage
|
|
if usageEmpty || model == "" {
|
|
scanned = scanCodexSessionUsage(startTime, codexHome, threadID, resumed)
|
|
}
|
|
if usageEmpty && scanned != nil {
|
|
usage = scanned.usage
|
|
}
|
|
if model == "" && scanned != nil {
|
|
model = scanned.model
|
|
}
|
|
if usage.InputTokens == 0 && usage.OutputTokens == 0 && usage.CacheReadTokens == 0 && usage.CacheWriteTokens == 0 {
|
|
return nil
|
|
}
|
|
if model == "" {
|
|
model = "unknown"
|
|
}
|
|
return map[string]TokenUsage{model: usage}
|
|
}
|
|
|
|
// findCodexSessionRollouts returns uncompressed rollout files owned by threadID
|
|
// in the two layouts supported by Codex 0.14x. filepath.Glob traverses a linked
|
|
// sessions root, unlike WalkDir, which treats a symlink root as a single file and
|
|
// never visits its children. The normal path uses Codex's filename contract and
|
|
// validates session_meta.id when present. If a future Codex version changes the
|
|
// filename, the metadata pass preserves exact ownership instead of silently
|
|
// dropping usage. Filtering after globbing also keeps a provider-supplied thread
|
|
// ID out of the glob expression.
|
|
func findCodexSessionRollouts(root, threadID string) []string {
|
|
threadID = strings.TrimSpace(threadID)
|
|
if root == "" || threadID == "" {
|
|
return nil
|
|
}
|
|
|
|
patterns := []string{
|
|
filepath.Join(root, "rollout-*.jsonl"),
|
|
filepath.Join(root, "*", "*", "*", "rollout-*.jsonl"),
|
|
}
|
|
seen := make(map[string]bool)
|
|
var candidates []string
|
|
for _, pattern := range patterns {
|
|
paths, err := filepath.Glob(pattern)
|
|
if err != nil {
|
|
continue
|
|
}
|
|
for _, path := range paths {
|
|
if seen[path] {
|
|
continue
|
|
}
|
|
seen[path] = true
|
|
candidates = append(candidates, path)
|
|
}
|
|
}
|
|
|
|
// Fast path for the current Codex contract. Reject a filename match if its
|
|
// canonical session_meta explicitly names a different thread.
|
|
suffix := "-" + threadID + ".jsonl"
|
|
var matches []string
|
|
for _, path := range candidates {
|
|
if !strings.HasSuffix(filepath.Base(path), suffix) {
|
|
continue
|
|
}
|
|
if metadataID, ok := readCodexRolloutThreadID(path); ok && metadataID != threadID {
|
|
continue
|
|
}
|
|
matches = append(matches, path)
|
|
}
|
|
if len(matches) > 0 {
|
|
return matches
|
|
}
|
|
|
|
// Compatibility path for a future filename format: the first session_meta
|
|
// remains the canonical owner of a rollout in Codex's own resume reader.
|
|
for _, path := range candidates {
|
|
if metadataID, ok := readCodexRolloutThreadID(path); ok && metadataID == threadID {
|
|
matches = append(matches, path)
|
|
}
|
|
}
|
|
return matches
|
|
}
|
|
|
|
// readCodexRolloutThreadID reads only the head of a rollout. Codex writes the
|
|
// canonical session_meta first; the line cap prevents a malformed legacy file
|
|
// without metadata from turning ownership checks into a full multi-GB scan.
|
|
func readCodexRolloutThreadID(path string) (string, bool) {
|
|
f, err := os.Open(path)
|
|
if err != nil {
|
|
return "", false
|
|
}
|
|
defer f.Close()
|
|
|
|
var evt struct {
|
|
Type string `json:"type"`
|
|
Payload *struct {
|
|
ID string `json:"id"`
|
|
} `json:"payload"`
|
|
}
|
|
scanner := bufio.NewScanner(f)
|
|
scanner.Buffer(make([]byte, 0, 64*1024), 1024*1024)
|
|
for lineCount := 0; lineCount < 64 && scanner.Scan(); lineCount++ {
|
|
line := scanner.Bytes()
|
|
if !bytesContainsStr(line, "session_meta") {
|
|
continue
|
|
}
|
|
if err := json.Unmarshal(line, &evt); err == nil && evt.Type == "session_meta" && evt.Payload != nil && evt.Payload.ID != "" {
|
|
return evt.Payload.ID, true
|
|
}
|
|
}
|
|
return "", false
|
|
}
|
|
|
|
// codexSessionRoot returns the Codex sessions directory. It prefers the
|
|
// explicit per-task codexHome the backend is running with (so usage is read
|
|
// from the same task-local sessions Codex actually wrote to), then the ambient
|
|
// CODEX_HOME, then ~/.codex.
|
|
func codexSessionRoot(codexHome string) string {
|
|
if codexHome = strings.TrimSpace(codexHome); codexHome == "" {
|
|
codexHome = os.Getenv("CODEX_HOME")
|
|
}
|
|
if codexHome != "" {
|
|
dir := filepath.Join(codexHome, "sessions")
|
|
if info, err := os.Stat(dir); err == nil && info.IsDir() {
|
|
return dir
|
|
}
|
|
}
|
|
|
|
home, err := os.UserHomeDir()
|
|
if err != nil {
|
|
return ""
|
|
}
|
|
|
|
dir := filepath.Join(home, ".codex", "sessions")
|
|
if info, err := os.Stat(dir); err == nil && info.IsDir() {
|
|
return dir
|
|
}
|
|
return ""
|
|
}
|
|
|
|
// codexRawTokenUsage retains ReasoningOutputTokens to mirror the provider payload
|
|
// and keep resume-baseline subtraction symmetric. No downstream consumer uses
|
|
// this breakdown; normalized usage reads OutputTokens, which already includes it.
|
|
type codexRawTokenUsage struct {
|
|
InputTokens int64 `json:"input_tokens"`
|
|
OutputTokens int64 `json:"output_tokens"`
|
|
CachedInputTokens int64 `json:"cached_input_tokens"`
|
|
CacheReadInputTokens int64 `json:"cache_read_input_tokens"`
|
|
CacheWriteInputTokens int64 `json:"cache_write_input_tokens"`
|
|
ReasoningOutputTokens int64 `json:"reasoning_output_tokens"`
|
|
}
|
|
|
|
// codexTokenUsage converts either event or rollout counters to the shared
|
|
// TokenUsage contract. Codex input includes cache reads/writes; its output
|
|
// already includes reasoning_output_tokens, which is only a breakdown.
|
|
func codexTokenUsage(raw codexRawTokenUsage) TokenUsage {
|
|
u := normalizeCodexRawTokenUsage(raw)
|
|
return TokenUsage{
|
|
InputTokens: codexPlainInputTokens(u.InputTokens, u.CachedInputTokens, u.CacheWriteInputTokens),
|
|
OutputTokens: u.OutputTokens,
|
|
CacheReadTokens: u.CachedInputTokens,
|
|
CacheWriteTokens: u.CacheWriteInputTokens,
|
|
}
|
|
}
|
|
|
|
// codexSessionTokenCount represents a token_count event in Codex JSONL.
|
|
type codexSessionTokenCount struct {
|
|
Timestamp time.Time `json:"timestamp"`
|
|
Type string `json:"type"`
|
|
Payload *struct {
|
|
Type string `json:"type"`
|
|
Info *struct {
|
|
TotalTokenUsage *codexRawTokenUsage `json:"total_token_usage"`
|
|
LastTokenUsage *codexRawTokenUsage `json:"last_token_usage"`
|
|
Model string `json:"model"`
|
|
} `json:"info"`
|
|
Model string `json:"model"`
|
|
} `json:"payload"`
|
|
}
|
|
|
|
// parseCodexSessionFile extracts the final token_count from a Codex session file.
|
|
func parseCodexSessionFile(path string) *codexSessionUsage {
|
|
return parseCodexSessionFileSince(path, time.Time{}, false)
|
|
}
|
|
|
|
// parseCodexSessionFileSince extracts usage accumulated after startTime. Codex
|
|
// reports total_token_usage cumulatively for the whole resumed session, so the
|
|
// last total before startTime is subtracted from the final total. Timestamp-less
|
|
// events in a resumed rollout are baseline-only until an explicit post-start
|
|
// timestamp establishes the boundary; fresh sessions retain the previous
|
|
// whole-file behavior because every event belongs to the new task.
|
|
func parseCodexSessionFileSince(path string, startTime time.Time, resumed bool) *codexSessionUsage {
|
|
f, err := os.Open(path)
|
|
if err != nil {
|
|
return nil
|
|
}
|
|
defer f.Close()
|
|
|
|
var result codexSessionUsage
|
|
var previousTotal, accumulated, finalUsage codexRawTokenUsage
|
|
previousTotalFound := false
|
|
finalUsageFound := false
|
|
afterStartBoundary := false
|
|
|
|
scanner := bufio.NewScanner(f)
|
|
scanner.Buffer(make([]byte, 0, 256*1024), 1024*1024)
|
|
|
|
for scanner.Scan() {
|
|
line := scanner.Bytes()
|
|
|
|
// Fast pre-filter.
|
|
if !bytesContainsStr(line, "token_count") && !bytesContainsStr(line, "turn_context") {
|
|
continue
|
|
}
|
|
|
|
var evt codexSessionTokenCount
|
|
if err := json.Unmarshal(line, &evt); err != nil || evt.Payload == nil {
|
|
continue
|
|
}
|
|
timestampAfterStart := !startTime.IsZero() && !evt.Timestamp.IsZero() && evt.Timestamp.After(startTime)
|
|
if timestampAfterStart {
|
|
afterStartBoundary = true
|
|
}
|
|
|
|
// Track model from turn_context events.
|
|
if evt.Type == "turn_context" && evt.Payload.Model != "" {
|
|
result.model = evt.Payload.Model
|
|
continue
|
|
}
|
|
|
|
// Extract token usage from token_count events.
|
|
if evt.Payload.Type == "token_count" && evt.Payload.Info != nil {
|
|
afterStart := startTime.IsZero() || timestampAfterStart ||
|
|
(evt.Timestamp.IsZero() && (!resumed || afterStartBoundary))
|
|
if usage := evt.Payload.Info.TotalTokenUsage; usage != nil {
|
|
current := normalizeCodexRawTokenUsage(*usage)
|
|
if afterStart {
|
|
delta := current
|
|
if previousTotalFound {
|
|
delta = subtractCodexRawTokenUsage(current, previousTotal)
|
|
}
|
|
accumulated = addCodexRawTokenUsage(accumulated, delta)
|
|
finalUsage = accumulated
|
|
finalUsageFound = true
|
|
}
|
|
previousTotal = current
|
|
previousTotalFound = true
|
|
} else if usage := evt.Payload.Info.LastTokenUsage; usage != nil && afterStart {
|
|
// Preserve event order: a later last_token_usage is the same
|
|
// fallback the old whole-file parser would have selected.
|
|
finalUsage = normalizeCodexRawTokenUsage(*usage)
|
|
finalUsageFound = true
|
|
}
|
|
if evt.Payload.Info.Model != "" {
|
|
result.model = evt.Payload.Info.Model
|
|
}
|
|
}
|
|
}
|
|
|
|
if !finalUsageFound && result.model == "" {
|
|
return nil
|
|
}
|
|
if finalUsageFound {
|
|
result.usage = codexTokenUsage(finalUsage)
|
|
}
|
|
return &result
|
|
}
|
|
|
|
func subtractCodexRawTokenUsage(total, baseline codexRawTokenUsage) codexRawTokenUsage {
|
|
total = normalizeCodexRawTokenUsage(total)
|
|
baseline = normalizeCodexRawTokenUsage(baseline)
|
|
// Guard each counter independently. Treating one field's reset as a reset of
|
|
// the entire snapshot would re-report still-monotonic fields and recreate the
|
|
// over-counting this fallback is meant to prevent.
|
|
return codexRawTokenUsage{
|
|
InputTokens: nonNegativeTokenDelta(total.InputTokens, baseline.InputTokens),
|
|
OutputTokens: nonNegativeTokenDelta(total.OutputTokens, baseline.OutputTokens),
|
|
CachedInputTokens: nonNegativeTokenDelta(total.CachedInputTokens, baseline.CachedInputTokens),
|
|
CacheWriteInputTokens: nonNegativeTokenDelta(total.CacheWriteInputTokens, baseline.CacheWriteInputTokens),
|
|
ReasoningOutputTokens: nonNegativeTokenDelta(total.ReasoningOutputTokens, baseline.ReasoningOutputTokens),
|
|
}
|
|
}
|
|
|
|
func normalizeCodexRawTokenUsage(usage codexRawTokenUsage) codexRawTokenUsage {
|
|
if usage.CachedInputTokens == 0 {
|
|
usage.CachedInputTokens = usage.CacheReadInputTokens
|
|
}
|
|
usage.CacheReadInputTokens = 0
|
|
return usage
|
|
}
|
|
|
|
func addCodexRawTokenUsage(a, b codexRawTokenUsage) codexRawTokenUsage {
|
|
return codexRawTokenUsage{
|
|
InputTokens: a.InputTokens + b.InputTokens,
|
|
OutputTokens: a.OutputTokens + b.OutputTokens,
|
|
CachedInputTokens: a.CachedInputTokens + b.CachedInputTokens,
|
|
CacheWriteInputTokens: a.CacheWriteInputTokens + b.CacheWriteInputTokens,
|
|
ReasoningOutputTokens: a.ReasoningOutputTokens + b.ReasoningOutputTokens,
|
|
}
|
|
}
|
|
|
|
func nonNegativeTokenDelta(total, baseline int64) int64 {
|
|
if total < baseline {
|
|
// A counter reset means the final value already belongs to the new span.
|
|
return total
|
|
}
|
|
return total - baseline
|
|
}
|
|
|
|
// bytesContainsStr checks if b contains the string s (without allocating).
|
|
func bytesContainsStr(b []byte, s string) bool {
|
|
return strings.Contains(string(b), s)
|
|
}
|
|
|
|
// ── Helpers ──
|
|
|
|
func extractThreadID(result json.RawMessage) string {
|
|
var r struct {
|
|
Thread struct {
|
|
ID string `json:"id"`
|
|
} `json:"thread"`
|
|
}
|
|
if err := json.Unmarshal(result, &r); err != nil {
|
|
return ""
|
|
}
|
|
return r.Thread.ID
|
|
}
|
|
|
|
func extractNestedString(m map[string]any, keys ...string) string {
|
|
current := any(m)
|
|
for _, key := range keys {
|
|
obj, ok := current.(map[string]any)
|
|
if !ok {
|
|
return ""
|
|
}
|
|
current = obj[key]
|
|
}
|
|
s, _ := current.(string)
|
|
return s
|
|
}
|
|
|
|
func nilIfEmpty(s string) any {
|
|
if s == "" {
|
|
return nil
|
|
}
|
|
return s
|
|
}
|