feat(session): protocol-safe checkpoint cut planner

agent/auto_compact.mbt

@@ -121,6 +121,15 @@ async fn AgentLoop::checkpoint_at_ceiling(
   steers : ArrayView[@agent_runtime.SteerInput],
 ) -> (@agent_session.Session, Bool) {
   let to_sequence = session.last_sequence()
+  // Warn-only invariant probe: every checkpoint site runs with its batch
+  // closed, so the planner must confirm the full range as a boundary. A
+  // clamp here means a sequencing bug let a checkpoint start mid-batch —
+  // full-range coverage still swallows the batch whole, but the tail and
+  // eviction policies this planner feeds must never inherit that state.
+  if session.checkpoint_cut(requested_to=to_sequence) != Some(to_sequence) {
+    @xlog.warn() <?
+      { "event": "auto_compaction_cut_clamped", "to_sequence": to_sequence }
+  }
   let covered = session
   let session = self.apply_steer_inputs(session, steers)
   @xlog.info() <?

agent_session/compact/compact.mbt

@@ -107,6 +107,14 @@ pub async fn compact_session(
   if to_sequence <= 0 {
     fail("session compaction requires at least one event")
   }
+  // Warn-only invariant probe: full-range coverage swallows batches whole,
+  // so this compaction proceeds either way — but a clamped planner result
+  // flags a crash-truncated log (a batch whose results never landed),
+  // which is worth surfacing next to the compaction it still allows.
+  if session.checkpoint_cut(requested_to=to_sequence) != Some(to_sequence) {
+    @xlog.warn() <?
+      { "event": "compaction_cut_clamped", "to_sequence": to_sequence }
+  }
   @xlog.info() <?
     {
       "event": "compaction_started",

agent_session/cut.mbt

@@ -0,0 +1,51 @@
+///|
+/// Largest protocol-safe coverage boundary at or below `requested_to`, or
+/// `None` when none exists.
+///
+/// A summary covering `1..to` SPLITS an assistant/tool batch when `to`
+/// falls between a tool call and its result: the covered call never enters
+/// the projection's pending set, so its uncovered sibling result is
+/// silently dropped — real output vanishes from every later request with
+/// nothing standing in for it, since the summary was only asked to cover
+/// the prefix. (Mid-range covers, a future incremental-checkpoint shape,
+/// additionally flush uncovered calls whose covered results can no longer
+/// answer them as synthetic repair text.) A boundary is safe exactly where
+/// no tool call is pending, so coverage always swallows batches whole.
+///
+/// The invariant is deliberately conservative at the log's edge: a batch
+/// whose results never arrived (a crashed process) keeps its calls pending
+/// forever, so every boundary from that batch on is rejected — for such
+/// logs the planner clamps below the batch even though covering the WHOLE
+/// open batch would not split it. Callers covering the full range may
+/// proceed identically either way; tail and eviction policies must respect
+/// the clamp, because a tail that starts mid-batch is exactly the split
+/// this planner exists to prevent.
+///
+/// Read-only single pass over raw events; `requested_to` beyond the last
+/// sequence is simply unconstraining.
+pub fn Session::checkpoint_cut(self : Session, requested_to~ : Int) -> Int? {
+  let mut best : Int? = None
+  let pending : Array[String] = []
+  for event in self.events() {
+    if event.sequence() > requested_to {
+      break
+    }
+    match event.item() {
+      Assistant(message) =>
+        for call in message.tool_calls() {
+          pending.push(call.id)
+        }
+      Tool(result) =>
+        // An orphaned result (no matching pending call) neither opens nor
+        // closes anything; projection drops it the same way.
+        if pending.search_by(id => id == result.tool_call_id()) is Some(index) {
+          ignore(pending.remove(index))
+        }
+      _ => ()
+    }
+    if pending.is_empty() {
+      best = Some(event.sequence())
+    }
+  }
+  best
+}

agent_session/cut_test.mbt

@@ -0,0 +1,165 @@
+///|
+fn batch_session() -> @agent_session.Session {
+  @agent_session.Session(
+    // 1 User, 2 Assistant(calls a+b), 3 Tool(a), 4 Tool(b), 5 Terminal.
+    SessionId("cut-batch"),
+    system_prompt="system",
+  )
+  .append(User(UserMessage("task")))
+  .append(
+    Assistant(
+      AssistantMessage("running two probes", tool_calls=[
+        ToolCall(id="call_a", name="probe", arguments="{}"),
+        ToolCall(id="call_b", name="probe", arguments="{}"),
+      ]),
+    ),
+  )
+  .append(
+    Tool(ToolResult(tool_call_id="call_a", tool_name="probe", content="ra")),
+  )
+  .append(
+    Tool(ToolResult(tool_call_id="call_b", tool_name="probe", content="rb")),
+  )
+  .append(Terminal(Finished("done")))
+}
+
+///|
+test "boundaries exist exactly where no tool call is pending" {
+  let session = batch_session()
+  assert_eq(session.checkpoint_cut(requested_to=5), Some(5))
+  assert_eq(session.checkpoint_cut(requested_to=4), Some(4))
+  // Sequences 2 and 3 sit inside the batch: the planner clamps to the
+  // boundary before it.
+  assert_eq(session.checkpoint_cut(requested_to=3), Some(1))
+  assert_eq(session.checkpoint_cut(requested_to=2), Some(1))
+  assert_eq(session.checkpoint_cut(requested_to=1), Some(1))
+  assert_eq(session.checkpoint_cut(requested_to=0), None)
+  // Beyond the last sequence is unconstraining.
+  assert_eq(session.checkpoint_cut(requested_to=100), Some(5))
+}
+
+///|
+test "result order inside a batch does not matter" {
+  // Results arrive b-then-a: the batch closes only once both landed.
+  let session = @agent_session.Session(
+      SessionId("cut-interleaved"),
+      system_prompt="system",
+    )
+    .append(User(UserMessage("task")))
+    .append(
+      Assistant(
+        AssistantMessage("two calls", tool_calls=[
+          ToolCall(id="call_a", name="probe", arguments="{}"),
+          ToolCall(id="call_b", name="probe", arguments="{}"),
+        ]),
+      ),
+    )
+    .append(
+      Tool(ToolResult(tool_call_id="call_b", tool_name="probe", content="rb")),
+    )
+    .append(
+      Tool(ToolResult(tool_call_id="call_a", tool_name="probe", content="ra")),
+    )
+  assert_eq(session.checkpoint_cut(requested_to=3), Some(1))
+  assert_eq(session.checkpoint_cut(requested_to=4), Some(4))
+}
+
+///|
+test "an open batch at the log edge clamps the boundary below it" {
+  let session = @agent_session.Session(
+      SessionId("cut-open-tail"),
+      system_prompt="system",
+    )
+    .append(User(UserMessage("task")))
+    .append(
+      Assistant(
+        AssistantMessage("crashed mid-batch", tool_calls=[
+          ToolCall(id="call_a", name="probe", arguments="{}"),
+        ]),
+      ),
+    )
+  assert_eq(session.checkpoint_cut(requested_to=2), Some(1))
+}
+
+///|
+test "a mid-log dangling call pins every later boundary" {
+  // The unanswered call at sequence 2 stays pending forever, so even the
+  // clean events after it never become boundaries.
+  let session = @agent_session.Session(
+      SessionId("cut-dangling"),
+      system_prompt="system",
+    )
+    .append(User(UserMessage("task")))
+    .append(
+      Assistant(
+        AssistantMessage("never answered", tool_calls=[
+          ToolCall(id="call_lost", name="probe", arguments="{}"),
+        ]),
+      ),
+    )
+    .append(User(UserMessage("resumed after a crash")))
+    .append(Assistant(AssistantMessage("plain answer")))
+    .append(Terminal(Finished("done")))
+  assert_eq(session.checkpoint_cut(requested_to=5), Some(1))
+}
+
+///|
+test "an orphaned tool result neither opens nor closes a batch" {
+  let session = @agent_session.Session(
+      SessionId("cut-orphan"),
+      system_prompt="system",
+    )
+    .append(User(UserMessage("task")))
+    .append(
+      Tool(
+        ToolResult(
+          tool_call_id="call_unknown",
+          tool_name="probe",
+          content="orphan",
+        ),
+      ),
+    )
+    .append(Terminal(Finished("done")))
+  assert_eq(session.checkpoint_cut(requested_to=3), Some(3))
+}
+
+///|
+/// The property the planner exists for: prefix coverage to a boundary
+/// leaves every uncovered event standing verbatim, while covering INSIDE a
+/// batch silently drops the batch's uncovered sibling result — the covered
+/// call never enters the projection's pending set, so pairing discards the
+/// result, and the summary does not stand in for it (it was only asked to
+/// cover the prefix).
+test "planner boundaries keep uncovered events; split cuts lose them" {
+  let session = batch_session()
+  let clean = session.compact(
+    content="checkpoint stand-in",
+    from_sequence=1,
+    to_sequence=4,
+    unix_ms=0,
+  )
+  let clean_projected = clean
+    .chat_messages()
+    .map(message => message.content)
+    .join("\n")
+  // The uncovered terminal survives verbatim next to the summary.
+  assert_true(clean_projected.contains("checkpoint stand-in"))
+  assert_true(clean_projected.contains("done"))
+  // Negative control: sequence 3 splits the batch between call_a's result
+  // and call_b's. rb (sequence 4) is NOT covered, yet it vanishes.
+  let split = session.compact(
+    content="checkpoint stand-in",
+    from_sequence=1,
+    to_sequence=3,
+    unix_ms=0,
+  )
+  let split_projected = split
+    .chat_messages()
+    .map(message => message.content)
+    .join("\n")
+  assert_true(split_projected.contains("done"))
+  assert_false(
+    split_projected.contains("rb"),
+    msg="split cut must drop the uncovered sibling result",
+  )
+}

agent_session/pkg.generated.mbti

@@ -55,6 +55,7 @@ pub fn Session::Session(SessionId, system_prompt~ : String, events? : @vector.Ve
 pub fn Session::append(Self, SessionItem, unix_ms? : Int64) -> Self
 pub fn Session::append_event(Self, SessionItem, unix_ms? : Int64) -> (Self, SessionEvent)
 pub fn Session::chat_messages(Self) -> Array[@deepseek.ChatMessage]
+pub fn Session::checkpoint_cut(Self, requested_to~ : Int) -> Int?
 pub fn Session::compact(Self, content~ : String, from_sequence~ : Int, to_sequence~ : Int, unix_ms? : Int64) -> Self raise
 pub fn Session::current_goal(Self) -> StandingGoal?
 pub fn Session::events(Self) -> @vector.Vector[SessionEvent]