Codex app-server: envelope transcript source (mapper + forward buffer)

src/Capacitor.Cli.Core/Models.cs

@@ -1230,6 +1230,23 @@ public static class AcpEventKind {
     /// cursor (verified against <c>CapacitorHub.AcpSessionEvents</c>'s unrecognised-Kind branch), so
     /// a newer daemon degrades to log-only rather than wedging the forwarder.</summary>
     public const string SystemNote         = "system_note";
+
+    /// <summary>A full plan snapshot (codex app-server <c>turn/plan/updated</c>, which always sends
+    /// complete revisions). Canonical, latest-snapshot-wins; the server maps it to
+    /// <c>PlanContentUpdatedEvent</c>, landing envelope-sourced sessions on the same native-plan path
+    /// <c>PlanArtifactExtractor</c> already consumes. Additive for other ACP vendors (their translator
+    /// may keep dropping plan updates until wired). An older server treats it as an unrecognised Kind
+    /// (dropped, cursor still advances).</summary>
+    public const string Plan               = "plan";
+
+    /// <summary>A per-event additive token-usage DELTA (codex app-server <c>thread/tokenUsage/updated</c>,
+    /// daemon-converted from cumulative to delta and attributed to the model resolved at that instant).
+    /// Distinct from <see cref="Usage"/>, which is context-window OCCUPANCY (the ACP context-usage
+    /// reading), not additive
+    /// billing buckets. The server stamps these buckets into Eventuous <c>$usage</c> metadata so the
+    /// existing additive folds (session totals, per-model attribution, cost) count them unchanged. An
+    /// older server treats it as an unrecognised Kind (dropped, cursor still advances).</summary>
+    public const string TokenUsage         = "token_usage";
 }
 
 /// <summary>
@@ -1285,7 +1302,33 @@ public readonly record struct AcpEventEnvelope(
         long?   ContextWindowTokens = null,
 
         // transcript-authoritative time (ISO-8601); server falls back to now if absent
-        string? TimestampIso      = null
+        string? TimestampIso      = null,
+
+        // Ephemeral live lane (codex app-server envelope transcript). Additive/default-false, so
+        // ContractVersion stays 1: an older server ignores both and its canonical-only path is
+        // unchanged. Ephemeral=true marks a transient live chunk (accumulated content-so-far for its
+        // item) that is relayed but NEVER persisted and carries NO seq — it consumes no canonical
+        // sequence number and is excluded from the dup/gap logic (the server relays it in batch order).
+        // The pure-replacement viewer rule (state[ItemId] = latest ephemeral payload; the item's
+        // canonical completed envelope replaces and finalizes it) makes a dropped/duplicated ephemeral
+        // harmless. ItemId is the app-server item id — the stable key a viewer uses to know which
+        // transient state a completed item supersedes; it rides BOTH the ephemeral envelopes and their
+        // item's canonical completed envelope, and the server stamps it into the canonical event's
+        // METADATA (the event records are not ours to change).
+        bool    Ephemeral         = false,
+        string? ItemId            = null,
+
+        // token_usage — a per-event additive token DELTA (codex app-server). Additive/nullable, so
+        // ContractVersion stays 1: an older server ignores them. The model rides the Model field
+        // above (attributed to the model resolved at the delta's instant — correct across a reroute).
+        // input is GROSS (server converts to net = input − cached before stamping $usage, matching
+        // UsageMetadataHelper's cross-vendor contract). cache-write is the cache-CREATION tier, billed
+        // separately from cached reads. total is derived server-side and not carried.
+        long?   UsageInputTokens       = null,
+        long?   UsageCachedInputTokens = null,
+        long?   UsageCacheWriteInputTokens = null,
+        long?   UsageOutputTokens      = null,
+        long?   UsageReasoningTokens   = null
     );
 
 /// <summary>

src/Capacitor.Cli.Daemon/Harness/Codex/CodexAppServerHostedAgentRuntime.cs

@@ -1,6 +1,7 @@
 using System.Runtime.CompilerServices;
 using System.Text.Json;
 using System.Text.Json.Nodes;
