ive harnessed the harness
Something went wrong. Try again.
42 kB · 1087 lines
Rust
at main
12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088//! The instruction header a transcript is running under, and the revisions it has observed.//!//! The prompt prefix is only reusable while it is byte-stable, and the one part of the prefix//! that gets edited on purpose is the head: the soul, the contract, the skill index. Rewriting//! that head invalidates every block after it, and - worse than the cost - it rewrites what the//! transcript claims to have been running under all along.//!//! So the head is pinned, and instruction changes are observations://!//! 1. A transcript pins the head it starts under, once.//! 2. Every later change to the instruction text is *observed* where it was noticed, with the//! actor when the change has one: a tool call in this transcript, or the host when the text//! changed with nothing here to attribute it to. The observation carries **verified patches**//! against the newest text this transcript knows, so a revision costs its own size instead of//! a second copy of the instructions.//! 3. A fold moves the head. Installing a context frontier rebuilds the prefix anyway, so the//! current text costs nothing extra there, and the observations it drains can no longer be//! rendered. This is also the safety net: a head whose only support was an observation the//! frontier swallowed must move, or the model silently loses instructions it is running under.//!//! Patches are verified by reconstruction before they are stored, and a patch that does not//! reproduce the text it claims to is replaced by the whole text. Two things follow: the//! transcript never carries a patch that would make a reader reconstruct an instruction text//! nobody wrote, and the reader can always ask the host for the exact current text//! (`runtime.instructions`) instead of applying patches by hand.
use serde::{Deserialize, Serialize};
use sha2::{Digest, Sha256};
use crate::text_patch::LinePatch;
/// Content identity of a header text.#[must_use]pub fn header_digest(text: &str) -> String { let mut hasher = Sha256::new(); hasher.update(text.as_bytes()); format!("{:x}", hasher.finalize())}
/// Event kind of the durable record that pins a transcript's header.pub const HEADER_PIN_EVENT: &str = "prompt.header";/// Event kind of an instruction revision observed inside the transcript that reads it.pub const HEADER_REVISION_EVENT: &str = "prompt.revision";
/// Why this transcript's head says what it says.#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)]#[serde(rename_all = "snake_case")]pub enum PinReason { /// The first request of this transcript, running under whatever was current then. SessionStart, /// A frontier was installed and the caller had the current text in hand. Fold, /// A frontier covered the observation that supported this head, so the head had to move. FoldCoveredRevision,}
impl PinReason { #[must_use] pub const fn as_str(self) -> &'static str { match self { Self::SessionStart => "session_start", Self::Fold => "fold", Self::FoldCoveredRevision => "fold_covered_revision", } }
#[must_use] pub fn parse(name: &str) -> Option<Self> { match name { "session_start" => Some(Self::SessionStart), "fold" => Some(Self::Fold), "fold_covered_revision" => Some(Self::FoldCoveredRevision), _ => None, } }}
/// Who changed the instructions, as far as this transcript can tell.#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)]#[serde(rename_all = "snake_case")]pub enum RevisionOrigin { /// A tool call inside this transcript. The change has a cause the model can point at. Session, /// The text changed with nothing in this transcript to attribute it to: a rebuild of the /// compiled constants, an edit made elsewhere, a skill installed out of band. The host /// reports the change and does not name an actor it did not observe. Host,}
/// Who caused an instruction revision, from inside this transcript.#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]pub struct RevisionActor { pub origin: RevisionOrigin, /// The execution whose tool call made the change, when a session made it. #[serde(default, skip_serializing_if = "Option::is_none")] pub execution_id: Option<String>, #[serde(default, skip_serializing_if = "Option::is_none")] pub step: Option<u32>, /// The path that was written, relative to the session's source root when derivable. #[serde(default, skip_serializing_if = "Option::is_none")] pub path: Option<String>, /// What the host knows about a change it did not cause. #[serde(default, skip_serializing_if = "Option::is_none")] pub reason: Option<String>,}
impl RevisionActor { /// A change this transcript made through a tool call. #[must_use] pub fn session( execution_id: impl Into<String>, step: Option<u32>, path: impl Into<String>, ) -> Self { Self { origin: RevisionOrigin::Session, execution_id: Some(execution_id.into()), step, path: Some(path.into()), reason: None, } }
/// A change with no actor inside this transcript. #[must_use] pub fn host(reason: impl Into<String>) -> Self { Self { origin: RevisionOrigin::Host, execution_id: None, step: None, path: None, reason: Some(reason.into()), } }
/// How the actor reads to the model, in the frame on the observation line. #[must_use] pub fn label(&self) -> String { match self.origin { RevisionOrigin::Session => { let mut label = String::from("instructions revised by this session"); if let Some(step) = self.step { label.push_str(&format!(" · step {step}")); } if let Some(path) = self.path.as_deref() { label.push_str(&format!(" · {path}")); } label } RevisionOrigin::Host => match self.reason.as_deref() { Some(reason) => format!("instructions changed for this runtime · {reason}"), None => "instructions changed for this runtime".into(), }, } }}
/// One instruction source, as it was when a revision was pinned or observed.////// Sources are kept apart rather than only as one joined text, because a change is attributable/// per source: "the contract changed" and "the soul changed" are different facts, and a patch/// that names its source lets a reader find the file it came from.#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]pub struct HeaderSource { pub name: String, pub digest: String, pub text: String,}
impl HeaderSource { #[must_use] pub fn new(name: impl Into<String>, text: impl Into<String>) -> Self { let text = text.into(); Self { digest: header_digest(&text), name: name.into(), text, } }}
/// The name used when a text's parts are unknown: the joined text is one source, honestly/// anonymous, instead of pretending the parts were known all along.pub const JOINED_SOURCE: &str = "instructions";
/// Whether two sets of sources line up by name, which is what makes a per-source patch possible.////// When they do not - a text whose parts were never recorded, a source that appeared or vanished -/// the change is patched over the joined texts instead. That is still exact, just less/// attributable, and it keeps a rename from reading as a rewrite of everything.#[must_use]pub fn sources_aligned(base: &[HeaderSource], current: &[HeaderSource]) -> bool { base.len() == current.len() && base.iter().all(|source| { current .iter() .any(|candidate| candidate.name == source.name) })}
/// Join sources into the text that stands at the head of a request.////// One rule in one place: a pin stores the joined text it produced *and* the parts it came from,/// so a later reader never has to re-derive the joining.#[must_use]pub fn join_sources(sources: &[HeaderSource]) -> String { sources .iter() .map(|source| source.text.as_str()) .filter(|text| !text.is_empty()) .collect::<Vec<_>>() .join("\n\n")}
/// A text this transcript knows, with the parts it was joined from.#[derive(Clone, Debug, Eq, PartialEq)]pub struct HeaderSnapshot { pub digest: String, pub text: String, pub sources: Vec<HeaderSource>,}
impl HeaderSnapshot { #[must_use] pub fn new(sources: Vec<HeaderSource>) -> Self { let text = join_sources(&sources); Self { digest: header_digest(&text), text, sources, } }
/// A snapshot of a text whose parts are not known: one honestly anonymous source. #[must_use] pub fn from_text(text: impl Into<String>) -> Self { let text = text.into(); Self { digest: header_digest(&text), sources: vec![HeaderSource::new(JOINED_SOURCE, &text)], text, } }
fn from_pin(pin: &HeaderPin) -> Self { Self { digest: pin.digest.clone(), text: pin.text.clone(), sources: if pin.sources.is_empty() { // A pin written before sources were recorded still anchors the chain: its text is // one anonymous source, so a change against it is a patch over that text. vec![HeaderSource::new(JOINED_SOURCE, &pin.text)] } else { pin.sources.clone() }, } }}
/// How a change reaches the model: as the whole text, or as patches against a known text.#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]#[serde(tag = "form", rename_all = "snake_case")]pub enum HeaderChange { /// The current text in full, used when there is no base to patch or a patch did not verify. Full { text: String }, /// Patches verified to turn `base_digest` into this revision's text. Diff { base_digest: String, patches: Vec<LinePatch>, },}
impl HeaderChange { #[must_use] pub const fn form(&self) -> &'static str { match self { Self::Full { .. } => "full", Self::Diff { .. } => "diff", } }
/// The bytes this change costs in a request, which is the number that motivated patching. #[must_use] pub fn body_len(&self) -> usize { match self { Self::Full { text } => text.len(), Self::Diff { patches, .. } => patches.iter().map(LinePatch::rendered_len).sum(), } }}
/// A change from `base` to `current`, as verified patches when they reconstruct `current`/// exactly and as the whole text when they do not.////// The verification is the point. A patch is a claim about a transformation, and a claim that/// does not reconstruct is worse than no claim: the reader would reconstruct an instruction text/// nobody ever wrote. Falling back to the whole text costs bytes and cannot be wrong.#[must_use]pub fn change_between(base: &HeaderSnapshot, current: &[HeaderSource]) -> HeaderChange { let current_text = join_sources(current); let patches = patches_between(base, current); if patches.is_empty() { return HeaderChange::Full { text: current_text }; } // Verify by reconstruction before storing the claim: apply the patches and re-join, exactly // as a reader would have to. match apply_patches_to_snapshot(base, &patches) { Some(reconstructed) if reconstructed == current_text => HeaderChange::Diff { base_digest: base.digest.clone(), patches, }, _ => HeaderChange::Full { text: current_text }, }}
/// Patches between a snapshot and current sources: per source when they line up, over the joined/// texts when they do not.////// Alignment decides, and it decides the same way on the reading side, so a patch is always/// applied to the text it was generated against.#[must_use]pub fn patches_between(base: &HeaderSnapshot, current: &[HeaderSource]) -> Vec<LinePatch> { if sources_aligned(&base.sources, current) { source_patches(&base.sources, current) } else { crate::text_patch::patches(&base.text, &join_sources(current), JOINED_SOURCE) }}
/// Apply patches to a snapshot, returning the text a reader would end up with.////// `None` when a patch does not fit: the caller must treat that as "no patch" rather than as a/// reason to guess a text.#[must_use]pub fn apply_patches_to_snapshot(base: &HeaderSnapshot, patches: &[LinePatch]) -> Option<String> { if !patches.is_empty() && patches.iter().all(|patch| patch.source == JOINED_SOURCE) { return crate::text_patch::apply(&base.text, patches); } let applied = apply_source_patches(&base.sources, patches)?; Some(join_sources(&applied))}
/// Patches per source between two sets of sources.////// A source that appeared is an addition from nothing, a source that vanished is a removal to/// nothing, and an unchanged source produces no patch at all.#[must_use]pub fn source_patches(base: &[HeaderSource], current: &[HeaderSource]) -> Vec<LinePatch> { let mut patches = Vec::new(); for source in current { let empty = HeaderSource::new(source.name.clone(), ""); let base_source = base .iter() .find(|candidate| candidate.name == source.name) .unwrap_or(&empty); if base_source.text == source.text { continue; } patches.extend(crate::text_patch::patches( &base_source.text, &source.text, &source.name, )); } for source in base { if current .iter() .any(|candidate| candidate.name == source.name) { continue; } patches.extend(crate::text_patch::patches(&source.text, "", &source.name)); } patches}
/// Apply patches to sources, returning `None` when a patch does not fit its source.////// Patches for one source are applied together, in the order they were generated: their line/// numbers are coordinates in the base text, so applying them one at a time would reset the/// search position between them and let a later patch match the wrong place.#[must_use]pub fn apply_source_patches( base: &[HeaderSource], patches: &[LinePatch],) -> Option<Vec<HeaderSource>> { let mut result: Vec<HeaderSource> = base.to_vec(); let mut names: Vec<&str> = Vec::new(); for patch in patches { if !names.contains(&patch.source.as_str()) { names.push(patch.source.as_str()); } } for name in names { let group: Vec<LinePatch> = patches .iter() .filter(|patch| patch.source == name) .cloned() .collect(); match result.iter().position(|source| source.name == name) { Some(index) => { let applied = crate::text_patch::apply(&result[index].text, &group)?; result[index] = HeaderSource::new(name, applied); } None => { // A source this revision introduces: an addition from nothing, which is only // legitimate when every patch for it is a pure insertion. if group.iter().any(|patch| !patch.old_lines.is_empty()) { return None; } let added = crate::text_patch::apply("", &group)?; result.push(HeaderSource::new(name, added)); } } } Some(result)}
/// The pinned text of a transcript's head, as it was recorded.#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]pub struct HeaderPin { pub digest: String, pub text: String, pub reason: String, pub pinned_at_ms: i64, /// The parts the text was joined from. Empty on pins written before sources were recorded, /// which is why a change against such a pin falls back to patching the joined text. #[serde(default)] pub sources: Vec<HeaderSource>,}
impl HeaderPin { #[must_use] pub fn new(text: &str, reason: PinReason, pinned_at_ms: i64) -> Self { Self::with_sources(text, Vec::new(), reason, pinned_at_ms) }
#[must_use] pub fn with_sources( text: &str, sources: Vec<HeaderSource>, reason: PinReason, pinned_at_ms: i64, ) -> Self { Self { digest: header_digest(text), text: text.to_string(), reason: reason.as_str().to_string(), pinned_at_ms, sources, } }
/// A pin for the sources a caller holds right now. #[must_use] pub fn of(sources: &[HeaderSource], reason: PinReason, pinned_at_ms: i64) -> Self { Self::with_sources( &join_sources(sources), sources.to_vec(), reason, pinned_at_ms, ) }
#[must_use] pub fn reason_kind(&self) -> Option<PinReason> { PinReason::parse(&self.reason) }
/// The durable payload for this pin. #[must_use] pub fn payload(&self) -> serde_json::Value { serde_json::json!({ "type": "header_pin", "digest": self.digest, "text": self.text, "reason": self.reason, "pinned_at_ms": self.pinned_at_ms, "sources": self.sources, }) }}
/// A revision to the instructions that this transcript can place in its own history.#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]pub struct HeaderRevision { /// Digest of the text this revision results in. pub digest: String, pub change: HeaderChange, /// The parts the resulting text is joined from, for later patches. #[serde(default)] pub sources: Vec<HeaderSource>, pub actor: RevisionActor, pub observed_at_ms: i64,}
/// A revision payload written before changes carried a form: the whole text sat at `text`.////// History is immutable, so a reader tolerates the older shape instead of a writer pretending it/// never happened. The live schema stays strict; projections recognize their legacy payloads.#[derive(Debug, Deserialize)]struct LegacyRevision { digest: String, text: String, actor: RevisionActor, observed_at_ms: i64,}
/// Read a revision payload, tolerating the pre-patch shape.#[must_use]pub fn read_revision(payload: &serde_json::Value) -> Option<HeaderRevision> { #[derive(Deserialize)] struct Wire { digest: String, #[serde(default)] change: Option<HeaderChange>, #[serde(default)] sources: Vec<HeaderSource>, actor: RevisionActor, observed_at_ms: i64, } let wire: Wire = serde_json::from_value(payload.clone()).ok()?; match wire.change { Some(change) => Some(HeaderRevision { digest: wire.digest, change, sources: wire.sources, actor: wire.actor, observed_at_ms: wire.observed_at_ms, }), None => { let legacy: LegacyRevision = serde_json::from_value(payload.clone()).ok()?; Some(HeaderRevision { digest: legacy.digest, change: HeaderChange::Full { text: legacy.text }, sources: Vec::new(), actor: legacy.actor, observed_at_ms: legacy.observed_at_ms, }) } }}
impl HeaderRevision { #[must_use] pub fn new( current_text: &str, sources: Vec<HeaderSource>, change: HeaderChange, actor: RevisionActor, observed_at_ms: i64, ) -> Self { Self { digest: header_digest(current_text), change, sources, actor, observed_at_ms, } }
#[must_use] pub const fn form(&self) -> &'static str { self.change.form() }
/// The parts a resolved text came from: the recorded ones when this revision has them, and one /// honestly anonymous source when it does not. A snapshot always has at least one source, so /// the next change has something to patch against. #[must_use] fn known_sources(&self, text: &str) -> Vec<HeaderSource> { if self.sources.is_empty() { vec![HeaderSource::new(JOINED_SOURCE, text)] } else { self.sources.clone() } }
#[must_use] pub fn payload(&self) -> serde_json::Value { serde_json::json!({ "type": "header_revision", "digest": self.digest, "change": serde_json::to_value(&self.change).unwrap_or(serde_json::Value::Null), "sources": self.sources, "actor": self.actor, "observed_at_ms": self.observed_at_ms, }) }
/// The text this revision results in, given a text it was patched against. /// /// A `Full` change states its own text. A `Diff` change is only resolvable against the base /// it names, which is why a reader that cannot reach that base treats the chain as broken /// rather than guessing. #[must_use] pub fn resolve(&self, base: &HeaderSnapshot) -> Option<HeaderSnapshot> { match &self.change { HeaderChange::Full { text } => Some(HeaderSnapshot { digest: self.digest.clone(), text: text.clone(), sources: self.known_sources(text), }), HeaderChange::Diff { base_digest, patches, } => { if base_digest != &base.digest { return None; } let text = apply_patches_to_snapshot(base, patches)?; (header_digest(&text) == self.digest).then_some(HeaderSnapshot { digest: self.digest.clone(), text: text.clone(), sources: self.known_sources(&text), }) } } }
/// A revision as it reaches the model: at the place it happened, with who did it. #[must_use] pub fn render(&self) -> String { let (note, body) = match &self.change { HeaderChange::Full { text } => ( "the header of this transcript is the revision it started under; the text below is current from here on.", text.clone(), ), HeaderChange::Diff { patches, .. } => ( "the header of this transcript is the revision it started under; the patch below is current from here on.", crate::text_patch::render(patches), ), }; format!( "[{} · {}]\n{note}\n{body}", crate::instant::format_instant(self.observed_at_ms), self.actor.label(), ) }}
/// The head this request runs under, and the observations on the way to it.#[derive(Clone, Debug, PartialEq)]pub struct HeaderState { /// Text of the leading block for this request. pub head_text: String, /// Digest of [`Self::head_text`]. pub head_digest: String, /// Digest of the text the runtime would use if it were starting now. pub current_digest: String, /// Whether the head is the current text or an earlier pinned revision. pub pinned: bool, /// Observations to render, each with the durable sequence that places it. pub revisions: Vec<(i64, HeaderRevision)>,}
impl HeaderState { /// Whether the head still matches what the runtime would write now. #[must_use] pub fn is_current(&self) -> bool { self.head_digest == self.current_digest }
/// Number of observations this request carries after the head. #[must_use] pub fn observation_count(&self) -> usize { self.revisions.len() }}
/// What the transcript has to record so that later requests agree with this one.#[derive(Clone, Debug, PartialEq)]pub enum HeaderRecord { /// The head itself, pinned for the first time or moved by a fold. Pin(HeaderPin), /// A change observed without moving the head. Revision(Box<HeaderRevision>),}
/// The head to use now, plus the record the caller must append before assembling.#[derive(Clone, Debug, PartialEq)]pub struct HeaderPlan { pub state: HeaderState, pub record: Option<HeaderRecord>,}
/// The newest instruction text this transcript can reconstruct, from its own records.////// This is what a reader gets when it asks the host for the current instructions instead of/// applying patches by hand: the chain is resolved here, and a link that does not reconstruct/// stops the chain at the last text that did rather than inventing one.#[must_use]pub fn known_snapshot<'a, I>(events: I) -> Option<HeaderSnapshot>where I: IntoIterator<Item = (i64, &'a str, &'a serde_json::Value)>,{ let mut known: Option<HeaderSnapshot> = None; for (_seq, kind, payload) in events { match kind { HEADER_PIN_EVENT => { if let Ok(pin) = serde_json::from_value::<HeaderPin>(payload.clone()) { known = Some(HeaderSnapshot::from_pin(&pin)); } } HEADER_REVISION_EVENT => { if let Some(revision) = read_revision(payload) { if let Some(resolved) = known.as_ref().and_then(|base| revision.resolve(base)) { known = Some(resolved); } } } _ => {} } } known}
/// The newest pin in a transcript, with the sequence that recorded it.////// The head a request runs under, for a reader that wants to name it. [`known_snapshot`] walks/// forward and resolves; this walks back and stops at the pin, which is the only record that says/// what the *head* is rather than how it changed.#[must_use]pub fn newest_pin<'a, I>(events: I) -> Option<(i64, HeaderPin)>where I: IntoIterator<Item = (i64, &'a str, &'a serde_json::Value)>,{ events .into_iter() .filter(|(_, kind, _)| *kind == HEADER_PIN_EVENT) .filter_map(|(seq, _, payload)| HeaderPin::deserialize(payload).ok().map(|pin| (seq, pin))) .max_by_key(|(seq, _)| *seq)}
/// Decide the head for a transcript.////// `events` is the session's durable history in order, each item carrying its sequence number,/// event kind, and payload. `current` is what the runtime would put at the head if this/// transcript were starting now. `drained_through` is the newest sequence a context frontier/// covers: an observation behind it can no longer be rendered, so it cannot be the reason a head/// stays pinned.#[must_use]pub fn plan_header<'a, I>(events: I, current: &[HeaderSource], drained_through: i64) -> HeaderPlanwhere I: IntoIterator<Item = (i64, &'a str, &'a serde_json::Value)>,{ let current_text = join_sources(current); let current_digest = header_digest(¤t_text); let mut pin: Option<HeaderPin> = None; let mut revisions: Vec<(i64, HeaderRevision)> = Vec::new(); let mut hidden: Vec<String> = Vec::new(); // The newest text this transcript can still reconstruct, and what it was joined from. A // patch is always against this, so a chain that breaks costs a bigger patch, never a wrong one. let mut known: Option<HeaderSnapshot> = None;
for (seq, kind, payload) in events { match kind { HEADER_PIN_EVENT => { if let Ok(recorded) = serde_json::from_value::<HeaderPin>(payload.clone()) { known = Some(HeaderSnapshot::from_pin(&recorded)); pin = Some(recorded); // A pin restarts the observation list: everything before it is behind it. revisions.retain(|(observed, _)| *observed > seq); hidden.clear(); } } HEADER_REVISION_EVENT => { if let Some(recorded) = read_revision(payload) { if seq > drained_through { // Resolve against the chain so a later patch has a base it can name. if let Some(resolved) = known.as_ref().and_then(|base| recorded.resolve(base)) { known = Some(resolved); } revisions.push((seq, recorded)); } else { hidden.push(recorded.digest.clone()); } } } _ => {} } }
let observed_now = revisions .iter() .any(|(_, revision)| revision.digest == current_digest);
let Some(pin) = pin else { // No pin yet: this transcript starts under what is current, and says so durably. return HeaderPlan { state: HeaderState { head_text: current_text, head_digest: current_digest.clone(), current_digest, pinned: false, revisions, }, record: Some(HeaderRecord::Pin(HeaderPin::of( current, PinReason::SessionStart, 0, ))), }; };
if pin.digest == current_digest || observed_now { // Either nothing changed, or the change is one this transcript can place in its own // history. Both keep the head's bytes. return HeaderPlan { state: HeaderState { head_text: pin.text, head_digest: pin.digest, current_digest, pinned: true, revisions, }, record: None, }; }
if hidden.iter().any(|digest| digest == ¤t_digest) { // The text is current because of an observation a frontier swallowed. Keeping the old // head would drop instructions the model runs under, so the head moves; the prefix is // being rebuilt by that fold anyway. return HeaderPlan { state: HeaderState { head_text: current_text, head_digest: current_digest.clone(), current_digest, pinned: false, revisions, }, record: Some(HeaderRecord::Pin(HeaderPin::of( current, PinReason::FoldCoveredRevision, 0, ))), }; }
// The text changed with nothing in this transcript to attribute it to. Record the change and // keep the head: an observation with an honest, unattributed actor is cheaper and truer than // rewriting what the transcript says it has been running under. let base = known.unwrap_or_else(|| HeaderSnapshot::from_pin(&pin)); let change = change_between(&base, current); let revision = HeaderRevision::new( ¤t_text, current.to_vec(), change, RevisionActor::host("changed outside this transcript"), 0, ); HeaderPlan { state: HeaderState { head_text: pin.text, head_digest: pin.digest, current_digest, pinned: true, revisions, }, record: Some(HeaderRecord::Revision(Box::new(revision))), }}
#[cfg(test)]mod tests { use super::*;
/// A text long enough that a patch's size is visibly its own, not the text's. const LONG_JOIN: &str = "instruction line 0 that nobody edited\ninstruction line 1 that nobody edited\ninstruction line 2 that nobody edited\ninstruction line 3 that nobody edited\ninstruction line 4 that nobody edited\ninstruction line 5 that nobody edited\ninstruction line 6 that nobody edited\ninstruction line 7 that nobody edited\ninstruction line 8 that nobody edited\ninstruction line 9 that nobody edited\ninstruction line 10 that nobody edited\ninstruction line 11 that nobody edited\ninstruction line 12 that nobody edited\ninstruction line 13 that nobody edited\ninstruction line 14 that nobody edited\ninstruction line 15 that nobody edited\ninstruction line 16 that nobody edited\ninstruction line 17 that nobody edited\ninstruction line 18 that nobody edited\ninstruction line 19 that nobody edited\n";
const LONG_JOIN_EDIT: &str = "instruction line 0 that nobody edited\ninstruction line 1 that nobody edited\ninstruction line 2 that nobody edited\ninstruction line 3 that nobody edited\ninstruction line 4 that nobody edited\ninstruction line 5 that nobody edited\ninstruction line 6 that nobody edited\ninstruction line 7 that nobody edited\ninstruction line 8 that nobody edited\ninstruction line 9 that nobody edited\ninstruction line 10 that nobody edited\ninstruction line 11 that nobody edited\ninstruction line 12 that nobody edited\ninstruction line 13 that nobody edited\ninstruction line 14 that nobody edited\ninstruction line 15 that nobody edited\ninstruction line 16 that nobody edited\ninstruction line 17 that somebody edited\ninstruction line 18 that nobody edited\ninstruction line 19 that nobody edited\n";
fn pin_event( text: &str, reason: PinReason, at: i64, sources: &[HeaderSource], ) -> serde_json::Value { HeaderPin::with_sources(text, sources.to_vec(), reason, at).payload() }
fn sources(soul: &str, contract: &str) -> Vec<HeaderSource> { vec![ HeaderSource::new("soul", soul), HeaderSource::new("contract", contract), ] }
fn plan( events: &[(i64, &str, serde_json::Value)], current: &[HeaderSource], drained_through: i64, ) -> HeaderPlan { plan_header( events .iter() .map(|(seq, kind, payload)| (*seq, *kind, payload)), current, drained_through, ) }
#[test] fn a_patch_reproduces_the_text_it_claims_to() { let base = HeaderSnapshot::new(sources("soul v1", &"line\n".repeat(40))); let current = sources( "soul v1", &("line\n".repeat(20) + "changed\n" + &"line\n".repeat(19)), ); let change = change_between(&base, ¤t); assert_eq!(change.form(), "diff"); let HeaderChange::Diff { patches, .. } = &change else { panic!("expected patches"); }; let applied = apply_source_patches(&base.sources, patches).expect("fits"); assert_eq!(join_sources(&applied), join_sources(¤t)); }
#[test] fn identical_texts_are_not_stored_as_an_empty_patch() { // Nothing changed, so there is no change to record: an empty patch would tell a reader // that something happened and leave it with nothing to apply. let base = HeaderSnapshot::new(sources("soul v1", "contract v1\n")); let change = change_between(&base, &sources("soul v1", "contract v1\n")); assert_eq!(change.form(), "full"); }
#[test] fn patches_for_one_source_apply_as_one_group() { // Two hunks in one source, the second only reachable after the first: applying them one // at a time would search from the top of a text that moved under it. let lines: Vec<String> = (0..40).map(|n| format!("line {n}\n")).collect(); let base = HeaderSnapshot::new(vec![HeaderSource::new("contract", lines.concat())]); let mut edited = lines.clone(); edited[2] = "line 2 changed\n".into(); edited[30] = "line 30 changed\n".into(); let current = vec![HeaderSource::new("contract", edited.concat())]; let change = change_between(&base, ¤t); assert_eq!(change.form(), "diff"); let HeaderChange::Diff { patches, .. } = &change else { panic!("expected patches"); }; assert!(patches.len() >= 2, "{patches:?}"); let applied = apply_source_patches(&base.sources, patches).expect("fits"); assert_eq!(join_sources(&applied), join_sources(¤t)); }
#[test] fn the_head_is_pinned_and_a_known_change_is_observed() { let pinned_text = "soul v1\n\ncontract v1\n"; let events = vec![( 5, HEADER_PIN_EVENT, pin_event( pinned_text, PinReason::SessionStart, 7, &sources("soul v1", "contract v1\n"), ), )]; let plan = plan(&events, &sources("soul v1", "contract v2\n"), 0); assert_eq!(plan.state.head_text, pinned_text); assert!(plan.state.pinned); let Some(HeaderRecord::Revision(revision)) = plan.record.as_ref() else { panic!("expected an observation"); }; assert_eq!(revision.form(), "diff"); assert_eq!(revision.actor.origin, RevisionOrigin::Host); assert_eq!( revision.digest, header_digest(&join_sources(&sources("soul v1", "contract v2\n"))) ); }
#[test] fn known_snapshot_resolves_the_chain_and_stops_at_a_broken_link() { let base = HeaderSnapshot::new(sources("soul v1", "contract v1\n")); let good = HeaderRevision::new( &join_sources(&sources("soul v1", "contract v2\n")), sources("soul v1", "contract v2\n"), change_between(&base, &sources("soul v1", "contract v2\n")), RevisionActor::session("exec-1", Some(1), "contract"), 1, ); let broken = serde_json::json!({ "type": "header_revision", "digest": "0".repeat(64), "change": {"form": "diff", "base_digest": "0".repeat(64), "patches": []}, "sources": [], "actor": {"origin": "host", "reason": "test"}, "observed_at_ms": 2, }); let events = [ ( 5, HEADER_PIN_EVENT, pin_event(&base.text, PinReason::SessionStart, 7, &base.sources), ), (6, HEADER_REVISION_EVENT, good.payload()), (7, HEADER_REVISION_EVENT, broken), ]; let known = known_snapshot( events .iter() .map(|(seq, kind, payload)| (*seq, *kind, payload)), ) .expect("a snapshot"); assert_eq!( known.text, "soul v1\n\ncontract v2\n", "the chain stops at the last text that reconstructs" ); }
#[test] fn a_pin_without_sources_still_anchors_the_chain() { // Pins written before sources were recorded keep working: they anchor the text, and a // change against them is patched per source only when sources are known. let events = vec![( 5, HEADER_PIN_EVENT, serde_json::json!({ "type": "header_pin", "digest": header_digest("old text\n"), "text": "old text\n", "reason": "session_start", "pinned_at_ms": 1, }), )]; let plan = plan(&events, &sources("soul v1", "contract v1\n"), 0); assert_eq!(plan.state.head_text, "old text\n"); let Some(HeaderRecord::Revision(revision)) = plan.record.as_ref() else { panic!("expected an observation"); }; assert_eq!( revision.form(), "diff", "unknown sources become additions, which is exact" ); } #[test] fn an_unaligned_source_set_is_patched_over_the_joined_text() { // A legacy pin (one anonymous source) against named sources today: the parts changed // names, so the change is patched over the joined texts rather than reading as "the soul // and the contract were each rewritten from nothing". let base = HeaderSnapshot::from_text(LONG_JOIN); assert_eq!(base.sources.len(), 1); assert_eq!(base.sources[0].name, JOINED_SOURCE); let current = sources("soul v1", LONG_JOIN_EDIT); let change = change_between(&base, ¤t); assert_eq!(change.form(), "diff"); let HeaderChange::Diff { patches, .. } = &change else { panic!("expected patches"); }; assert!(patches.iter().all(|patch| patch.source == JOINED_SOURCE)); let applied = apply_patches_to_snapshot(&base, patches).expect("fits"); assert_eq!(applied, join_sources(¤t)); assert!( change.body_len() < join_sources(¤t).len(), "a small change against an anonymous base is still small" ); }
#[test] fn a_resolved_revision_carries_the_parts_it_was_written_with() { let base = HeaderSnapshot::new(sources("soul v1", "contract v1\n")); let current = sources("soul v1", "contract v2\n"); let change = change_between(&base, ¤t); let revision = HeaderRevision::new( &join_sources(¤t), current.clone(), change, RevisionActor::host("test"), 1, ); let resolved = revision.resolve(&base).expect("resolves"); assert_eq!(resolved.text, join_sources(¤t)); assert_eq!( resolved.sources.len(), 2, "the next change has named parts to patch against" ); }}