//! The session's active context view: an installed LCM frontier.
//!
//! A frontier is what the session has *decided* to carry forward. Raw events below its
//! `drained_through_seq` are represented by `summary` plus the source children they were built
//! from, not by replaying them into every provider request. Everything above it is still raw.
//!
//! Two properties are structural rather than asserted:
//!
//! * Installation is ONE durable append. A crash before it leaves the previous view active; a
//! crash after it leaves the new one. There is no window in which the session has neither.
//! * `plan` is pure and takes the previous frontier explicitly, so the same event page always
//! yields the same decision. Nothing about wall-clock or a mutable counter can change a view
//! that was already installed.
use anyhow::Result;
use serde::{Deserialize, Serialize};
use serde_json::{json, Value};
use crate::store::Event;
/// The event kind an installed frontier is recorded as.
pub const FRONTIER_KIND: &str = "lcm_frontier";
/// Scratchpad record types that carry conversation. The daemon assembles a request from `input`
/// events and from scratchpad records of these types, and from nothing else, so these are what a
/// frontier's size is measured in. Kernel placement, cell output and transient scratchpad noise
/// are covered by the range without deciding where it ends.
const CONVERSATION_SCRATCHPAD_TYPES: &[&str] = &["model_response", "tool_result"];
/// Whether an event carries conversation, and therefore counts toward the kept window.
fn carries_conversation(event: &Event) -> bool {
match event.kind.as_str() {
"input" => true,
"scratchpad" => event
.payload
.get("type")
.and_then(Value::as_str)
.is_some_and(|kind| CONVERSATION_SCRATCHPAD_TYPES.contains(&kind)),
"tool.result" => true,
_ => false,
}
}
/// Bounded size of a derived recollection, in characters.
const RECOLLECTION_BUDGET: usize = 4096;
/// Conversation-carrying events that stay raw below a newly installed frontier.
///
/// The daemon passes the operator's configured session window here, so there is one window in the
/// system rather than a second constant that disagrees the moment config changes.
pub const DEFAULT_KEEP_EVENTS: usize = 20;
/// One archived group of covered events, as the frontier remembers them.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct SourceChild {
pub kind: String,
pub first_seq: i64,
pub last_seq: i64,
pub count: usize,
}
/// Where exact actor testimony came from. Testimony without its source is a claim, not testimony.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct ActorTestimonySource {
pub actor_id: String,
pub session_id: String,
pub source_ref: String,
}
/// An installed frontier: the active view of a session's context.
///
/// Deliberately NOT `Serialize`/`Deserialize`. The canonical wire form is `to_payload` /
/// `from_payload`, which names the fields the record actually stores (`created_at_ms`, and
/// testimony as one nested object). A derived shape would be a SECOND serialization of the same
/// type that disagrees with the stored one - `to_value` would look reasonable and produce a
/// payload `from_payload` rejects - so the type has exactly one.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct Frontier {
/// Monotone per session. Every install increments it, so a stale install is detectable.
pub epoch: u64,
/// The covered prefix, by event sequence. Everything at or below is represented by the
/// summary; everything above is still raw.
pub drained_through_seq: i64,
/// The derived summary. Never presented as original evidence.
pub summary: String,
/// How the summary was derived, so a reader is never misled about its provenance.
pub derived_by: String,
pub created_ms: i64,
pub source_children: Vec<SourceChild>,
/// Exact actor testimony, carried as itself when one is installed. Absent stays absent.
pub actor_testimony: Option<String>,
/// The identity the testimony came from, carried beside it so an exact quote is attributable.
pub actor_testimony_source: Option<ActorTestimonySource>,
}
impl Frontier {
/// A view with nothing in it, for tests that care about one field.
#[cfg(test)]
fn for_test() -> Self {
Self {
epoch: 1,
drained_through_seq: 1,
summary: String::new(),
derived_by: "test".to_string(),
created_ms: 0,
source_children: Vec::new(),
actor_testimony: None,
actor_testimony_source: None,
}
}
/// The payload shape recorded in the event log.
pub fn to_payload(&self) -> Value {
json!({
"type": "lcm_frontier",
"epoch": self.epoch,
"drained_through_seq": self.drained_through_seq,
"summary": self.summary,
"derived_by": self.derived_by,
"created_at_ms": self.created_ms,
"source_children": self.source_children,
"actor_testimony": testimony_payload(
self.actor_testimony.as_deref(),
self.actor_testimony_source.as_ref(),
),
})
}
/// Reads a frontier back out of a recorded payload. A malformed record is not a frontier:
/// it is refused rather than half-read, because a half-read view silently loses context.
pub fn from_payload(payload: &Value) -> Result<Self> {
let is_frontier = payload.get("type").and_then(Value::as_str) == Some("lcm_frontier")
|| payload.get("kind").and_then(Value::as_str) == Some("lcm_frontier");
if !is_frontier {
anyhow::bail!("not an lcm_frontier payload");
}
let epoch = payload
.get("epoch")
.and_then(Value::as_u64)
.ok_or_else(|| anyhow::anyhow!("frontier has no epoch"))?;
let drained_through_seq = payload
.get("drained_through_seq")
.and_then(Value::as_i64)
.ok_or_else(|| anyhow::anyhow!("frontier has no drained_through_seq"))?;
let summary = payload
.get("summary")
.and_then(Value::as_str)
.ok_or_else(|| anyhow::anyhow!("frontier has no summary"))?
.to_string();
let derived_by = payload
.get("derived_by")
.and_then(Value::as_str)
.unwrap_or("unknown")
.to_string();
let created_ms = payload
.get("created_at_ms")
.and_then(Value::as_i64)
.unwrap_or(0);
let source_children: Vec<SourceChild> = payload
.get("source_children")
.map(|value| serde_json::from_value(value.clone()))
.transpose()?
.unwrap_or_default();
let testimony = payload.get("actor_testimony");
let actor_testimony = testimony
.and_then(|t| t.get("exact_text"))
.and_then(Value::as_str)
.map(str::to_string);
let actor_testimony_source = testimony
.and_then(|t| t.get("source"))
.map(|source| serde_json::from_value(source.clone()))
.transpose()?;
Ok(Self {
epoch,
drained_through_seq,
summary,
derived_by,
created_ms,
source_children,
actor_testimony,
actor_testimony_source,
})
}
}
/// The `actor_testimony` object: absent when there is no testimony, and otherwise the exact text
/// and the identity it came from, so a reader never has to guess either.
fn testimony_payload(exact: Option<&str>, source: Option<&ActorTestimonySource>) -> Value {
if exact.is_none() && source.is_none() {
return Value::Null;
}
let mut object = serde_json::Map::new();
if let Some(text) = exact {
object.insert("exact_text".into(), Value::String(text.to_string()));
}
if let Some(source) = source {
object.insert(
"source".into(),
serde_json::to_value(source).unwrap_or(Value::Null),
);
}
Value::Object(object)
}
/// Plans the frontier a session's event page implies, or `None` when nothing should be installed.
///
/// `keep` is the number of most recent conversation-carrying events that stay raw. A frontier is
/// installed only when at least that many would be drained: finite thresholds, so a session does
/// not reinstall a view on every turn (the hysteresis the LCM keeps for the same reason).
pub fn plan_frontier(
events: &[Event],
previous: Option<&Frontier>,
keep: usize,
) -> Option<Frontier> {
let covered = previous
.map(|frontier| frontier.drained_through_seq)
.unwrap_or(0);
let uncovered: Vec<&Event> = events.iter().filter(|event| event.id > covered).collect();
let conversation: Vec<&Event> = uncovered
.iter()
.copied()
.filter(|event| carries_conversation(event))
.collect();
if conversation.len() <= keep {
return None;
}
// Finite threshold with hysteresis: an install must drain at least as much conversation as it
// keeps. Draining one event per turn would reinstall a view on practically every turn, which
// is the continual tiny compaction the LCM thresholds exist to prevent.
let drained_conversation = conversation.len() - keep;
if drained_conversation < keep.max(1) {
return None;
}
// The drain boundary is the last event before the kept window. Coverage is a prefix by
// sequence, so it also covers the non-conversation records in between.
let last_drained = conversation[conversation.len() - keep - 1].id;
let drained: Vec<&Event> = uncovered
.iter()
.copied()
.filter(|event| event.id <= last_drained)
.collect();
if drained.is_empty() {
return None;
}
// A frontier's summary is CUMULATIVE: it must describe everything the session carries, not
// just what this install added. Summarizing only the delta would silently drop the previous
// view's content the moment a newer view replaced it, which is a context loss disguised as an
// install. The oldest lines are what fall off when the budget is reached.
let summary = carry_forward(previous, &drained);
if summary.trim().is_empty() {
// An empty recollection is not a view. Refusing here is what keeps the previous view
// active instead of installing one that says nothing.
return None;
}
let mut source_children: Vec<SourceChild> = Vec::new();
for event in &drained {
match source_children.last_mut() {
Some(child) if child.kind == event.kind => {
child.last_seq = event.id;
child.count += 1;
}
_ => source_children.push(SourceChild {
kind: event.kind.clone(),
first_seq: event.id,
last_seq: event.id,
count: 1,
}),
}
}
let created_ms = drained.last().map(|event| event.created_ms).unwrap_or(0);
Some(Frontier {
epoch: previous.map(|frontier| frontier.epoch).unwrap_or(0) + 1,
drained_through_seq: last_drained,
summary,
derived_by: "deterministic_prune".to_string(),
created_ms,
source_children,
// A prune does not author testimony, and inventing one would be exactly the misrepresentation
// this field exists to prevent. Installed testimony is therefore carried forward UNCHANGED
// for as long as it stands: it is exact text the session is keeping, so a summary of it is
// not a replacement for it, and dropping it because a drain happened to cover its source
// would silently delete the one thing that is not allowed to be paraphrased.
// Supersession is explicit: an installer that means to replace it passes its own testimony,
// which overrides these fields exactly as an explicit orientation change overrides a carrier.
actor_testimony: previous.and_then(|frontier| frontier.actor_testimony.clone()),
actor_testimony_source: previous
.and_then(|frontier| frontier.actor_testimony_source.clone()),
})
}
/// A bounded, deterministic recollection of a drained range: this install's delta.
///
/// It is a summary, so it is never presented as original evidence: each line keeps the event id it
/// came from, and the exact text remains reachable from the store.
fn recollection(drained: &[&Event]) -> String {
let mut out = String::new();
for event in drained {
if !carries_conversation(event) {
continue;
}
let text = event
.payload
.get("content")
.or_else(|| event.payload.get("text"))
.or_else(|| event.payload.get("output"))
.and_then(Value::as_str)
.unwrap_or("")
.to_string();
let text = text.trim();
if text.is_empty() {
continue;
}
let clipped: String = text.chars().take(240).collect();
let label = event
.payload
.get("type")
.and_then(Value::as_str)
.unwrap_or(event.kind.as_str());
out.push_str(&format!("- [{label}#{}] {clipped}\n", event.id));
}
out
}
/// The previous view's summary followed by this install's delta, trimmed to the budget from the
/// front: the newest content is what must survive, and the covered records stay in the store.
fn carry_forward(previous: Option<&Frontier>, drained: &[&Event]) -> String {
let delta = recollection(drained);
let mut combined = match previous {
Some(frontier) if !frontier.summary.trim().is_empty() => {
format!("{}\n{}", frontier.summary.trim_end(), delta)
}
_ => delta,
};
if combined.len() > RECOLLECTION_BUDGET {
let overflow = combined.len() - RECOLLECTION_BUDGET;
// A byte offset is not a character boundary. Slicing at one panics the runner task, and the
// text that triggers it is the ordinary case - any non-ASCII history, an emoticon in a
// message - so the cut walks forward to the next boundary before it is used.
let mut start = overflow;
while start < combined.len() && !combined.is_char_boundary(start) {
start += 1;
}
let cut = combined[start..]
.find('\n')
.map(|offset| start + offset + 1)
.unwrap_or(start);
combined = format!(
"- (older recollection trimmed; those records remain in the store)\n{}",
&combined[cut..]
);
}
combined
}
#[cfg(test)]
mod tests {
use super::*;
fn conversation(id: i64, kind: &str, text: &str) -> Event {
Event {
id,
kind: kind.to_string(),
payload: json!({"type": "model_response", "content": text}),
created_ms: id,
}
}
/// Non-ASCII history is the ordinary case, and a budget that trims it must not panic.
///
/// The prefix is multi-byte on purpose: the trim crosses it at a byte offset that is almost
/// never a character boundary, which is exactly what used to panic the runner task.
#[test]
fn a_multi_byte_recollection_is_trimmed_to_a_character_boundary() {
let previous = Frontier {
summary: format!("{}\n- [input#1] earlier", "😀ことば".repeat(400)),
..Frontier::for_test()
};
let events = [
conversation(9, "input", "the newest turn"),
conversation(10, "input", "the newest turn continued"),
];
let drained: Vec<&Event> = events.iter().collect();
let carried = carry_forward(Some(&previous), &drained);
assert!(
carried.starts_with("- (older recollection trimmed"),
"an over-budget recollection must say it was trimmed"
);
assert!(
carried.len() <= RECOLLECTION_BUDGET + 128,
"the trimmed recollection must be bounded, got {}",
carried.len()
);
// The cut lands on a line boundary, and the newest content is what survives it.
assert!(
carried.trim_end().ends_with("the newest turn continued"),
"the newest recollection must survive the trim, got {:?}",
&carried[carried.len().saturating_sub(80)..]
);
}
/// Testimony is carried until an explicit supersession, and a plan does not invent one.
#[test]
fn testimony_is_carried_forward_until_superseded() {
let previous = Frontier {
summary: "- [input#1] earlier".to_string(),
actor_testimony: Some("I said the exact words".to_string()),
actor_testimony_source: Some(ActorTestimonySource {
actor_id: "dawn".to_string(),
session_id: "s".to_string(),
source_ref: "ref-1".to_string(),
}),
..Frontier::for_test()
};
let events: Vec<Event> = (1..8)
.map(|id| {
conversation(
id,
"input",
&format!("turn {id} of ordinary history that will be drained"),
)
})
.collect();
let planned = plan_frontier(&events, Some(&previous), 2).expect("a view is earned");
assert_eq!(
planned.actor_testimony.as_deref(),
Some("I said the exact words"),
"exact testimony must survive an install that did not author one"
);
assert_eq!(
planned
.actor_testimony_source
.as_ref()
.map(|s| s.actor_id.as_str()),
Some("dawn"),
"testimony stays attributable"
);
// A different testimony replaces it, which is the only way it changes.
let superseded = Frontier {
actor_testimony: Some("the newer exact words".to_string()),
..planned.clone()
};
assert_eq!(
superseded.actor_testimony.as_deref(),
Some("the newer exact words")
);
}
/// The threshold, stated exactly: an install must drain at least as much as it keeps.
#[test]
fn a_tiny_drain_does_not_earn_a_new_view() {
let one_past = |count: i64| -> Vec<Event> {
(1..=count)
.map(|id| conversation(id, "input", "one turn"))
.collect()
};
// One event past a keep of two drains a single event: not a view.
assert!(
plan_frontier(&one_past(3), None, 2).is_none(),
"draining one event per turn is the continual tiny compaction the threshold exists to prevent"
);
// Two past it drains as much as it keeps: that is a view.
assert!(
plan_frontier(&one_past(4), None, 2).is_some(),
"a drain as large as the keep is what earns a view"
);
}
}