+using System.Threading.Channels;
 using Capacitor.Cli.Core;
 using Capacitor.Cli.Core.Acp;
 using Capacitor.Cli.Daemon.Acp;
@@ -41,11 +42,13 @@ internal sealed record CodexAppServerLaunch(
     string                ClientVersion);
 
 /// <summary>
-/// A <see cref="IHostedAgentRuntime"/> that hosts a Codex reviewer over the <c>codex app-server</c>
-/// JSON-RPC protocol instead of the interactive PTY. It is control-plane only: the transcript is
-/// still recorded through the Codex hooks + <c>kcap watch</c> rollout path (which is why the
-/// hook-trust preflight is load-bearing), so this runtime never aggregates transcript from the
-/// protocol stream and <see cref="EmitsTerminalOutput"/> is <see langword="false"/>.
+/// A <see cref="IHostedAgentRuntime"/> that hosts a Codex agent over the <c>codex app-server</c>
+/// JSON-RPC protocol instead of the interactive PTY. It exposes <see cref="IAcpTranscriptSource"/>
+/// (§2.4): every app-server notification is translated by <see cref="CodexNotificationMapper"/> into the
+/// canonical/ephemeral <see cref="AcpEventEnvelope"/> vocabulary and drained through a bounded
+/// <see cref="CodexForwardBuffer"/>, so the envelope transcript — not the hooks + <c>kcap watch</c>
+/// rollout path — is the ingestion source for hosted app-server sessions. <see cref="EmitsTerminalOutput"/>
+/// stays <see langword="false"/> (this is a structured, not terminal, view).
 ///
 /// <para>Lifecycle (<see cref="StartAsync"/>, called by the factory): spawn → <c>initialize</c> →
 /// <c>hooks/list</c> trust preflight (seed + one restart when required) → <c>thread/start</c> →
