Keep monitoring and session controls compatible with Codex 0.164

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

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

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=

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)
+	}
+}

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

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,21 +252,27 @@ 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()
 	}
 	if err != nil {
 		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()
 	}
 }