diff --git a/README.md b/README.md
index 6156f81..96107e1 100644
--- a/README.md
+++ b/README.md
@@ -462,6 +462,12 @@ Set up the recommended arrangement as follows.
reconstructed after the fact. No additional Codexometer option is required;
it automatically probes the same default socket under `CODEX_HOME`.
+ Codexometer offers Codex's `zstd-stream-v1` WebSocket compression on that
+ connection. Compression is used only when the server acknowledges it;
+ older servers and servers with compression disabled continue using JSON
+ text. Encoder/decoder history is reset on every connection, with bounded
+ messages and strict compressed-message framing. No server setting is changed.
+
If the latest Codexometer build is already running, leave it open: it retries
the connection and discovers newly connected sessions automatically. Allow
a refresh cycle. Restart it if you have replaced its binary or changed its
@@ -763,7 +769,8 @@ new evidence accumulates. Future improvement: consume the actual response
prefer that evidence to requested settings.
Codexometer prefers Codex's durable **`token_usage_record`** entries for each
-newly completed response. Exact records are counted independently of cumulative
+reported response, including usage that Codex persists for failed/incomplete
+requests. Exact records are counted independently of cumulative
context counters, with bounded response-ID deduplication; inherited parent
records are excluded from a child's own usage. Startup establishes a historical
baseline without charging past responses, using a bounded 4 MiB tail scan.
@@ -773,6 +780,38 @@ another billable response. Older rollouts retain the legacy token-count reader.
an existing thread establishes a new baseline rather than charging copied
history again; session settings come from its newest observed rollout.
+Compressed `.jsonl.zst` histories are streamed into bounded in-memory baselines;
+no decompressed transcript copy is written to disk. Their decoded offsets and
+recent response identities are retained. When Codex restores a plain rollout,
+Codexometer verifies the entire historical prefix before consuming appended
+responses. Rewritten histories establish a fresh baseline. Source and decoded
+histories are each limited to 256 MiB per compressed file, with bounded decoder
+windows and record buffers; unreadable or oversized histories report incomplete
+coverage instead of silently qualifying quota-learning intervals.
+
+Codex 0.164.0's experimental background compaction can consume tokens without
+publishing the corresponding exact foreground response record. Codexometer
+checks the loaded thread's effective `async_compaction` and `token_budget`
+feature state through a read-only inventory. A potentially unobserved background
+path pauses API-EQ quota learning, clears prior capacity estimates, and displays
+an **INCOMPLETE COVERAGE** diagnostic in terminal/browser Quota. Child sessions
+are checked separately. Known risks survive a daemon disconnect; a verified
+recovery starts a fresh learning interval. Observed token/cost totals remain
+available and no missing response cost is invented.
+
+Without effective daemon feature state, relevant local configuration is checked
+read-only. Enabled flags, including flags in potentially selected profiles, are
+conservative coverage warnings rather than proof of the current runtime mode.
+This fallback cannot establish hidden CLI overrides, remote configuration or
+unobserved usage from another machine. Those existing observation limits remain;
+Codexometer does not change Codex's compaction configuration.
+
+Model, speed and permission changes are confirmed against the running thread
+rather than treating a queued settings acknowledgement as success. Verification
+uses resume without overrides, stays bound to the original connection, and does
+not retry mutations or block approval/event processing. Account changes refresh
+the model/options catalogue while retaining the observed conversation.
+
The requested model remains the fallback when actual per-response model evidence
is unavailable; durable usage records do not supply model or billed-tier fields.
Diagnostics explicitly distinguish requested-model attribution from a matching
@@ -3450,7 +3489,7 @@ virtual clock rather than depending on millisecond wall-clock scheduling.
Codexometer uses:
-- Go 1.27.2+
+- Go 1.27.3+
- Bubble Tea v2 for the terminal event loop and declarative terminal modes
- Lip Gloss v2 for adaptive ANSI styling and layout
- Starlark for deterministic, hermetic benchmark-code evaluation
diff --git a/go.mod b/go.mod
index 855572a..a8ee175 100644
--- a/go.mod
+++ b/go.mod
@@ -1,6 +1,6 @@
module github.com/merefield/codexometer
-go 1.27.2
+go 1.27.3
require (
charm.land/bubbles/v2 v2.2.1
@@ -11,6 +11,8 @@ require (
github.com/charmbracelet/x/ansi v0.11.8
github.com/charmbracelet/x/term v0.2.2
github.com/gorilla/websocket v1.5.3
+ github.com/klauspost/compress v1.20.3
+ github.com/pelletier/go-toml/v2 v2.4.3
go.starlark.net v0.0.0-20260930220527-d7438c5a85ac
golang.org/x/sync v0.23.0
golang.org/x/sys v0.48.0
diff --git a/go.sum b/go.sum
index 057603a..6bd8508 100644
--- a/go.sum
+++ b/go.sum
@@ -32,12 +32,16 @@ github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8=
github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU=
github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aNNg=
github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
+github.com/klauspost/compress v1.20.3 h1:zWNoFqq/lSuYQRim3dUXlXeA0IqwuZGicdG5ZTfY5j4=
+github.com/klauspost/compress v1.20.3/go.mod h1:LUdAzn7YLVvxLpc7y3V1m40wESHTgc1422pwwBSKYuI=
github.com/lucasb-eyer/go-colorful v1.4.1 h1:1EO+WB73+EH8EVbzlrG3KLAfEypQWVHIBqlTf+2hNss=
github.com/lucasb-eyer/go-colorful v1.4.1/go.mod h1:R4dSotOR9KMtayYi1e77YzuveK+i7ruzyGqttikkLy0=
github.com/mattn/go-runewidth v0.0.27 h1:Feg/Oou5zI/wnpgDF6omIU0OokC9GxLC/WRknhVlIR0=
github.com/mattn/go-runewidth v0.0.27/go.mod h1:3qAiGCV4Koz/yuveO58qUefmUTRm8r0IGEXZ9jeHp/8=
github.com/muesli/cancelreader v0.2.2 h1:3I4Kt4BQjOR54NavqnDogx/MIoWBFa0StPA8ELUXHmA=
github.com/muesli/cancelreader v0.2.2/go.mod h1:3XuTXfFS2VjM+HTLZY9Ak0l6eUKfijIfMUZ4EgX0QYo=
+github.com/pelletier/go-toml/v2 v2.4.3 h1:GTRvJQutkOSftxIFD5xw9aepkYNuPWmVJpffdDPYVpY=
+github.com/pelletier/go-toml/v2 v2.4.3/go.mod h1:2gIqNv+qfxSVS7cM2xJQKtLSTLUE9V8t9Stt+h56mCY=
github.com/rivo/uniseg v0.4.7 h1:WUdvkW8uEhrYfLC4ZzdpI2ztxP1I582+49Oc5Mq64VQ=
github.com/rivo/uniseg v0.4.7/go.mod h1:FN3SvrM+Zdj16jyLfmOkMNblXMcoc8DfTHruCPUcx88=
github.com/xo/terminfo v0.0.0-20220910002029-abceb7e1c41e h1:JVG44RsyaB9T2KIHavMF/ppJZNG9ZpyihvCd0w101no=
diff --git a/internal/codex/daemon_native_unix_test.go b/internal/codex/daemon_native_unix_test.go
new file mode 100644
index 0000000..6f8972a
--- /dev/null
+++ b/internal/codex/daemon_native_unix_test.go
@@ -0,0 +1,87 @@
+//go:build unix
+
+package codex
+
+import (
+ "context"
+ "os"
+ "os/exec"
+ "path/filepath"
+ "strings"
+ "testing"
+ "time"
+)
+
+// Opt-in integration check: no credentials, user configuration, model requests,
+// thread creation or shared daemon are used. CI otherwise uses wire fixtures.
+func TestNativeCodex164Compression(t *testing.T) {
+ if os.Getenv("CODEXOMETER_TEST_CODEX_164") != "1" {
+ t.Skip("set CODEXOMETER_TEST_CODEX_164=1 with Codex 0.164.0 installed")
+ }
+ binary, err := exec.LookPath("codex")
+ if err != nil {
+ t.Fatal(err)
+ }
+ version, err := exec.Command(binary, "--version").Output()
+ if err != nil || strings.TrimSpace(string(version)) != "codex-cli 0.164.0" {
+ t.Fatal("expected Codex 0.164.0", string(version), err)
+ }
+ home, err := os.MkdirTemp("/tmp", "cxm-native-")
+ if err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(func() { os.RemoveAll(home) })
+ ctx, cancel := context.WithCancel(context.Background())
+ command := exec.CommandContext(ctx, binary, "-c", "features.app_server_websocket_compression=true", "app-server", "--listen", "unix://")
+ command.Dir = home
+ command.Env = environmentWith(environmentWithout(os.Environ(), "CODEX_HOME", "OPENAI_API_KEY", "CODEX_API_KEY"), "CODEX_HOME", home)
+ log, err := os.Create(filepath.Join(home, "server.log"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ command.Stderr = log
+ command.Stdout = log
+ if err := command.Start(); err != nil {
+ cancel()
+ log.Close()
+ t.Fatal(err)
+ }
+ t.Cleanup(func() { cancel(); command.Wait(); log.Close() })
+ p := newSessionStatusProvider(home).(*daemonStatusProvider)
+ t.Cleanup(func() { p.disconnect(nil) })
+ deadline := time.Now().Add(10 * time.Second)
+ for {
+ err = p.ensureConnected(context.Background())
+ if err == nil {
+ break
+ }
+ if time.Now().After(deadline) {
+ t.Fatal("native handshake failed", err)
+ }
+ time.Sleep(20 * time.Millisecond)
+ }
+ p.mu.Lock()
+ compressed := p.wire != nil && p.wire.decoder != nil
+ p.mu.Unlock()
+ if !compressed {
+ t.Fatal("Codex did not acknowledge zstd-stream-v1")
+ }
+ for range 5 {
+ var result struct {
+ Data []struct {
+ Name string
+ Enabled bool
+ }
+ }
+ if err := p.request(context.Background(), "experimentalFeature/list", map[string]any{}, &result); err != nil {
+ t.Fatal("native compressed RPC", err)
+ }
+ if len(result.Data) == 0 {
+ t.Fatal("native feature inventory empty")
+ }
+ }
+ loaded, err := p.loadedThreads(context.Background())
+ if err != nil || len(loaded) != 0 {
+ t.Fatal("isolated server unexpectedly has loaded threads", loaded, err)
+ }
+}
diff --git a/internal/codex/daemon_status.go b/internal/codex/daemon_status.go
index bbd54e7..df3dbb6 100644
--- a/internal/codex/daemon_status.go
+++ b/internal/codex/daemon_status.go
@@ -18,6 +18,7 @@ type sessionStatusProvider interface {
}
type sessionDaemonSnapshot struct {
+ UsageCoverage map[string]string
Contexts map[string]SessionContext
WorkingDirectories map[string]string
Statuses map[string]sessionRuntimeStatus
diff --git a/internal/codex/daemon_status_unix.go b/internal/codex/daemon_status_unix.go
index 7eba0a4..6e9e54f 100644
--- a/internal/codex/daemon_status_unix.go
+++ b/internal/codex/daemon_status_unix.go
@@ -31,6 +31,7 @@ type daemonStatusProvider struct {
mu sync.Mutex
connection *websocket.Conn
+ wire *daemonWire
nextRequestID int64
pending map[int64]chan daemonEnvelope
subscribed map[string]struct{}
@@ -40,6 +41,7 @@ type daemonStatusProvider struct {
lastStatusAt time.Time
statusThreads map[string]struct{}
statuses map[string]sessionRuntimeStatus
+ usageCoverage map[string]string
threadSettings map[string]QuotaSession
threadDirectories map[string]string
settingsVersions map[string]uint64
@@ -98,6 +100,7 @@ func (p *daemonStatusProvider) Fetch(ctx context.Context, threadIDs []string) (s
}
}
+ coverage := p.readUsageCoverage(ctx, threadIDs)
statuses := make(map[string]sessionRuntimeStatus, len(threadIDs))
for _, threadID := range threadIDs {
var result struct {
@@ -121,6 +124,7 @@ func (p *daemonStatusProvider) Fetch(ctx context.Context, threadIDs []string) (s
}
p.mu.Lock()
+ p.usageCoverage = coverage
p.lastStatusAt = time.Now()
p.statusThreads = stringStructSet(threadIDs)
for id, status := range statuses {
@@ -161,7 +165,12 @@ func (p *daemonStatusProvider) statusSnapshotLocked(threadIDs []string) sessionD
for threadID := range p.subscribed {
subscribed[threadID] = struct{}{}
}
+ coverage := make(map[string]string, len(p.usageCoverage))
+ for id, issue := range p.usageCoverage {
+ coverage[id] = issue
+ }
return sessionDaemonSnapshot{
+ UsageCoverage: coverage,
Contexts: daemonContextSnapshot(p.contexts, threadIDs, statuses),
WorkingDirectories: directories,
Statuses: statuses,
@@ -243,7 +252,7 @@ func (p *daemonStatusProvider) ensureConnected(ctx context.Context) error {
return (&net.Dialer{Timeout: daemonStatusTimeout}).DialContext(ctx, "unix", p.socketPath)
},
}
- connection, response, err := dialer.DialContext(ctx, "ws://localhost/", http.Header{})
+ connection, response, err := dialer.DialContext(ctx, "ws://localhost/", http.Header{http.CanonicalHeaderKey(daemonCompressionHeader): []string{daemonCompressionProtocol}})
if response != nil && response.Body != nil {
_ = response.Body.Close()
}
@@ -251,13 +260,19 @@ func (p *daemonStatusProvider) ensureConnected(ctx context.Context) error {
return err
}
+ wire, err := newDaemonWire(connection, response.Header)
+ if err != nil {
+ _ = connection.Close()
+ return err
+ }
p.mu.Lock()
if p.connection != nil {
p.mu.Unlock()
- _ = connection.Close()
+ wire.close()
return nil
}
p.connection = connection
+ p.wire = wire
p.commandGeneration++
p.commandTransitions = make(map[string]uint64)
p.pending = make(map[int64]chan daemonEnvelope)
@@ -268,7 +283,7 @@ func (p *daemonStatusProvider) ensureConnected(ctx context.Context) error {
p.settingsVersions = make(map[string]uint64)
p.settingsSignal = make(chan struct{})
p.mu.Unlock()
- go p.readLoop(connection)
+ go p.readLoop(connection, wire)
var initialized struct {
UserAgent string `json:"userAgent"`
@@ -385,6 +400,13 @@ func (p *daemonStatusProvider) write(connection *websocket.Conn, value any) erro
p.writeMu.Lock()
defer p.writeMu.Unlock()
_ = connection.SetWriteDeadline(time.Now().Add(daemonStatusTimeout))
+ p.mu.Lock()
+ wire := p.wire
+ current := p.connection
+ p.mu.Unlock()
+ if wire != nil && current == connection {
+ return wire.writeJSON(value)
+ }
return connection.WriteJSON(value)
}
@@ -394,16 +416,24 @@ func (p *daemonStatusProvider) dropPending(id int64) {
p.mu.Unlock()
}
-func (p *daemonStatusProvider) readLoop(connection *websocket.Conn) {
+func (p *daemonStatusProvider) readLoop(connection *websocket.Conn, wires ...*daemonWire) {
+ wire := &daemonWire{conn: connection}
+ if len(wires) > 0 {
+ wire = wires[0]
+ }
for {
var envelope daemonEnvelope
- if err := connection.ReadJSON(&envelope); err != nil {
+ if err := wire.readJSON(&envelope); err != nil {
p.disconnect(connection)
return
}
var responseID int64
if envelope.Method == "" && len(envelope.ID) > 0 && json.Unmarshal(envelope.ID, &responseID) == nil {
p.mu.Lock()
+ if p.connection != connection {
+ p.mu.Unlock()
+ return
+ }
response := p.pending[responseID]
delete(p.pending, responseID)
p.mu.Unlock()
@@ -416,6 +446,10 @@ func (p *daemonStatusProvider) readLoop(connection *websocket.Conn) {
continue
}
p.mu.Lock()
+ if p.connection != connection {
+ p.mu.Unlock()
+ return
+ }
if envelope.Method == "thread/queue/changed" {
p.queueRevision++
}
@@ -441,6 +475,11 @@ func (p *daemonStatusProvider) handleNotification(method string, params json.Raw
}
}
switch method {
+ case "account/updated":
+ p.mu.Lock()
+ p.commandGeneration++
+ p.lastStatusAt = time.Time{}
+ p.mu.Unlock()
case "thread/settings/updated":
var notification struct {
ThreadID string `json:"threadId"`
@@ -541,6 +580,8 @@ func (p *daemonStatusProvider) disconnect(connection *websocket.Conn) {
return
}
current := p.connection
+ wire := p.wire
+ p.wire = nil
p.connection = nil
for id, response := range p.pending {
delete(p.pending, id)
@@ -557,13 +598,16 @@ func (p *daemonStatusProvider) disconnect(connection *websocket.Conn) {
p.statuses = nil
p.commandTransitions = nil
p.threadSettings = nil
+ p.usageCoverage = nil
p.threadDirectories = nil
if p.settingsSignal != nil {
close(p.settingsSignal)
p.settingsSignal = nil
}
p.mu.Unlock()
- if current != nil {
+ if wire != nil {
+ wire.close()
+ } else if current != nil {
_ = current.Close()
}
}
diff --git a/internal/codex/daemon_wire.go b/internal/codex/daemon_wire.go
new file mode 100644
index 0000000..21478d8
--- /dev/null
+++ b/internal/codex/daemon_wire.go
@@ -0,0 +1,260 @@
+package codex
+
+import (
+ "bytes"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "io"
+ "net/http"
+ "sync"
+ "unicode/utf8"
+
+ "github.com/gorilla/websocket"
+ "github.com/klauspost/compress/zstd"
+)
+
+const daemonCompressionHeader = "x-codex-app-server-compression"
+const daemonCompressionProtocol = "zstd-stream-v1"
+const daemonMessageLimit = 16 << 20
+const daemonZstdWindow = 1 << 20
+
+// This is Codex's application-level, flushed zstd stream protocol, not the
+// WebSocket permessage-deflate extension. Streams belong to one connection.
+type daemonWire struct {
+ conn *websocket.Conn
+ mu sync.Mutex
+ encoder *zstd.Encoder
+ encoded bytes.Buffer
+ decoder *daemonStreamDecoder
+}
+
+func newDaemonWire(conn *websocket.Conn, headers http.Header) (*daemonWire, error) {
+ w := &daemonWire{conn: conn}
+ conn.SetReadLimit(daemonMessageLimit + (1 << 20))
+ values := headers.Values(daemonCompressionHeader)
+ if len(values) == 0 {
+ return w, nil
+ }
+ if len(values) != 1 || values[0] != daemonCompressionProtocol {
+ return nil, errors.New("unsupported daemon WebSocket compression acknowledgement")
+ }
+ encoder, err := zstd.NewWriter(&w.encoded, zstd.WithEncoderConcurrency(1), zstd.WithWindowSize(daemonZstdWindow), zstd.WithEncoderLevel(zstd.SpeedFastest))
+ if err != nil {
+ return nil, err
+ }
+ w.encoder = encoder
+ w.decoder = newDaemonStreamDecoder()
+ return w, nil
+}
+
+func (w *daemonWire) writeJSON(value any) error {
+ data, err := json.Marshal(value)
+ if err != nil {
+ return err
+ }
+ if len(data) > daemonMessageLimit {
+ return errors.New("daemon message exceeds size limit")
+ }
+ w.mu.Lock()
+ defer w.mu.Unlock()
+ if w.encoder == nil {
+ return w.conn.WriteMessage(websocket.TextMessage, data)
+ }
+ w.encoded.Reset()
+ if _, err = w.encoder.Write(data); err != nil {
+ return err
+ }
+ if err = w.encoder.Flush(); err != nil {
+ return err
+ }
+ return w.conn.WriteMessage(websocket.BinaryMessage, w.encoded.Bytes())
+}
+
+func (w *daemonWire) readJSON(value any) error {
+ kind, data, err := w.conn.ReadMessage()
+ if err != nil {
+ return err
+ }
+ if w.decoder != nil {
+ if kind != websocket.BinaryMessage {
+ return errors.New("compressed daemon sent a non-binary message")
+ }
+ data, err = w.decoder.decode(data)
+ if err != nil {
+ return err
+ }
+ } else if kind != websocket.TextMessage {
+ return errors.New("unnegotiated binary daemon message")
+ }
+ if len(data) > daemonMessageLimit || !utf8.Valid(data) {
+ return errors.New("invalid or oversized daemon JSON message")
+ }
+ return json.Unmarshal(data, value)
+}
+
+func (w *daemonWire) close() {
+ // Closing the socket interrupts its reader/writer before waiting for codec IO.
+ _ = w.conn.Close()
+ w.mu.Lock()
+ defer w.mu.Unlock()
+ if w.encoder != nil {
+ _ = w.encoder.Close()
+ }
+ if w.decoder != nil {
+ w.decoder.close()
+ }
+}
+
+type daemonDecodedChunk struct {
+ data []byte
+ err error
+}
+
+// The synchronous decoder writes each decoded block before requesting the
+// next compressed block. At a validated flush boundary the reader hands back
+// the collected message, then waits for the next WebSocket message without
+// returning EOF (which would discard stream history). One worker per connection
+// owns all decoder state and exits when the connection closes.
+type daemonStreamDecoder struct {
+ chunks chan []byte
+ results chan daemonDecodedChunk
+ stop chan struct{}
+ done chan struct{}
+ once sync.Once
+ inFrame bool
+ checksum bool
+}
+
+type daemonChunkReader struct {
+ stream *daemonStreamDecoder
+ input []byte
+ active bool
+ output bytes.Buffer
+}
+
+func newDaemonStreamDecoder() *daemonStreamDecoder {
+ d := &daemonStreamDecoder{chunks: make(chan []byte), results: make(chan daemonDecodedChunk), stop: make(chan struct{}), done: make(chan struct{})}
+ go func() {
+ defer close(d.done)
+ reader := &daemonChunkReader{stream: d}
+ decoder, err := zstd.NewReader(reader, zstd.WithDecoderConcurrency(1), zstd.WithDecoderMaxWindow(daemonZstdWindow), zstd.WithDecoderMaxMemory(8<<20))
+ if err == nil {
+ _, err = decoder.WriteTo(reader)
+ decoder.Close()
+ }
+ if err == nil {
+ err = io.ErrUnexpectedEOF
+ }
+ select {
+ case d.results <- daemonDecodedChunk{err: err}:
+ case <-d.stop:
+ }
+ }()
+ return d
+}
+
+func (r *daemonChunkReader) Read(p []byte) (int, error) {
+ if len(r.input) == 0 {
+ if r.active {
+ result := daemonDecodedChunk{data: bytes.Clone(r.output.Bytes())}
+ r.output.Reset()
+ r.active = false
+ select {
+ case r.stream.results <- result:
+ case <-r.stream.stop:
+ return 0, io.EOF
+ }
+ }
+ select {
+ case r.input = <-r.stream.chunks:
+ r.active = true
+ case <-r.stream.stop:
+ return 0, io.EOF
+ }
+ }
+ n := copy(p, r.input)
+ r.input = r.input[n:]
+ return n, nil
+}
+
+func (r *daemonChunkReader) Write(p []byte) (int, error) {
+ if len(p) > daemonMessageLimit-r.output.Len() {
+ return 0, errors.New("decompressed daemon message exceeds size limit")
+ }
+ return r.output.Write(p)
+}
+
+func (d *daemonStreamDecoder) decode(data []byte) ([]byte, error) {
+ if err := d.validateFlush(data); err != nil {
+ return nil, err
+ }
+ select {
+ case d.chunks <- data:
+ case <-d.done:
+ return nil, errors.New("daemon decompressor closed")
+ case <-d.stop:
+ return nil, io.EOF
+ }
+ select {
+ case result := <-d.results:
+ return result.data, result.err
+ case <-d.done:
+ return nil, errors.New("daemon decompressor failed")
+ case <-d.stop:
+ return nil, io.EOF
+ }
+}
+
+func (d *daemonStreamDecoder) close() { d.once.Do(func() { close(d.stop) }); <-d.done }
+
+// A partial block must never be joined to the next RPC message. Validate block
+// framing first; the library then validates actual compressed data and checksums.
+func (d *daemonStreamDecoder) validateFlush(data []byte) error {
+ if len(data) == 0 || len(data) > daemonMessageLimit+(1<<20) {
+ return errors.New("invalid compressed daemon message size")
+ }
+ inFrame, checksum := d.inFrame, d.checksum
+ for len(data) > 0 {
+ if !inFrame {
+ var header zstd.Header
+ if err := header.Decode(data); err != nil {
+ return fmt.Errorf("daemon zstd header: %w", err)
+ }
+ if header.Skippable || header.DictionaryID != 0 || header.WindowSize > daemonZstdWindow || header.SingleSegment && header.FrameContentSize > daemonZstdWindow {
+ return errors.New("unsupported daemon zstd frame")
+ }
+ data = data[header.HeaderSize:]
+ inFrame = true
+ checksum = header.HasCheckSum
+ }
+ if len(data) < 3 {
+ return errors.New("truncated daemon zstd block header")
+ }
+ block := uint32(data[0]) | uint32(data[1])<<8 | uint32(data[2])<<16
+ size := int(block >> 3)
+ kind := (block >> 1) & 3
+ last := block&1 != 0
+ if kind == 3 || size > 128<<10 {
+ return errors.New("invalid daemon zstd block")
+ }
+ if kind == 1 {
+ size = 1
+ }
+ if len(data) < 3+size {
+ return errors.New("truncated daemon zstd block")
+ }
+ data = data[3+size:]
+ if last {
+ if checksum {
+ if len(data) < 4 {
+ return errors.New("truncated daemon zstd checksum")
+ }
+ data = data[4:]
+ }
+ inFrame = false
+ }
+ }
+ d.inFrame, d.checksum = inFrame, checksum
+ return nil
+}
diff --git a/internal/codex/daemon_wire_test.go b/internal/codex/daemon_wire_test.go
new file mode 100644
index 0000000..3a23971
--- /dev/null
+++ b/internal/codex/daemon_wire_test.go
@@ -0,0 +1,196 @@
+package codex
+
+import (
+ "bytes"
+ "encoding/hex"
+ "encoding/json"
+ "net/http"
+ "net/http/httptest"
+ "os"
+ "strings"
+ "testing"
+ "time"
+
+ "github.com/gorilla/websocket"
+ "github.com/klauspost/compress/zstd"
+)
+
+func TestDaemonDecoderNativeZstdFlushes(t *testing.T) {
+ // Generated with native libzstd ZSTD_compressStream2(ZSTD_e_flush), the same
+ // stream framing used by Codex's Rust zstd crate. Later messages reuse history.
+ var fixtures []struct{ JSON, Hex string }
+ data, err := os.ReadFile("testdata/zstd-stream-v1.json")
+ if err != nil {
+ t.Fatal(err)
+ }
+ if err = json.Unmarshal(data, &fixtures); err != nil {
+ t.Fatal(err)
+ }
+ d := newDaemonStreamDecoder()
+ defer d.close()
+ for _, fixture := range fixtures {
+ input, err := hex.DecodeString(fixture.Hex)
+ if err != nil {
+ t.Fatal(err)
+ }
+ got, err := d.decode(input)
+ if err != nil || string(got) != fixture.JSON {
+ t.Fatal("native stream failed", string(got), err)
+ }
+ }
+ // Reconnect must discard the previous connection's dictionary/history.
+ fresh := newDaemonStreamDecoder()
+ defer fresh.close()
+ second, _ := hex.DecodeString(fixtures[1].Hex)
+ if _, err := fresh.decode(second); err == nil {
+ t.Fatal("continuation accepted without first frame")
+ }
+}
+
+func TestDaemonWireNegotiationAndStreaming(t *testing.T) {
+ for _, compressed := range []bool{false, true} {
+ t.Run(map[bool]string{false: "legacy", true: "zstd"}[compressed], func(t *testing.T) {
+ serverErr := make(chan error, 1)
+ server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ header := http.Header{}
+ if compressed {
+ header.Set(daemonCompressionHeader, daemonCompressionProtocol)
+ }
+ conn, err := (&websocket.Upgrader{}).Upgrade(w, r, header)
+ if err != nil {
+ serverErr <- err
+ return
+ }
+ wire, err := newDaemonWire(conn, header)
+ if err != nil {
+ serverErr <- err
+ return
+ }
+ defer wire.close()
+ for i := 0; i < 8; i++ {
+ var request map[string]any
+ if err = wire.readJSON(&request); err != nil {
+ serverErr <- err
+ return
+ }
+ if err = wire.writeJSON(request); err != nil {
+ serverErr <- err
+ return
+ }
+ }
+ serverErr <- nil
+ }))
+ defer server.Close()
+ conn, response, err := websocket.DefaultDialer.Dial("ws"+strings.TrimPrefix(server.URL, "http"), http.Header{http.CanonicalHeaderKey(daemonCompressionHeader): []string{daemonCompressionProtocol}})
+ if err != nil {
+ t.Fatal(err)
+ }
+ wire, err := newDaemonWire(conn, response.Header)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer wire.close()
+ for i := 0; i < 8; i++ {
+ text := strings.Repeat("retained stream history ", 15000)
+ if err = wire.writeJSON(map[string]any{"id": i, "text": text}); err != nil {
+ t.Fatal(err)
+ }
+ var result struct {
+ ID int
+ Text string
+ }
+ if err = wire.readJSON(&result); err != nil || result.ID != i || result.Text != text {
+ t.Fatal("round trip", result.ID, err)
+ }
+ }
+ if err := <-serverErr; err != nil {
+ t.Fatal(err)
+ }
+ })
+ }
+}
+
+func TestDaemonDecoderRejectsBrokenAndOversizedFrames(t *testing.T) {
+ var encoded bytes.Buffer
+ encoder, err := zstd.NewWriter(&encoded, zstd.WithEncoderConcurrency(1), zstd.WithWindowSize(daemonZstdWindow))
+ if err != nil {
+ t.Fatal(err)
+ }
+ encoder.Write([]byte(`{"result":"ok"}`))
+ encoder.Flush()
+ valid := bytes.Clone(encoded.Bytes())
+ encoder.Close()
+ for _, bad := range [][]byte{nil, []byte("invalid zstd"), valid[:len(valid)-1], append(bytes.Clone(valid), 1)} {
+ d := newDaemonStreamDecoder()
+ if _, err := d.decode(bad); err == nil {
+ t.Error("broken frame accepted")
+ }
+ d.close()
+ }
+ encoded.Reset()
+ encoder, _ = zstd.NewWriter(&encoded, zstd.WithEncoderConcurrency(1), zstd.WithWindowSize(daemonZstdWindow))
+ encoder.Write(bytes.Repeat([]byte{'x'}, daemonMessageLimit+1))
+ encoder.Flush()
+ d := newDaemonStreamDecoder()
+ if _, err := d.decode(encoded.Bytes()); err == nil {
+ t.Fatal("decompression limit ignored")
+ }
+ d.close()
+ encoder.Close()
+}
+
+func TestDaemonDecoderCloseUnblocksIdleStream(t *testing.T) {
+ d := newDaemonStreamDecoder()
+ done := make(chan struct{})
+ go func() { d.close(); close(done) }()
+ select {
+ case <-done:
+ case <-time.After(time.Second):
+ t.Fatal("idle decoder leaked")
+ }
+ d.close()
+}
+
+func TestDaemonWireRejectsUnknownAcknowledgement(t *testing.T) {
+ conn := &websocket.Conn{}
+ for _, values := range [][]string{{"future"}, {daemonCompressionProtocol, daemonCompressionProtocol}} {
+ if _, err := newDaemonWire(conn, http.Header{http.CanonicalHeaderKey(daemonCompressionHeader): values}); err == nil {
+ t.Fatal("unknown negotiation accepted")
+ }
+ }
+}
+
+func TestDaemonWireRejectsWrongMessageType(t *testing.T) {
+ for _, compressed := range []bool{false, true} {
+ server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ headers := http.Header{}
+ if compressed {
+ headers.Set(daemonCompressionHeader, daemonCompressionProtocol)
+ }
+ conn, err := (&websocket.Upgrader{}).Upgrade(w, r, headers)
+ if err != nil {
+ return
+ }
+ defer conn.Close()
+ kind := websocket.BinaryMessage
+ if compressed {
+ kind = websocket.TextMessage
+ }
+ _ = conn.WriteMessage(kind, []byte(`{"result":{}}`))
+ }))
+ conn, response, err := websocket.DefaultDialer.Dial("ws"+strings.TrimPrefix(server.URL, "http"), nil)
+ if err != nil {
+ t.Fatal(err)
+ }
+ wire, err := newDaemonWire(conn, response.Header)
+ if err != nil {
+ t.Fatal(err)
+ }
+ var value any
+ if err := wire.readJSON(&value); err == nil {
+ t.Error("message type accepted without matching negotiation")
+ }
+ wire.close()
+ server.Close()
+ }
+}
diff --git a/internal/codex/live_usage.go b/internal/codex/live_usage.go
index b58a9c6..a72e3e9 100644
--- a/internal/codex/live_usage.go
+++ b/internal/codex/live_usage.go
@@ -52,6 +52,7 @@ type LiveUsageSnapshot struct {
APIEqTierPremiumUSD float64
APIEqUnknownTierCalls int64
APIEqDiagnostics []APIEqDiagnostic
+ APIEqCoverageIssue string
}
// LiveUsageSession is one independently started local Codex session. Token
@@ -126,15 +127,18 @@ type LiveTurnTiming struct {
// sessions. It also extracts bounded display-only replies and request context;
// reasoning and arbitrary tool output are never retained.
type LiveUsageReader struct {
- sessionNames map[string]string
- nameIndexInfo os.FileInfo
- daemonContexts map[string]SessionContext
- daemonDirectories map[string]string
- SessionsRoot string
- WriterLocksRoot string
- statusProvider sessionStatusProvider
- commandMu sync.Mutex
- commandMenus *sessionCommandCache
+ knownUsageCoverage map[string]string
+ localFeatureChecks map[string]localFeatureCheck
+ discoveryCoverageIssue string
+ sessionNames map[string]string
+ nameIndexInfo os.FileInfo
+ daemonContexts map[string]SessionContext
+ daemonDirectories map[string]string
+ SessionsRoot string
+ WriterLocksRoot string
+ statusProvider sessionStatusProvider
+ commandMu sync.Mutex
+ commandMenus *sessionCommandCache
mu sync.Mutex
initialized bool
@@ -181,6 +185,10 @@ type rolloutCursor struct {
modelCalls []LiveModelCall
turnTimings []LiveTurnTiming
attention sessionAttentionState
+ compressed bool
+ compressedInfo os.FileInfo
+ compressedTail []byte
+ compressedDigest [32]byte
exactUsage bool
exactUsageUncertain bool
exactResponses map[string]BenchmarkUsage
@@ -378,6 +386,7 @@ func (r *LiveUsageReader) fetchTokenUsage(ctx context.Context, forceFullDiscover
}
exactStatuses := map[string]sessionRuntimeStatus(nil)
+ var daemonCoverage map[string]string
r.daemonContexts = nil
r.daemonDirectories = nil
appServerUp := false
@@ -387,6 +396,7 @@ func (r *LiveUsageReader) fetchTokenUsage(ctx context.Context, forceFullDiscover
appServerUp = true
exactStatuses = daemonSnapshot.Statuses
r.daemonContexts = daemonSnapshot.Contexts
+ daemonCoverage = daemonSnapshot.UsageCoverage
r.daemonDirectories = daemonSnapshot.WorkingDirectories
r.ingestResolvedModelObservations(daemonSnapshot.ModelObservations)
r.daemonSubscribedThreads = daemonSnapshot.SubscribedThreads
@@ -400,11 +410,36 @@ func (r *LiveUsageReader) fetchTokenUsage(ctx context.Context, forceFullDiscover
if err := contextErr(ctx); err != nil {
return LiveUsageSnapshot{}, err
}
+ if cursor.compressed {
+ info, err := os.Stat(path)
+ if errors.Is(err, os.ErrNotExist) {
+ continue
+ }
+ if err == nil {
+ err = r.restoreCompressedCursor(path, info, cursor)
+ }
+ if err != nil {
+ if firstReadErr == nil {
+ firstReadErr = err
+ }
+ continue
+ }
+ cursor = r.files[path]
+ }
if now.Sub(cursor.lastModified) > activeRolloutHorizon {
continue
}
if err := r.consume(path, cursor); err != nil {
if errors.Is(err, os.ErrNotExist) {
+ if info, compressedErr := os.Stat(path + ".zst"); compressedErr == nil {
+ if compressedErr = r.addCompressedFile(ctx, path+".zst", info); compressedErr != nil {
+ r.discoveryCoverageIssue = "compressed rollout history could not be verified"
+ if firstReadErr == nil {
+ firstReadErr = compressedErr
+ }
+ }
+ continue
+ }
delete(r.files, path)
continue
}
@@ -425,6 +460,10 @@ func (r *LiveUsageReader) fetchTokenUsage(ctx context.Context, forceFullDiscover
for i := range sessions {
sessions[i].Name = r.sessionNames[sessions[i].ID]
}
+ coverageIssue := r.usageCoverage(ctx, sessions, daemonCoverage)
+ if firstReadErr != nil {
+ coverageIssue = "local response telemetry could not be completely read"
+ }
codexStatusKnown, codexUp, codexWorking := codexRuntimeHealth(
appServerUp, len(liveWriters) > 0, sessionWorking, writerLocksSupported,
)
@@ -443,6 +482,7 @@ func (r *LiveUsageReader) fetchTokenUsage(ctx context.Context, forceFullDiscover
APIEqTierPremiumUSD: r.apiEqTierPremiumUSD,
APIEqUnknownTierCalls: r.apiEqUnknownTierCalls,
APIEqDiagnostics: append([]APIEqDiagnostic(nil), r.apiEqDiagnostics...),
+ APIEqCoverageIssue: coverageIssue,
}, nil
}
@@ -485,6 +525,9 @@ func (r *LiveUsageReader) initialize(ctx context.Context, now time.Time) error {
}
func (r *LiveUsageReader) discover(ctx context.Context, now time.Time, full bool) error {
+ if full {
+ r.discoveryCoverageIssue = ""
+ }
roots := []string{r.SessionsRoot}
if !full {
roots = roots[:0]
@@ -511,7 +554,7 @@ func (r *LiveUsageReader) discover(ctx context.Context, now time.Time, full bool
if err := contextErr(ctx); err != nil {
return err
}
- if entry.IsDir() || entry.Type()&os.ModeSymlink != 0 || !isRolloutFile(entry.Name()) {
+ if entry.IsDir() || entry.Type()&os.ModeSymlink != 0 || (!isRolloutFile(entry.Name()) && !isCompressedRolloutFile(entry.Name())) {
return nil
}
info, err := entry.Info()
@@ -521,7 +564,19 @@ func (r *LiveUsageReader) discover(ctx context.Context, now time.Time, full bool
}
return nil
}
+ if isCompressedRolloutFile(entry.Name()) {
+ if err := r.addCompressedFile(ctx, path, info); err != nil {
+ r.discoveryCoverageIssue = "compressed rollout history could not be verified"
+ if firstFileErr == nil {
+ firstFileErr = err
+ }
+ }
+ return nil
+ }
if cursor, exists := r.files[path]; exists {
+ if cursor.compressed {
+ return r.restoreCompressedCursor(path, info, cursor)
+ }
cursor.lastModified = info.ModTime()
return nil
}
@@ -604,6 +659,11 @@ func (r *LiveUsageReader) addFile(path string, info os.FileInfo) error {
r.apiEqAccountingAdjustments++
}
}
+ // Replacing this path changes the historical baseline, not the activity
+ // already observed by this process. Keep it only for the same thread.
+ if previous := r.files[path]; previous != nil && cursor.threadID != "" && previous.threadID == cursor.threadID {
+ cursor.observedTokens, cursor.modelCalls, cursor.turnTimings = previous.observedTokens, previous.modelCalls, previous.turnTimings
+ }
r.files[path] = cursor
return nil
}
@@ -1236,27 +1296,8 @@ func readRolloutMetadata(path string) (rolloutMetadata, error) {
reader := bufio.NewReader(file)
for record := 0; record < 64; record++ {
line, readErr := reader.ReadBytes('\n')
- if bytes.Contains(line, []byte(`"session_meta"`)) {
- var event rolloutMetadataEvent
- if json.Unmarshal(bytes.TrimSpace(line), &event) == nil && event.Type == "session_meta" {
- var payload rolloutMetadataPayload
- if json.Unmarshal(event.Payload, &payload) == nil {
- parent, nonRoot := sessionSourceParent(payload.Source)
- if payload.ParentThreadID != "" {
- parent = payload.ParentThreadID
- nonRoot = true
- }
- if payload.ThreadSource == "subagent" {
- nonRoot = true
- }
- return rolloutMetadata{
- ID: payload.ID, ParentThreadID: parent,
- WorkingDirectory: payload.WorkingDirectory, NonRoot: nonRoot,
- StartedAt: firstNonZeroTime(payload.Timestamp, event.Timestamp),
- SubagentHistoryStartOrdinal: payload.SubagentHistoryStartOrdinal,
- }, nil
- }
- }
+ if metadata, ok := rolloutMetadataRecord(line); ok {
+ return metadata, nil
}
if readErr != nil {
if errors.Is(readErr, io.EOF) {
@@ -1268,6 +1309,32 @@ func readRolloutMetadata(path string) (rolloutMetadata, error) {
return rolloutMetadata{}, nil
}
+func rolloutMetadataRecord(line []byte) (rolloutMetadata, bool) {
+ if bytes.Contains(line, []byte(`"session_meta"`)) {
+ var event rolloutMetadataEvent
+ if json.Unmarshal(bytes.TrimSpace(line), &event) == nil && event.Type == "session_meta" {
+ var payload rolloutMetadataPayload
+ if json.Unmarshal(event.Payload, &payload) == nil {
+ parent, nonRoot := sessionSourceParent(payload.Source)
+ if payload.ParentThreadID != "" {
+ parent = payload.ParentThreadID
+ nonRoot = true
+ }
+ if payload.ThreadSource == "subagent" {
+ nonRoot = true
+ }
+ return rolloutMetadata{
+ ID: payload.ID, ParentThreadID: parent,
+ WorkingDirectory: payload.WorkingDirectory, NonRoot: nonRoot,
+ StartedAt: firstNonZeroTime(payload.Timestamp, event.Timestamp),
+ SubagentHistoryStartOrdinal: payload.SubagentHistoryStartOrdinal,
+ }, true
+ }
+ }
+ }
+ return rolloutMetadata{}, false
+}
+
func sessionSourceParent(raw json.RawMessage) (string, bool) {
if len(raw) == 0 || bytes.Equal(bytes.TrimSpace(raw), []byte("null")) {
return "", false
diff --git a/internal/codex/live_usage_compressed.go b/internal/codex/live_usage_compressed.go
new file mode 100644
index 0000000..07740d0
--- /dev/null
+++ b/internal/codex/live_usage_compressed.go
@@ -0,0 +1,190 @@
+package codex
+
+import (
+ "bufio"
+ "bytes"
+ "context"
+ "crypto/sha256"
+ "errors"
+ "fmt"
+ "io"
+ "os"
+ "strings"
+
+ "github.com/klauspost/compress/zstd"
+)
+
+const compressedHistoryLimit = 256 << 20
+const rolloutBaselineLineLimit = 4 << 20
+
+func isCompressedRolloutFile(name string) bool {
+ return strings.HasSuffix(name, ".zst") && isRolloutFile(strings.TrimSuffix(name, ".zst"))
+}
+
+// Compressed histories are immutable baselines, never new spending. Keep their
+// logical .jsonl identity and decoded offset so restoring them preserves both
+// observed totals and the boundary between history and concurrent appends.
+// No transcript copy is written to disk and retained memory is bounded.
+func (r *LiveUsageReader) addCompressedFile(ctx context.Context, path string, info os.FileInfo) error {
+ plain := strings.TrimSuffix(path, ".zst")
+ if _, err := os.Stat(plain); err == nil {
+ return nil
+ } else if !errors.Is(err, os.ErrNotExist) {
+ return err
+ }
+ old := r.files[plain]
+ if old != nil && old.compressed && old.compressedInfo.Size() == info.Size() && old.compressedInfo.ModTime() == info.ModTime() {
+ return nil
+ }
+ cursor, err := compressedRolloutBaseline(ctx, path, info)
+ if err != nil {
+ return fmt.Errorf("read compressed rollout: %w", err)
+ }
+ if old != nil && old.threadID == cursor.threadID {
+ cursor.observedTokens, cursor.modelCalls, cursor.turnTimings = old.observedTokens, old.modelCalls, old.turnTimings
+ if old.offset != cursor.offset {
+ r.apiEqAccountingAdjustments++
+ }
+ }
+ r.files[plain] = cursor
+ return nil
+}
+
+func compressedRolloutBaseline(ctx context.Context, path string, info os.FileInfo) (*rolloutCursor, error) {
+ if info.Size() > compressedHistoryLimit {
+ return nil, errors.New("compressed history exceeds 256 MiB source limit")
+ }
+ file, err := os.Open(path)
+ if err != nil {
+ return nil, err
+ }
+ defer file.Close()
+ decoder, err := zstd.NewReader(io.LimitReader(file, compressedHistoryLimit+1), zstd.WithDecoderConcurrency(1), zstd.WithDecoderMaxWindow(128<<20), zstd.WithDecoderMaxMemory(128<<20))
+ if err != nil {
+ return nil, err
+ }
+ defer decoder.Close()
+ reader := bufio.NewReaderSize(io.LimitReader(decoder, compressedHistoryLimit+1), 64<<10)
+ cursor := &rolloutCursor{lastModified: info.ModTime(), compressed: true, compressedInfo: info}
+ var line, tail []byte
+ digest := sha256.New()
+ var decoded int64
+ oversized := false
+ for {
+ if err := ctx.Err(); err != nil {
+ return nil, err
+ }
+ chunk, readErr := reader.ReadSlice('\n')
+ decoded += int64(len(chunk))
+ digest.Write(chunk)
+ if decoded > compressedHistoryLimit {
+ return nil, errors.New("decoded history exceeds 256 MiB baseline limit")
+ }
+ tail = append(tail, chunk...)
+ if len(tail) > 4096 {
+ tail = append(tail[:0], tail[len(tail)-4096:]...)
+ }
+ if len(line)+len(chunk) <= rolloutBaselineLineLimit && !oversized {
+ line = append(line, chunk...)
+ } else {
+ oversized = true
+ cursor.exactUsageUncertain = true
+ line = nil
+ }
+ if errors.Is(readErr, bufio.ErrBufferFull) {
+ continue
+ }
+ if len(chunk) > 0 && chunk[len(chunk)-1] == '\n' {
+ cursor.offset = decoded
+ copy(cursor.compressedDigest[:], digest.Sum(nil))
+ cursor.compressedTail = append(cursor.compressedTail[:0], tail...)
+ if !oversized {
+ baselineRolloutRecord(cursor, line)
+ }
+ line = line[:0]
+ oversized = false
+ }
+ if readErr != nil {
+ if !errors.Is(readErr, io.EOF) {
+ return nil, readErr
+ }
+ break
+ }
+ }
+ cursor.exactUsageUncertain = cursor.exactUsageUncertain && !cursor.exactUsage
+ if cursor.threadID == "" {
+ return nil, errors.New("compressed history omitted session metadata")
+ }
+ return cursor, nil
+}
+
+func baselineRolloutRecord(c *rolloutCursor, line []byte) {
+ if metadata, ok := rolloutMetadataRecord(line); ok && c.threadID == "" {
+ c.threadID, c.parentThreadID, c.workingDirectory = metadata.ID, metadata.ParentThreadID, metadata.WorkingDirectory
+ c.startedAt, c.nonRoot, c.subagentHistoryStartOrdinal = metadata.StartedAt, metadata.NonRoot, metadata.SubagentHistoryStartOrdinal
+ }
+ if settings, ordinal, at, ok := rolloutSessionSettingsRecord(line, c.modelSettings); ok && tokenRecordIsOwned(ordinal, c.subagentHistoryStartOrdinal, at, c.nonRoot, c.startedAt) {
+ c.modelSettings = settings
+ }
+ if tier, ordinal, at, ok := rolloutTierRecord(line); ok && tokenRecordIsOwned(ordinal, c.subagentHistoryStartOrdinal, at, c.nonRoot, c.startedAt) {
+ c.nextServiceTier = tier
+ }
+ if model, ordinal, at, ok := rolloutModelRecord(line); ok && tokenRecordIsOwned(ordinal, c.subagentHistoryStartOrdinal, at, c.nonRoot, c.startedAt) {
+ c.currentModel, c.currentServiceTier = model, c.nextServiceTier
+ }
+ if turn, ordinal, at, ok := rolloutTurnIDRecord(line); ok && tokenRecordIsOwned(ordinal, c.subagentHistoryStartOrdinal, at, c.nonRoot, c.startedAt) {
+ c.currentTurnID = turn
+ }
+ if attention, ordinal, at, ok := sessionAttentionRecord(line); ok && tokenRecordIsOwned(ordinal, c.subagentHistoryStartOrdinal, at, c.nonRoot, c.startedAt) {
+ c.attention = attention
+ }
+ if preview, ok := rolloutContextRecord(line, c); ok {
+ c.preview = preview
+ }
+ if record, ok := tokenUsageRecord(line); ok {
+ c.totalTokens = record.total
+ }
+ if record, ok := exactUsageRecord(line); ok && c.ownsExactRecord(record) {
+ c.exactUsage = true
+ usage := BenchmarkUsage{}
+ if record.Payload.Usage != nil {
+ usage = record.Payload.Usage.benchmarkUsage()
+ }
+ if id := strings.TrimSpace(record.Payload.ResponseID); id != "" {
+ c.rememberExactResponse(id, usage)
+ }
+ }
+}
+
+func (r *LiveUsageReader) restoreCompressedCursor(path string, info os.FileInfo, cursor *rolloutCursor) error {
+ file, err := os.Open(path)
+ if err != nil {
+ return err
+ }
+ tail := make([]byte, len(cursor.compressedTail))
+ n, readErr := file.ReadAt(tail, cursor.offset-int64(len(tail)))
+ // The tail is a quick mismatch check; hash the entire bounded prefix before
+ // trusting its settings and offset. Reverts may change earlier context while
+ // retaining an identical tail and file length.
+ matches := false
+ if readErr == nil && n == len(tail) && bytes.Equal(tail, cursor.compressedTail) && info.Size() >= cursor.offset {
+ digest := sha256.New()
+ _, err = io.Copy(digest, io.NewSectionReader(file, 0, cursor.offset))
+ if err != nil {
+ file.Close()
+ return err
+ }
+ matches = bytes.Equal(digest.Sum(nil), cursor.compressedDigest[:])
+ }
+ file.Close()
+ if matches {
+ cursor.compressed = false
+ cursor.compressedInfo = nil
+ cursor.compressedTail = nil
+ cursor.lastModified = info.ModTime()
+ return nil
+ }
+ // A changed/truncated restoration cannot reuse an offset from different
+ // history. Existing-thread discovery rebases it rather than billing copies.
+ return r.addFile(path, info)
+}
diff --git a/internal/codex/live_usage_compressed_test.go b/internal/codex/live_usage_compressed_test.go
new file mode 100644
index 0000000..80002a0
--- /dev/null
+++ b/internal/codex/live_usage_compressed_test.go
@@ -0,0 +1,253 @@
+package codex
+
+import (
+ "context"
+ "os"
+ "path/filepath"
+ "strings"
+ "testing"
+
+ "github.com/klauspost/compress/zstd"
+)
+
+func compressTestRollout(t *testing.T, path string, remove bool) []byte {
+ t.Helper()
+ plain, err := os.ReadFile(path)
+ if err != nil {
+ t.Fatal(err)
+ }
+ encoder, err := zstd.NewWriter(nil)
+ if err != nil {
+ t.Fatal(err)
+ }
+ data := encoder.EncodeAll(plain, nil)
+ encoder.Close()
+ if err := os.WriteFile(path+".zst", data, 0600); err != nil {
+ t.Fatal(err)
+ }
+ if remove {
+ if err := os.Remove(path); err != nil {
+ t.Fatal(err)
+ }
+ }
+ return plain
+}
+
+func TestCompressedHistoryRestorationCountsOnlyFreshResponses(t *testing.T) {
+ _, path, now := astraUsageReader(t, 100)
+ usage := BenchmarkUsage{InputTokens: 1000, OutputTokens: 100, TotalTokens: 1100}
+ old := exactLine(t, now, "astra-accounting", "turn", "historical", usage) + "\n"
+ appendRollout(t, path, old)
+ plain := compressTestRollout(t, path, true)
+ home := strings.Split(path, string(filepath.Separator)+"sessions"+string(filepath.Separator))[0]
+ r, err := NewLiveUsageReader(home)
+ if err != nil {
+ t.Fatal(err)
+ }
+ r.statusProvider = nil
+ baseline, err := r.FetchTokenUsage(context.Background())
+ if err != nil || baseline.TotalTokens != 0 || len(r.files) != 1 || !r.files[path].compressed || !r.files[path].exactUsage {
+ t.Fatal("compressed history was lost or billed", baseline, err)
+ }
+ // Codex restores history and can append before our next discovery pass.
+ if err := os.WriteFile(path, plain, 0600); err != nil {
+ t.Fatal(err)
+ }
+ appendRollout(t, path, old+exactLine(t, now, "astra-accounting", "turn", "new", usage)+"\n")
+ fresh, err := r.FetchTokenUsageFresh(context.Background())
+ if err != nil || fresh.TotalTokens != 1100 || fresh.APIEqPricedCalls != 1 || len(r.files) != 1 || r.files[path].compressed {
+ t.Fatal("restoration billed history or missed append", fresh, err)
+ }
+ next, err := r.FetchTokenUsageFresh(context.Background())
+ if err != nil || next.TotalTokens != fresh.TotalTokens || next.APIEqUSD != fresh.APIEqUSD {
+ t.Fatal("plain/compressed siblings duplicated accounting", next, err)
+ }
+}
+
+func TestCompressionWhileObservedPreservesSessionTotals(t *testing.T) {
+ r, path, now := astraUsageReader(t, 100)
+ r.statusProvider = nil
+ usage := BenchmarkUsage{InputTokens: 1000, OutputTokens: 100, TotalTokens: 1100}
+ appendRollout(t, path, exactLine(t, now, "astra-accounting", "turn", "first", usage)+"\n")
+ before, err := r.FetchTokenUsage(context.Background())
+ if err != nil {
+ t.Fatal(err)
+ }
+ plain := compressTestRollout(t, path, true)
+ compressed, err := r.FetchTokenUsageFresh(context.Background())
+ if err != nil || compressed.TotalTokens != before.TotalTokens || compressed.APIEqUSD != before.APIEqUSD || len(r.files) != 1 || !r.files[path].compressed {
+ t.Fatal("compression discarded or recounted observed history", compressed, err)
+ }
+ if err := os.WriteFile(path, plain, 0600); err != nil {
+ t.Fatal(err)
+ }
+ appendRollout(t, path, exactLine(t, now, "astra-accounting", "turn", "second", usage)+"\n")
+ restored, err := r.FetchTokenUsage(context.Background())
+ if err != nil || restored.TotalTokens != 2200 || restored.APIEqPricedCalls != 2 || restored.Sessions[0].TotalTokens != 2200 {
+ t.Fatal("restoration lost existing totals", restored, err)
+ }
+}
+
+func TestChangedCompressedRestorationRebasesCopiedHistory(t *testing.T) {
+ _, path, now := astraUsageReader(t, 100)
+ usage := BenchmarkUsage{InputTokens: 1000, OutputTokens: 100, TotalTokens: 1100}
+ appendRollout(t, path, exactLine(t, now, "astra-accounting", "turn", "old", usage)+"\n")
+ plain := compressTestRollout(t, path, true)
+ home := strings.Split(path, string(filepath.Separator)+"sessions"+string(filepath.Separator))[0]
+ r, err := NewLiveUsageReader(home)
+ if err != nil {
+ t.Fatal(err)
+ }
+ r.statusProvider = nil
+ if _, err := r.FetchTokenUsage(context.Background()); err != nil {
+ t.Fatal(err)
+ }
+ // Revert/rewrite changes the baseline prefix; none of this is fresh spend.
+ if err := os.WriteFile(path, []byte(strings.ReplaceAll(string(plain), "old", "rewritten")), 0600); err != nil {
+ t.Fatal(err)
+ }
+ rebased, err := r.FetchTokenUsageFresh(context.Background())
+ if err != nil || rebased.TotalTokens != 0 || rebased.APIEqPricedCalls != 0 || rebased.APIEqAccountingAdjustments != 1 {
+ t.Fatal("changed restoration billed copied history", rebased, err)
+ }
+}
+
+func TestCorruptCompressedHistoryIsRejected(t *testing.T) {
+ _, path, _ := astraUsageReader(t, 100)
+ if err := os.Remove(path); err != nil {
+ t.Fatal(err)
+ }
+ if err := os.WriteFile(path+".zst", []byte("invalid zstd"), 0600); err != nil {
+ t.Fatal(err)
+ }
+ home := strings.Split(path, string(filepath.Separator)+"sessions"+string(filepath.Separator))[0]
+ r, err := NewLiveUsageReader(home)
+ if err != nil {
+ t.Fatal(err)
+ }
+ r.statusProvider = nil
+ if _, err := r.FetchTokenUsage(context.Background()); err == nil {
+ t.Fatal("corrupt compressed history accepted")
+ }
+ if len(r.files) != 0 {
+ t.Fatal("partial compressed baseline retained")
+ }
+}
+
+func TestCompressedRestorationChecksContextBeforeUnchangedTail(t *testing.T) {
+ _, path, now := astraUsageReader(t, 100)
+ // Put changed model context outside the suffix check, retaining byte length.
+ appendRollout(t, path, strings.Repeat(`{"type":"response_item","payload":"padding"}`+"\n", 200))
+ usage := BenchmarkUsage{InputTokens: 1000, OutputTokens: 100, TotalTokens: 1100}
+ appendRollout(t, path, exactLine(t, now, "astra-accounting", "turn", "old", usage)+"\n")
+ plain := compressTestRollout(t, path, true)
+ home := strings.Split(path, string(filepath.Separator)+"sessions"+string(filepath.Separator))[0]
+ r, err := NewLiveUsageReader(home)
+ if err != nil {
+ t.Fatal(err)
+ }
+ r.statusProvider = nil
+ if _, err := r.FetchTokenUsage(context.Background()); err != nil {
+ t.Fatal(err)
+ }
+ changed := strings.ReplaceAll(string(plain), "gpt-6-astra", "gpt-6.1-sol")
+ if len(changed) != len(plain) || changed == string(plain) {
+ t.Fatal("test did not preserve history size")
+ }
+ if err := os.WriteFile(path, []byte(changed), 0600); err != nil {
+ t.Fatal(err)
+ }
+ got, err := r.FetchTokenUsageFresh(context.Background())
+ if err != nil || got.APIEqAccountingAdjustments != 1 || got.APIEqPricedCalls != 0 || r.files[path].currentModel != "gpt-6.1-sol" {
+ t.Fatal("changed context trusted because suffix matched", got, err)
+ }
+}
+
+func TestCompressedDiscoveryWarningSurvivesEarlierPlainReadFailure(t *testing.T) {
+ r, _, _ := astraUsageReader(t, 100)
+ r.statusProvider = nil
+ // WalkDir visits the unreadable plain rollout before the corrupt compressed
+ // one. A third, healthy cursor keeps discovery usable rather than fatal.
+ denied := filepath.Join(r.SessionsRoot, "rollout-a.jsonl")
+ if err := os.WriteFile(denied, []byte("unreadable"), 0600); err != nil {
+ t.Fatal(err)
+ }
+ if err := os.Chmod(denied, 0); err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(func() { _ = os.Chmod(denied, 0600) })
+ if file, err := os.Open(denied); err == nil {
+ file.Close()
+ t.Skip("filesystem permits reading files with mode 000")
+ }
+ corrupt := filepath.Join(r.SessionsRoot, "rollout-z.jsonl.zst")
+ if err := os.WriteFile(corrupt, []byte("corrupt zstd"), 0600); err != nil {
+ t.Fatal(err)
+ }
+ got, err := r.FetchTokenUsageFresh(context.Background())
+ if err != nil || !strings.Contains(got.APIEqCoverageIssue, "compressed rollout history") || len(r.files) != 1 {
+ t.Fatal("earlier read failure suppressed compressed coverage warning", got.APIEqCoverageIssue, err)
+ }
+}
+
+func TestRewrittenCompressedRestorePreservesObservedActivity(t *testing.T) {
+ r, path, now := astraUsageReader(t, 100)
+ r.statusProvider = nil
+ usage := BenchmarkUsage{InputTokens: 1000, OutputTokens: 100, TotalTokens: 1100}
+ appendRollout(t, path, exactLine(t, now, "astra-accounting", "turn", "first", usage)+"\n"+turnTimingLine(now, 42)+"\n")
+ before, err := r.FetchTokenUsage(context.Background())
+ if err != nil || len(before.Sessions) != 1 || len(before.Sessions[0].ModelCalls) != 1 || len(before.Sessions[0].TurnTimings) != 1 {
+ t.Fatal("observed activity fixture missing", before, err)
+ }
+ observed := before.Sessions[0]
+ plain := compressTestRollout(t, path, true)
+ if _, err := r.FetchTokenUsageFresh(context.Background()); err != nil {
+ t.Fatal(err)
+ }
+ rewritten := strings.ReplaceAll(string(plain), "first", "rewritten")
+ if err := os.WriteFile(path, []byte(rewritten), 0600); err != nil {
+ t.Fatal(err)
+ }
+ rebased, err := r.FetchTokenUsageFresh(context.Background())
+ if err != nil || rebased.APIEqAccountingAdjustments != 1 || rebased.TotalTokens != before.TotalTokens || rebased.APIEqUSD != before.APIEqUSD || len(rebased.Sessions) != 1 {
+ t.Fatal("rewrite changed reader-wide accounting", rebased, err)
+ }
+ session := rebased.Sessions[0]
+ if session.TotalTokens != observed.TotalTokens || len(session.ModelCalls) != 1 || session.ModelCalls[0] != observed.ModelCalls[0] || len(session.TurnTimings) != 1 || session.TurnTimings[0] != observed.TurnTimings[0] {
+ t.Fatal("rewrite discarded observed session history", session)
+ }
+ appendRollout(t, path, exactLine(t, now, "astra-accounting", "turn", "fresh", usage)+"\n")
+ fresh, err := r.FetchTokenUsage(context.Background())
+ if err != nil || fresh.TotalTokens != 2200 || fresh.APIEqPricedCalls != 2 || fresh.Sessions[0].TotalTokens != 2200 || len(fresh.Sessions[0].ModelCalls) != 2 || len(fresh.Sessions[0].TurnTimings) != 1 {
+ t.Fatal("fresh response did not extend preserved history", fresh, err)
+ }
+}
+
+func TestReplacingRolloutWithDifferentThreadDoesNotInheritActivity(t *testing.T) {
+ r, path, now := astraUsageReader(t, 100)
+ r.statusProvider = nil
+ usage := BenchmarkUsage{InputTokens: 1000, OutputTokens: 100, TotalTokens: 1100}
+ appendRollout(t, path, exactLine(t, now, "astra-accounting", "turn", "first", usage)+"\n"+turnTimingLine(now, 42)+"\n")
+ if _, err := r.FetchTokenUsage(context.Background()); err != nil {
+ t.Fatal(err)
+ }
+ original, err := os.ReadFile(path)
+ if err != nil {
+ t.Fatal(err)
+ }
+ changed := strings.ReplaceAll(string(original), "astra-accounting", "different-thread")
+ if err := os.WriteFile(path, []byte(changed), 0600); err != nil {
+ t.Fatal(err)
+ }
+ info, err := os.Stat(path)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if err := r.addFile(path, info); err != nil {
+ t.Fatal(err)
+ }
+ cursor := r.files[path]
+ if cursor.threadID != "different-thread" || cursor.observedTokens != 0 || len(cursor.modelCalls) != 0 || len(cursor.turnTimings) != 0 {
+ t.Fatal("observed activity crossed thread identities", cursor)
+ }
+}
diff --git a/internal/codex/live_usage_exact_test.go b/internal/codex/live_usage_exact_test.go
index eac8339..696348c 100644
--- a/internal/codex/live_usage_exact_test.go
+++ b/internal/codex/live_usage_exact_test.go
@@ -278,3 +278,43 @@ func TestExactUsageOlderThanStartupTailCannotEnableLegacyBilling(t *testing.T) {
}
}
}
+
+// Codex 0.163+ persists spending from incomplete Responses API requests even
+// when the turn fails. These records must not depend on a successful turn or
+// a matching (potentially recomputed) context counter.
+func TestExactIncompleteResponseSpendingSurvivesFailedTurn(t *testing.T) {
+ r, path, now := astraUsageReader(t, 100)
+ r.statusProvider = nil
+ usage := BenchmarkUsage{InputTokens: 1000, CachedInputTokens: 200, OutputTokens: 100, ReasoningOutputTokens: 80, TotalTokens: 1100}
+ response := exactLine(t, now, "astra-accounting", "failed-turn", "incomplete-response", usage) + "\n"
+ appendRollout(t, path, response+`{"type":"event_msg","payload":{"type":"error","message":"Incomplete response returned, reason: max_output_tokens"}}`+"\n"+richTokenCountLine(now, 123456, 1000, 200, 0, 100)+"\n")
+ failed, err := r.FetchTokenUsage(context.Background())
+ want, _, _ := EstimateStandardAPIEqCost("gpt-6-astra", usage)
+ if err != nil || failed.APIEqPricedCalls != 1 || failed.TotalTokens != 1100 || math.Abs(failed.APIEqUSD-want) > 1e-12 || failed.APIEqInconsistentUsageCalls != 0 {
+ t.Fatal("failed response spending was dropped or tied to context", failed, err)
+ }
+ // A reconnect replay and a retained-reasoning context correction are not
+ // additional spending, even though the context total moves backwards.
+ appendRollout(t, path, response+richTokenCountLine(now, 120000, 1000, 200, 0, 100)+"\n")
+ replayed, err := r.FetchTokenUsage(context.Background())
+ if err != nil || replayed.TotalTokens != failed.TotalTokens || replayed.APIEqUSD != failed.APIEqUSD || replayed.APIEqPricedCalls != 1 || replayed.APIEqAccountingAdjustments != 0 {
+ t.Fatal("failed response replay or context correction was billed twice", replayed, err)
+ }
+}
+
+func TestExactIncompleteResponseWithoutUsageRemainsMissing(t *testing.T) {
+ r, path, now := astraUsageReader(t, 100)
+ r.statusProvider = nil
+ line := exactLine(t, now, "astra-accounting", "failed-turn", "incomplete-response", BenchmarkUsage{})
+ var record map[string]any
+ if err := json.Unmarshal([]byte(line), &record); err != nil {
+ t.Fatal(err)
+ }
+ record["payload"].(map[string]any)["usage"] = nil
+ raw, _ := json.Marshal(record)
+ appendRollout(t, path, string(raw)+"\n"+richTokenCountLine(now, 1200, 1000, 0, 0, 100)+"\n")
+ got, err := r.FetchTokenUsage(context.Background())
+ if err != nil || got.APIEqMissingUsageCalls != 1 || got.APIEqPricedCalls != 0 || got.APIEqUnpricedCalls != 0 || got.TotalTokens != 0 || len(got.APIEqDiagnostics) != 1 || got.APIEqDiagnostics[0].ResponseID != "incomplete-response" {
+ t.Fatal("missing incomplete-response usage was guessed from context", got, err)
+ }
+}
diff --git a/internal/codex/session_commands_unix.go b/internal/codex/session_commands_unix.go
index 9eb4450..b6e15fb 100644
--- a/internal/codex/session_commands_unix.go
+++ b/internal/codex/session_commands_unix.go
@@ -478,16 +478,26 @@ func (p *daemonStatusProvider) ExecuteSessionCommand(ctx context.Context, id, pa
}
}
}
- if m.Picker {
+ if permission, ok := o.params["permissions"].(string); ok {
+ return p.writeVerifiedPermissions(ctx, conn, id, permission, o.params)
+ }
+ _, changingModel := o.params["model"]
+ if m.Picker || changingModel {
current, err := p.readQuotaSession(ctx, id)
if err != nil {
return err
}
+ if model, ok := o.params["model"].(string); ok {
+ current.Model = model
+ }
+ if effort, ok := o.params["effort"].(string); ok {
+ current.Effort = effort
+ }
current.Tier = nil
if tier, ok := o.params["serviceTier"].(string); ok {
current.Tier = &tier
}
- return p.writeQuotaSettings(ctx, current, o.params)
+ return p.writeQuotaSettingsOn(ctx, conn, current, o.params)
}
return p.requestOn(ctx, conn, o.method, o.params, nil)
}
@@ -516,7 +526,7 @@ func (p *daemonStatusProvider) commandCatalogue(ctx context.Context, t commandTh
switch path {
case "skills", "hooks":
params["cwds"] = []string{t.Cwd}
- case "apps", "mcp":
+ case "apps", "mcp", "experimental":
params["threadId"] = t.ID
}
rows, err := p.commandRows(ctx, method, params)
diff --git a/internal/codex/session_compatibility_unix_test.go b/internal/codex/session_compatibility_unix_test.go
new file mode 100644
index 0000000..4c45b8f
--- /dev/null
+++ b/internal/codex/session_compatibility_unix_test.go
@@ -0,0 +1,138 @@
+//go:build unix
+
+package codex
+
+import (
+ "context"
+ "encoding/json"
+ "errors"
+ "strings"
+ "testing"
+ "time"
+)
+
+func TestSettingsCompatibilityAcrossClientsAndReconnect(t *testing.T) {
+ for _, compressed := range []bool{false, true} {
+ t.Run(map[bool]string{false: "plain", true: "zstd"}[compressed], func(t *testing.T) {
+ p, f := newQuotaDaemonTransport(t, compressed)
+ f.catalogues = map[string]map[string]any{
+ "permissionProfile/list": {"data": []any{map[string]any{"id": "restricted", "description": "Workspace access only", "allowed": true}}},
+ "experimentalFeature/list": {"data": []any{}},
+ }
+ ctx := context.Background()
+ // A second attached client reads the same persisted runtime selections.
+ second := &daemonStatusProvider{socketPath: p.socketPath}
+ t.Cleanup(func() { second.disconnect(nil) })
+ model, err := p.SessionCommands(ctx, "one", "model/small")
+ if err != nil {
+ t.Fatal(err)
+ }
+ if err = p.ExecuteSessionCommand(ctx, "one", model.Path, model.Revision, model.Choices[0].ID); err != nil {
+ t.Fatal(err)
+ }
+ permissions, err := p.SessionCommands(ctx, "one", "permissions")
+ if err != nil || len(permissions.Choices) != 1 {
+ t.Fatal(permissions, err)
+ }
+ if err = p.ExecuteSessionCommand(ctx, "one", permissions.Path, permissions.Revision, permissions.Choices[0].ID); err != nil {
+ t.Fatal(err)
+ }
+ if err = second.ensureConnected(ctx); err != nil {
+ t.Fatal(err)
+ }
+ current, err := second.readQuotaSession(ctx, "one")
+ if err != nil || current.Model != "small" || current.Effort != "medium" || current.Tier != nil {
+ t.Fatal("second client lost applied selections", current, err)
+ }
+ p.disconnect(nil)
+ if err = p.ensureConnected(ctx); err != nil {
+ t.Fatal(err)
+ }
+ resumed, err := p.readQuotaSession(ctx, "one")
+ if err != nil || !sameQuotaSettings(current, resumed) {
+ t.Fatal("reconnect replaced saved settings", resumed, err)
+ }
+ f.mu.Lock()
+ defer f.mu.Unlock()
+ if f.permissions["one"] != "restricted" || f.sessions["two"].Model != "large" || len(f.writes) != 2 {
+ t.Fatal("settings escaped selected thread", f.writes)
+ }
+ for _, read := range f.resumeReads {
+ if len(read) != 2 || read["threadId"] == nil || read["excludeTurns"] != true {
+ t.Fatal("read-only resume supplied overrides", read)
+ }
+ }
+ })
+ }
+}
+
+func TestQueuedModelAndPermissionAcknowledgementsAreNotApplied(t *testing.T) {
+ for _, path := range []string{"model/small", "permissions"} {
+ t.Run(path, func(t *testing.T) {
+ p, f := newQuotaDaemonTransport(t, true)
+ f.queued = true
+ f.catalogues = map[string]map[string]any{"permissionProfile/list": {"data": []any{map[string]any{"id": "restricted", "description": "Workspace access only", "allowed": true}}}}
+ menu, err := p.SessionCommands(context.Background(), "one", path)
+ if err != nil || len(menu.Choices) != 1 {
+ t.Fatal(menu, err)
+ }
+ ctx, cancel := context.WithTimeout(context.Background(), 300*time.Millisecond)
+ defer cancel()
+ done := make(chan error, 1)
+ go func() { done <- p.ExecuteSessionCommand(ctx, "one", menu.Path, menu.Revision, menu.Choices[0].ID) }()
+ // State updates for another attached session and approval state remain
+ // responsive while the selected thread's queued write is being verified.
+ p.handleNotification("thread/settings/updated", json.RawMessage(`{"threadId":"two","threadSettings":{"model":"other","effort":"high","serviceTier":"priority"}}`))
+ responsive := make(chan struct{})
+ go func() { p.SessionPrompt("two"); close(responsive) }()
+ select {
+ case <-responsive:
+ case <-time.After(time.Second):
+ t.Fatal("settings verification froze event state")
+ }
+ err = <-done
+ if err == nil {
+ t.Fatal("queued acknowledgement reported as applied")
+ }
+ if path == "model/small" && !errors.Is(err, ErrQuotaProfileUnverified) {
+ t.Fatal("model failure lost unverified result", err)
+ }
+ if path == "permissions" && !strings.Contains(err.Error(), "unconfirmed") {
+ t.Fatal("permission failure lost unconfirmed result", err)
+ }
+ f.mu.Lock()
+ writes := len(f.writes)
+ f.mu.Unlock()
+ if writes != 1 {
+ t.Fatal("queued mutation was retried", writes)
+ }
+ })
+ }
+}
+
+func TestAccountChangeInvalidatesCatalogueWithoutDiscardingConversation(t *testing.T) {
+ p, f := newQuotaDaemon(t)
+ f.catalogues = map[string]map[string]any{"experimentalFeature/list": {"data": []any{}}}
+ if _, ok := p.Fetch(context.Background(), []string{"one"}); !ok {
+ t.Fatal("daemon unavailable")
+ }
+ c := Client{LiveUsage: &LiveUsageReader{statusProvider: p}}
+ if _, err := c.SessionCommands(context.Background(), "one", "model/small"); err != nil {
+ t.Fatal(err)
+ }
+ p.mu.Lock()
+ if p.contexts == nil {
+ p.contexts = make(map[string]*daemonContextState)
+ }
+ p.contexts["one"] = &daemonContextState{latest: SessionContext{Text: "retained reply"}}
+ p.mu.Unlock()
+ p.handleNotification("account/updated", json.RawMessage(`{"authMode":"chatgpt","planType":"pro"}`))
+ if _, ok := c.CachedSessionCommands("one", "model/small"); ok {
+ t.Fatal("account change retained model catalogue")
+ }
+ p.mu.Lock()
+ defer p.mu.Unlock()
+ if p.contexts["one"].latest.Text != "retained reply" || !p.lastStatusAt.IsZero() {
+ t.Fatal("account change discarded conversation or skipped refresh")
+ }
+}
diff --git a/internal/codex/session_permissions_unix.go b/internal/codex/session_permissions_unix.go
new file mode 100644
index 0000000..42bbbb7
--- /dev/null
+++ b/internal/codex/session_permissions_unix.go
@@ -0,0 +1,49 @@
+//go:build unix
+
+package codex
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "time"
+
+ "github.com/gorilla/websocket"
+)
+
+// A settings acknowledgement means queued. Read the loaded thread without
+// overrides until native permission provenance confirms the chosen profile.
+// This neither waits under the event-state lock nor retries the mutation.
+func (p *daemonStatusProvider) writeVerifiedPermissions(ctx context.Context, conn *websocket.Conn, id, permission string, params map[string]any) error {
+ if err := p.requestOn(ctx, conn, "thread/settings/update", params, nil); err != nil {
+ return err
+ }
+ verify, cancel := context.WithTimeout(ctx, 3*time.Second)
+ defer cancel()
+ ticker := time.NewTicker(50 * time.Millisecond)
+ defer ticker.Stop()
+ for {
+ loaded, err := p.loadedThreads(verify)
+ if err != nil {
+ return fmt.Errorf("permission update unconfirmed: %w", err)
+ }
+ if _, ok := loaded[id]; !ok {
+ return errors.New("permission update unconfirmed: session unloaded")
+ }
+ var result struct {
+ Thread struct{ ID string }
+ ActivePermissionProfile *struct{ ID string }
+ }
+ if err := p.requestOn(verify, conn, "thread/resume", map[string]any{"threadId": id, "excludeTurns": true}, &result); err != nil {
+ return fmt.Errorf("permission update unconfirmed: %w", err)
+ }
+ if result.Thread.ID == id && result.ActivePermissionProfile != nil && result.ActivePermissionProfile.ID == permission {
+ return nil
+ }
+ select {
+ case <-verify.Done():
+ return fmt.Errorf("permission update queued but application unconfirmed: %w", verify.Err())
+ case <-ticker.C:
+ }
+ }
+}
diff --git a/internal/codex/session_settings_unix.go b/internal/codex/session_settings_unix.go
index 0091bda..26272c3 100644
--- a/internal/codex/session_settings_unix.go
+++ b/internal/codex/session_settings_unix.go
@@ -7,6 +7,7 @@ import (
"encoding/json"
"errors"
"fmt"
+ "github.com/gorilla/websocket"
"sort"
"strings"
"time"
@@ -145,12 +146,19 @@ func (p *daemonStatusProvider) validateQuotaStep(ctx context.Context, step Quota
// The RPC acknowledgement means queued, not applied. Observe the actual
// configured settings before calling an update successful.
-func (p *daemonStatusProvider) writeQuotaSettings(ctx context.Context, expected QuotaSession, params map[string]any) (err error) {
+func (p *daemonStatusProvider) writeQuotaSettings(ctx context.Context, expected QuotaSession, params map[string]any) error {
+ p.mu.Lock()
+ conn := p.connection
+ p.mu.Unlock()
+ return p.writeQuotaSettingsOn(ctx, conn, expected, params)
+}
+
+func (p *daemonStatusProvider) writeQuotaSettingsOn(ctx context.Context, conn *websocket.Conn, expected QuotaSession, params map[string]any) (err error) {
if err := ctx.Err(); err != nil {
return err
}
_, _, before, _ := p.observedQuotaSettings(expected.ID)
- if err := p.request(ctx, "thread/settings/update", params, nil); err != nil {
+ if err := p.requestOn(ctx, conn, "thread/settings/update", params, nil); err != nil {
var rejection *daemonResponseError
if !errors.As(err, &rejection) {
return errors.Join(ErrQuotaProfileUncertain, err)
@@ -165,6 +173,12 @@ func (p *daemonStatusProvider) writeQuotaSettings(ctx context.Context, expected
deadline := time.NewTimer(3 * time.Second)
defer deadline.Stop()
for {
+ p.mu.Lock()
+ sameConnection := p.connection == conn
+ p.mu.Unlock()
+ if !sameConnection {
+ return errors.New("connection changed before settings were verified")
+ }
observed, changed, version, ok := p.observedQuotaSettings(expected.ID)
if ok && version > before && sameQuotaSettings(observed, expected) {
return nil
diff --git a/internal/codex/session_settings_unix_test.go b/internal/codex/session_settings_unix_test.go
index 88539ff..e8fbfc6 100644
--- a/internal/codex/session_settings_unix_test.go
+++ b/internal/codex/session_settings_unix_test.go
@@ -33,6 +33,8 @@ type quotaDaemonFixture struct {
cwds map[string]string
ignoreCwd bool
environments []any
+ permissions map[string]string
+ resumeReads []map[string]any
}
func TestQuotaEffortDefaultFallback(t *testing.T) {
@@ -58,6 +60,10 @@ func TestQuotaEffortDefaultFallback(t *testing.T) {
}
func newQuotaDaemon(t *testing.T) (*daemonStatusProvider, *quotaDaemonFixture) {
+ return newQuotaDaemonTransport(t, false)
+}
+
+func newQuotaDaemonTransport(t *testing.T, compressed bool) (*daemonStatusProvider, *quotaDaemonFixture) {
t.Helper()
fixture := "aDaemonFixture{sessions: map[string]QuotaSession{"one": {ID: "one", Model: "large", Effort: "high"}, "two": {ID: "two", Model: "large", Effort: "high"}}}
dir, err := os.MkdirTemp("", "cxq-")
@@ -75,18 +81,27 @@ func newQuotaDaemon(t *testing.T) (*daemonStatusProvider, *quotaDaemonFixture) {
}
upgrader := websocket.Upgrader{}
server := &http.Server{Handler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
- conn, err := upgrader.Upgrade(w, r, nil)
+ headers := http.Header{}
+ if compressed {
+ headers.Set(daemonCompressionHeader, daemonCompressionProtocol)
+ }
+ conn, err := upgrader.Upgrade(w, r, headers)
if err != nil {
return
}
- defer conn.Close()
+ wire, err := newDaemonWire(conn, headers)
+ if err != nil {
+ conn.Close()
+ return
+ }
+ defer wire.close()
for {
var req struct {
ID *int64
Method string
Params map[string]any
}
- if conn.ReadJSON(&req) != nil {
+ if wire.readJSON(&req) != nil {
return
}
if req.ID == nil {
@@ -109,13 +124,14 @@ func newQuotaDaemon(t *testing.T) (*daemonStatusProvider, *quotaDaemonFixture) {
}
result["data"] = ids
case "thread/resume":
+ fixture.resumeReads = append(fixture.resumeReads, req.Params)
s, ok := fixture.sessions[id]
failure = !ok || fixture.fail == id
cwd := fixture.cwds[id]
if cwd == "" {
cwd = "/work"
}
- result = map[string]any{"cwd": cwd, "model": s.Model, "reasoningEffort": s.Effort, "serviceTier": s.Tier, "thread": map[string]any{"id": id}}
+ result = map[string]any{"cwd": cwd, "model": s.Model, "reasoningEffort": s.Effort, "serviceTier": s.Tier, "thread": map[string]any{"id": id}, "activePermissionProfile": map[string]any{"id": fixture.permissions[id]}}
case "thread/read":
status := fixture.commandStatus
if status == "" {
@@ -159,6 +175,12 @@ func newQuotaDaemon(t *testing.T) (*daemonStatusProvider, *quotaDaemonFixture) {
}
}
if !fixture.queued {
+ if permission, ok := req.Params["permissions"].(string); ok {
+ if fixture.permissions == nil {
+ fixture.permissions = make(map[string]string)
+ }
+ fixture.permissions[id] = permission
+ }
fixture.sessions[id] = desired
}
if fixture.notify {
@@ -193,10 +215,10 @@ func newQuotaDaemon(t *testing.T) (*daemonStatusProvider, *quotaDaemonFixture) {
delete(response, "result")
response["error"] = map[string]any{"code": -1, "message": "test failure"}
}
- if conn.WriteJSON(response) != nil {
+ if wire.writeJSON(response) != nil {
return
}
- if settingsNotification != nil && conn.WriteJSON(settingsNotification) != nil {
+ if settingsNotification != nil && wire.writeJSON(settingsNotification) != nil {
return
}
}
diff --git a/internal/codex/session_stream_test.go b/internal/codex/session_stream_test.go
index a06d391..6010f56 100644
--- a/internal/codex/session_stream_test.go
+++ b/internal/codex/session_stream_test.go
@@ -107,3 +107,26 @@ func TestStreamingCommentaryKeepsCommandAndApprovalPriority(t *testing.T) {
t.Fatal("stream revived after interruption")
}
}
+
+func TestSettingsNotificationsDoNotResolveOtherClientApprovals(t *testing.T) {
+ states := map[string]*daemonContextState{}
+ now := time.Now()
+ for _, thread := range []string{"one", "two"} {
+ raw := `{"threadId":"` + thread + `","turnId":"turn","itemId":"item","command":"go test","cwd":"/work"}`
+ daemonContextEvent(states, "item/commandExecution/requestApproval", json.RawMessage(`"same-id"`), json.RawMessage(raw), now)
+ }
+ daemonContextEvent(states, "thread/settings/updated", nil, json.RawMessage(`{"threadId":"one","threadSettings":{"model":"changed","effort":"high"}}`), now)
+ if len(states["one"].requests) != 1 || len(states["two"].requests) != 1 {
+ t.Fatal("settings discarded pending approvals")
+ }
+ daemonContextEvent(states, "serverRequest/resolved", nil, json.RawMessage(`{"threadId":"one","requestId":"same-id"}`), now)
+ if len(states["one"].requests) != 0 || len(states["two"].requests) != 1 {
+ t.Fatal("resolution crossed thread/client boundaries")
+ }
+ daemonContextEvent(states, "turn/started", nil, json.RawMessage(`{"threadId":"one","turn":{"id":"next"}}`), now)
+ daemonContextEvent(states, "item/started", nil, json.RawMessage(`{"threadId":"one","turnId":"next","item":{"type":"agentMessage","id":"reply","phase":"final_answer"}}`), now)
+ daemonContextEvent(states, "item/agentMessage/delta", nil, json.RawMessage(`{"threadId":"one","turnId":"next","itemId":"reply","delta":"Still updating"}`), now)
+ if states["one"].latest.Text != "Still updating" || len(states["two"].requests) != 1 {
+ t.Fatal("stream stalled after settings/approval resolution")
+ }
+}
diff --git a/internal/codex/testdata/zstd-stream-v1.json b/internal/codex/testdata/zstd-stream-v1.json
new file mode 100644
index 0000000..1dd9935
--- /dev/null
+++ b/internal/codex/testdata/zstd-stream-v1.json
@@ -0,0 +1,14 @@
+[
+ {
+ "json": "{\"id\":0,\"result\":{\"text\":\"Native Codex-compatible stream history abcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdef\"}}",
+ "hex": "28b52ffd00482c0300c4057b226964223a302c22726573756c74223a7b2274657874223a224e617469766520436f6465782d636f6d70617469626c652073747265616d20686973746f727920616263646566616263646566616263646566616263646566227d7d0100598fad6614"
+ },
+ {
+ "json": "{\"id\":1,\"result\":{\"text\":\"Native Codex-compatible stream history abcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdef\"}}",
+ "hex": "7c0000387b226964223a310100923e55c208"
+ },
+ {
+ "json": "{\"id\":2,\"result\":{\"text\":\"Native Codex-compatible stream history abcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdef\"}}",
+ "hex": "740000387b226964223a320100922a6004"
+ }
+]
diff --git a/internal/codex/usage_coverage.go b/internal/codex/usage_coverage.go
new file mode 100644
index 0000000..b32604b
--- /dev/null
+++ b/internal/codex/usage_coverage.go
@@ -0,0 +1,162 @@
+package codex
+
+import (
+ "context"
+ "errors"
+ "io"
+ "os"
+ "path/filepath"
+ "sort"
+ "time"
+
+ "github.com/pelletier/go-toml/v2"
+)
+
+const asyncCompactionCoverage = "background compaction is enabled; Codex does not publish all background response usage"
+
+type localFeatureCheck struct {
+ info os.FileInfo
+ enabled bool
+ issue string
+}
+
+// A positive local flag is enough to qualify file-only accounting. A negative
+// file read cannot establish hidden per-process CLI overrides; authoritative
+// loaded-thread feature state is preferred whenever the daemon supplies it.
+func (r *LiveUsageReader) usageCoverage(ctx context.Context, sessions []LiveUsageSession, daemon map[string]string) string {
+ if r.knownUsageCoverage == nil {
+ r.knownUsageCoverage = make(map[string]string)
+ }
+ for id, issue := range daemon {
+ r.knownUsageCoverage[id] = issue
+ }
+ issues := []string{}
+ // Children are folded into their root's display row. Preserve independent
+ // feature coverage for loaded/recent children before that grouping hides IDs.
+ candidates := append([]LiveUsageSession(nil), sessions...)
+ seen := make(map[string]bool)
+ for _, session := range sessions {
+ seen[session.ID] = true
+ }
+ for _, cursor := range r.files {
+ recent := time.Since(cursor.lastModified) <= activeRolloutHorizon
+ if !seen[cursor.threadID] && cursor.threadID != "" && (recent || r.knownUsageCoverage[cursor.threadID] != "") {
+ candidates = append(candidates, LiveUsageSession{ID: cursor.threadID, WorkingDirectory: cursor.workingDirectory, Active: recent})
+ seen[cursor.threadID] = true
+ }
+ }
+ for id := range r.knownUsageCoverage {
+ if !seen[id] {
+ delete(r.knownUsageCoverage, id)
+ }
+ }
+ for _, session := range candidates {
+ issue, verified := daemon[session.ID]
+ if !verified {
+ issue = r.knownUsageCoverage[session.ID]
+ }
+ // Missing writer/status signals after disconnect do not prove recovery.
+ // Keep known risks even when their retained session row becomes inactive.
+ if !session.Active && issue == "" {
+ continue
+ }
+ if issue == "" && !verified {
+ issue = r.localUsageCoverage(ctx, session.WorkingDirectory)
+ }
+ if issue != "" {
+ id := []rune(SanitizeSessionContext(session.ID))
+ issues = append(issues, "SESSION "+string(id[:min(12, len(id))])+": "+issue)
+ }
+ }
+ if r.discoveryCoverageIssue != "" {
+ issues = append(issues, r.discoveryCoverageIssue)
+ }
+ if len(issues) == 0 {
+ return ""
+ }
+ sort.Strings(issues)
+ return issues[0]
+}
+
+func (r *LiveUsageReader) localUsageCoverage(ctx context.Context, cwd string) string {
+ paths := []string{filepath.Join(filepath.Dir(r.SessionsRoot), "config.toml")}
+ if filepath.IsAbs(cwd) {
+ for dir := filepath.Clean(cwd); ; dir = filepath.Dir(dir) {
+ paths = append(paths, filepath.Join(dir, ".codex", "config.toml"))
+ if filepath.Dir(dir) == dir {
+ break
+ }
+ }
+ }
+ for _, path := range paths {
+ if err := ctx.Err(); err != nil {
+ return "background usage configuration check interrupted"
+ }
+ enabled, issue := r.localAsyncCompaction(path)
+ if issue != "" {
+ return issue
+ }
+ if enabled {
+ return "local configuration can enable background compaction; effective usage coverage is unverified"
+ }
+ }
+ return ""
+}
+
+func (r *LiveUsageReader) localAsyncCompaction(path string) (bool, string) {
+ info, err := os.Stat(path)
+ if errors.Is(err, os.ErrNotExist) {
+ return false, ""
+ }
+ if err != nil {
+ return false, "background usage configuration could not be read"
+ }
+ if r.localFeatureChecks == nil {
+ r.localFeatureChecks = make(map[string]localFeatureCheck)
+ }
+ if previous, ok := r.localFeatureChecks[path]; ok && os.SameFile(previous.info, info) && previous.info.ModTime() == info.ModTime() && previous.info.Size() == info.Size() {
+ return previous.enabled, previous.issue
+ }
+ file, err := os.Open(path)
+ if err != nil {
+ return false, "background usage configuration could not be read"
+ }
+ data, err := io.ReadAll(io.LimitReader(file, 1<<20+1))
+ file.Close()
+ check := localFeatureCheck{info: info}
+ var config map[string]any
+ if err != nil || len(data) > 1<<20 || toml.Unmarshal(data, &config) != nil {
+ check.issue = "background usage configuration could not be verified"
+ } else {
+ check.enabled = hasAsyncCompaction(config)
+ }
+ // Only active-session configuration is relevant; retain a bounded cache.
+ if len(r.localFeatureChecks) >= 256 {
+ clear(r.localFeatureChecks)
+ }
+ r.localFeatureChecks[path] = check
+ return check.enabled, check.issue
+}
+
+func hasAsyncCompaction(config map[string]any) bool {
+ if features, ok := config["features"].(map[string]any); ok {
+ switch value := features["async_compaction"].(type) {
+ case bool:
+ if value {
+ return true
+ }
+ case map[string]any:
+ if enabled, ok := value["enabled"].(bool); ok && enabled {
+ return true
+ }
+ }
+ }
+ if profiles, ok := config["profiles"].(map[string]any); ok {
+ for _, profile := range profiles {
+ if values, ok := profile.(map[string]any); ok && hasAsyncCompaction(values) {
+ return true
+ }
+ }
+ }
+ return false
+}
diff --git a/internal/codex/usage_coverage_test.go b/internal/codex/usage_coverage_test.go
new file mode 100644
index 0000000..7a29242
--- /dev/null
+++ b/internal/codex/usage_coverage_test.go
@@ -0,0 +1,104 @@
+package codex
+
+import (
+ "context"
+ "os"
+ "path/filepath"
+ "strings"
+ "testing"
+ "time"
+)
+
+func TestUsageCoverageDetectsFileOnlyConfigurationAndPrefersEffectiveState(t *testing.T) {
+ home := t.TempDir()
+ project := filepath.Join(home, "project")
+ if err := os.MkdirAll(filepath.Join(project, ".codex"), 0700); err != nil {
+ t.Fatal(err)
+ }
+ path := filepath.Join(project, ".codex", "config.toml")
+ r := &LiveUsageReader{SessionsRoot: filepath.Join(home, "sessions")}
+ sessions := []LiveUsageSession{{ID: "thread", WorkingDirectory: project, Active: true}}
+ for _, config := range []string{"[features]\nasync_compaction = true\n", "[features.async_compaction]\nenabled = true\n", "[profiles.experimental.features]\nasync_compaction = true\n"} {
+ if err := os.WriteFile(path, []byte(config), 0600); err != nil {
+ t.Fatal(err)
+ }
+ r.localFeatureChecks = nil
+ if got := r.usageCoverage(context.Background(), sessions, nil); !strings.Contains(got, "background compaction") {
+ t.Fatal("file-only risk ignored", got)
+ }
+ if got := r.usageCoverage(context.Background(), sessions, map[string]string{"thread": ""}); got != "" {
+ t.Fatal("local setting overrode effective server state", got)
+ }
+ }
+ if err := os.WriteFile(path, []byte("[features]\nasync_compaction = false\n"), 0600); err != nil {
+ t.Fatal(err)
+ }
+ r.localFeatureChecks = nil
+ if got := r.usageCoverage(context.Background(), sessions, nil); got != "" {
+ t.Fatal("disabled flag blocked accounting", got)
+ }
+}
+
+func TestUsageCoverageKeepsKnownRiskAcrossDisconnectAndIncludesChildren(t *testing.T) {
+ home := t.TempDir()
+ r := &LiveUsageReader{SessionsRoot: filepath.Join(home, "sessions"), files: map[string]*rolloutCursor{"child": {threadID: "child", workingDirectory: home, lastModified: time.Now()}}}
+ sessions := []LiveUsageSession{{ID: "root", WorkingDirectory: home, Active: true}}
+ for _, state := range []map[string]string{{"child": asyncCompactionCoverage}, nil} {
+ if got := r.usageCoverage(context.Background(), sessions, state); !strings.Contains(got, "child") || !strings.Contains(got, "background compaction") {
+ t.Fatal("child coverage vanished", got)
+ }
+ }
+ if got := r.usageCoverage(context.Background(), sessions, map[string]string{"child": ""}); got != "" {
+ t.Fatal("verified recovery retained stale risk", got)
+ }
+}
+
+func TestUsageCoverageRejectsUnreadableConfigurationAndPartialDiscovery(t *testing.T) {
+ home := t.TempDir()
+ path := filepath.Join(home, "config.toml")
+ r := &LiveUsageReader{SessionsRoot: filepath.Join(home, "sessions")}
+ if err := os.WriteFile(path, []byte("malformed = ["), 0600); err != nil {
+ t.Fatal(err)
+ }
+ sessions := []LiveUsageSession{{ID: "thread", Active: true}}
+ if got := r.usageCoverage(context.Background(), sessions, nil); !strings.Contains(got, "could not be verified") {
+ t.Fatal("invalid configuration silently accepted", got)
+ }
+ r.discoveryCoverageIssue = "compressed rollout history could not be verified"
+ if got := r.usageCoverage(context.Background(), nil, nil); got != r.discoveryCoverageIssue {
+ t.Fatal("partial discovery hidden", got)
+ }
+}
+
+func TestUsageCoverageRetainsInactiveSessionRiskUntilVerifiedRecovery(t *testing.T) {
+ for _, child := range []bool{false, true} {
+ t.Run(map[bool]string{false: "root", true: "old folded child"}[child], func(t *testing.T) {
+ home := t.TempDir()
+ r := &LiveUsageReader{SessionsRoot: filepath.Join(home, "sessions")}
+ sessions := []LiveUsageSession{{ID: "root", WorkingDirectory: home, Active: true, TotalTokens: 1100}}
+ id := "root"
+ if child {
+ id = "child"
+ r.files = map[string]*rolloutCursor{"child": {threadID: id, parentThreadID: "root", workingDirectory: home, lastModified: time.Now().Add(-48 * time.Hour), observedTokens: 1100}}
+ }
+ if got := r.usageCoverage(context.Background(), sessions, map[string]string{id: asyncCompactionCoverage}); !strings.Contains(got, "background compaction") {
+ t.Fatal("known risk missing", got)
+ }
+ // An idle, previously observed row remains visible, but loses Active once
+ // daemon/writer signals disappear. That is not verified feature recovery.
+ sessions[0].Active = false
+ for range 2 {
+ got := r.usageCoverage(context.Background(), sessions, nil)
+ if !strings.Contains(got, id) || !strings.Contains(got, "background compaction") {
+ t.Fatal("inactive/disconnected row hid cached risk", got)
+ }
+ }
+ if got := r.usageCoverage(context.Background(), sessions, map[string]string{id: ""}); got != "" {
+ t.Fatal("verified recovery did not clear inactive risk", got)
+ }
+ if got := r.usageCoverage(context.Background(), sessions, nil); got != "" {
+ t.Fatal("recovered risk reappeared", got)
+ }
+ })
+ }
+}
diff --git a/internal/codex/usage_coverage_unix.go b/internal/codex/usage_coverage_unix.go
new file mode 100644
index 0000000..c76115d
--- /dev/null
+++ b/internal/codex/usage_coverage_unix.go
@@ -0,0 +1,69 @@
+//go:build unix
+
+package codex
+
+import (
+ "context"
+ "errors"
+)
+
+func (p *daemonStatusProvider) readUsageCoverage(ctx context.Context, ids []string) map[string]string {
+ result := make(map[string]string)
+ for _, id := range ids {
+ if !p.isSubscribed(id) {
+ continue
+ }
+ issue, err := p.threadUsageCoverage(ctx, id)
+ if err != nil {
+ issue = "effective background-compaction feature state could not be verified"
+ }
+ result[id] = issue
+ }
+ return result
+}
+
+func (p *daemonStatusProvider) threadUsageCoverage(ctx context.Context, id string) (string, error) {
+ params := map[string]any{"threadId": id, "limit": 256}
+ seen := make(map[string]bool)
+ async, tokenBudget := false, false
+ for page := 0; page < 32; page++ {
+ var response struct {
+ Data *[]struct {
+ Name string
+ Enabled *bool
+ }
+ NextCursor *string
+ }
+ if err := p.request(ctx, "experimentalFeature/list", params, &response); err != nil {
+ return "", err
+ }
+ if response.Data == nil {
+ return "", errors.New("feature inventory omitted data")
+ }
+ for _, feature := range *response.Data {
+ if feature.Name != "async_compaction" && feature.Name != "token_budget" {
+ continue
+ }
+ if feature.Enabled == nil {
+ return "", errors.New("feature inventory omitted enablement")
+ }
+ if feature.Name == "async_compaction" {
+ async = *feature.Enabled
+ } else {
+ tokenBudget = *feature.Enabled
+ }
+ }
+ if response.NextCursor == nil || *response.NextCursor == "" {
+ if async && !tokenBudget {
+ return asyncCompactionCoverage, nil
+ }
+ return "", nil
+ }
+ if seen[*response.NextCursor] {
+ return "", errors.New("feature inventory repeated a cursor")
+ }
+ seen[*response.NextCursor] = true
+ params["cursor"] = *response.NextCursor
+ }
+ return "", errors.New("feature inventory exceeded pagination limit")
+}
diff --git a/internal/codex/usage_coverage_unix_test.go b/internal/codex/usage_coverage_unix_test.go
new file mode 100644
index 0000000..0a4b495
--- /dev/null
+++ b/internal/codex/usage_coverage_unix_test.go
@@ -0,0 +1,39 @@
+//go:build unix
+
+package codex
+
+import (
+ "context"
+ "strings"
+ "testing"
+)
+
+func TestThreadUsageCoverageUsesEffectiveFeatureInventory(t *testing.T) {
+ for _, tc := range []struct {
+ name string
+ data any
+ next any
+ blocked bool
+ invalid bool
+ }{
+ {name: "older inventory", data: []any{}},
+ {name: "async enabled", data: []any{map[string]any{"name": "async_compaction", "enabled": true}}, blocked: true},
+ {name: "disabled", data: []any{map[string]any{"name": "async_compaction", "enabled": false}}},
+ {name: "token budget disables async path", data: []any{map[string]any{"name": "async_compaction", "enabled": true}, map[string]any{"name": "token_budget", "enabled": true}}},
+ {name: "missing enablement", data: []any{map[string]any{"name": "async_compaction"}}, invalid: true},
+ {name: "missing data", invalid: true},
+ {name: "repeated cursor", data: []any{}, next: "same", invalid: true},
+ } {
+ t.Run(tc.name, func(t *testing.T) {
+ p, f := newQuotaDaemon(t)
+ f.catalogues = map[string]map[string]any{"experimentalFeature/list": {"data": tc.data, "nextCursor": tc.next}}
+ if err := p.ensureConnected(context.Background()); err != nil {
+ t.Fatal(err)
+ }
+ issue, err := p.threadUsageCoverage(context.Background(), "one")
+ if (err != nil) != tc.invalid || strings.Contains(issue, "background compaction") != tc.blocked {
+ t.Fatal(issue, err)
+ }
+ })
+ }
+}
diff --git a/internal/codex/usage_diagnostics.go b/internal/codex/usage_diagnostics.go
index bab07d1..753e828 100644
--- a/internal/codex/usage_diagnostics.go
+++ b/internal/codex/usage_diagnostics.go
@@ -8,6 +8,7 @@ import (
type APIEqIssueKind string
const (
+ UsageIssueCoverage APIEqIssueKind = "INCOMPLETE COVERAGE"
UsageIssueUnpriced APIEqIssueKind = "UNPRICED USAGE"
UsageIssueMissing APIEqIssueKind = "MISSING RESPONSE USAGE"
UsageIssueInconsistent APIEqIssueKind = "INCONSISTENT RESPONSE USAGE"
@@ -61,6 +62,9 @@ func DiagnosticText(issue APIEqDiagnostic) string {
}
return value
}
+ if issue.Kind == UsageIssueCoverage {
+ return string(issue.Kind) + " // " + field(issue.Reason, 240) + "\nSOURCE " + field(issue.Source, 48)
+ }
basis := issue.ModelBasis
if basis == "" {
basis = "requested"
diff --git a/internal/ui/model.go b/internal/ui/model.go
index f79e2db..2d51b6d 100644
--- a/internal/ui/model.go
+++ b/internal/ui/model.go
@@ -216,6 +216,7 @@ type Model struct {
quotaAPIIssues map[string]string
quotaAPIAccount string
quotaAPITelemetryIssue string
+ quotaAPICoverageIssue string
}
type fetchedMsg struct {
@@ -1090,6 +1091,9 @@ func (m Model) Update(message tea.Msg) (tea.Model, tea.Cmd) {
if observedAt.IsZero() {
observedAt = time.Now()
}
+ if message.usage.APIEqCoverageIssue != "" {
+ m.observeQuotaAPICoverage(message.usage.APIEqCoverageIssue)
+ }
m.quotaGraphs.Observe(message.snapshot, message.err != nil, observedAt)
var quotaStepCommand tea.Cmd
if message.err == nil {
@@ -2274,16 +2278,24 @@ func (m Model) fetch() tea.Cmd {
message := fetchedMsg{
resetRevision: m.resetRevision,
snapshot: snapshot, err: err, at: time.Now(),
+ usage: codex.LiveUsageSnapshot{APIEqCoverageIssue: before.APIEqCoverageIssue},
benchmarkQuotaRevision: m.benchmarkQuotaAccounting.revision,
}
if err == nil && m.usageFetcher != nil {
after, afterErr := m.usageFetcher.FetchTokenUsage(context.Background())
after = m.benchmarkQuotaAccounting.combine(after)
+ if after.APIEqCoverageIssue != "" {
+ message.usage.APIEqCoverageIssue = after.APIEqCoverageIssue
+ }
+ coverageIssue := message.usage.APIEqCoverageIssue
switch {
case beforeErr != nil:
message.usageErr = beforeErr
case afterErr != nil:
message.usageErr = afterErr
+ case coverageIssue != "":
+ message.usage = after
+ message.usage.APIEqCoverageIssue = coverageIssue
case !quotaAPIAccountingEqual(before, after) || after.APIEqPendingCalls > 0:
message.usageErr = errQuotaObservationChanged
default:
diff --git a/internal/ui/model_test.go b/internal/ui/model_test.go
index ff728b3..2f7969d 100644
--- a/internal/ui/model_test.go
+++ b/internal/ui/model_test.go
@@ -22,19 +22,28 @@ type stubFetcher struct {
}
type quotaBracketFetcher struct {
- snapshot codex.Snapshot
- usages []codex.LiveUsageSnapshot
- usageErr error
- calls []string
+ snapshot codex.Snapshot
+ usages []codex.LiveUsageSnapshot
+ usageErr error
+ usageErrors []error
+ err error
+ calls []string
}
func (f *quotaBracketFetcher) Fetch(context.Context) (codex.Snapshot, error) {
f.calls = append(f.calls, "quota")
- return f.snapshot, nil
+ return f.snapshot, f.err
}
func (f *quotaBracketFetcher) FetchTokenUsage(context.Context) (codex.LiveUsageSnapshot, error) {
f.calls = append(f.calls, "usage")
+ if len(f.usageErrors) > 0 {
+ err := f.usageErrors[0]
+ f.usageErrors = f.usageErrors[1:]
+ if err != nil {
+ return codex.LiveUsageSnapshot{}, err
+ }
+ }
if f.usageErr != nil {
return codex.LiveUsageSnapshot{}, f.usageErr
}
@@ -487,3 +496,97 @@ func footerButtonByID(model Model, id footerButtonID) (footerButton, bool) {
}
return footerButton{}, false
}
+
+func TestQuotaFetchCoverageRecoveryCannotSpanUnobservedGap(t *testing.T) {
+ now := time.Now()
+ reset := now.Add(time.Hour).Unix()
+ clean := codex.LiveUsageSnapshot{APIEqUSD: 2, APIEqPricedCalls: 2}
+ gap := clean
+ gap.APIEqCoverageIssue = "SESSION child: background compaction usage is unobserved"
+ for _, bracket := range []struct {
+ name string
+ before, after codex.LiveUsageSnapshot
+ }{{"recovers during quota read", gap, clean}, {"appears during quota read", clean, gap}} {
+ t.Run(bracket.name, func(t *testing.T) {
+ f := "aBracketFetcher{snapshot: apiEqSnapshot(20, reset), usages: []codex.LiveUsageSnapshot{bracket.before, bracket.after}}
+ m := New(f, time.Minute)
+ m.observeQuotaAPIEq(apiEqSnapshot(10, reset), codex.LiveUsageSnapshot{}, now)
+ m.observeQuotaAPIEq(apiEqSnapshot(15, reset), codex.LiveUsageSnapshot{APIEqUSD: 1, APIEqPricedCalls: 1}, now.Add(time.Minute))
+ if len(m.quotaAPIEvidence) != 1 {
+ t.Fatal("test did not establish an estimate")
+ }
+ message := m.fetch()().(fetchedMsg)
+ if message.usageErr != nil || message.usage.APIEqCoverageIssue == "" {
+ t.Fatal("quota bracket dropped a coverage gap", message.usageErr, message.usage)
+ }
+ updated, _ := m.Update(message)
+ m = updated.(Model)
+ if len(m.quotaAPIEvidence) != 0 || len(m.quotaAPIAnchors) != 0 || m.quotaAPICoverageIssue == "" {
+ t.Fatal("coverage gap retained previous evidence")
+ }
+ // A later clean bracket starts a new baseline even when the recovered
+ // totals already include substantial activity from the blocked interval.
+ recovered := codex.LiveUsageSnapshot{APIEqUSD: 10, APIEqPricedCalls: 3}
+ f.snapshot = apiEqSnapshot(25, reset)
+ f.usages = []codex.LiveUsageSnapshot{recovered, recovered}
+ updated, _ = m.Update(m.fetch()())
+ m = updated.(Model)
+ if len(m.quotaAPIEvidence) != 0 || m.quotaAPICoverageIssue != "" {
+ t.Fatal("recovery learned across the gap")
+ }
+ recovered.APIEqUSD++
+ recovered.APIEqPricedCalls++
+ f.snapshot = apiEqSnapshot(30, reset)
+ f.usages = []codex.LiveUsageSnapshot{recovered, recovered}
+ updated, _ = m.Update(m.fetch()())
+ m = updated.(Model)
+ if len(m.quotaAPIEvidence) != 1 || m.quotaAPIEvidence[0].CapacityUSD != 20 {
+ t.Fatal("next clean interval failed to learn independently", m.quotaAPIEvidence)
+ }
+ })
+ }
+}
+
+func TestQuotaCoverageInvalidationSurvivesObservationErrorsAndBenchmarkDeferral(t *testing.T) {
+ for _, scenario := range []string{"after telemetry fails", "before telemetry fails", "quota fails", "benchmark active"} {
+ t.Run(scenario, func(t *testing.T) {
+ now := time.Now()
+ reset := now.Add(time.Hour).Unix()
+ clean := codex.LiveUsageSnapshot{APIEqUSD: 2, APIEqPricedCalls: 2}
+ gap := clean
+ gap.APIEqCoverageIssue = "known coverage gap"
+ unavailable := errors.New("observation unavailable")
+ f := "aBracketFetcher{snapshot: apiEqSnapshot(20, reset), usages: []codex.LiveUsageSnapshot{gap, clean}}
+ switch scenario {
+ case "after telemetry fails":
+ f.usageErrors = []error{nil, unavailable}
+ case "before telemetry fails":
+ f.usages = []codex.LiveUsageSnapshot{gap}
+ f.usageErrors = []error{unavailable, nil}
+ case "quota fails":
+ f.err = unavailable
+ }
+ m := New(f, time.Minute)
+ m.observeQuotaAPIEq(apiEqSnapshot(10, reset), codex.LiveUsageSnapshot{}, now)
+ m.observeQuotaAPIEq(apiEqSnapshot(15, reset), codex.LiveUsageSnapshot{APIEqUSD: 1, APIEqPricedCalls: 1}, now.Add(time.Minute))
+ if scenario == "benchmark active" {
+ m.benchmarkQuotaAccounting.active = true
+ }
+ message := m.fetch()().(fetchedMsg)
+ if message.usage.APIEqCoverageIssue == "" {
+ t.Fatal("gap was lost alongside another deferral")
+ }
+ if strings.Contains(scenario, "telemetry") && !errors.Is(message.usageErr, unavailable) {
+ t.Fatal("telemetry error was hidden", message.usageErr)
+ }
+ if scenario == "quota fails" && !errors.Is(message.err, unavailable) {
+ t.Fatal("quota error was hidden", message.err)
+ }
+ updated, _ := m.Update(message)
+ m = updated.(Model)
+ if len(m.quotaAPIEvidence) != 0 || len(m.quotaAPIAnchors) != 0 || m.quotaAPICoverageIssue == "" {
+ t.Fatal("deferred observation retained stale capacity evidence")
+ }
+ })
+ }
+}
diff --git a/internal/ui/quota_api_eq.go b/internal/ui/quota_api_eq.go
index 0470c7e..1b26b6a 100644
--- a/internal/ui/quota_api_eq.go
+++ b/internal/ui/quota_api_eq.go
@@ -22,7 +22,8 @@ const (
var errQuotaObservationChanged = fmt.Errorf("local accounting changed during quota read")
func quotaAPIAccountingEqual(left, right codex.LiveUsageSnapshot) bool {
- return left.APIEqUSD == right.APIEqUSD &&
+ return left.APIEqCoverageIssue == right.APIEqCoverageIssue &&
+ left.APIEqUSD == right.APIEqUSD &&
left.APIEqTierPremiumUSD == right.APIEqTierPremiumUSD &&
left.APIEqUnknownTierCalls == right.APIEqUnknownTierCalls &&
left.APIEqPricedCalls == right.APIEqPricedCalls &&
@@ -86,6 +87,18 @@ type quotaAPIEstimate struct {
unknownTier bool
}
+// A known coverage gap invalidates evidence even when another error or an
+// active benchmark prevents using the associated quota observation.
+func (m *Model) observeQuotaAPICoverage(issue string) {
+ previous := m.quotaAPICoverageIssue
+ m.quotaAPICoverageIssue = issue
+ if previous != "" || issue != "" {
+ m.quotaAPIAnchors = make(map[string]quotaAPIAnchor)
+ m.quotaAPIEvidence = nil
+ m.quotaAPIIssues = make(map[string]string)
+ }
+}
+
func (m *Model) observeQuotaAPIEq(snapshot codex.Snapshot, usage codex.LiveUsageSnapshot, at time.Time) {
if at.IsZero() {
at = time.Now()
@@ -101,6 +114,11 @@ func (m *Model) observeQuotaAPIEq(snapshot codex.Snapshot, usage codex.LiveUsage
m.quotaAPIIssues = make(map[string]string)
}
m.quotaAPIAccount = account
+ m.observeQuotaAPICoverage(usage.APIEqCoverageIssue)
+ if usage.APIEqCoverageIssue != "" {
+ return
+ }
+
if m.quotaAPIAnchors == nil {
m.quotaAPIAnchors = make(map[string]quotaAPIAnchor)
}
@@ -312,6 +330,9 @@ func trimQuotaAPISamples(samples []quotaAPISample, key string) []quotaAPISample
}
func (m Model) quotaAPIEstimate(snapshot codex.Snapshot, meter codex.Meter) (quotaAPIEstimate, bool) {
+ if m.quotaAPICoverageIssue != "" {
+ return quotaAPIEstimate{}, false
+ }
key := quotaAPIKey(snapshot, meter)
matching := make([]quotaAPISample, 0, quotaAPIMaxSamples)
for _, sample := range validQuotaAPISamples(m.quotaAPIEvidence, time.Now()) {
@@ -404,6 +425,9 @@ func (m Model) quotaAPILine(meter codex.Meter, width int) (line string) {
if m.benchmarkQuotaAccounting.deferred() {
return m.benchmarkQuotaAccounting.deferredLabel(width)
}
+ if m.quotaAPICoverageIssue != "" {
+ return "API-EQ // " + i18n.Text(quotaAPIRestartCoverageGap)
+ }
if m.quotaAPITelemetryIssue != "" {
if width < 44 {
return "API-EQ // " + m.quotaAPITelemetryIssue
@@ -534,6 +558,9 @@ func (m Model) quotaAPIDiagnostic() codex.APIEqDiagnostic {
if !m.meterView.isQuota() {
return codex.APIEqDiagnostic{}
}
+ if m.quotaAPICoverageIssue != "" {
+ return codex.APIEqDiagnostic{Kind: codex.UsageIssueCoverage, Source: "usage coverage", Reason: m.quotaAPICoverageIssue}
+ }
for _, meter := range m.snapshot.Meters() {
if quotaAPIMeterEligible(m.snapshot, meter) {
if issue := m.quotaAPIAnchors[quotaAPIKey(m.snapshot, meter)].restartIssue; issue.Kind != "" {
diff --git a/internal/ui/quota_api_eq_test.go b/internal/ui/quota_api_eq_test.go
index 4916e2b..5612348 100644
--- a/internal/ui/quota_api_eq_test.go
+++ b/internal/ui/quota_api_eq_test.go
@@ -493,3 +493,38 @@ func TestQuotaDiagnosticPreservesFooterAcrossTerminalSizes(t *testing.T) {
}
}
}
+
+func TestQuotaAPICoverageGapDiscardsEvidenceAndRestartsCleanly(t *testing.T) {
+ now := time.Now()
+ reset := now.Add(time.Hour).Unix()
+ m := New(nil, time.Minute)
+ m.snapshot = apiEqSnapshot(10, reset)
+ m.observeQuotaAPIEq(m.snapshot, codex.LiveUsageSnapshot{}, now)
+ m.snapshot = apiEqSnapshot(15, reset)
+ m.observeQuotaAPIEq(m.snapshot, codex.LiveUsageSnapshot{APIEqUSD: 1, APIEqPricedCalls: 1}, now.Add(time.Minute))
+ if len(m.quotaAPIEvidence) != 1 {
+ t.Fatal("test failed to learn")
+ }
+ for _, used := range []int{20, 25} {
+ m.snapshot = apiEqSnapshot(used, reset)
+ m.observeQuotaAPIEq(m.snapshot, codex.LiveUsageSnapshot{APIEqUSD: 2, APIEqPricedCalls: 2, APIEqCoverageIssue: "SESSION child: background compaction usage is unobserved"}, now.Add(2*time.Minute))
+ if len(m.quotaAPIEvidence) != 0 || len(m.quotaAPIAnchors) != 0 || !strings.Contains(m.quotaAPILine(m.snapshot.Meters()[0], 100), "COVERAGE GAP") {
+ t.Fatal("incomplete interval learned or showed old estimates")
+ }
+ if _, ok := m.quotaAPIEstimate(m.snapshot, m.snapshot.Meters()[0]); ok {
+ t.Fatal("stale estimate remained usable")
+ }
+ }
+ // Recovery creates a fresh baseline; accumulated costs from the blocked
+ // interval cannot be attributed to the next clean quota movement.
+ m.snapshot = apiEqSnapshot(30, reset)
+ m.observeQuotaAPIEq(m.snapshot, codex.LiveUsageSnapshot{APIEqUSD: 10, APIEqPricedCalls: 3}, now.Add(3*time.Minute))
+ if len(m.quotaAPIEvidence) != 0 {
+ t.Fatal("recovery learned across blocked interval")
+ }
+ m.snapshot = apiEqSnapshot(35, reset)
+ m.observeQuotaAPIEq(m.snapshot, codex.LiveUsageSnapshot{APIEqUSD: 11, APIEqPricedCalls: 4}, now.Add(4*time.Minute))
+ if len(m.quotaAPIEvidence) != 1 || math.Abs(m.quotaAPIEvidence[0].CapacityUSD-20) > 1e-9 {
+ t.Fatal("clean recovery failed", m.quotaAPIEvidence)
+ }
+}
diff --git a/internal/web/dist/assets/index-D1LQiLkx.js b/internal/web/dist/assets/index-Bg6CQ-Bn.js
similarity index 61%
rename from internal/web/dist/assets/index-D1LQiLkx.js
rename to internal/web/dist/assets/index-Bg6CQ-Bn.js
index 3518d45..d338386 100644
--- a/internal/web/dist/assets/index-D1LQiLkx.js
+++ b/internal/web/dist/assets/index-Bg6CQ-Bn.js
@@ -1,5 +1,5 @@
-(function(){let e=document.createElement(`link`).relList;if(e&&e.supports&&e.supports(`modulepreload`))return;for(let e of document.querySelectorAll(`link[rel="modulepreload"]`))n(e);new MutationObserver(e=>{for(let t of e)if(t.type===`childList`)for(let e of t.addedNodes)e.tagName===`LINK`&&e.rel===`modulepreload`&&n(e)}).observe(document,{childList:!0,subtree:!0});function t(e){let t={};return e.integrity&&(t.integrity=e.integrity),e.referrerPolicy&&(t.referrerPolicy=e.referrerPolicy),t.credentials=e.crossOrigin===`use-credentials`?`include`:e.crossOrigin===`anonymous`?`omit`:`same-origin`,t}function n(e){if(e.ep)return;e.ep=!0;let n=t(e);fetch(e.href,n)}})();var e=Array.isArray,t=Array.prototype.indexOf,n=Array.prototype.includes,r=Array.from,i=Object.defineProperty,a=Object.getOwnPropertyDescriptor,o=Object.getOwnPropertyDescriptors,s=Object.prototype,c=Array.prototype,l=Object.getPrototypeOf,u=Object.isExtensible;function d(e){return typeof e==`function`}var f=()=>{};function p(e){return e()}function m(e){for(var t=0;t Quota refresh failed. Policy state is based on the last successful
observation. The longest Codex quota window selects one active model profile. ASK
@@ -9,7 +9,7 @@
redemption order. Cycle duration or reset date unavailable — position cannot be
plotted. No quota windows reported yet. All reported windows are shown. API-equivalent learning and quota status
scoring remain in the terminal for this first preview. Loading / sending… Review expired. Press Enter to review again. No available matching commands. Review expired. Review the option again. Loading / sending… Review expired. Press Enter to review again. No available matching commands. Review expired. Review the option again.0){var E=i&4&&s===0?n:null;if(o){for(v=0;v0){var E=i&4&&s===0?n:null;if(o){for(v=0;v
Session controls temporarily unavailable. Check Codex for current state.
`),Jo=J(`Checking session controls…
`),Yo=J(``),Xo=J(`
Controls require a supported live request from a connected shared app-server session. Local observations alone cannot provide them.
`),$o=J(`Grants permission beyond this one action. Check the scope
carefully.`),es=J(``),ts=J(``),ns=J(``),rs=J(``),is=J(`Not before this time; when quota and idle allow.
`),as=J(``,1),os=J(`
The existing trigger remains active until saved or deleted. If - it changes or sends while you edit, saving is blocked.
`),ss=J(` `,1),cs=J(``),ls=J(``),us=J(``),ds=J(` `,1),fs=J(`Type one of these choices exactly. Your answer stays masked.
`),gs=J(`
`,1),_s=J(` `,1),vs=J(``),ys=J(`
`,1),bs=J(`
Type one of these choices exactly. Your answer stays masked.
`),gs=J(`
`,1),_s=J(` `,1),vs=J(``),ys=J(`
`,1),bs=J(`
Some quota profile checks are unavailable. Only sessions with freshly verified quota and settings can be updated; previous outcome notices remain visible.
`),Ws=J(` `),Gs=J(` `),Ks=J(` `),qs=J(``),Js=J(`Read only — reply or approve in Codex.
`),Xs=J(`QUOTA THRESHOLD // REVIEW PROFILE ↗`),Zs=J(`
This session is no longer in the current observation. Return to sessions.
`),ec=J(``),tc=J(``),nc=J(``),rc=J(` `),ic=J(`
`),ac=J(`QUOTA THRESHOLD // REVIEW PROFILE ↗`),oc=J(` `,1),sc=J(`
No locally observed sessions yet. Keep Codex running alongside - Codexometer.
`),dc=J(`↑ ↓ SELECT SESSION // ← LESS DETAIL // → MORE DETAIL // ESC BACK
`,1);function pc(e,t){Ue(t,!0);let n=(e,t=f,n,r)=>{let i=pt(()=>g(n?.(),!0)),a=pt(()=>g(r?.(),!1));var o=Bs(),s=B(o),c=e=>{var n=ks(),r=B(n),i=V(r,!0),a=V(H(r,2),!0);U(()=>{X(i,t().contextKind===`LAST REPLY`&&!t().streaming?`TASK`:`CURRENT TASK`),X(a,t().currentTask)}),Y(e,n)};Z(s,e=>{t().currentTask&&e(c)});var l=H(s,2),u=e=>{var n=As(),r=V(H(B(n),2),!0);U(()=>X(r,t().latestGuidance)),Y(e,n)};Z(l,e=>{t().latestGuidance&&e(u)});var d=H(l,2),p=e=>{var n=js(),r=V(H(B(n),2),!0);xe(2),U(()=>X(r,t().approvalContext)),Y(e,n)};Z(d,e=>{K(a)&&t().contextKind===`APPROVAL REQUEST`&&t().approvalContext&&e(p)});var m=H(d,2),h=e=>{var n=Ms();let r;var i=V(n,!0);U(e=>{r=ci(n,1,`attention-note`,null,r,{inferred:t().status===`CHECK SESSION`}),X(i,e)},[()=>w(t())]),Y(e,n)},_=F(()=>w(t())&&(!K(a)||K(v)||t().status===`CHECK SESSION`));Z(m,e=>{K(_)&&e(h)});var y=H(m,2),b=e=>{var n=Ns(),r=V(n,!0);U(e=>X(r,e),[()=>T(t())]),Y(e,n)};Z(y,e=>{K(i)&&e(b)});var x=H(y,2),S=e=>{var n=Ps(),r=V(n,!0);U(()=>X(r,t().text||`No session context available.`)),Y(e,n)};Z(x,e=>{(t().text||!t().workingCommand)&&e(S)});var C=H(x,2),E=e=>{var n=Fs(),r=H(B(n),2),i=z(r),a=H(i),o=e=>{var n=Mr();U(()=>X(n,`// +${(t().runningCommands||0)-+(t().commandStatus===`running`)} RUNNING`)),Y(e,n)};Z(a,e=>{!K(v)&&(t().runningCommands||0)>+(t().commandStatus===`running`)&&e(o)});var s=H(a,2),c=e=>{Y(e,Mr(`+`))};Z(s,e=>{t().runningLimited&&e(c)}),P(r);var l=V(H(r,2),!0);U(e=>{X(i,`COMMAND // ${e??``} `),X(l,t().workingCommand)},[()=>K(v)?`UNKNOWN`:(t().commandStatus||`unknown`).toUpperCase()]),Y(e,n)};Z(C,e=>{t().workingCommand&&e(E)});var D=H(C,2),ee=e=>{Ro(e,{get lines(){return t().fileChanges}})},te=e=>{var n=Rs(),r=H(B(n),2),i=V(r,!0),a=H(r,2),o=e=>{var n=Is(),r=V(n,!0);U(()=>X(r,t().command)),Y(e,n)},s=e=>{Y(e,Ls())};Z(a,e=>{t().command?e(o):e(s,-1)}),U(()=>X(i,K(v)||t().status!==`APPROVAL NEEDED`?`LAST OBSERVED COMMAND`:$.data?.control?`COMMAND REQUEST`:`COMMAND TO APPROVE IN CODEX`)),Y(e,n)};Z(D,e=>{(!K(a)||!$.data?.control)&&t().fileChanges?.length?e(ee):(!K(a)||!$.data?.control)&&(t().command||t().status===`APPROVAL NEEDED`)&&e(te,1)});var O=H(D,2),k=e=>{var n=zs(),r=V(n);U(()=>X(r,`CONTEXT SOURCE // ${(t().source||`LOCAL`)??``} // OBSERVATION`)),Y(e,n)};Z(O,e=>{K(a)||e(k)}),Y(e,o)},r=L(void 0),i=Wi(t,`params`,19,()=>({})),a=F(()=>$.data?.sessions||[]),o=F(()=>$.data?.control&&$.data.profiles||[]),s=F(()=>($.data?.triggers||[]).filter(e=>K(a).some(t=>t.id===e.session)));function c(e,t){if(i().id&&i().id!==t&&K(u)){e.preventDefault();return}x(t)}function l(e){return[`TURN COMPLETE`,`IDLE`,`ACTIVE`].includes(e.status)&&K(s).some(t=>t.session===e.id)?`TRIGGER SET`:e.status}let u=L(!1);function d(e,t){if(i().id&&i().id!==t&&K(u)){e.preventDefault();return}x(t)}vn(()=>{let e=i().id;e&&fr(()=>{aa.selected=e,fa(e,2)})});let p=F(()=>K(a).find(e=>e.id===i().id)),m=F(()=>new URLSearchParams(Yi.querystring).get(`review`)===`profile`&&K(o).some(e=>e.session===i().id&&e.pending)),h=F(()=>K(a).some(e=>e.id===aa.selected)?aa.selected:K(a)[0]?.id),v=F(()=>!$.connected||!!$.data?.sessionsError),y=F(()=>[[`OBSERVED TOKENS`,pa(K(a).reduce((e,t)=>e+t.tokens,0))],[`LISTED SESSIONS`,pa(K(a).length)],...[[`WORKING`,`WORKING`],[`AWAITING APPROVAL`,`APPROVAL NEEDED`],[`AWAITING INPUT`,`INPUT NEEDED`],[`CHECK · INFERRED`,`CHECK SESSION`]].map(([e,t])=>[e,K(v)?`—`:pa(K(a).filter(e=>e.status===t).length)])]),b=F(()=>K(a).filter(e=>[`INPUT NEEDED`,`APPROVAL NEEDED`,`CHECK SESSION`].includes(e.status)));function x(e){aa.selected=e,lr().then(()=>document.querySelector(`.session-row.selected`)?.scrollIntoView({block:`nearest`}))}function S(e,t){if(x(e),i().id){t<0&&(fa(e,2),location.hash=`/sessions`);return}let n=ua(e)+t;n>2?location.hash=`/sessions/`+encodeURIComponent(e):fa(e,n)}function C(e){let t=e.target;if(!(e.altKey||e.ctrlKey||e.metaKey||e.shiftKey||t.closest(`input, textarea, select, [contenteditable="true"]`))){if(e.key===`Escape`&&i().id)e.preventDefault(),S(i().id,-1);else if([`ArrowLeft`,`ArrowRight`].includes(e.key)&&(i().id||K(h)))e.preventDefault(),S(i().id||K(h),e.key===`ArrowLeft`?-1:1);else if(!i().id&&[`ArrowUp`,`ArrowDown`].includes(e.key)&&K(a).length){e.preventDefault();let t=K(a).findIndex(e=>e.id===K(h));x(K(a)[Math.max(0,Math.min(K(a).length-1,t+(e.key===`ArrowUp`?-1:1)))].id)}}}function w(e){return K(v)?`Last observation only — refresh unavailable. Check Codex for current state.`:e.status===`CHECK SESSION`?`INFERRED INACTIVITY — a quiet session, not a confirmed input or approval request. Local tools may still be running.`:$.data?.control&&[`APPROVAL NEEDED`,`INPUT NEEDED`].includes(e.status)?`Open full detail for supported live controls; otherwise reply in Codex.`:e.status===`APPROVAL NEEDED`?`OBSERVED APPROVAL SIGNAL — approve or decline in Codex.`:e.status===`INPUT NEEDED`?`OBSERVED INPUT SIGNAL — reply in Codex.`:e.status===`TURN COMPLETE`?`OBSERVED TURN COMPLETION — informational, not an approval request.`:``}function T(e){return e.streaming&&e.contextKind===`LAST REPLY`?`REPLY // STREAMING`:e.contextKind||`LAST ACTIVITY`}var E=fc();xr(`keydown`,Qt,C);var D=B(E),ee=V(H(z(D),2));P(D);var te=H(D,2);let O;Xr(te,21,()=>K(y),Kr,(e,t)=>{var n=F(()=>_(K(t),2));let r=()=>K(n)[0],i=()=>K(n)[1];var a=Vs(),o=z(a),s=V(o,!0),c=V(H(o,2),!0);P(a),U(()=>{X(s,r()),X(c,i())}),Y(e,a)}),P(te);var k=H(te,2),ne=V(k),re=H(k,2),ie=e=>{Y(e,Hs())};Z(re,e=>{K(v)&&e(ie)});var ae=H(re,2),oe=e=>{Y(e,Us())};Z(ae,e=>{$.data?.profileError&&e(oe)});var se=H(ae,2),A=e=>{var t=qs(),n=z(t);Xr(n,17,()=>K(b),Kr,(e,t)=>{var n=Ws();let r;var i=V(n);U((e,a)=>{r=ci(n,1,`button`,null,r,{approval:K(t).status===`APPROVAL NEEDED`}),Q(n,`href`,e),X(i,`${K(t).status??``} + Codexometer.`),dc=J(`
↑ ↓ SELECT SESSION // ← LESS DETAIL // → MORE DETAIL // ESC BACK
`,1);function pc(e,t){Ue(t,!0);let n=(e,t=f,n,r)=>{let i=pt(()=>g(n?.(),!0)),a=pt(()=>g(r?.(),!1));var o=Bs(),s=B(o),c=e=>{var n=ks(),r=B(n),i=V(r,!0),a=V(H(r,2),!0);U(()=>{X(i,t().contextKind===`LAST REPLY`&&!t().streaming?`TASK`:`CURRENT TASK`),X(a,t().currentTask)}),Y(e,n)};Z(s,e=>{t().currentTask&&e(c)});var l=H(s,2),u=e=>{var n=As(),r=V(H(B(n),2),!0);U(()=>X(r,t().latestGuidance)),Y(e,n)};Z(l,e=>{t().latestGuidance&&e(u)});var d=H(l,2),p=e=>{var n=js(),r=V(H(B(n),2),!0);xe(2),U(()=>X(r,t().approvalContext)),Y(e,n)};Z(d,e=>{K(a)&&t().contextKind===`APPROVAL REQUEST`&&t().approvalContext&&e(p)});var m=H(d,2),h=e=>{var n=Ms();let r;var i=V(n,!0);U(e=>{r=ci(n,1,`attention-note`,null,r,{inferred:t().status===`CHECK SESSION`}),X(i,e)},[()=>w(t())]),Y(e,n)},_=F(()=>w(t())&&(!K(a)||K(v)||t().status===`CHECK SESSION`));Z(m,e=>{K(_)&&e(h)});var y=H(m,2),b=e=>{var n=Ns(),r=V(n,!0);U(e=>X(r,e),[()=>T(t())]),Y(e,n)};Z(y,e=>{K(i)&&e(b)});var x=H(y,2),S=e=>{var n=Ps(),r=V(n,!0);U(()=>X(r,t().text||`No session context available.`)),Y(e,n)};Z(x,e=>{(t().text||!t().workingCommand)&&e(S)});var C=H(x,2),E=e=>{var n=Fs(),r=H(B(n),2),i=z(r),a=H(i),o=e=>{var n=Mr();U(()=>X(n,`// +${(t().runningCommands||0)-+(t().commandStatus===`running`)} RUNNING`)),Y(e,n)};Z(a,e=>{!K(v)&&(t().runningCommands||0)>+(t().commandStatus===`running`)&&e(o)});var s=H(a,2),c=e=>{Y(e,Mr(`+`))};Z(s,e=>{t().runningLimited&&e(c)}),P(r);var l=V(H(r,2),!0);U(e=>{X(i,`COMMAND // ${e??``} `),X(l,t().workingCommand)},[()=>K(v)?`UNKNOWN`:(t().commandStatus||`unknown`).toUpperCase()]),Y(e,n)};Z(C,e=>{t().workingCommand&&e(E)});var D=H(C,2),ee=e=>{Ro(e,{get lines(){return t().fileChanges}})},te=e=>{var n=Rs(),r=H(B(n),2),i=V(r,!0),a=H(r,2),o=e=>{var n=Is(),r=V(n,!0);U(()=>X(r,t().command)),Y(e,n)},s=e=>{Y(e,Ls())};Z(a,e=>{t().command?e(o):e(s,-1)}),U(()=>X(i,K(v)||t().status!==`APPROVAL NEEDED`?`LAST OBSERVED COMMAND`:$.data?.control?`COMMAND REQUEST`:`COMMAND TO APPROVE IN CODEX`)),Y(e,n)};Z(D,e=>{(!K(a)||!$.data?.control)&&t().fileChanges?.length?e(ee):K(a)&&$.data?.control||!t().command&&t().status!==`APPROVAL NEEDED`||e(te,1)});var O=H(D,2),k=e=>{var n=zs(),r=V(n);U(()=>X(r,`CONTEXT SOURCE // ${(t().source||`LOCAL`)??``} // OBSERVATION`)),Y(e,n)};Z(O,e=>{K(a)||e(k)}),Y(e,o)},r=L(void 0),i=Wi(t,`params`,19,()=>({})),a=F(()=>$.data?.sessions||[]),o=F(()=>$.data?.control&&$.data.profiles||[]),s=F(()=>($.data?.triggers||[]).filter(e=>K(a).some(t=>t.id===e.session)));function c(e,t){i().id&&i().id!==t&&K(u)?e.preventDefault():x(t)}function l(e){return[`TURN COMPLETE`,`IDLE`,`ACTIVE`].includes(e.status)&&K(s).some(t=>t.session===e.id)?`TRIGGER SET`:e.status}let u=L(!1);function d(e,t){i().id&&i().id!==t&&K(u)?e.preventDefault():x(t)}vn(()=>{let e=i().id;e&&fr(()=>{aa.selected=e,fa(e,2)})});let p=F(()=>K(a).find(e=>e.id===i().id)),m=F(()=>new URLSearchParams(Yi.querystring).get(`review`)===`profile`&&K(o).some(e=>e.session===i().id&&e.pending)),h=F(()=>K(a).some(e=>e.id===aa.selected)?aa.selected:K(a)[0]?.id),v=F(()=>!$.connected||!!$.data?.sessionsError),y=F(()=>[[`OBSERVED TOKENS`,pa(K(a).reduce((e,t)=>e+t.tokens,0))],[`LISTED SESSIONS`,pa(K(a).length)],...[[`WORKING`,`WORKING`],[`AWAITING APPROVAL`,`APPROVAL NEEDED`],[`AWAITING INPUT`,`INPUT NEEDED`],[`CHECK · INFERRED`,`CHECK SESSION`]].map(([e,t])=>[e,K(v)?`—`:pa(K(a).filter(e=>e.status===t).length)])]),b=F(()=>K(a).filter(e=>[`INPUT NEEDED`,`APPROVAL NEEDED`,`CHECK SESSION`].includes(e.status)));function x(e){aa.selected=e,lr().then(()=>document.querySelector(`.session-row.selected`)?.scrollIntoView({block:`nearest`}))}function S(e,t){if(x(e),i().id){t<0&&(fa(e,2),location.hash=`/sessions`);return}let n=ua(e)+t;n>2?location.hash=`/sessions/`+encodeURIComponent(e):fa(e,n)}function C(e){let t=e.target;if(!(e.altKey||e.ctrlKey||e.metaKey||e.shiftKey||t.closest(`input, textarea, select, [contenteditable="true"]`))){if(e.key===`Escape`&&i().id)e.preventDefault(),S(i().id,-1);else if([`ArrowLeft`,`ArrowRight`].includes(e.key)&&(i().id||K(h)))e.preventDefault(),S(i().id||K(h),e.key===`ArrowLeft`?-1:1);else if(!i().id&&[`ArrowUp`,`ArrowDown`].includes(e.key)&&K(a).length){e.preventDefault();let t=K(a).findIndex(e=>e.id===K(h));x(K(a)[Math.max(0,Math.min(K(a).length-1,t+(e.key===`ArrowUp`?-1:1)))].id)}}}function w(e){return K(v)?`Last observation only — refresh unavailable. Check Codex for current state.`:e.status===`CHECK SESSION`?`INFERRED INACTIVITY — a quiet session, not a confirmed input or approval request. Local tools may still be running.`:$.data?.control&&[`APPROVAL NEEDED`,`INPUT NEEDED`].includes(e.status)?`Open full detail for supported live controls; otherwise reply in Codex.`:e.status===`APPROVAL NEEDED`?`OBSERVED APPROVAL SIGNAL — approve or decline in Codex.`:e.status===`INPUT NEEDED`?`OBSERVED INPUT SIGNAL — reply in Codex.`:e.status===`TURN COMPLETE`?`OBSERVED TURN COMPLETION — informational, not an approval request.`:``}function T(e){return e.streaming&&e.contextKind===`LAST REPLY`?`REPLY // STREAMING`:e.contextKind||`LAST ACTIVITY`}var E=fc();xr(`keydown`,Qt,C);var D=B(E),ee=V(H(z(D),2));P(D);var te=H(D,2);let O;Xr(te,21,()=>K(y),Kr,(e,t)=>{var n=F(()=>_(K(t),2));let r=()=>K(n)[0],i=()=>K(n)[1];var a=Vs(),o=z(a),s=V(o,!0),c=V(H(o,2),!0);P(a),U(()=>{X(s,r()),X(c,i())}),Y(e,a)}),P(te);var k=H(te,2),ne=V(k),re=H(k,2),ie=e=>{Y(e,Hs())};Z(re,e=>{K(v)&&e(ie)});var ae=H(re,2),oe=e=>{Y(e,Us())};Z(ae,e=>{$.data?.profileError&&e(oe)});var se=H(ae,2),A=e=>{var t=qs(),n=z(t);Xr(n,17,()=>K(b),Kr,(e,t)=>{var n=Ws();let r;var i=V(n);U((e,a)=>{r=ci(n,1,`button`,null,r,{approval:K(t).status===`APPROVAL NEEDED`}),Q(n,`href`,e),X(i,`${K(t).status??``} ${a??``} // ${(K(t).name||K(t).directory||K(t).id)??``}`)},[()=>`#/sessions/`+encodeURIComponent(K(t).id),()=>Array.from(K(t).id).slice(-5).join(``).toUpperCase()]),q(`click`,n,()=>x(K(t).id)),Y(e,n)});var r=H(n,2);Xr(r,17,()=>K(o).filter(e=>e.pending&&K(a).some(t=>t.id===e.session)),Kr,(e,t)=>{var n=Gs(),r=V(n);U((e,a,o)=>{Q(n,`title`,i().id&&i().id!==K(t).session&&K(u)?`Finish sending or clear your current draft before switching sessions.`:`Review this session’s quota threshold`),Q(n,`href`,e),X(r,`QUOTA THRESHOLD ${a??``} // ${o??``}`)},[()=>`#/sessions/`+encodeURIComponent(K(t).session)+`?review=profile`,()=>Array.from(K(t).session).slice(-5).join(``).toUpperCase(),()=>K(a).find(e=>e.id===K(t).session)?.name||K(a).find(e=>e.id===K(t).session)?.directory||K(t).session]),q(`click`,n,e=>d(e,K(t).session)),Y(e,n)}),Xr(H(r,2),17,()=>K(s),Kr,(e,t)=>{var n=Ks(),r=V(n);U((e,t,i)=>{Q(n,`href`,e),X(r,`TRIGGER SET ${t??``} // ${i??``}`)},[()=>`#/sessions/`+encodeURIComponent(K(t).session),()=>Array.from(K(t).session).slice(-5).join(``).toUpperCase(),()=>K(a).find(e=>e.id===K(t).session)?.name||K(a).find(e=>e.id===K(t).session)?.directory]),q(`click`,n,e=>c(e,K(t).session)),Y(e,n)}),P(t),Y(e,t)},ce=F(()=>(K(b).length||K(o).some(e=>e.pending)||K(s).length)&&!K(v));Z(se,e=>{K(ce)&&e(A)});var le=H(se,2),ue=e=>{var t=Nr(),a=B(t),s=e=>{var t=Qs(),a=z(t),s=z(a),c=z(s);let d;var f=H(c);P(s);var h=H(s,2);P(a);var g=H(a,2),_=e=>{var t=zs(),n=V(t,!0);U(()=>X(n,K(p).directory)),Y(e,t)};Z(g,e=>{K(p).name&&K(p).directory&&e(_)});var y=H(g,2),b=V(y),S=H(y,2);let C;var w=z(S),T=z(w);n(T,()=>K(p),()=>!0,()=>!0),P(w);var E=H(w,2),D=e=>{var t=Nr();Gr(B(t),()=>K(p).id,e=>{{let t=F(()=>new URLSearchParams(Yi.querystring).get(`schedule`)===`edit`),n=F(()=>K(p).fileChanges||[]);xs(e,{get session(){return K(p).id},oncopy:()=>K(r)?.copy(),get openScheduler(){return K(t)},get observedCommand(){return K(p).command},get observedFileChanges(){return K(n)},get suspended(){return K(m)},onProtectedChange:e=>{R(u,e,!0)}})}}),Y(e,t)},ee=e=>{var t=Js();Gr(H(z(t),2),()=>K(p).id,e=>{Po(e,{get session(){return K(p).id},localOnly:!0,oncopy:()=>K(r)?.copy()})}),P(t),Y(e,t)};Z(E,e=>{$.data?.control?e(D):e(ee,-1)}),P(S);var te=H(S,2);Xr(te,17,()=>K(o).filter(e=>e.session===K(p).id),e=>e.session,(e,t)=>{var n=Zs(),r=z(n),i=e=>{var n=Ys(),r=V(n,!0);U(()=>X(r,K(t).notice)),Y(e,n)};Z(r,e=>{K(t).notice&&e(i)});var a=H(r,2),o=e=>{xs(e,{get session(){return K(p).id},review:`profile`})},s=e=>{var t=Xs();U(e=>Q(t,`href`,e),[()=>`#/sessions/`+encodeURIComponent(K(p).id)+`?review=profile`]),Y(e,t)};Z(a,e=>{K(t).pending&&K(m)?e(o):K(t).pending&&e(s,1)}),P(n),U(()=>Q(n,`id`,`quota-profile-`+K(p).id)),Y(e,n)});var O=H(te,2),k=e=>{var t=Nr();Gr(B(t),()=>K(p).id,e=>{Ii(Os(e,{get session(){return K(p)},active:!0}),e=>R(r,e,!0),()=>K(r))}),Y(e,t)};Z(O,e=>{K(m)||e(k)}),Gr(H(O,2),()=>K(p).id,e=>{Es(e,{get session(){return K(p)},get stale(){return K(v)}})}),P(t),U((e,t)=>{d=ci(c,1,`lamp lit`,null,d,{working:K(p).status===`WORKING`&&!K(v)}),X(f,`${e??``} // ${(K(p).name||K(p).directory)??``}`),X(b,`${t??``} TOKENS // ${K(p).id??``} // CONTEXT SOURCE // ${(K(p).source||`LOCAL`)??``}`),C=ui(S,``,C,{display:K(m)?`none`:void 0})},[()=>K(v)?`STALE`:K(m)?`QUOTA THRESHOLD`:l(K(p)),()=>pa(K(p).tokens)]),q(`click`,h,()=>{x(i().id),fa(i().id,2)}),Y(e,t)},c=e=>{Y(e,$s())};Z(a,e=>{K(p)?e(s):e(c,-1)}),Y(e,t)},de=e=>{var t=dc(),r=B(t),i=H(z(r),2),s=z(i),c=H(s,2);P(i),P(r);var u=H(r,2);Xr(u,17,()=>K(a),e=>e.id,(e,t)=>{let r=F(()=>ua(K(t).id));var i=lc();let a;var s=z(i),c=z(s),u=e=>{var n=ec(),r=z(n),i=V(r);P(n),U(e=>{Q(r,`aria-pressed`,K(h)===K(t).id),X(i,`${e??``} // ${K(t).name??``}`)},[()=>Array.from(K(t).id).slice(-5).join(``).toUpperCase()]),q(`click`,r,()=>x(K(t).id)),Y(e,n)};Z(c,e=>{K(t).name&&e(u)});var f=H(c,2),p=z(f),m=z(p);let g;var _=H(m,1,!0);P(p),P(f);var y=H(f,2),b=e=>{var n=tc(),r=V(n,!0);U(()=>{Q(n,`aria-pressed`,K(h)===K(t).id),X(r,K(t).directory||K(t).id)}),q(`click`,n,()=>x(K(t).id)),Y(e,n)};Z(y,e=>{K(t).name||e(b)});var C=H(y,2),w=z(C);xe(),P(C);var E=H(C,2),D=e=>{var n=zs(),r=V(n,!0);U(()=>X(r,K(t).directory)),Y(e,n)};Z(E,e=>{K(t).name&&K(t).directory&&e(D)});var ee=H(E,2),te=V(ee),O=H(ee,2),k=V(O),ne=H(O,2),re=z(ne),ie=H(re,2),ae=V(ie,!0),oe=H(ie,2);P(ne);var se=H(ne,2),A=H(se,2),ce=e=>{var n=nc(),r=V(n,!0);U(()=>X(r,K(t).status===`CHECK SESSION`?`INFERRED INACTIVITY`:$.data?.control?`OPEN FULL DETAIL OR REPLY IN CODEX`:K(t).status===`APPROVAL NEEDED`?`APPROVE OR DECLINE IN CODEX`:`REPLY IN CODEX`)),Y(e,n)},le=F(()=>!K(v)&&[`APPROVAL NEEDED`,`INPUT NEEDED`,`CHECK SESSION`].includes(K(t).status));Z(A,e=>{K(le)&&e(ce)}),P(s);var ue=H(s,2),de=e=>{var r=sc(),i=z(r),a=z(i),s=V(a,!0),c=H(a,2),l=e=>{var n=rc(),r=V(n,!0);U(e=>{Q(n,`href`,e),X(r,$.data?.control?`REVIEW IN FULL DETAIL ↗`:`APPROVAL IN CODEX ↗`)},[()=>`#/sessions/`+encodeURIComponent(K(t).id)]),Y(e,n)};Z(c,e=>{!K(v)&&K(t).status===`APPROVAL NEEDED`&&e(l)}),P(i);var u=H(i,2);n(u,()=>K(t),()=>!1);var f=H(u,2);Xr(f,17,()=>K(o).filter(e=>e.session===K(t).id),Kr,(e,n)=>{var r=oc(),i=B(r),a=e=>{var t=ic(),r=V(t,!0);U(()=>X(r,K(n).notice)),Y(e,t)};Z(i,e=>{K(n).notice&&e(a)});var o=H(i,2),s=e=>{var n=ac();U(e=>Q(n,`href`,e),[()=>`#/sessions/`+encodeURIComponent(K(t).id)+`?review=profile`]),q(`click`,n,e=>d(e,K(t).id)),Y(e,n)};Z(o,e=>{K(n).pending&&e(s)}),Y(e,r)});var p=H(f,2);{let e=F(()=>K(h)===K(t).id);Os(p,{get session(){return K(t)},get active(){return K(e)}})}P(r),U(e=>X(s,e),[()=>T(K(t))]),Y(e,r)};Z(ue,e=>{K(r)>0&&e(de)});var fe=H(ue,2),j=e=>{var n=cc(),r=H(z(n),2);{let e=F(()=>(K(t).samples||[]).map(e=>e.tokens));io(r,{get values(){return K(e)},capacity:120})}var i=V(H(r,2));P(n),U(()=>X(i,`LAST ${(K(t).samples?.length||0)??``} SAMPLES // AUTO SCALE`)),Y(e,n)};Z(fe,e=>{K(r)<2&&e(j)}),P(i),U((e,n,o,s,c)=>{a=ci(i,1,`session-row`,null,a,{selected:K(h)===K(t).id,wide:K(r)===2,split:K(r)===1}),Q(i,`aria-label`,`Session `+(K(t).name||K(t).directory||K(t).id)),Q(p,`href`,e),g=ci(m,1,`lamp lit`,null,g,{working:K(t).status===`WORKING`&&!K(v)}),X(_,n),X(w,`${o??``} `),X(te,`${K(t).agents??``} LINKED AGENTS`),X(k,`ACTIVE // ${s??``}`),re.disabled=K(r)===0,Q(ie,`aria-expanded`,K(r)>0),X(ae,K(r)?`HIDE DETAIL`:`SHOW DETAIL`),Q(se,`href`,c)},[()=>`#/sessions/`+encodeURIComponent(K(t).id),()=>K(v)?`STALE`:l(K(t)),()=>pa(K(t).tokens),()=>ma(K(t).activity),()=>`#/sessions/`+encodeURIComponent(K(t).id)]),q(`click`,p,()=>x(K(t).id)),q(`click`,re,()=>S(K(t).id,-1)),q(`click`,ie,()=>{x(K(t).id),fa(K(t).id,+!K(r))}),q(`click`,oe,()=>S(K(t).id,1)),q(`click`,se,()=>x(K(t).id)),Y(e,i)});var f=H(u,2),p=e=>{Y(e,uc())};Z(f,e=>{K(a).length||e(p)}),q(`click`,s,()=>da(1)),q(`click`,c,()=>da(0)),Y(e,t)};Z(le,e=>{i().id?e(ue):e(de,-1)}),U(e=>{X(ee,`OBSERVED ${e??``}`),O=ci(te,1,`session-totals`,null,O,{stale:K(v)}),X(ne,`${K(v)?`LAST KNOWN TOTALS — live state counts unavailable. `:``}Tokens observed since this server started for currently listed sessions; linked agents are already included. Totals can decrease when a session leaves the diff --git a/internal/web/dist/index.html b/internal/web/dist/index.html index ebf4529..9845735 100644 --- a/internal/web/dist/index.html +++ b/internal/web/dist/index.html @@ -5,7 +5,7 @@