From c98785d84af4464df63b6caa35d0aed9ca6fcf88 Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Mon, 14 Sep 2026 12:58:31 -0600 Subject: [PATCH] Pin pebble a39f43e and refuse an older stored session record clearly Pebble main a39f43e26effdf99635eaf343f095c17157c9c93 (pebble #22) carries an assistant turn's usage as Usage in the session record and moves the record format to version 5. CodingRuntime::from_record refuses a record in another format with UnsupportedRecord { version, supported } before it reads the route. Fabro persists those records in SQLite for Ask Fabro resume, and old runs get no migration, so a record written by an older build is read back as stored and refused on the next turn. Two tests pin that down. The store reads pebble's own version 4 fixture back through get without a parse error and reports it unsupported. A resumed Ask Fabro session whose stored record declares the previous format fails its next turn with the agent_error code and the message "session record format version 4 is not supported (this build requires 5)", runs no turn, and leaves the stored record in place. Co-Authored-By: Claude Fable 5.1 --- Cargo.lock | 6 +- Cargo.toml | 6 +- .../src/server/handler/sessions.rs | 110 +++++++++++++++++ .../src/fixtures/session_record_v4.json | 116 ++++++++++++++++++ .../src/run_session_record_store.rs | 39 ++++++ 5 files changed, 271 insertions(+), 6 deletions(-) create mode 100644 lib/components/fabro-store/src/fixtures/session_record_v4.json diff --git a/Cargo.lock b/Cargo.lock index 4612fda23..987d2eb24 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5869,7 +5869,7 @@ dependencies = [ [[package]] name = "pebble-agent" version = "0.1.0" -source = "git+https://github.com/lithoscomputer/pebble?rev=c91810fe51aece80359b9cd8efea971af0c46925#c91810fe51aece80359b9cd8efea971af0c46925" +source = "git+https://github.com/lithoscomputer/pebble?rev=a39f43e26effdf99635eaf343f095c17157c9c93#a39f43e26effdf99635eaf343f095c17157c9c93" dependencies = [ "async-trait", "futures-util", @@ -5886,7 +5886,7 @@ dependencies = [ [[package]] name = "pebble-cli-core" version = "0.1.0" -source = "git+https://github.com/lithoscomputer/pebble?rev=c91810fe51aece80359b9cd8efea971af0c46925#c91810fe51aece80359b9cd8efea971af0c46925" +source = "git+https://github.com/lithoscomputer/pebble?rev=a39f43e26effdf99635eaf343f095c17157c9c93#a39f43e26effdf99635eaf343f095c17157c9c93" dependencies = [ "anyhow", "async-trait", @@ -5915,7 +5915,7 @@ dependencies = [ [[package]] name = "pebble-coding-agent" version = "0.1.0" -source = "git+https://github.com/lithoscomputer/pebble?rev=c91810fe51aece80359b9cd8efea971af0c46925#c91810fe51aece80359b9cd8efea971af0c46925" +source = "git+https://github.com/lithoscomputer/pebble?rev=a39f43e26effdf99635eaf343f095c17157c9c93#a39f43e26effdf99635eaf343f095c17157c9c93" dependencies = [ "async-trait", "futures-util", diff --git a/Cargo.toml b/Cargo.toml index fd625c341..4fa4f8650 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -122,9 +122,9 @@ sandbox-driver-testing = { git = "https://github.com/lithoscomputer/sandbox-driv # sandbox, so the pebble and sandbox-driver pins move independently. Pebble # pins the same lithos-llm rev as fabro, and its lockfile policy is that # every shared crate resolves to the version lithos-llm locks. -pebble-agent = { git = "https://github.com/lithoscomputer/pebble", rev = "c91810fe51aece80359b9cd8efea971af0c46925" } -pebble-coding-agent = { git = "https://github.com/lithoscomputer/pebble", rev = "c91810fe51aece80359b9cd8efea971af0c46925", features = ["mcp", "search-providers"] } -pebble-cli-core = { git = "https://github.com/lithoscomputer/pebble", rev = "c91810fe51aece80359b9cd8efea971af0c46925" } +pebble-agent = { git = "https://github.com/lithoscomputer/pebble", rev = "a39f43e26effdf99635eaf343f095c17157c9c93" } +pebble-coding-agent = { git = "https://github.com/lithoscomputer/pebble", rev = "a39f43e26effdf99635eaf343f095c17157c9c93", features = ["mcp", "search-providers"] } +pebble-cli-core = { git = "https://github.com/lithoscomputer/pebble", rev = "a39f43e26effdf99635eaf343f095c17157c9c93" } sentry = { version = "0.35", default-features = false, features = ["backtrace", "contexts", "ureq", "rustls"] } fork = "0.2" exec = "0.3" diff --git a/lib/apps/fabro-server/src/server/handler/sessions.rs b/lib/apps/fabro-server/src/server/handler/sessions.rs index 94eadaba8..04ce2c082 100644 --- a/lib/apps/fabro-server/src/server/handler/sessions.rs +++ b/lib/apps/fabro-server/src/server/handler/sessions.rs @@ -1973,6 +1973,7 @@ mod resume_tests { use fabro_static::EnvVars; use fabro_test::{TwinScenario, TwinScenarios, twin_openai}; use fabro_types::{RunId, SessionId}; + use pebble_coding_agent::state::SESSION_RECORD_FORMAT_VERSION; use tower::ServiceExt; use crate::server::{AppState, spawn_scheduler}; @@ -2122,6 +2123,115 @@ mod resume_tests { events } + /// A record written by an older build is refused, not read: pebble checks + /// the format version before it reads the route, and fabro turns that + /// refusal into a turn failure naming both versions. Old runs get no + /// migration, so this is what a session persisted before the record + /// format moved sees on its next turn. + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn a_stored_record_in_an_older_format_fails_the_turn_naming_both_versions() { + let twin = twin_openai().await; + let namespace = format!("{}::{}", module_path!(), line!()); + TwinScenarios::new(namespace.clone()) + .scenario( + TwinScenario::responses(MODEL) + .input_contains("First question") + .text("First answer"), + ) + .load(twin) + .await; + let state = twin_backed_state(twin.base_url.clone(), &namespace); + spawn_scheduler(Arc::clone(&state)); + let app = build_test_router(Arc::clone(&state)); + let workspace = tempfile::tempdir().unwrap(); + let run_id = completed_run(&app, workspace.path()).await; + + let created = json_response( + &app, + post_json( + &format!("/runs/{run_id}/sessions"), + &serde_json::json!({ "title": "Ask Fabro", "model": MODEL }), + ), + StatusCode::CREATED, + ) + .await; + let session_id: SessionId = created["id"].as_str().unwrap().parse().unwrap(); + + turn(&app, session_id, "First question").await; + let stored = state + .stores + .session_records + .get(session_id) + .await + .unwrap() + .expect("the first turn persists the record"); + + // The record as an older build wrote it: the previous format version. + // The store parses it, and the version check refuses it on resume. + let previous = SESSION_RECORD_FORMAT_VERSION - 1; + let mut older = stored.record.clone(); + older.format_version = previous; + state + .stores + .session_records + .put(session_id, run_id, &older, chrono::Utc::now()) + .await + .unwrap(); + state + .session_runtimes() + .load_or_create_runtime(session_id) + .clear_agent() + .await; + + let response = app + .clone() + .oneshot(post_json( + &format!("/sessions/{session_id}/turns"), + &serde_json::json!({ "input": "Second question" }), + )) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::OK); + let bytes = to_bytes(response.into_body(), usize::MAX).await.unwrap(); + let body = String::from_utf8(bytes.to_vec()).unwrap(); + let events: Vec = body + .lines() + .filter_map(|line| line.strip_prefix("data: ")) + .map(|data| serde_json::from_str(data).unwrap()) + .collect(); + let failed = events + .iter() + .find(|event| event["event"] == "run.session.turn.failed") + .unwrap_or_else(|| panic!("the resumed turn fails: {events:#?}")); + assert_eq!(failed["properties"]["code"], "agent_error"); + assert_eq!(failed["properties"]["retryable"], false); + let error = failed["properties"]["error"].as_str().unwrap(); + let expected = format!( + "session record format version {previous} is not supported (this build requires {SESSION_RECORD_FORMAT_VERSION})" + ); + assert!( + error.contains(&expected), + "the failure names both versions: {error}" + ); + assert!( + !events + .iter() + .any(|event| event["event"] == "run.session.turn.succeeded"), + "a refused record runs no turn: {events:#?}" + ); + + // The stored record is untouched, so a build that reads its format + // can still resume it. + let after = state + .stores + .session_records + .get(session_id) + .await + .unwrap() + .expect("the refused record stays stored"); + assert_eq!(after.record.format_version, previous); + } + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn a_resumed_session_continues_its_conversation_past_the_event_log() { let twin = twin_openai().await; diff --git a/lib/components/fabro-store/src/fixtures/session_record_v4.json b/lib/components/fabro-store/src/fixtures/session_record_v4.json new file mode 100644 index 000000000..47dfaaec4 --- /dev/null +++ b/lib/components/fabro-store/src/fixtures/session_record_v4.json @@ -0,0 +1,116 @@ +{ + "format_version": 4, + "scope": { + "session_id": "ses_root", + "root_session_id": "ses_root", + "parent_session_id": null, + "depth": 0 + }, + "provider": "anthropic", + "model": "claude-sonnet-5", + "created_at": "2026-01-01T00:00:00.500Z", + "updated_at": "2026-01-01T00:00:00.500Z", + "last_event_seq": 41, + "messages": [ + { + "kind": "user", + "content": [ + { + "type": "text", + "text": "read the crate root" + } + ], + "timestamp": "2026-01-01T00:00:00.500Z" + }, + { + "kind": "assistant", + "content": "Reading it now.", + "tool_calls": [ + { + "id": "call_1", + "name": "read_file", + "input": { + "type": "function", + "value": "{\"path\":\"src/lib.rs\"}" + }, + "provider_metadata": { + "openai": { + "item_id": "fc_1" + } + } + } + ], + "provider_parts": [ + { + "type": "reasoning", + "text": "the root is a facade", + "signature": "sig_1", + "signature_origin": "anthropic" + }, + { + "type": "opaque", + "kind": "openai.reasoning", + "data": { + "id": "rs_1" + } + } + ], + "usage": { + "input": 1200, + "output": 340, + "reasoning": 96, + "cache_read": 800, + "cache_write": 64 + }, + "response_id": "resp_1", + "timestamp": "2026-01-01T00:00:00.500Z" + }, + { + "kind": "tool_results", + "results": [ + { + "tool_call_id": "call_1", + "name": "read_file", + "content": [ + { + "type": "text", + "text": "//! Pebble is a coding-agent loop library." + } + ], + "is_error": false + } + ], + "timestamp": "2026-01-01T00:00:00.500Z" + }, + { + "kind": "compaction", + "summary": "[Context Summary]\nThe session read the crate root.", + "reason": "manual", + "original_turn_count": 8, + "preserved_turn_count": 2, + "estimated_tokens_before": 12000, + "summary_token_estimate": 24, + "tracked_file_count": 1, + "summary_truncated": false, + "usage": { + "input": 1200, + "output": 340, + "reasoning": 96, + "cache_read": 800, + "cache_write": 64 + }, + "cost_usd_micros": 1250, + "timestamp": "2026-01-01T00:00:00.500Z" + }, + { + "kind": "steering", + "content": [ + { + "type": "text", + "text": "also update the changelog" + } + ], + "timestamp": "2026-01-01T00:00:00.500Z" + } + ] +} diff --git a/lib/components/fabro-store/src/run_session_record_store.rs b/lib/components/fabro-store/src/run_session_record_store.rs index bcfa613e6..05e78da69 100644 --- a/lib/components/fabro-store/src/run_session_record_store.rs +++ b/lib/components/fabro-store/src/run_session_record_store.rs @@ -110,6 +110,7 @@ mod tests { use chrono::TimeZone; use fabro_types::fixtures; use pebble_coding_agent::SessionScope; + use pebble_coding_agent::state::SESSION_RECORD_FORMAT_VERSION; use super::*; use crate::test_support; @@ -149,6 +150,44 @@ mod tests { assert_eq!(stored.updated_at, updated_at); } + /// A record an older build wrote is read back as it was stored: the + /// store parses the previous format, and pebble's version check, not a + /// parse error, is what refuses it on resume. Old runs get no migration. + #[tokio::test] + async fn get_reads_a_record_in_the_previous_format_for_pebble_to_refuse() { + const PREVIOUS_RECORD: &str = include_str!("fixtures/session_record_v4.json"); + + let pool = + test_support::in_memory_pool_with(&[fabro_db::RUN_SESSION_RECORDS_MIGRATION_SQL]); + let store = RunSessionRecordStore::new(pool.clone()); + let session_id = SessionId::new(); + sqlx::query( + "INSERT INTO run_session_records (session_id, run_id, record_json, updated_at_ms) \ + VALUES (?, ?, ?, ?)", + ) + .bind(session_id.to_string()) + .bind(fixtures::RUN_1.to_string()) + .bind(PREVIOUS_RECORD) + .bind(1_789_156_874_678_i64) + .execute(&pool) + .await + .unwrap(); + + let stored = store.get(session_id).await.unwrap().expect("stored record"); + assert_eq!(stored.run_id, fixtures::RUN_1); + assert_eq!( + stored.record.format_version, + SESSION_RECORD_FORMAT_VERSION - 1, + "the fixture is the previous format" + ); + assert!( + !stored.record.is_supported(), + "pebble refuses the previous format on resume rather than reading it" + ); + assert_eq!(stored.record.provider.as_deref(), Some("anthropic")); + assert_eq!(stored.record.model.as_deref(), Some("claude-sonnet-5")); + } + #[tokio::test] async fn put_replaces_an_earlier_record() { let store = RunSessionRecordStore::new(test_support::in_memory_pool_with(&[