scriptoriumadmin/fabro
1
0
Fork 0
mirror of https://github.com/fabro-sh/fabro.git synced 2026-09-10 22:43:37 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 3

Remove Context logs, replace with RunNotice events + tracing

Browse source 
Context::append_log / logs_snapshot was a write-only audit trail that
was never surfaced — not in events, CLI output, or tracing. Errors like
"checkpoint save failed" were silently swallowed.

Replace all append_log call sites with RunNotice events (which are
automatically traced and visible in progress.jsonl / CLI). Remove the
logs field from both Context types, the Checkpoint struct, the OpenAPI
spec, and the TS client. Old checkpoints containing a logs field are
silently ignored during deserialization.

Also make git_diff return Result<String, String> with structured error
info (exit code + stderr) instead of Option<String>.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
Bryan Helmkamp 2026-03-24 09:46:08 -04:00
parent 2aa3dcfaf6
commit 20250a71c0
No known key found for this signature in database
11 changed files with 89 additions and 153 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
docs/api-reference/fabro-api.yaml
View file

@ -2426,7 +2426,6 @@ components:
- completed_nodes
- node_retries
- context_values
- logs
properties:
timestamp:
type: string
@ -2449,11 +2448,6 @@ components:
type: object
additionalProperties: true
description: Key-value context map accumulated during execution.
logs:
type: array
items:
type: string
description: Log entries recorded during execution.
node_outcomes:
type: object
additionalProperties: true

3
docs/execution/checkpoints.mdx
View file

@ -65,7 +65,6 @@ The `checkpoint.json` captures everything needed to resume a run:
| `node_retries` | How many retry attempts each node has used |
| `node_outcomes` | Full outcome (status, context updates, usage) for each completed node |
| `context_values` | Snapshot of the entire [run context](/execution/context) |
| `logs` | Internal log entries |
| `git_commit_sha` | SHA of the run branch commit at this checkpoint |
| `loop_failure_signatures` | Failure signature counts for loop detection |
| `restart_failure_signatures` | Failure signature counts across loop-restart edges |
@ -131,7 +130,7 @@ Here's the full sequence that runs after every node completes:
3. **Commit to run branch** — Stage all file changes, commit with structured trailers linking to the shadow commit SHA
4. **Update checkpoint** — Re-save `checkpoint.json` with the `git_commit_sha` field set
Steps 2-4 are best-effort — if any Git operation fails, the run continues and logs a warning. The disk checkpoint from step 1 is always available as a fallback.
Steps 2-4 are best-effort — if any Git operation fails, the run continues and emits a `RunNotice` warning event. The disk checkpoint from step 1 is always available as a fallback.
## Inspecting run history

36
lib/crates/fabro-core/src/context.rs
View file

