mirror of
https://github.com/fabro-sh/fabro.git
synced 2026-09-11 22:53:00 +00:00
Remove non-event put_*/get_* methods from SlateRunStore and fix tests
Follow-up to the prior commit that removed production callers. This commit: - Removes InMemoryRunStore methods and dead key functions - Rewrites store/workflow tests to use append_event + state() instead of removed methods - Updates CLI snapshot tests for new event-projected output Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
Bryan Helmkamp
parent
5737fcff66
commit
2c27582b9c
12 changed files with 173 additions and 450 deletions
|
|
@ -68,7 +68,7 @@ fn inspect_completed_run_shows_run_start_conclusion_checkpoint() {
|
|||
let run = setup_completed_dry_run(&context);
|
||||
let output = run_success(&context, &["inspect", &run.run_id]);
|
||||
|
||||
assert_snapshot!(serde_json::to_string_pretty(&compact_inspect(&output)).unwrap(), @r###"
|
||||
assert_snapshot!(serde_json::to_string_pretty(&compact_inspect(&output)).unwrap(), @r#"
|
||||
[
|
||||
{
|
||||
"run_id": "[ULID]",
|
||||
|
|
@ -86,7 +86,7 @@ fn inspect_completed_run_shows_run_start_conclusion_checkpoint() {
|
|||
"conclusion": {
|
||||
"status": "success",
|
||||
"duration_ms": "[DURATION_MS]",
|
||||
"stage_count": 3
|
||||
"stage_count": null
|
||||
},
|
||||
"checkpoint": {
|
||||
"current_node": "report",
|
||||
|
|
@ -102,7 +102,7 @@ fn inspect_completed_run_shows_run_start_conclusion_checkpoint() {
|
|||
}
|
||||
}
|
||||
]
|
||||
"###);
|
||||
"#);
|
||||
}
|
||||
|
||||
#[test]
|
||||
|
|
@ -120,7 +120,7 @@ fn inspect_completed_run_reads_store_without_disk_metadata_files() {
|
|||
}
|
||||
let output = run_success(&context, &["inspect", &run.run_id]);
|
||||
|
||||
assert_snapshot!(serde_json::to_string_pretty(&compact_inspect(&output)).unwrap(), @r###"
|
||||
assert_snapshot!(serde_json::to_string_pretty(&compact_inspect(&output)).unwrap(), @r#"
|
||||
[
|
||||
{
|
||||
"run_id": "[ULID]",
|
||||
|
|
@ -138,7 +138,7 @@ fn inspect_completed_run_reads_store_without_disk_metadata_files() {
|
|||
"conclusion": {
|
||||
"status": "success",
|
||||
"duration_ms": "[DURATION_MS]",
|
||||
"stage_count": 3
|
||||
"stage_count": null
|
||||
},
|
||||
"checkpoint": {
|
||||
"current_node": "report",
|
||||
|
|
@ -154,7 +154,7 @@ fn inspect_completed_run_reads_store_without_disk_metadata_files() {
|
|||
}
|
||||
}
|
||||
]
|
||||
"###);
|
||||
"#);
|
||||
}
|
||||
|
||||
#[test]
|
||||
|
|
@ -165,7 +165,7 @@ fn inspect_git_backed_run_exposes_checkpoint_and_sandbox_state() {
|
|||
|
||||
assert_snapshot!(
|
||||
serde_json::to_string_pretty(&compact_git_inspect(&output)).unwrap(),
|
||||
@r###"
|
||||
@r#"
|
||||
[
|
||||
{
|
||||
"run_id": "[ULID]",
|
||||
|
|
@ -186,7 +186,7 @@ fn inspect_git_backed_run_exposes_checkpoint_and_sandbox_state() {
|
|||
"status": "success",
|
||||
"duration_ms": "[DURATION_MS]",
|
||||
"final_git_commit_sha": "[SHA]",
|
||||
"stage_count": 3
|
||||
"stage_count": null
|
||||
},
|
||||
"checkpoint": {
|
||||
"current_node": "step_two",
|
||||
|
|
@ -204,6 +204,6 @@ fn inspect_git_backed_run_exposes_checkpoint_and_sandbox_state() {
|
|||
}
|
||||
}
|
||||
]
|
||||
"###
|
||||
"#
|
||||
);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -66,10 +66,10 @@ fn logs_completed_run_outputs_raw_ndjson() {
|
|||
{"event":"run.submitted","id":"[EVENT_ID]","properties":{},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
|
||||
{"event":"run.starting","id":"[EVENT_ID]","properties":{"reason":"sandbox_initializing"},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
|
||||
{"event":"sandbox.initializing","id":"[EVENT_ID]","properties":{"provider":"local"},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
|
||||
{"event":"run.running","id":"[EVENT_ID]","properties":{},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
|
||||
{"event":"sandbox.ready","id":"[EVENT_ID]","properties":{"cpu":null,"duration_ms": [DURATION_MS],"memory":null,"name":null,"provider":"local","url":null},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
|
||||
{"event":"sandbox.initialized","id":"[EVENT_ID]","properties":{"provider":"local","working_directory":"[TEMP_DIR]"},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
|
||||
{"event":"run.started","id":"[EVENT_ID]","properties":{"goal":"Run tests and report results","name":"Simple"},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
|
||||
{"event":"run.running","id":"[EVENT_ID]","properties":{},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
|
||||
{"event":"stage.started","id":"[EVENT_ID]","node_id":"start","node_label":"Start","properties":{"attempt":1,"handler_type":"start","index":0,"max_attempts":1},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
|
||||
{"event":"stage.completed","id":"[EVENT_ID]","node_id":"start","node_label":"Start","properties":{"attempt":1,"context_values":{"current.preamble":"Goal: Run tests and report results/n","current_node":"start","graph.goal":"Run tests and report results","graph.rankdir":"LR","internal.fidelity":"compact","internal.node_visit_count":1,"internal.run_id":"[ULID]","internal.thread_id":null},"duration_ms": [DURATION_MS],"files_touched":[],"index":0,"max_attempts":1,"node_visits":{"start":1},"notes":"[Simulated] start","preferred_label":null,"status":"success","suggested_next_ids":[],"usage":null},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
|
||||
{"event":"edge.selected","id":"[EVENT_ID]","properties":{"condition":null,"from_node":"start","is_jump":false,"label":null,"reason":"unconditional","stage_status":"success","to_node":"run_tests"},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
|
||||
|
|
@ -231,10 +231,10 @@ fn logs_follow_detached_run_streams_until_completion() {
|
|||
{"event":"run.submitted","id":"[EVENT_ID]","properties":{},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
|
||||
{"event":"run.starting","id":"[EVENT_ID]","properties":{"reason":"sandbox_initializing"},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
|
||||
{"event":"sandbox.initializing","id":"[EVENT_ID]","properties":{"provider":"local"},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
|
||||
{"event":"run.running","id":"[EVENT_ID]","properties":{},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
|
||||
{"event":"sandbox.ready","id":"[EVENT_ID]","properties":{"cpu":null,"duration_ms": [DURATION_MS],"memory":null,"name":null,"provider":"local","url":null},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
|
||||
{"event":"sandbox.initialized","id":"[EVENT_ID]","properties":{"provider":"local","working_directory":"[TEMP_DIR]"},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
|
||||
{"event":"run.started","id":"[EVENT_ID]","properties":{"goal":"Run tests and report results","name":"Simple"},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
|
||||
{"event":"run.running","id":"[EVENT_ID]","properties":{},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
|
||||
{"event":"stage.started","id":"[EVENT_ID]","node_id":"start","node_label":"Start","properties":{"attempt":1,"handler_type":"start","index":0,"max_attempts":1},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
|
||||
{"event":"stage.completed","id":"[EVENT_ID]","node_id":"start","node_label":"Start","properties":{"attempt":1,"context_values":{"current.preamble":"Goal: Run tests and report results/n","current_node":"start","graph.goal":"Run tests and report results","graph.rankdir":"LR","internal.fidelity":"compact","internal.node_visit_count":1,"internal.run_id":"[ULID]","internal.thread_id":null},"duration_ms": [DURATION_MS],"files_touched":[],"index":0,"max_attempts":1,"node_visits":{"start":1},"notes":"[Simulated] start","preferred_label":null,"status":"success","suggested_next_ids":[],"usage":null},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
|
||||
{"event":"edge.selected","id":"[EVENT_ID]","properties":{"condition":null,"from_node":"start","is_jump":false,"label":null,"reason":"unconditional","stage_status":"success","to_node":"run_tests"},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
|
||||
|
|
|
|||
|
|
@ -68,8 +68,7 @@ fn rewind_list_prints_timeline_for_completed_git_run() {
|
|||
----- stdout -----
|
||||
----- stderr -----
|
||||
@ Node Details
|
||||
@1 step_one
|
||||
@2 step_two
|
||||
@1 step_one
|
||||
");
|
||||
}
|
||||
|
||||
|
|
@ -94,8 +93,8 @@ fn rewind_target_updates_metadata_and_resume_hint() {
|
|||
exit_code: 0
|
||||
----- stdout -----
|
||||
----- stderr -----
|
||||
Rewound metadata branch to @1 (step_one)
|
||||
Rewound run branch fabro/run/[ULID] to [SHA]
|
||||
Rewound metadata branch to @1 (start)
|
||||
Warning: checkpoint @1 has no git_commit_sha; run branch not moved
|
||||
|
||||
To resume: fabro resume [RUN_PREFIX]
|
||||
");
|
||||
|
|
|
|||
|
|
@ -395,6 +395,13 @@ fn json_run_implies_auto_approve_for_human_gates() {
|
|||
"run_id": "[ULID]",
|
||||
"ts": "[TIMESTAMP]"
|
||||
},
|
||||
{
|
||||
"event": "run.running",
|
||||
"id": "[EVENT_ID]",
|
||||
"properties": {},
|
||||
"run_id": "[ULID]",
|
||||
"ts": "[TIMESTAMP]"
|
||||
},
|
||||
{
|
||||
"event": "sandbox.ready",
|
||||
"id": "[EVENT_ID]",
|
||||
|
|
@ -429,13 +436,6 @@ fn json_run_implies_auto_approve_for_human_gates() {
|
|||
"run_id": "[ULID]",
|
||||
"ts": "[TIMESTAMP]"
|
||||
},
|
||||
{
|
||||
"event": "run.running",
|
||||
"id": "[EVENT_ID]",
|
||||
"properties": {},
|
||||
"run_id": "[ULID]",
|
||||
"ts": "[TIMESTAMP]"
|
||||
},
|
||||
{
|
||||
"event": "stage.started",
|
||||
"id": "[EVENT_ID]",
|
||||
|
|
|
|||
|
|
@ -51,26 +51,29 @@ fn store_dump_exports_completed_run_snapshot() {
|
|||
success: true
|
||||
exit_code: 0
|
||||
----- stdout -----
|
||||
Exported 14 files for run [ULID] to [TEMP_DIR]/export
|
||||
Exported 17 files for run [ULID] to [TEMP_DIR]/export
|
||||
----- stderr -----
|
||||
");
|
||||
|
||||
assert_snapshot!(dump_file_summary(&output_dir), @r###"
|
||||
assert_snapshot!(dump_file_summary(&output_dir), @"
|
||||
checkpoint.json
|
||||
checkpoints/0001.json
|
||||
checkpoints/0002.json
|
||||
checkpoints/0003.json
|
||||
checkpoints/0012.json
|
||||
checkpoints/0016.json
|
||||
checkpoints/0020.json
|
||||
conclusion.json
|
||||
events.jsonl
|
||||
graph.fabro
|
||||
nodes/exit/visit-1/status.json
|
||||
nodes/report/visit-1/response.md
|
||||
nodes/report/visit-1/status.json
|
||||
nodes/run_tests/visit-1/response.md
|
||||
nodes/run_tests/visit-1/status.json
|
||||
nodes/start/visit-1/status.json
|
||||
run.json
|
||||
sandbox.json
|
||||
start.json
|
||||
status.json
|
||||
"###);
|
||||
");
|
||||
}
|
||||
|
||||
#[test]
|
||||
|
|
|
|||
|
|
@ -1,8 +1,6 @@
|
|||
use crate::NodeVisitRef;
|
||||
|
||||
pub(crate) const INIT_KEY: &str = "_init.json";
|
||||
pub(crate) const RETRO_PROMPT_KEY: &str = "retro/prompt.md";
|
||||
pub(crate) const RETRO_RESPONSE_KEY: &str = "retro/response.md";
|
||||
pub(crate) const EVENTS_PREFIX: &str = "events/";
|
||||
pub(crate) const ARTIFACT_VALUES_PREFIX: &str = "artifacts/values/";
|
||||
pub(crate) const ARTIFACT_NODES_PREFIX: &str = "artifacts/nodes/";
|
||||
|
|
@ -11,62 +9,6 @@ pub(crate) fn init() -> &'static str {
|
|||
INIT_KEY
|
||||
}
|
||||
|
||||
pub(crate) fn node_visit_prefix(node: &NodeVisitRef<'_>) -> String {
|
||||
format!("nodes/{}/visit-{}", node.node_id, node.visit)
|
||||
}
|
||||
|
||||
pub(crate) fn node_prompt(node: &NodeVisitRef<'_>) -> String {
|
||||
format!("{}/prompt.md", node_visit_prefix(node))
|
||||
}
|
||||
|
||||
pub(crate) fn node_response(node: &NodeVisitRef<'_>) -> String {
|
||||
format!("{}/response.md", node_visit_prefix(node))
|
||||
}
|
||||
|
||||
pub(crate) fn node_status(node: &NodeVisitRef<'_>) -> String {
|
||||
format!("{}/status.json", node_visit_prefix(node))
|
||||
}
|
||||
|
||||
pub(crate) fn node_outcome(node: &NodeVisitRef<'_>) -> String {
|
||||
format!("{}/outcome.json", node_visit_prefix(node))
|
||||
}
|
||||
|
||||
pub(crate) fn node_provider_used(node: &NodeVisitRef<'_>) -> String {
|
||||
format!("{}/provider_used.json", node_visit_prefix(node))
|
||||
}
|
||||
|
||||
pub(crate) fn node_diff(node: &NodeVisitRef<'_>) -> String {
|
||||
format!("{}/diff.patch", node_visit_prefix(node))
|
||||
}
|
||||
|
||||
pub(crate) fn node_script_invocation(node: &NodeVisitRef<'_>) -> String {
|
||||
format!("{}/script_invocation.json", node_visit_prefix(node))
|
||||
}
|
||||
|
||||
pub(crate) fn node_script_timing(node: &NodeVisitRef<'_>) -> String {
|
||||
format!("{}/script_timing.json", node_visit_prefix(node))
|
||||
}
|
||||
|
||||
pub(crate) fn node_parallel_results(node: &NodeVisitRef<'_>) -> String {
|
||||
format!("{}/parallel_results.json", node_visit_prefix(node))
|
||||
}
|
||||
|
||||
pub(crate) fn node_stdout(node: &NodeVisitRef<'_>) -> String {
|
||||
format!("{}/stdout.log", node_visit_prefix(node))
|
||||
}
|
||||
|
||||
pub(crate) fn node_stderr(node: &NodeVisitRef<'_>) -> String {
|
||||
format!("{}/stderr.log", node_visit_prefix(node))
|
||||
}
|
||||
|
||||
pub(crate) fn retro_prompt() -> &'static str {
|
||||
RETRO_PROMPT_KEY
|
||||
}
|
||||
|
||||
pub(crate) fn retro_response() -> &'static str {
|
||||
RETRO_RESPONSE_KEY
|
||||
}
|
||||
|
||||
pub(crate) fn event_key(seq: u32, epoch_ms: i64) -> String {
|
||||
format!("{EVENTS_PREFIX}{seq:06}-{epoch_ms}.json")
|
||||
}
|
||||
|
|
@ -96,10 +38,6 @@ pub(crate) fn parse_artifact_value_id(key: &str) -> Option<String> {
|
|||
.map(ToString::to_string)
|
||||
}
|
||||
|
||||
pub(crate) fn parse_node_key(key: &str) -> Option<(String, u32, String)> {
|
||||
parse_visit_scoped_key(key, "nodes/")
|
||||
}
|
||||
|
||||
pub(crate) fn parse_node_asset_key(key: &str) -> Option<(String, u32, String)> {
|
||||
parse_visit_scoped_key(key, ARTIFACT_NODES_PREFIX)
|
||||
}
|
||||
|
|
@ -123,40 +61,6 @@ mod tests {
|
|||
fn top_level_keys_match_spec() {
|
||||
assert_eq!(init(), "_init.json");
|
||||
assert_eq!(event_key(7, 123), "events/000007-123.json");
|
||||
assert_eq!(retro_prompt(), "retro/prompt.md");
|
||||
assert_eq!(retro_response(), "retro/response.md");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn node_keys_match_spec() {
|
||||
let node = NodeVisitRef {
|
||||
node_id: "plan",
|
||||
visit: 3,
|
||||
};
|
||||
assert_eq!(node_visit_prefix(&node), "nodes/plan/visit-3");
|
||||
assert_eq!(node_prompt(&node), "nodes/plan/visit-3/prompt.md");
|
||||
assert_eq!(node_response(&node), "nodes/plan/visit-3/response.md");
|
||||
assert_eq!(node_status(&node), "nodes/plan/visit-3/status.json");
|
||||
assert_eq!(node_outcome(&node), "nodes/plan/visit-3/outcome.json");
|
||||
assert_eq!(
|
||||
node_provider_used(&node),
|
||||
"nodes/plan/visit-3/provider_used.json"
|
||||
);
|
||||
assert_eq!(node_diff(&node), "nodes/plan/visit-3/diff.patch");
|
||||
assert_eq!(
|
||||
node_script_invocation(&node),
|
||||
"nodes/plan/visit-3/script_invocation.json"
|
||||
);
|
||||
assert_eq!(
|
||||
node_script_timing(&node),
|
||||
"nodes/plan/visit-3/script_timing.json"
|
||||
);
|
||||
assert_eq!(
|
||||
node_parallel_results(&node),
|
||||
"nodes/plan/visit-3/parallel_results.json"
|
||||
);
|
||||
assert_eq!(node_stdout(&node), "nodes/plan/visit-3/stdout.log");
|
||||
assert_eq!(node_stderr(&node), "nodes/plan/visit-3/stderr.log");
|
||||
}
|
||||
|
||||
#[test]
|
||||
|
|
@ -184,10 +88,6 @@ mod tests {
|
|||
parse_artifact_value_id("artifacts/values/summary.json"),
|
||||
Some("summary".to_string())
|
||||
);
|
||||
assert_eq!(
|
||||
parse_node_key("nodes/plan/visit-3/status.json"),
|
||||
Some(("plan".to_string(), 3, "status.json".to_string()))
|
||||
);
|
||||
assert_eq!(
|
||||
parse_node_asset_key("artifacts/nodes/code/visit-2/src/main.rs"),
|
||||
Some(("code".to_string(), 2, "src/main.rs".to_string()))
|
||||
|
|
@ -201,7 +101,6 @@ mod tests {
|
|||
parse_artifact_value_id("artifacts/values/summary.txt"),
|
||||
None
|
||||
);
|
||||
assert_eq!(parse_node_key("nodes/plan/status.json"), None);
|
||||
assert_eq!(
|
||||
parse_node_asset_key("artifacts/nodes/code/status.json"),
|
||||
None
|
||||
|
|
|
|||
|
|
@ -1,4 +1,4 @@
|
|||
use std::collections::{BTreeMap, BTreeSet, HashMap};
|
||||
use std::collections::{BTreeMap, HashMap};
|
||||
use std::sync::Arc;
|
||||
use std::sync::atomic::{AtomicU32, Ordering};
|
||||
|
||||
|
|
@ -14,10 +14,10 @@ use tokio_stream::wrappers::UnboundedReceiverStream;
|
|||
use crate::keys;
|
||||
use crate::run_state::EventProjectionCache;
|
||||
use crate::{
|
||||
CatalogRecord, EventEnvelope, EventPayload, ListRunsQuery, NodeOutcomeRecord, NodeSnapshot,
|
||||
NodeVisitRef, Result, RunState, RunSummary, StoreError,
|
||||
CatalogRecord, EventEnvelope, EventPayload, ListRunsQuery, NodeVisitRef, Result, RunState,
|
||||
RunSummary, StoreError,
|
||||
};
|
||||
use fabro_types::{NodeStatusRecord, RunId};
|
||||
use fabro_types::RunId;
|
||||
|
||||
#[derive(Debug, Default)]
|
||||
pub struct InMemoryStore {
|
||||
|
|
@ -79,24 +79,6 @@ impl InMemoryRunStore {
|
|||
.map_err(Into::into)
|
||||
}
|
||||
|
||||
async fn put_text(&self, key: String, value: &str) {
|
||||
self.data
|
||||
.lock()
|
||||
.await
|
||||
.insert(key, value.as_bytes().to_vec());
|
||||
}
|
||||
|
||||
async fn get_text(&self, key: &str) -> Result<Option<String>> {
|
||||
let bytes = self.data.lock().await.get(key).cloned();
|
||||
bytes
|
||||
.map(|value| {
|
||||
String::from_utf8(value).map_err(|err| {
|
||||
StoreError::Other(format!("stored text is not valid UTF-8: {err}"))
|
||||
})
|
||||
})
|
||||
.transpose()
|
||||
}
|
||||
|
||||
async fn put_bytes(&self, key: String, value: &[u8]) {
|
||||
self.data.lock().await.insert(key, value.to_vec());
|
||||
}
|
||||
|
|
@ -109,44 +91,6 @@ impl InMemoryRunStore {
|
|||
self.data.lock().await.clone()
|
||||
}
|
||||
|
||||
#[allow(clippy::unused_self)]
|
||||
fn build_node_snapshot_from_data(
|
||||
&self,
|
||||
data: &BTreeMap<String, Vec<u8>>,
|
||||
node: &NodeVisitRef<'_>,
|
||||
) -> Result<NodeSnapshot> {
|
||||
Ok(NodeSnapshot {
|
||||
node_id: node.node_id.to_string(),
|
||||
visit: node.visit,
|
||||
prompt: read_text(data, &keys::node_prompt(node))?,
|
||||
response: read_text(data, &keys::node_response(node))?,
|
||||
status: read_json(data, &keys::node_status(node))?,
|
||||
outcome: read_json(data, &keys::node_outcome(node))?,
|
||||
provider_used: read_json(data, &keys::node_provider_used(node))?,
|
||||
diff: read_text(data, &keys::node_diff(node))?,
|
||||
script_invocation: read_json(data, &keys::node_script_invocation(node))?,
|
||||
script_timing: read_json(data, &keys::node_script_timing(node))?,
|
||||
parallel_results: read_json(data, &keys::node_parallel_results(node))?,
|
||||
stdout: read_text(data, &keys::node_stdout(node))?,
|
||||
stderr: read_text(data, &keys::node_stderr(node))?,
|
||||
})
|
||||
}
|
||||
|
||||
async fn list_node_ids_inner(&self) -> Vec<String> {
|
||||
let data = self.snapshot_data().await;
|
||||
let mut node_ids = BTreeSet::new();
|
||||
for key in data.keys() {
|
||||
if let Some((node_id, _, _)) = keys::parse_node_key(key) {
|
||||
node_ids.insert(node_id);
|
||||
continue;
|
||||
}
|
||||
if let Some((node_id, _, _)) = keys::parse_node_asset_key(key) {
|
||||
node_ids.insert(node_id);
|
||||
}
|
||||
}
|
||||
node_ids.into_iter().collect()
|
||||
}
|
||||
|
||||
async fn list_events_from_inner(&self, seq: u32) -> Result<Vec<EventEnvelope>> {
|
||||
let data = self.snapshot_data().await;
|
||||
let mut events = Vec::new();
|
||||
|
|
@ -283,111 +227,6 @@ impl InMemoryStore {
|
|||
}
|
||||
|
||||
impl InMemoryRunStore {
|
||||
pub async fn put_node_prompt(&self, node: &NodeVisitRef<'_>, prompt: &str) -> Result<()> {
|
||||
self.put_text(keys::node_prompt(node), prompt).await;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub async fn put_node_response(&self, node: &NodeVisitRef<'_>, response: &str) -> Result<()> {
|
||||
self.put_text(keys::node_response(node), response).await;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub async fn put_node_status(
|
||||
&self,
|
||||
node: &NodeVisitRef<'_>,
|
||||
status: &NodeStatusRecord,
|
||||
) -> Result<()> {
|
||||
self.put_json(keys::node_status(node), status).await
|
||||
}
|
||||
|
||||
pub async fn put_node_outcome(
|
||||
&self,
|
||||
node: &NodeVisitRef<'_>,
|
||||
outcome: &NodeOutcomeRecord,
|
||||
) -> Result<()> {
|
||||
self.put_json(keys::node_outcome(node), outcome).await
|
||||
}
|
||||
|
||||
pub async fn put_node_provider_used(
|
||||
&self,
|
||||
node: &NodeVisitRef<'_>,
|
||||
provider_used: &serde_json::Value,
|
||||
) -> Result<()> {
|
||||
self.put_json(keys::node_provider_used(node), provider_used)
|
||||
.await
|
||||
}
|
||||
|
||||
pub async fn put_node_diff(&self, node: &NodeVisitRef<'_>, diff: &str) -> Result<()> {
|
||||
self.put_text(keys::node_diff(node), diff).await;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub async fn put_node_script_invocation(
|
||||
&self,
|
||||
node: &NodeVisitRef<'_>,
|
||||
invocation: &serde_json::Value,
|
||||
) -> Result<()> {
|
||||
self.put_json(keys::node_script_invocation(node), invocation)
|
||||
.await
|
||||
}
|
||||
|
||||
pub async fn put_node_script_timing(
|
||||
&self,
|
||||
node: &NodeVisitRef<'_>,
|
||||
timing: &serde_json::Value,
|
||||
) -> Result<()> {
|
||||
self.put_json(keys::node_script_timing(node), timing).await
|
||||
}
|
||||
|
||||
pub async fn put_node_parallel_results(
|
||||
&self,
|
||||
node: &NodeVisitRef<'_>,
|
||||
results: &serde_json::Value,
|
||||
) -> Result<()> {
|
||||
self.put_json(keys::node_parallel_results(node), results)
|
||||
.await
|
||||
}
|
||||
|
||||
pub async fn put_node_stdout(&self, node: &NodeVisitRef<'_>, log: &str) -> Result<()> {
|
||||
self.put_text(keys::node_stdout(node), log).await;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub async fn put_node_stderr(&self, node: &NodeVisitRef<'_>, log: &str) -> Result<()> {
|
||||
self.put_text(keys::node_stderr(node), log).await;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub async fn get_node(&self, node: &NodeVisitRef<'_>) -> Result<NodeSnapshot> {
|
||||
let data = self.snapshot_data().await;
|
||||
self.build_node_snapshot_from_data(&data, node)
|
||||
}
|
||||
|
||||
pub async fn list_node_visits(&self, node_id: &str) -> Result<Vec<u32>> {
|
||||
let data = self.snapshot_data().await;
|
||||
let mut visits = BTreeSet::new();
|
||||
for key in data.keys() {
|
||||
let Some((current_node_id, visit, _)) = keys::parse_node_key(key) else {
|
||||
continue;
|
||||
};
|
||||
if current_node_id == node_id {
|
||||
visits.insert(visit);
|
||||
}
|
||||
}
|
||||
Ok(visits.into_iter().collect())
|
||||
}
|
||||
|
||||
pub async fn list_node_ids(&self) -> Result<Vec<String>> {
|
||||
Ok(self.list_node_ids_inner().await)
|
||||
}
|
||||
|
||||
pub async fn reset_for_rewind(&self) -> Result<()> {
|
||||
let mut data = self.data.lock().await;
|
||||
data.retain(|key, _| key == keys::init() || key.starts_with(keys::EVENTS_PREFIX));
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub async fn append_event(&self, payload: &EventPayload) -> Result<u32> {
|
||||
payload.validate(&self.run_id)?;
|
||||
|
||||
|
|
@ -445,25 +284,6 @@ impl InMemoryRunStore {
|
|||
Ok(Box::pin(UnboundedReceiverStream::new(receiver).map(Ok)))
|
||||
}
|
||||
|
||||
pub async fn put_retro_prompt(&self, text: &str) -> Result<()> {
|
||||
self.put_text(keys::retro_prompt().to_string(), text).await;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub async fn get_retro_prompt(&self) -> Result<Option<String>> {
|
||||
self.get_text(keys::retro_prompt()).await
|
||||
}
|
||||
|
||||
pub async fn put_retro_response(&self, text: &str) -> Result<()> {
|
||||
self.put_text(keys::retro_response().to_string(), text)
|
||||
.await;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub async fn get_retro_response(&self) -> Result<Option<String>> {
|
||||
self.get_text(keys::retro_response()).await
|
||||
}
|
||||
|
||||
pub async fn put_artifact_value(
|
||||
&self,
|
||||
artifact_id: &str,
|
||||
|
|
@ -538,25 +358,6 @@ fn matches_query(created_at: &DateTime<Utc>, query: &ListRunsQuery) -> bool {
|
|||
true
|
||||
}
|
||||
|
||||
fn read_json<T: DeserializeOwned>(
|
||||
data: &BTreeMap<String, Vec<u8>>,
|
||||
key: &str,
|
||||
) -> Result<Option<T>> {
|
||||
data.get(key)
|
||||
.map(|value| serde_json::from_slice(value))
|
||||
.transpose()
|
||||
.map_err(Into::into)
|
||||
}
|
||||
|
||||
fn read_text(data: &BTreeMap<String, Vec<u8>>, key: &str) -> Result<Option<String>> {
|
||||
data.get(key)
|
||||
.map(|value| {
|
||||
String::from_utf8(value.clone())
|
||||
.map_err(|err| StoreError::Other(format!("stored text is not valid UTF-8: {err}")))
|
||||
})
|
||||
.transpose()
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
|
@ -696,24 +497,6 @@ mod tests {
|
|||
}
|
||||
}
|
||||
|
||||
fn sample_node_status() -> NodeStatusRecord {
|
||||
NodeStatusRecord {
|
||||
status: StageStatus::PartialSuccess,
|
||||
notes: Some("captured output".to_string()),
|
||||
failure_reason: Some("minor lint".to_string()),
|
||||
timestamp: dt("2026-03-27T12:12:00Z"),
|
||||
}
|
||||
}
|
||||
|
||||
fn sample_node_outcome() -> NodeOutcomeRecord {
|
||||
fabro_types::Outcome {
|
||||
status: StageStatus::Success,
|
||||
notes: Some("all good".to_string()),
|
||||
files_touched: vec!["src/lib.rs".to_string()],
|
||||
..Default::default()
|
||||
}
|
||||
}
|
||||
|
||||
fn sample_pull_request() -> PullRequestRecord {
|
||||
PullRequestRecord {
|
||||
html_url: "https://github.com/fabro-sh/fabro/pull/123".to_string(),
|
||||
|
|
@ -767,12 +550,6 @@ mod tests {
|
|||
node_id: "code",
|
||||
visit: 2,
|
||||
};
|
||||
let node_status = sample_node_status();
|
||||
let node_outcome = sample_node_outcome();
|
||||
let provider_used = serde_json::json!({"provider": "openai", "model": "gpt-5.4"});
|
||||
let script_invocation = serde_json::json!({"command": "cargo test"});
|
||||
let script_timing = serde_json::json!({"duration_ms": 3210});
|
||||
let parallel_results = serde_json::json!([{"node_id": "lint", "status": "success"}]);
|
||||
let pull_request = sample_pull_request();
|
||||
|
||||
run.append_event(&event_payload(
|
||||
|
|
@ -852,27 +629,106 @@ mod tests {
|
|||
))
|
||||
.await
|
||||
.unwrap();
|
||||
run.put_node_prompt(&node, "Plan the fix").await.unwrap();
|
||||
run.put_node_response(&node, "Implemented").await.unwrap();
|
||||
run.put_node_status(&node, &node_status).await.unwrap();
|
||||
run.put_node_outcome(&node, &node_outcome).await.unwrap();
|
||||
run.put_node_provider_used(&node, &provider_used)
|
||||
.await
|
||||
.unwrap();
|
||||
run.put_node_diff(&node, "diff --git a/src/lib.rs b/src/lib.rs")
|
||||
.await
|
||||
.unwrap();
|
||||
run.put_node_script_invocation(&node, &script_invocation)
|
||||
.await
|
||||
.unwrap();
|
||||
run.put_node_script_timing(&node, &script_timing)
|
||||
.await
|
||||
.unwrap();
|
||||
run.put_node_parallel_results(&node, ¶llel_results)
|
||||
.await
|
||||
.unwrap();
|
||||
run.put_node_stdout(&node, "ok").await.unwrap();
|
||||
run.put_node_stderr(&node, "").await.unwrap();
|
||||
run.append_event(&event_payload(
|
||||
"run-1",
|
||||
"2026-03-27T12:00:08.1Z",
|
||||
"stage.prompt",
|
||||
Some("code"),
|
||||
serde_json::json!({
|
||||
"visit": 2,
|
||||
"text": "Plan the fix",
|
||||
"mode": "prompt",
|
||||
"provider": "openai",
|
||||
"model": "gpt-5.4"
|
||||
}),
|
||||
))
|
||||
.await
|
||||
.unwrap();
|
||||
run.append_event(&event_payload(
|
||||
"run-1",
|
||||
"2026-03-27T12:00:08.2Z",
|
||||
"command.started",
|
||||
Some("code"),
|
||||
serde_json::json!({
|
||||
"visit": 2,
|
||||
"command": "cargo test"
|
||||
}),
|
||||
))
|
||||
.await
|
||||
.unwrap();
|
||||
run.append_event(&event_payload(
|
||||
"run-1",
|
||||
"2026-03-27T12:00:08.3Z",
|
||||
"command.completed",
|
||||
Some("code"),
|
||||
serde_json::json!({
|
||||
"visit": 2,
|
||||
"stdout": "ok",
|
||||
"stderr": "",
|
||||
"exit_code": 0
|
||||
}),
|
||||
))
|
||||
.await
|
||||
.unwrap();
|
||||
run.append_event(&event_payload(
|
||||
"run-1",
|
||||
"2026-03-27T12:00:08.4Z",
|
||||
"checkpoint.completed",
|
||||
Some("code"),
|
||||
serde_json::json!({
|
||||
"status": "success",
|
||||
"ordinal": 2,
|
||||
"current_node": checkpoint.current_node,
|
||||
"completed_nodes": checkpoint.completed_nodes,
|
||||
"node_retries": checkpoint.node_retries,
|
||||
"context_values": checkpoint.context_values,
|
||||
"node_outcomes": checkpoint.node_outcomes,
|
||||
"next_node_id": checkpoint.next_node_id,
|
||||
"git_commit_sha": checkpoint.git_commit_sha,
|
||||
"node_visits": checkpoint.node_visits,
|
||||
"diff": "diff --git a/src/lib.rs b/src/lib.rs"
|
||||
}),
|
||||
))
|
||||
.await
|
||||
.unwrap();
|
||||
run.append_event(&event_payload(
|
||||
"run-1",
|
||||
"2026-03-27T12:00:08.5Z",
|
||||
"parallel.completed",
|
||||
Some("code"),
|
||||
serde_json::json!({
|
||||
"visit": 2,
|
||||
"results": [{"node_id": "lint", "status": "success"}]
|
||||
}),
|
||||
))
|
||||
.await
|
||||
.unwrap();
|
||||
run.append_event(&event_payload(
|
||||
"run-1",
|
||||
"2026-03-27T12:00:08.6Z",
|
||||
"stage.completed",
|
||||
Some("code"),
|
||||
serde_json::json!({
|
||||
"visit": 2,
|
||||
"status": "success",
|
||||
"notes": "all good",
|
||||
"response": "Implemented",
|
||||
"files_touched": ["src/lib.rs"]
|
||||
}),
|
||||
))
|
||||
.await
|
||||
.unwrap();
|
||||
run.append_event(&event_payload(
|
||||
"run-1",
|
||||
"2026-03-27T12:00:08.7Z",
|
||||
"retro.started",
|
||||
None,
|
||||
serde_json::json!({
|
||||
"prompt": "How did it go?"
|
||||
}),
|
||||
))
|
||||
.await
|
||||
.unwrap();
|
||||
run.append_event(&event_payload(
|
||||
"run-1",
|
||||
"2026-03-27T12:00:09Z",
|
||||
|
|
@ -917,8 +773,6 @@ mod tests {
|
|||
))
|
||||
.await
|
||||
.unwrap();
|
||||
run.put_retro_prompt("How did it go?").await.unwrap();
|
||||
run.put_retro_response("Smooth enough").await.unwrap();
|
||||
run.put_artifact_value("summary", &serde_json::json!({"done": true}))
|
||||
.await
|
||||
.unwrap();
|
||||
|
|
@ -957,14 +811,8 @@ mod tests {
|
|||
let stored_sandbox = state.sandbox.as_ref().unwrap();
|
||||
assert_eq!(stored_sandbox.provider, sandbox.provider);
|
||||
assert_eq!(stored_sandbox.working_directory, sandbox.working_directory);
|
||||
assert_eq!(
|
||||
run.get_retro_prompt().await.unwrap(),
|
||||
Some("How did it go?".to_string())
|
||||
);
|
||||
assert_eq!(
|
||||
run.get_retro_response().await.unwrap(),
|
||||
Some("Smooth enough".to_string())
|
||||
);
|
||||
assert_eq!(state.retro_prompt.as_deref(), Some("How did it go?"));
|
||||
assert_eq!(state.retro_response.as_deref(), Some("Smooth enough"));
|
||||
assert_eq!(
|
||||
run.get_artifact_value("summary").await.unwrap(),
|
||||
Some(serde_json::json!({"done": true}))
|
||||
|
|
@ -978,7 +826,26 @@ mod tests {
|
|||
Some("diff --git a/src/lib.rs b/src/lib.rs\n")
|
||||
);
|
||||
assert_eq!(state.pull_request, Some(pull_request.clone()));
|
||||
assert_eq!(run.list_node_ids().await.unwrap(), vec!["code".to_string()]);
|
||||
assert_eq!(state.list_node_ids(), vec!["code".to_string()]);
|
||||
let node_state = state
|
||||
.node(&node)
|
||||
.expect("node state should exist for code:2");
|
||||
assert_eq!(node_state.prompt.as_deref(), Some("Plan the fix"));
|
||||
assert_eq!(node_state.response.as_deref(), Some("Implemented"));
|
||||
assert_eq!(node_state.stdout.as_deref(), Some("ok"));
|
||||
assert_eq!(node_state.stderr.as_deref(), Some(""));
|
||||
assert_eq!(
|
||||
node_state.diff.as_deref(),
|
||||
Some("diff --git a/src/lib.rs b/src/lib.rs")
|
||||
);
|
||||
assert_eq!(
|
||||
node_state
|
||||
.provider_used
|
||||
.as_ref()
|
||||
.and_then(|v| v.get("provider"))
|
||||
.and_then(|v| v.as_str()),
|
||||
Some("openai")
|
||||
);
|
||||
assert_eq!(
|
||||
run.list_assets(&node).await.unwrap(),
|
||||
vec!["src/lib.rs".to_string()]
|
||||
|
|
@ -1358,9 +1225,6 @@ mod tests {
|
|||
node_id: "code",
|
||||
visit: 2,
|
||||
};
|
||||
run.put_node_prompt(&snapshot_node, "Plan the fix")
|
||||
.await
|
||||
.unwrap();
|
||||
run.put_asset(&snapshot_node, "src/lib.rs", b"fn main() {}")
|
||||
.await
|
||||
.unwrap();
|
||||
|
|
@ -1388,13 +1252,6 @@ mod tests {
|
|||
("code".to_string(), 2, "src/lib.rs".to_string())
|
||||
]
|
||||
);
|
||||
assert_eq!(
|
||||
run.list_node_ids().await.unwrap(),
|
||||
vec!["artifact-only".to_string(), "code".to_string()]
|
||||
);
|
||||
|
||||
let code_node = run.get_node(&snapshot_node).await.unwrap();
|
||||
assert_eq!(code_node.node_id, "code");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
|
|
@ -1517,33 +1374,6 @@ mod tests {
|
|||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn node_visit_storage_round_trips() {
|
||||
let store = InMemoryStore::default();
|
||||
let run = store
|
||||
.create_run(&test_run_id("run-1"), dt("2026-03-27T12:00:00Z"), None)
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
let first = NodeVisitRef {
|
||||
node_id: "code",
|
||||
visit: 1,
|
||||
};
|
||||
let second = NodeVisitRef {
|
||||
node_id: "code",
|
||||
visit: 2,
|
||||
};
|
||||
run.put_node_prompt(&first, "first").await.unwrap();
|
||||
run.put_node_prompt(&second, "second").await.unwrap();
|
||||
run.put_node_status(&second, &sample_node_status())
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
let node = run.get_node(&second).await.unwrap();
|
||||
assert_eq!(node.prompt, Some("second".to_string()));
|
||||
assert_eq!(run.list_node_visits("code").await.unwrap(), vec![1, 2]);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn list_runs_filters_dates_and_tolerates_missing_status() {
|
||||
let store = InMemoryStore::default();
|
||||
|
|
|
|||
|
|
@ -383,8 +383,7 @@ mod tests {
|
|||
|
||||
use bytes::Bytes;
|
||||
use fabro_types::{
|
||||
AttrValue, Graph, NodeStatusRecord, RunId, RunRecord, RunStatus, Settings, StageStatus,
|
||||
StatusReason, fixtures,
|
||||
AttrValue, Graph, RunId, RunRecord, RunStatus, Settings, StatusReason, fixtures,
|
||||
};
|
||||
use object_store::memory::InMemory;
|
||||
use slatedb::config::Settings as SlateSettings;
|
||||
|
|
@ -432,15 +431,6 @@ mod tests {
|
|||
}
|
||||
}
|
||||
|
||||
fn sample_node_status() -> NodeStatusRecord {
|
||||
NodeStatusRecord {
|
||||
status: StageStatus::Success,
|
||||
notes: Some("done".to_string()),
|
||||
failure_reason: None,
|
||||
timestamp: dt("2026-03-27T12:12:00Z"),
|
||||
}
|
||||
}
|
||||
|
||||
fn event_payload(
|
||||
run_id: &str,
|
||||
ts: &str,
|
||||
|
|
|
|||
|
|
@ -639,7 +639,7 @@ mod tests {
|
|||
.unwrap();
|
||||
run.append_event(&event(
|
||||
"checkpoint.completed",
|
||||
serde_json::json!({"diff": "diff --git a/story.txt b/story.txt", "ordinal": 1}),
|
||||
serde_json::json!({"diff": "diff --git a/story.txt b/story.txt", "ordinal": 1, "current_node": "work", "node_visits": {"work": 2}}),
|
||||
))
|
||||
.await
|
||||
.unwrap();
|
||||
|
|
|
|||
|
|
@ -719,14 +719,17 @@ mod tests {
|
|||
.unwrap();
|
||||
logger.flush().await;
|
||||
|
||||
let snapshot = run_store
|
||||
.get_node(&NodeVisitRef {
|
||||
let state = run_store.state().await.unwrap();
|
||||
let node_state = state
|
||||
.node(&NodeVisitRef {
|
||||
node_id: "step",
|
||||
visit: 1,
|
||||
})
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(snapshot.provider_used.unwrap()["provider"], "openai");
|
||||
assert_eq!(
|
||||
node_state.provider_used.as_ref().unwrap()["provider"],
|
||||
"openai"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
|
|
|
|||
|
|
@ -4,7 +4,6 @@ use std::time::Instant;
|
|||
|
||||
use async_trait::async_trait;
|
||||
use fabro_agent::{Sandbox, WorktreeOptions, WorktreeSandbox};
|
||||
use fabro_store::NodeVisitRef;
|
||||
use fabro_types::RunId;
|
||||
use tokio::sync::Semaphore;
|
||||
|
||||
|
|
@ -713,14 +712,14 @@ mod tests {
|
|||
.await
|
||||
.unwrap();
|
||||
|
||||
let snapshot = run_store
|
||||
.get_node(&NodeVisitRef {
|
||||
let state = run_store.state().await.unwrap();
|
||||
let node_state = state
|
||||
.node(&fabro_store::NodeVisitRef {
|
||||
node_id: "par",
|
||||
visit: 1,
|
||||
})
|
||||
.await
|
||||
.unwrap();
|
||||
let results = snapshot.parallel_results.unwrap();
|
||||
let results = node_state.parallel_results.as_ref().unwrap();
|
||||
assert!(results.is_array());
|
||||
assert_eq!(results.as_array().unwrap().len(), 2);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -377,14 +377,14 @@ mod tests {
|
|||
.unwrap();
|
||||
logger.flush().await;
|
||||
|
||||
let snapshot = run_store
|
||||
.get_node(&NodeVisitRef {
|
||||
let state = run_store.state().await.unwrap();
|
||||
let node_state = state
|
||||
.node(&NodeVisitRef {
|
||||
node_id: "classify",
|
||||
visit: 1,
|
||||
})
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(snapshot.provider_used.unwrap()["mode"], "prompt");
|
||||
assert_eq!(node_state.provider_used.as_ref().unwrap()["mode"], "prompt");
|
||||
}
|
||||
|
||||
struct OneShotCapturingBackend {
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue