Close an import batch on bytes and report what it drops

docs/CHANGES.md

@@ -6,6 +6,20 @@ diff. `CLAUDE.md` holds the invariants; `docs/superpowers/specs/` holds the full
 Not release notes. Each entry is written as of the change that produced it and is not revised as the
 code moves on; where an entry disagrees with the code, the code wins.
 
+## An import batch closes on bytes as well as lines
+
+**#814** closes a transcript batch at 4 MiB of line content as well as at 100 lines, and reports
+what an import leaves behind. A Codex rollout's `compacted` line re-carries every earlier user
+message, so a hundred of them could exceed the 30 MB body Kestrel binds by default; the server
+answered 413 and the importer, which read the status only for a strict caller, counted the batch as
+sent. The budget is measured on raw UTF-8 bytes and sits a third of the limit under the worst JSON
+escaping expansion, so it holds whatever a line contains. A single line over the budget cannot be
+split and is skipped rather than posted in a batch that would be refused. Both the skip and a
+refused batch travel on the import progress channel as warnings carrying their session id: a routed
+source shares one sink across the run, and a stderr line written under the live progress region is
+erased by its next frame. The lines of a dropped batch still count as sent; the figure is what was
+posted, and the warning above it is the record of what was not kept.
+
 ## Read a linked pull request inside the workspace
 
 The work-context pane now has a compact PR card and a wide native reader beside

src/Capacitor.Cli/Commands/IImportSource.cs

@@ -49,12 +49,16 @@ internal sealed record ClassifyContext(
 /// effective --private flag from the orchestrator; DefaultVisibility carries
 /// the Step 3 setup visibility choice, or null for standalone `kcap import`.
 /// Neither is read directly by a source — <see cref="VisibilityStampFor"/> is.
+/// Progress is the sink a source hands to every <see cref="SessionImporter.SendTranscriptBatches"/>
+/// call so an <see cref="ImportWarning"/> reaches the user; it is shared by every session of the
+/// run, which is why a warning carries its own session id.
 /// </summary>
 internal sealed record ImportContext(
-    HttpClient HttpClient,
-    string     BaseUrl,
-    bool       ForcePrivate,
-    string?    DefaultVisibility = null) {
+    HttpClient                 HttpClient,
+    string                     BaseUrl,
+    bool                       ForcePrivate,
+    string?                    DefaultVisibility = null,
+    IProgress<ImportProgress>? Progress          = null) {
     /// <summary>
     /// The <c>default_visibility</c> to stamp on a session-start, or null to leave the field off.
     /// <b>An omitted stamp is not "no default"</b> — the server coalesces an absent one to

src/Capacitor.Cli/Commands/ImportCommand.cs

@@ -1308,6 +1308,11 @@ or ClassificationStatus.AlreadyLoaded
         var       summaryFailures    = new ConcurrentBag<(string SessionId, string Reason)>();
         var       importedSessionIds = new ConcurrentBag<string>();
 
+        void WarnSession(string sessionId, string message) => display.Line(
+            $"! {sessionId}: {message}",
+            $"[yellow]![/] [cyan]{Markup.Escape(sessionId)}[/] {Markup.Escape(message)}"
+        );
+
         var events = new ChainWorkerEvents {
             OnSessionStarted  = (_, _) => { },    // non-TTY: session start is silent; TTY overrides below
             OnSubagentStarted = (_, _, _) => { }, // non-TTY: subagent start is silent; TTY overrides below
@@ -1320,6 +1325,7 @@ or ClassificationStatus.AlreadyLoaded
                 $"Skipping {sid} [{reason}]",
                 FormatSkippedReasonMarkup(sid, reason)
             ),
+            OnSessionWarning = (_, sid, message) => WarnSession(sid, message),
             OnSessionEnded = (_, c, outcome, lines) => {
                 importedSessionIds.Add(c.SessionId);
 
@@ -1673,7 +1679,10 @@ string LoadingDesc(int slot, string sid, string verb) =>
                 HttpClient: httpClient,
                 BaseUrl: baseUrl,
                 ForcePrivate: forcePrivate,
-                DefaultVisibility: defaultVisibility
+                DefaultVisibility: defaultVisibility,
+                Progress: new CallbackProgress(ev => {
+                    if (ev is ImportWarning w) WarnSession(w.SessionId, w.Message);
+                })
             );
 
             async Task<ImportSessionResult> ImportOne(SessionClassification c) {
@@ -2843,6 +2852,9 @@ internal sealed record ChainWorkerEvents {
         /// <summary>Fired when a session import fails on a worker slot.</summary>
         public required Action<int, string, string> OnSessionErrored { get; init; } // slot, sessionId, reason
 
+        /// <summary>Fired for an <see cref="ImportWarning"/>: content the import left behind without failing.</summary>
+        public required Action<int, string, string> OnSessionWarning { get; init; } // slot, sessionId, message
+
         /// <summary>
         /// Fired after a session import completes (loaded or resumed). The slot is
         /// available for the next session as soon as this returns.
@@ -3001,6 +3013,7 @@ internal enum SessionImportOutcome { Loaded, Resumed, Errored }
                     case BatchFlushed { AgentId: null } bf: events.OnSessionProgress(slot, bf.LinesAdded, sendableTotal); break;
                     case SubagentStarted ss:               events.OnSubagentStarted(slot, session.SessionId, ss.AgentId); break;
                     case SubagentFinished sf:              events.OnSubagentFinished(slot, session.SessionId, sf.AgentId, sf.LinesSent); break;
+                    case ImportWarning w:                  events.OnSessionWarning(slot, w.SessionId, w.Message); break;
                 }
             }
         );

src/Capacitor.Cli/Commands/ImportProgress.cs

@@ -19,3 +19,27 @@ public sealed record SubagentStarted(string AgentId) : ImportProgress;
 
 /// <summary>Fired after a subagent's transcript has been fully streamed.</summary>
 public sealed record SubagentFinished(string AgentId, int LinesSent) : ImportProgress;
+
+/// <summary>
+/// Content the importer left behind without failing the session. Every sink must show these:
+/// an import that reports success while a warning went unrendered is a silent loss. The session
+/// id travels on the event because a routed source's sink is shared by every session of the run.
+/// </summary>
+public abstract record ImportWarning(string SessionId, string? AgentId) : ImportProgress {
+    public abstract string Message { get; }
+
+    protected string Scope => AgentId is null ? "" : $"subagent {AgentId} ";
+}
+
+/// <summary>A line that alone exceeds the batch byte budget; it cannot be split, so it is not sent.</summary>
+public sealed record LineSkipped(string SessionId, string? AgentId, int LineNumber, int Bytes)
+    : ImportWarning(SessionId, AgentId) {
+    public override string Message =>
+        $"{Scope}line {LineNumber} skipped: {Bytes} bytes exceeds the {TranscriptBatchBuffer.MaxBytes >> 20} MiB per-line limit";
+}
+
+/// <summary>A batch the server refused, or that never reached it, on a path that keeps going past the loss.</summary>
+public sealed record BatchDropped(string SessionId, string? AgentId, int FirstLineNumber, int LastLineNumber, string Reason)
+    : ImportWarning(SessionId, AgentId) {
+    public override string Message => $"{Scope}lines {FirstLineNumber}-{LastLineNumber} dropped: {Reason}";
+}