@ -52,7 +52,6 @@ impl ContextStore for InMemoryStore {
#[derive(Clone)]
pub struct Context {
store: Arc<dyn ContextStore>,
logs: Arc<RwLock<Vec<String>>>,
}
impl Default for Context {
@ -65,22 +64,11 @@ impl Context {
pub fn new() -> Self {
Self {
store: Arc::new(InMemoryStore::new()),
logs: Arc::new(RwLock::new(Vec::new())),
}
}
pub fn with_store(store: Arc<dyn ContextStore>) -> Self {
Self {
store,
logs: Arc::new(RwLock::new(Vec::new())),
}
}
pub fn with_store_and_logs(
store: Arc<dyn ContextStore>,
logs: Arc<RwLock<Vec<String>>>,
) -> Self {
Self { store, logs }
Self { store }
}
pub fn set(&self, key: impl Into<String>, value: Value) {
@ -107,18 +95,9 @@ impl Context {
self.store.snapshot()
}
pub fn append_log(&self, entry: impl Into<String>) {
self.logs.write().unwrap().push(entry.into());
}
pub fn logs_snapshot(&self) -> Vec<String> {
self.logs.read().unwrap().clone()
}
pub fn clone_context(&self) -> Self {
Self {
store: self.store.fork(),
logs: Arc::new(RwLock::new(self.logs.read().unwrap().clone())),
}
}
@ -211,15 +190,6 @@ mod tests {
assert_eq!(cloned.get("x"), Some(json!(2)));
}
#[test]
fn context_append_and_snapshot_logs() {
let ctx = Context::new();
ctx.append_log("step 1");
ctx.append_log("step 2");
let logs = ctx.logs_snapshot();
assert_eq!(logs, vec!["step 1", "step 2"]);
}
#[test]
fn context_with_custom_store() {
struct CountingStore {
@ -257,14 +227,10 @@ mod tests {
fn context_fork_is_independent() {
let ctx = Context::new();
ctx.set("shared", json!("original"));
ctx.append_log("log1");
let forked = ctx.clone_context();
forked.set("shared", json!("modified"));
forked.append_log("log2");
assert_eq!(ctx.get("shared"), Some(json!("original")));
assert_eq!(ctx.logs_snapshot().len(), 1);
assert_eq!(forked.get("shared"), Some(json!("modified")));
assert_eq!(forked.logs_snapshot().len(), 2);
}
#[test]

2
lib/crates/fabro-workflows/README.md
View file

@ -11,7 +11,7 @@ A DOT-based pipeline runner for multi-stage AI workflows. Define workflows as Gr
- **Outcome** -- The result of executing a handler, carrying a `StageStatus` (Success, Fail, PartialSuccess, Retry, Skipped), optional routing hints (`preferred_label`, `suggested_next_ids`), and context updates.
- **Context** -- A thread-safe key-value store shared across pipeline stages, supporting snapshots and isolated cloning for parallel branches.
- **Interviewer** -- A trait for human-in-the-loop interactions. Implementations include `AutoApproveInterviewer`, `QueueInterviewer`, `CallbackInterviewer`, `ConsoleInterviewer`, and `RecordingInterviewer`.
- **Checkpoint** -- A serializable snapshot of execution state (completed nodes, context values, logs) for crash recovery and resume.
- **Checkpoint** -- A serializable snapshot of execution state (completed nodes, context values) for crash recovery and resume.
## Pipeline Definition

40
lib/crates/fabro-workflows/src/checkpoint.rs
View file

@ -17,7 +17,6 @@ pub struct Checkpoint {
pub completed_nodes: Vec<String>,
pub node_retries: HashMap<String, u32>,
pub context_values: HashMap<String, Value>,
pub logs: Vec<String>,
/// Persisted node outcomes for goal gate checks after resume.
#[serde(default, skip_serializing_if = "HashMap::is_empty")]
pub node_outcomes: HashMap<String, Outcome>,
@ -58,7 +57,6 @@ impl Checkpoint {
completed_nodes,
node_retries,
context_values: context.snapshot(),
logs: context.logs_snapshot(),
node_outcomes,
next_node_id,
git_commit_sha: None,
@ -97,7 +95,6 @@ mod tests {
fn from_context_captures_state() {
let ctx = Context::new();
ctx.set("key", serde_json::json!("value"));
ctx.append_log("started");
let cp = Checkpoint::from_context(
&ctx,
@ -119,8 +116,6 @@ mod tests {
cp.context_values.get("key"),
Some(&serde_json::json!("value"))
);
assert_eq!(cp.logs.len(), 1);
assert_eq!(cp.logs[0], "started");
assert!(cp.node_retries.is_empty());
assert!(cp.node_outcomes.is_empty());
assert!(cp.next_node_id.is_none());
@ -133,7 +128,6 @@ mod tests {
let ctx = Context::new();
ctx.set("goal", serde_json::json!("test"));
ctx.append_log("log entry");
let mut retries = HashMap::new();
retries.insert("work".to_string(), 2u32);
@ -161,7 +155,6 @@ mod tests {
loaded.context_values.get("goal"),
Some(&serde_json::json!("test"))
);
assert_eq!(loaded.logs, vec!["log entry"]);
assert_eq!(
loaded.node_outcomes.get("start").map(|o| &o.status),
Some(&crate::outcome::StageStatus::Success)
@ -274,4 +267,37 @@ mod tests {
assert!(cp.loop_failure_signatures.is_empty());
assert!(cp.restart_failure_signatures.is_empty());
}
#[test]
fn backward_compat_old_checkpoint_with_logs_ignored() {
// Old checkpoints that contain a `logs` field should deserialize fine (field is ignored)
let json = r#"{
"timestamp": "2025-01-01T00:00:00Z",
"current_node": "work",
"completed_nodes": ["start"],
"node_retries": {},
"context_values": {},
"logs": ["old entry 1", "old entry 2"]
}"#;
let cp: Checkpoint = serde_json::from_str(json).unwrap();
assert_eq!(cp.current_node, "work");
}
#[test]
fn new_checkpoint_does_not_serialize_logs() {
let ctx = Context::new();
let cp = Checkpoint::from_context(
&ctx,
"n1",
vec![],
HashMap::new(),
HashMap::new(),
None,
HashMap::new(),
HashMap::new(),
HashMap::new(),
);
let json = serde_json::to_string(&cp).unwrap();
assert!(!json.contains("\"logs\""));
}
}

47
lib/crates/fabro-workflows/src/context/mod.rs
View file

@ -9,7 +9,6 @@ use serde_json::Value;
#[derive(Debug, Clone)]
pub struct Context {
values: Arc<RwLock<HashMap<String, Value>>>,
logs: Arc<RwLock<Vec<String>>>,
}
impl Default for Context {
@ -23,7 +22,6 @@ impl Context {
pub fn new() -> Self {
Self {
values: Arc::new(RwLock::new(HashMap::new())),
logs: Arc::new(RwLock::new(Vec::new())),
}
}
@ -61,18 +59,6 @@ impl Context {
.unwrap_or_else(|| default.to_string())
}
/// Append a log entry.
///
/// # Panics
///
/// Panics if the internal lock is poisoned.
pub fn append_log(&self, entry: impl Into<String>) {
self.logs
.write()
.expect("context lock poisoned")
.push(entry.into());
}
/// Return a snapshot (clone) of all current context values.
///
/// # Panics
@ -83,24 +69,12 @@ impl Context {
self.values.read().expect("context lock poisoned").clone()
}
/// Return a snapshot of the logs.
///
/// # Panics
///
/// Panics if the internal lock is poisoned.
#[must_use]
pub fn logs_snapshot(&self) -> Vec<String> {
self.logs.read().expect("context lock poisoned").clone()
}
/// Deep copy for parallel branch isolation.
#[must_use]
pub fn clone_context(&self) -> Self {
let values = self.snapshot();
let logs = self.logs_snapshot();
Self {
values: Arc::new(RwLock::new(values)),
logs: Arc::new(RwLock::new(logs)),
}
}
@ -121,7 +95,6 @@ impl Context {
pub(crate) fn from_values(values: HashMap<String, Value>) -> Self {
Self {
values: Arc::new(RwLock::new(values)),
logs: Arc::new(RwLock::new(Vec::new())),
}
}
@ -129,10 +102,6 @@ impl Context {
self.values.clone()
}
pub(crate) fn logs_arc(&self) -> Arc<RwLock<Vec<String>>> {
self.logs.clone()
}
// --- Typed accessors ---
#[must_use]
@ -179,7 +148,6 @@ mod tests {
fn new_context_is_empty() {
let ctx = Context::new();
assert!(ctx.snapshot().is_empty());
assert!(ctx.logs_snapshot().is_empty());
}
#[test]
@ -215,17 +183,6 @@ mod tests {
assert_eq!(ctx.get_string("num", "default"), "default");
}
#[test]
fn append_and_snapshot_logs() {
let ctx = Context::new();
ctx.append_log("first entry");
ctx.append_log("second entry");
let logs = ctx.logs_snapshot();
assert_eq!(logs.len(), 2);
assert_eq!(logs[0], "first entry");
assert_eq!(logs[1], "second entry");
}
#[test]
fn snapshot_is_independent() {
let ctx = Context::new();
@ -241,19 +198,15 @@ mod tests {
fn clone_context_is_independent() {
let ctx = Context::new();
ctx.set("shared", serde_json::json!("original"));
ctx.append_log("log1");
let cloned = ctx.clone_context();
cloned.set("shared", serde_json::json!("modified"));
cloned.append_log("log2");
// original should be unchanged
assert_eq!(ctx.get("shared"), Some(serde_json::json!("original")));
assert_eq!(ctx.logs_snapshot().len(), 1);
// cloned has the modification
assert_eq!(cloned.get("shared"), Some(serde_json::json!("modified")));
assert_eq!(cloned.logs_snapshot().len(), 2);
}
#[test]

18
lib/crates/fabro-workflows/src/core_adapter/context.rs
View file

@ -40,13 +40,13 @@ impl ContextStore for WfContextStore {
}
}
/// Create a fabro_core::Context that shares the same underlying values and logs
/// Create a fabro_core::Context that shares the same underlying values
/// as the given wf::Context. Writes through either are visible to both.
pub fn bridge_context(wf_ctx: &WfContext) -> CoreContext {
let store = Arc::new(WfContextStore {
values: wf_ctx.values_arc(),
});
CoreContext::with_store_and_logs(store, wf_ctx.logs_arc())
CoreContext::with_store(store)
}
/// Extension trait providing typed domain accessors on a fabro_core::Context.
@ -97,20 +97,6 @@ mod tests {
assert_eq!(wf.get("key2"), Some(json!("from_core")));
}
#[test]
fn bridge_shares_logs() {
let wf = WfContext::new();
let core = bridge_context(&wf);
// Append via wf, read via core
wf.append_log("wf_log");
assert_eq!(core.logs_snapshot(), vec!["wf_log"]);
// Append via core, read via wf
core.append_log("core_log");
assert_eq!(wf.logs_snapshot(), vec!["wf_log", "core_log"]);
}
#[test]
fn bridge_fork_is_independent() {
let wf = WfContext::new();

9
lib/crates/fabro-workflows/src/core_adapter/lifecycle.rs
View file

@ -483,7 +483,6 @@ impl RunLifecycle<WorkflowGraph> for WorkflowLifecycle {
node_outcomes,
node_retries: state.node_retries.clone(),
context_values: state.context.snapshot(),
logs: state.context.logs_snapshot(),
next_node_id: next_node_id.map(String::from),
git_commit_sha: None,
node_visits: state.node_visits.clone(),
@ -494,9 +493,11 @@ impl RunLifecycle<WorkflowGraph> for WorkflowLifecycle {
// Write checkpoint.json
let checkpoint_path = self.run_dir.join("checkpoint.json");
if let Err(e) = checkpoint.save(&checkpoint_path) {
state
.context
.append_log(format!("checkpoint save failed: {e}"));
self.emitter.emit(&WorkflowRunEvent::RunNotice {
level: crate::event::RunNoticeLevel::Warn,
code: "checkpoint_disk_save_failed".to_string(),
message: format!("[node: {}] checkpoint save failed: {e}", node.id()),
});
}
// Emit CheckpointCompleted event

73
lib/crates/fabro-workflows/src/engine.rs
View file

@ -735,11 +735,12 @@ pub async fn git_push_host(
}
/// Run a git diff via the sandbox.
async fn git_diff(sandbox: &dyn Sandbox, base: &str) -> Option<String> {
async fn git_diff(sandbox: &dyn Sandbox, base: &str) -> std::result::Result<String, String> {
let cmd = format!("{GIT_REMOTE} diff {base} HEAD");
match sandbox.exec_command(&cmd, 30_000, None, None, None).await {
Ok(r) if r.exit_code == 0 => Some(r.stdout),
_ => None,
Ok(r) if r.exit_code == 0 => Ok(r.stdout),
Ok(r) => Err(format!("exit {}: {}", r.exit_code, r.stderr.trim())),
Err(e) => Err(e.to_string()),
}
}
@ -1473,10 +1474,6 @@ impl WorkflowRunEngine {
for (k, v) in &cp.context_values {
s.context.set(k.clone(), v.clone());
}
// Restore logs
for log in &cp.logs {
s.context.append_log(log.clone());
}
s.completed_nodes = cp.completed_nodes.clone();
s.node_retries = cp.node_retries.clone();
s.node_visits = cp.node_visits.clone();
@ -1495,9 +1492,6 @@ impl WorkflowRunEngine {
for (k, v) in seed.snapshot() {
s.context.set(k, v);
}
for log in seed.logs_snapshot() {
s.context.append_log(log);
}
s
} else {
RunState::new(&wf_graph).map_err(|e| FabroError::engine(e.to_string()))?
@ -1739,9 +1733,6 @@ impl WorkflowRunEngine {
for (key, value) in &cp.context_values {
context.set(key.clone(), value.clone());
}
for log_entry in &cp.logs {
context.append_log(log_entry.clone());
}
completed_nodes = cp.completed_nodes.clone();
// Use persisted node_visits; fall back to reconstruction for old checkpoints
if cp.node_visits.is_empty() {
@ -2149,14 +2140,22 @@ impl WorkflowRunEngine {
// Offload large context values to artifact store before recording
if let Err(e) = offload_large_values(&mut outcome.context_updates, &artifact_store) {
context.append_log(format!("artifact offload failed: {e}"));
self.services.emitter.emit(&WorkflowRunEvent::RunNotice {
level: crate::event::RunNoticeLevel::Warn,
code: "artifact_offload_failed".to_string(),
message: format!("[node: {}] artifact offload failed: {e}", node.id),
});
}
// Sync artifact files to the sandbox (no-op for local envs)
if let Err(e) =
sync_artifacts_to_env(&mut outcome.context_updates, &*self.services.sandbox).await
{
context.append_log(format!("artifact sync failed: {e}"));
self.services.emitter.emit(&WorkflowRunEvent::RunNotice {
level: crate::event::RunNoticeLevel::Warn,
code: "artifact_sync_failed".to_string(),
message: format!("[node: {}] artifact sync failed: {e}", node.id),
});
}
// Step 3: Record completion
@ -2276,7 +2275,11 @@ impl WorkflowRunEngine {
);
let checkpoint_path = config.run_dir.join("checkpoint.json");
if let Err(e) = checkpoint.save(&checkpoint_path) {
context.append_log(format!("checkpoint save failed: {e}"));
self.services.emitter.emit(&WorkflowRunEvent::RunNotice {
level: crate::event::RunNoticeLevel::Warn,
code: "checkpoint_disk_save_failed".to_string(),
message: format!("[node: {}] checkpoint save failed: {e}", node.id),
});
}
// Step 6b: Write shadow branch first, then run branch commit with trailer
@ -2309,9 +2312,14 @@ impl WorkflowRunEngine {
match store.write_checkpoint(&config.run_id, &cp_json, &extra_refs) {
Ok(sha) => Some(sha),
Err(e) => {
context.append_log(format!(
"metadata checkpoint write failed: {e}"
));
self.services.emitter.emit(&WorkflowRunEvent::RunNotice {
level: crate::event::RunNoticeLevel::Warn,
code: "checkpoint_metadata_write_failed".to_string(),
message: format!(
"[node: {}] metadata checkpoint write failed: {e}",
node.id
),
});
None
}
}
@ -2338,7 +2346,14 @@ impl WorkflowRunEngine {
Ok(sha) => {
checkpoint.git_commit_sha = Some(sha.clone());
if let Err(e) = checkpoint.save(&checkpoint_path) {
context.append_log(format!("checkpoint re-save with SHA failed: {e}"));
self.services.emitter.emit(&WorkflowRunEvent::RunNotice {
level: crate::event::RunNoticeLevel::Warn,
code: "checkpoint_resave_failed".to_string(),
message: format!(
"[node: {}] checkpoint re-save with SHA failed: {e}",
node.id
),
});
}
self.services
.emitter
@ -2404,13 +2419,18 @@ impl WorkflowRunEngine {
let diff_dest =
node_dir(&config.run_dir, &node.id, visit).join("diff.patch");
let diff_result = git_diff(&*self.services.sandbox, &diff_base).await;
if let Some(patch) = diff_result {
if !patch.is_empty() {
match git_diff(&*self.services.sandbox, &diff_base).await {
Ok(patch) if !patch.is_empty() => {
let _ = std::fs::write(&diff_dest, patch);
}
} else {
context.append_log("git diff failed".to_string());
Ok(_) => {} // empty diff, nothing to write
Err(err) => {
self.services.emitter.emit(&WorkflowRunEvent::RunNotice {
level: crate::event::RunNoticeLevel::Warn,
code: "git_diff_failed".to_string(),
message: format!("[node: {}] git diff failed: {err}", node.id),
});
}
}
last_git_sha = Some(sha);
@ -2589,8 +2609,7 @@ impl WorkflowRunEngine {
// Write final.patch: comprehensive diff from base_sha to HEAD
if config.git_checkpoint_enabled {
if let Some(ref base) = config.base_sha {
let patch = git_diff(&*self.services.sandbox, base).await;
if let Some(patch) = patch {
if let Ok(patch) = git_diff(&*self.services.sandbox, base).await {
if !patch.is_empty() {
let _ = std::fs::write(config.run_dir.join("final.patch"), patch);
}

4
lib/crates/fabro-workflows/tests/integration.rs
View file

@ -1375,9 +1375,6 @@ fn checkpoint_save_and_resume_roundtrip() {
let ctx = Context::new();
ctx.set("goal", serde_json::json!("Test checkpoint"));
ctx.set("progress", serde_json::json!(42));
ctx.append_log("started");
ctx.append_log("step_1 completed");
let mut retries = std::collections::HashMap::new();
retries.insert("step_1".to_string(), 1u32);
let checkpoint = Checkpoint::from_context(
@ -1408,7 +1405,6 @@ fn checkpoint_save_and_resume_roundtrip() {
loaded.context_values.get("progress"),
Some(&serde_json::json!(42))
);
assert_eq!(loaded.logs.len(), 2);
}
// ---------------------------------------------------------------------------

4
lib/packages/fabro-api-client/src/models/run-checkpoint.ts
View file

@ -38,10 +38,6 @@ export interface RunCheckpoint {
* Key-value context map accumulated during execution.
*/
'context_values': { [key: string]: any; };
/**
* Log entries recorded during execution.
*/
'logs': Array<string>;
/**
* Map of node identifier to outcome data for goal gate checks after resume.
*/