ive harnessed the harness
Something went wrong. Try again.
19 kB · 471 lines
Rust
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472//! 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" ); }}