forge_workspace/store/
handoffs.rs

1//! The handoff log: one row per swap, with the numbers `at_percent` is
2//! tuned from.
3//!
4//! A row is written at the swap and completed once its successor has run
5//! its first turns: the fill it fired at, the cycle it ended, and how much
6//! of the predecessor's work the successor had to re-derive. That last half
7//! is what separates a premature handoff (a successor re-reading
8//! everything) from a late one (a compaction that fired first).
9
10use anyhow::Context;
11use forge_primitives::{HandoffRecord, HandoffSuccessor, SessionSlot};
12use redb::{ReadableTable, TableDefinition};
13
14use super::Db;
15
16const HANDOFFS: TableDefinition<&str, &[u8]> = TableDefinition::new("handoffs");
17
18/// One stored row: the record, and what the successor's re-reads are
19/// measured against.
20#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
21struct StoredHandoff {
22    record: HandoffRecord,
23    /// The predecessor's read paths. Kept here rather than on the record:
24    /// no view draws it, and it is only ever read against the successor.
25    #[serde(default)]
26    predecessor_reads: Vec<String>,
27}
28
29/// The key every row of one seat shares. Its rows sort by it first, so a
30/// seat's newest row is its last.
31fn seat_prefix(slot: &SessionSlot) -> String {
32    format!("{}\u{0}{}\u{0}{}\u{0}", slot.org(), slot.project(), slot.label())
33}
34
35/// One row's key: the seat, then the instant (RFC 3339 sorts by time), then
36/// the successor. The successor is in the key because two handoffs at one
37/// instant are still two rows.
38fn row_key(record: &HandoffRecord) -> String {
39    format!("{}{}\u{0}{}", seat_prefix(&record.slot), record.at, record.to)
40}
41
42/// Write one handoff.
43pub fn put(db: &Db, record: &HandoffRecord, predecessor_reads: &[String]) -> anyhow::Result<()> {
44    let stored =
45        StoredHandoff { record: record.clone(), predecessor_reads: predecessor_reads.to_vec() };
46    let value = serde_json::to_vec(&stored).context("serialize handoff")?;
47    let key = row_key(record);
48    let txn = db.database().begin_write()?;
49    {
50        let mut table = txn.open_table(HANDOFFS)?;
51        table.insert(key.as_str(), value.as_slice())?;
52    }
53    txn.commit()?;
54    Ok(())
55}
56
57/// The handoff `occupant` was the successor of, with what its predecessor
58/// had read. `None` when this occupant did not arrive by a handoff.
59pub fn for_successor(
60    db: &Db,
61    slot: &SessionSlot,
62    occupant: &str,
63) -> anyhow::Result<Option<(HandoffRecord, Vec<String>)>> {
64    Ok(rows_for_slot(db, slot)?
65        .into_iter()
66        .find(|stored| stored.record.to == occupant)
67        .map(|stored| (stored.record, stored.predecessor_reads)))
68}
69
70/// Record how the successor's first turns went.
71pub fn set_successor(
72    db: &Db,
73    slot: &SessionSlot,
74    occupant: &str,
75    successor: HandoffSuccessor,
76) -> anyhow::Result<()> {
77    let prefix = seat_prefix(slot);
78    let txn = db.database().begin_write()?;
79    {
80        let mut table = txn.open_table(HANDOFFS)?;
81        let mut found: Option<(String, StoredHandoff)> = None;
82        for entry in table.range(prefix.as_str()..)? {
83            let (key, value) = entry?;
84            if !key.value().starts_with(&prefix) {
85                break;
86            }
87            let Ok(mut stored) = serde_json::from_slice::<StoredHandoff>(value.value()) else {
88                continue;
89            };
90            if stored.record.to == occupant {
91                stored.record.successor = Some(successor);
92                found = Some((key.value().to_owned(), stored));
93                break;
94            }
95        }
96        if let Some((key, stored)) = found {
97            let value = serde_json::to_vec(&stored).context("serialize handoff")?;
98            table.insert(key.as_str(), value.as_slice())?;
99        }
100    }
101    txn.commit()?;
102    Ok(())
103}
104
105/// A seat's rows, oldest first. A row that fails to decode is skipped
106/// loudly rather than failing the read: a damaged record is not a reason to
107/// stop drawing the handoff that did decode.
108fn rows_for_slot(db: &Db, slot: &SessionSlot) -> anyhow::Result<Vec<StoredHandoff>> {
109    let txn = db.database().begin_read()?;
110    let table = match txn.open_table(HANDOFFS) {
111        Ok(table) => table,
112        Err(redb::TableError::TableDoesNotExist(_)) => return Ok(Vec::new()),
113        Err(error) => return Err(error.into()),
114    };
115    let prefix = seat_prefix(slot);
116    let mut out = Vec::new();
117    for entry in table.range(prefix.as_str()..)? {
118        let (key, value) = entry?;
119        if !key.value().starts_with(&prefix) {
120            break;
121        }
122        match serde_json::from_slice::<StoredHandoff>(value.value()) {
123            Ok(stored) => out.push(stored),
124            Err(error) => tracing::warn!(
125                target: "forge_workspace::store::handoffs",
126                row = %key.value(),
127                %error,
128                "skipping a handoff row that failed to decode",
129            ),
130        }
131    }
132    Ok(out)
133}
134
135#[cfg(test)]
136mod tests {
137    use std::path::Path;
138
139    use super::*;
140    use forge_primitives::{HandoffBrief, HandoffCycle, HandoffReason};
141
142    fn open_db(dir: &Path) -> Db {
143        Db::open(&dir.join("db.redb")).expect("open db")
144    }
145
146    fn record(to: &str, at: &str) -> HandoffRecord {
147        HandoffRecord {
148            slot: SessionSlot::worker("Busytools", "forge", "handoff-cut"),
149            from: "old".to_owned(),
150            to: to.to_owned(),
151            reason: HandoffReason::Fill,
152            fill_percent: Some(85),
153            total_tokens: Some(850_000),
154            max_tokens: Some(1_000_000),
155            at_percent: 85,
156            brief: HandoffBrief::default(),
157            transcript_path: "/tmp/old.jsonl".to_owned(),
158            at: at.to_owned(),
159            cycle: HandoffCycle { requests: 10, turns: 2, prompt_tokens: 100 },
160            successor: None,
161        }
162    }
163
164    /// A successor is found by the id it runs under, and the readings its
165    /// predecessor had done ride with it.
166    #[test]
167    fn a_successor_is_found_by_its_occupant_id() {
168        let dir = tempfile::tempdir().expect("tempdir");
169        let db = open_db(dir.path());
170        let written = record("new-id", "2026-10-24T08:00:00Z");
171        put(&db, &written, &["src/a.rs".to_owned()]).expect("put");
172
173        let (found, reads) = for_successor(&db, &written.slot, "new-id")
174            .expect("read")
175            .expect("the handoff this occupant arrived by");
176        assert_eq!(found.to, "new-id");
177        assert_eq!(reads, vec!["src/a.rs".to_owned()], "the predecessor's reads ride along");
178        assert!(
179            for_successor(&db, &written.slot, "someone-else").expect("read").is_none(),
180            "an occupant that arrived another way is nobody's successor",
181        );
182    }
183
184    /// The counts are written onto the row that named the successor, and
185    /// nothing else moves.
186    #[test]
187    fn the_successors_counts_land_on_their_own_row() {
188        let dir = tempfile::tempdir().expect("tempdir");
189        let db = open_db(dir.path());
190        let written = record("new-id", "2026-10-24T08:00:00Z");
191        put(&db, &written, &[]).expect("put");
192
193        let counts = HandoffSuccessor { turns: 10, reread_files: 2, questions: 1, corrections: 0 };
194        set_successor(&db, &written.slot, "new-id", counts).expect("write");
195
196        let (measured, _) = for_successor(&db, &written.slot, "new-id")
197            .expect("read")
198            .expect("the handoff this occupant arrived by");
199        assert_eq!(measured.successor, Some(counts), "the measurement is kept");
200        assert_eq!(measured.fill_percent, Some(85), "and the firing numbers are untouched");
201    }
202
203    #[test]
204    fn an_absent_table_reads_as_none() {
205        let dir = tempfile::tempdir().expect("tempdir");
206        let db = open_db(dir.path());
207        let slot = SessionSlot::lead("Busytools", "forge");
208        assert!(for_successor(&db, &slot, "anyone").expect("read").is_none());
209    }
210
211    /// One seat's rows never leak into another's, label included.
212    #[test]
213    fn a_workers_handoff_is_not_its_leads() {
214        let dir = tempfile::tempdir().expect("tempdir");
215        let db = open_db(dir.path());
216        let worker = record("w-new", "2026-10-24T08:00:00Z");
217        put(&db, &worker, &[]).expect("put");
218
219        assert!(
220            for_successor(&db, &SessionSlot::lead("Busytools", "forge"), "w-new")
221                .expect("read")
222                .is_none(),
223            "the lead's seat holds no row of its worker's",
224        );
225    }
226
227    /// A later handoff that never reached its successor does not hide the
228    /// row of the one running: the lookup is by occupant, not by recency.
229    #[test]
230    fn an_earlier_successors_row_survives_a_later_attempt() {
231        let dir = tempfile::tempdir().expect("tempdir");
232        let db = open_db(dir.path());
233        let reached = record("reached-id", "2026-10-24T08:00:00Z");
234        put(&db, &reached, &[]).expect("put reached");
235        let never = record("never-id", "2026-10-24T09:00:00Z");
236        put(&db, &never, &[]).expect("put never");
237
238        let (found, _) = for_successor(&db, &reached.slot, "reached-id")
239            .expect("read")
240            .expect("the running occupant's own handoff");
241        assert_eq!(found.to, "reached-id");
242    }
243}
forge_workspace/store/handoffs.rs

forge_workspace/store/
handoffs.rs