@@ -57,7 +60,7 @@ internal sealed record CodexAppServerLaunch(
 /// request is a protocol violation — answered with a valid on-the-wire <c>decline</c> (never a
 /// JSON-RPC error, whose turn/process effect is Codex's to define) and logged at Error.</para>
 /// </summary>
-internal sealed partial class CodexAppServerHostedAgentRuntime : IHostedAgentRuntime {
+internal sealed partial class CodexAppServerHostedAgentRuntime : IHostedAgentRuntime, IAcpTranscriptSource {
     // Reviewers need lifecycle + usage only; opting out of the high-volume delta streams is a
     // performance choice (the reader always drains regardless — the app-server blocks rather than
     // dropping), not a correctness one.
@@ -72,6 +75,13 @@ internal sealed partial class CodexAppServerHostedAgentRuntime : IHostedAgentRun
     const string ClientName = "kcap-daemon";
     const int    BackpressureMaxAttempts = 6;
 
+    // §2.4 forward buffer sizing. The capacity trades memory for the SignalR-outage window a canonical
+    // burst can ride before the read loop blocks; the stall timeout bounds that block before a
+    // deterministic terminal fault. `codex.appServer.forwardStallSeconds` (spec-named) maps to the
+    // injectable stall timeout below; the default engages when the caller passes none.
+    const int    ForwardBufferCapacity      = 1024;
+    const double DefaultForwardStallSeconds = 30;
+
     readonly CodexAppServerSpawn  _spawn;
     readonly CodexAppServerLaunch _launch;
     readonly AgentActivityClock?  _clock;
@@ -90,6 +100,11 @@ internal sealed partial class CodexAppServerHostedAgentRuntime : IHostedAgentRun
     // clock, WaitForTurnIdle) is derived from it — the runtime no longer tracks turns itself.
     readonly CodexTurnInputDispatcher _dispatcher;
 
+    // §2.4 envelope transcript: the mapper translates every app-server notification into
+    // canonical/ephemeral envelopes, drained through the bounded forward buffer (IAcpTranscriptSource).
+    readonly CodexNotificationMapper _mapper;
+    readonly CodexForwardBuffer      _forwardBuffer;
+
     CodexAppServerConnection? _connection;
     IAcpProcess?              _process;
     Task                      _runLoop = Task.CompletedTask;
@@ -101,17 +116,32 @@ internal sealed partial class CodexAppServerHostedAgentRuntime : IHostedAgentRun
     CodexTokenUsage?          _usage;
     int                       _disposed;
 
+    // §2.4 envelope emission is gated OFF until §2.5 activates the envelope-source ingestion (deferred
+    // first-turn, source claim, and the hooks/watch dedup). Feeding the buffer while it is NOT drained
+    // by an attached AcpTranscriptForwarder would fill it and stall the read loop; draining it WITHOUT
+    // the §2.5 dedup would double-ingest a reviewer session (hooks + envelopes). So the transcript
+    // surface (IAcpTranscriptSource, mapper, buffer) ships here dormant; §2.5 flips this one flag
+    // together with the factory's Transcript wiring and the dedup guards.
+    readonly bool _emitEnvelopeTranscript;
+
     public CodexAppServerHostedAgentRuntime(
             CodexAppServerSpawn spawn, CodexAppServerLaunch launch, AgentActivityClock? clock,
-            ILogger logger, TimeProvider? timeProvider = null) {
+            ILogger logger, TimeProvider? timeProvider = null, TimeSpan? forwardStallTimeout = null,
+            bool emitEnvelopeTranscript = false) {
         _spawn   = spawn;
         _launch  = launch;
         _clock   = clock;
         _logger  = logger;
         _time    = timeProvider ?? TimeProvider.System;
+        _emitEnvelopeTranscript = emitEnvelopeTranscript;
         _dispatcher = new CodexTurnInputDispatcher(
             startTurn: IssueTurnStartAsync, steerTurn: IssueTurnSteerAsync,
             logger: logger, ct: _cts.Token, onTurnInFlight: flip => _clock?.SetTurnInFlight(flip));
+        _mapper = new CodexNotificationMapper(() => _resolvedModel, logger);
+        _forwardBuffer = new CodexForwardBuffer(
+            ForwardBufferCapacity,
+            forwardStallTimeout ?? TimeSpan.FromSeconds(DefaultForwardStallSeconds),
+            _cts.Token, OnForwardStall);
     }
 
     // ── IHostedAgentRuntime: identity / lifecycle observables ──────────────────────────────────
@@ -134,6 +164,20 @@ public CodexAppServerHostedAgentRuntime(
     /// if none has arrived yet.</summary>
     public CodexTokenUsage? Usage => _usage;
 
+    // ── IAcpTranscriptSource (§2.4 envelope transcript) ──────────────────────────────────────────
+    // The canonical session id is the app-server thread id (== the rollout filename id and the hook
+    // payload session_id, established by the app-server protocol spike), read only after the
+    // thread/start handshake sets it.
+
+    /// <inheritdoc cref="IAcpTranscriptSource.AcpSessionId"/>
+    public string AcpSessionId => _threadId ?? "";
+
+    /// <inheritdoc cref="IAcpTranscriptSource.Cwd"/>
+    string IAcpTranscriptSource.Cwd => _launch.Cwd;
+
+    /// <inheritdoc cref="IAcpTranscriptSource.Envelopes"/>
+    public ChannelReader<AcpEventEnvelope> Envelopes => _forwardBuffer.Reader;
+
     // ── Startup (called by the factory, not part of the interface) ─────────────────────────────
 
     /// <summary>
@@ -199,10 +243,14 @@ async Task SpawnAndInitializeAsync(string? hookStateSeed, CancellationToken ct)
         _runLoop  = RunConnectionAsync(connection, _childCts.Token);
         _clock?.SetLaunchStage("spawned");
 
+        // With the envelope transcript ON, the mapper's ephemeral lane consumes exactly the delta streams
+        // DeltaOptOut suppresses, so opting out would silence the live lane the gate is meant to enable —
+        // receive everything. With it OFF (reviewer path), keep the perf opt-out.
+        var optOut = _emitEnvelopeTranscript ? Array.Empty<string>() : DeltaOptOut;
         var initParams = new JsonObject {
             ["clientInfo"]   = new JsonObject { ["name"] = ClientName, ["version"] = _launch.ClientVersion },
             ["capabilities"] = new JsonObject {
-                ["optOutNotificationMethods"] = new JsonArray(DeltaOptOut.Select(m => (JsonNode?) m).ToArray()),
+                ["optOutNotificationMethods"] = new JsonArray(optOut.Select(m => (JsonNode?) m).ToArray()),
             },
         };
         await RequestAsync("initialize", initParams, ct).ConfigureAwait(false);
@@ -339,10 +387,34 @@ void HandleNotification(AcpNotification n) {
                 _usage = ParseUsage(n.Params);
                 _clock?.Advance();
                 break;
+            case "model/rerouted":
+                // The resolved model changed mid-thread; the mapper attributes each subsequent token
+                // delta to the model-at-instant, so per-interval attribution across a reroute is correct.
+                if (n.Params?.Str("model") is { } rerouted) _resolvedModel = rerouted;
+                _clock?.Advance();
+                break;
             default:
                 _clock?.Advance();
                 break;
         }
+
+        // §2.4 envelope transcript: translate every notification into canonical/ephemeral envelopes and
+        // drain them through the forward buffer. Runs on the read-loop thread, so a full buffer blocks
+        // here (a canonical emit) — the intended lossless backpressure onto the app-server's stdout.
+        // Gated off until §2.5 activates envelope-source ingestion (see _emitEnvelopeTranscript).
+        if (_emitEnvelopeTranscript)
+            foreach (var env in _mapper.Map(n.Method, n.Params))
+                _forwardBuffer.Emit(env);
+    }
+
+    // §2.4 stall watchdog: the forward buffer stayed full of canonical envelopes past the stall timeout
+    // (the SignalR leg is wedged). Fault the runtime deterministically rather than wedging the read loop
+    // forever — the truncated canonical tail is reported here, never dropped silently.
+    void OnForwardStall(TimeSpan stallTimeout) {
+        _logger.LogError(
+            "codex app-server: forward buffer stalled for {StallSeconds}s (SignalR forwarding wedged); faulting the session — canonical tail truncated.",
+            stallTimeout.TotalSeconds);
+        _runtimeTerminal.TrySetResult();
     }
 
     /// <summary>
@@ -442,6 +514,7 @@ public async ValueTask DisposeAsync() {
         // the loop started) so a ReadOutputAsync consumer never hangs.
         _runtimeTerminal.TrySetResult();
         _dispatcher.FaultAll(new ObjectDisposedException(nameof(CodexAppServerHostedAgentRuntime)));
+        _forwardBuffer.Complete(); // end the envelope stream so a draining forwarder's ReadAllAsync completes
         _cts.Dispose();
     }
 
@@ -480,19 +553,36 @@ static string MapEffort(string effort) =>
         if (paramsEl is not { } p || p.Obj("tokenUsage") is not { } u || u.Obj("total") is not { } total)
             return null;
 
-        return new CodexTokenUsage(
-            InputTokens:           total.Num("inputTokens")           ?? 0,
-            CachedInputTokens:     total.Num("cachedInputTokens")     ?? 0,
-            OutputTokens:          total.Num("outputTokens")          ?? 0,
-            ReasoningOutputTokens: total.Num("reasoningOutputTokens") ?? 0,
-            TotalTokens:           total.Num("totalTokens")           ?? 0);
+        return CodexTokenUsage.FromTotal(total);
     }
 
     // JsonNode → JsonElement without reflection (AOT-safe): the node writes its own JSON.
     static JsonElement ToElement(JsonNode node) =>
         JsonDocument.Parse(node.ToJsonString()).RootElement.Clone();
 }
 
-/// <summary>Cumulative thread token usage from <c>thread/tokenUsage/updated.total</c>.</summary>
+/// <summary>Cumulative thread token usage from <c>thread/tokenUsage/updated.total</c> (the app-server
+/// <c>TokenUsageBreakdown</c>). <see cref="CacheWriteInputTokens"/> is the cache-CREATION tier, billed
+/// separately from <see cref="CachedInputTokens"/> reads — kept so the delta converter never silently
+/// drops a billed bucket.</summary>
 internal readonly record struct CodexTokenUsage(
-    long InputTokens, long CachedInputTokens, long OutputTokens, long ReasoningOutputTokens, long TotalTokens);
+    long InputTokens, long CachedInputTokens, long CacheWriteInputTokens,
+    long OutputTokens, long ReasoningOutputTokens, long TotalTokens) {
+
+    /// <summary>Reads a <c>TokenUsageBreakdown</c> object (<c>thread/tokenUsage/updated.total</c> or
+    /// <c>.last</c>). Missing fields read as 0 (the schema defaults <c>cacheWriteInputTokens</c> to 0
+    /// and the others are required).</summary>
+    public static CodexTokenUsage FromTotal(JsonElement total) => new(
+        InputTokens:           total.Num("inputTokens")           ?? 0,
+        CachedInputTokens:     total.Num("cachedInputTokens")     ?? 0,
+        CacheWriteInputTokens: total.Num("cacheWriteInputTokens") ?? 0,
+        OutputTokens:          total.Num("outputTokens")          ?? 0,
+        ReasoningOutputTokens: total.Num("reasoningOutputTokens") ?? 0,
+        TotalTokens:           total.Num("totalTokens")           ?? 0);
+
+    /// <summary>True when every additive bucket is zero — a delta carrying no information, which the
+    /// mapper drops rather than stamping an empty <c>$usage</c>.</summary>
+    public bool IsZero =>
+        InputTokens == 0 && CachedInputTokens == 0 && CacheWriteInputTokens == 0
+     && OutputTokens == 0 && ReasoningOutputTokens == 0 && TotalTokens == 0;
+}

src/Capacitor.Cli.Daemon/Harness/Codex/CodexEphemeralAccumulator.cs

@@ -0,0 +1,37 @@
+using System.Text;
+
+namespace Capacitor.Cli.Daemon.Harness.Codex;
+
+/// <summary>
+/// Accumulates codex app-server per-item deltas (<c>item/agentMessage/delta</c>, reasoning deltas,
+/// <c>item/commandExecution/outputDelta</c>, …) into the CUMULATIVE content-so-far that each ephemeral
+/// envelope carries. §2.4 fixes the contract: ephemeral payloads are the item's whole accumulated
+/// content (idempotent replacement at the viewer), never an increment — so a dropped or duplicated
+/// ephemeral is harmless, and the canonical <c>item/completed</c> snapshot supersedes it. There is no
+/// increment reassembly anywhere; <see cref="Complete"/> drops the item's transient state once its
+/// authoritative snapshot has been mapped.
+///
+/// <para>Not thread-safe: driven from the single notification-handling path.</para>
+/// </summary>
+internal sealed class CodexEphemeralAccumulator {
+    readonly Dictionary<string, StringBuilder> _byItem = new(StringComparer.Ordinal);
+
+    /// <summary>Appends a delta to the item's buffer and returns the accumulated content so far — the
+    /// payload of the ephemeral envelope for this item.</summary>
+    public string Accumulate(string itemId, string chunk) {
+        if (!_byItem.TryGetValue(itemId, out var sb)) {
+            sb = new StringBuilder();
+            _byItem[itemId] = sb;
+        }
+        sb.Append(chunk);
+        return sb.ToString();
+    }
+
+    /// <summary>Drops the item's transient buffer — called once the canonical completed envelope for it
+    /// has been produced (that snapshot finalizes the viewer's state for the item).</summary>
+    public void Complete(string itemId) => _byItem.Remove(itemId);
+
+    /// <summary>Number of items with live transient state (for the forward buffer's eviction bookkeeping
+    /// and tests).</summary>
+    public int ActiveItems => _byItem.Count;
+}