use std::path::Path;
use keel_journal::{FlowId, SqliteJournal, SystemClock};
use serde::Serialize;
use crate::render::to_json;
use crate::{EXIT_OK, Rendered, evidence, exec, flows};
#[derive(Debug, Serialize)]
struct ForceReport {
armed: bool,
flow_id: String,
}
pub fn run(project: &Path, flow: &str) -> (Option<Rendered>, i32) {
let journal_path = evidence::resolved_journal(project).path;
if !journal_path.exists() {
return soft_pair(
"no journal yet (.keel/journal.db). Run a flow first with `keel run` or `keel exec`.",
);
}
let resolved = {
let conn = match flows::open_ro(&journal_path) {
Ok(c) => c,
Err(e) => return soft_pair(&e),
};
match flows::resolve_flow(&conn, flow) {
Ok(r) => r,
Err(e) => return soft_pair(&e),
}
};
match resolved.status.as_str() {
"completed" => {
return soft_pair(&format!(
"flow {id} is already completed; there is nothing to force \u{2014} a completed \
flow's command never re-runs (KEEL-E033 only gates a running/failed \
re-dispatch). Inspect it with `keel replay {id}` or `keel trace {id}`.",
id = resolved.flow_id
));
}
"dead" => {
return soft_pair(&format!(
"flow {id} is dead (KEEL-E032); forcing the KEEL-E033 side-effect gate cannot \
revive it \u{2014} a dead flow is never re-dispatched. See `keel explain \
KEEL-E032`, then start a new flow with a fresh identity instead.",
id = resolved.flow_id
));
}
_ => {}
}
let journal = match SqliteJournal::open(&journal_path, SystemClock) {
Ok(j) => j,
Err(e) => {
return soft_pair(&format!(
"could not open the journal at {} read-write: {e}",
journal_path.display()
));
}
};
let flow_id = FlowId::new(resolved.flow_id.as_str());
if let Err(e) = exec::request_force_override(&journal, &flow_id) {
return soft_pair(&format!(
"could not arm the force override for flow {}: {e}",
resolved.flow_id
));
}
let human = format!(
"keel \u{25b8} flows force: {} armed \u{2014} the next re-dispatch that would otherwise be \
KEEL-E033-refused (its declared side-effect files changed) will proceed once, then this \
one-shot clears itself.",
resolved.flow_id
);
let report = ForceReport {
armed: true,
flow_id: resolved.flow_id,
};
(Some(Rendered::ok(human, to_json(&report))), EXIT_OK)
}
fn soft_pair(message: &str) -> (Option<Rendered>, i32) {
let r = flows::soft_error(message);
let code = r.exit;
(Some(r), code)
}
#[cfg(test)]
mod tests {
use super::*;
use rusqlite::{Connection, params};
const T0: i64 = 1_783_728_000_000;
fn project_with_fixtures(fixtures: &[&str]) -> (tempfile::TempDir, std::path::PathBuf) {
let root = Path::new(env!("CARGO_MANIFEST_DIR")).join("../..");
let dir = tempfile::TempDir::new().unwrap();
let keel = dir.path().join(".keel");
std::fs::create_dir_all(&keel).unwrap();
let conn = Connection::open(keel.join("journal.db")).unwrap();
let schema = std::fs::read_to_string(root.join("contracts/journal.sql")).unwrap();
conn.execute_batch(&schema).unwrap();
for f in fixtures {
let sql =
std::fs::read_to_string(root.join("conformance/fixtures/journal").join(f)).unwrap();
conn.execute_batch(&sql).unwrap();
}
let project = dir.path().to_path_buf();
(dir, project)
}
fn seed_failed_cmd_flow(project: &Path, flow_id: &str) {
let conn = Connection::open(project.join(".keel/journal.db")).unwrap();
conn.execute(
"INSERT INTO flows (flow_id, entrypoint, args_hash, status, created_at, updated_at) \
VALUES (?1, 'cmd:trade', 'ah', 'failed', ?2, ?2)",
params![flow_id, T0],
)
.unwrap();
}
#[test]
fn missing_journal_is_a_soft_error() {
let dir = tempfile::TempDir::new().unwrap();
let (r, code) = run(dir.path(), "anything");
assert_eq!(code, crate::EXIT_FAILURE);
assert!(r.unwrap().human.contains("no journal yet"));
}
#[test]
fn unknown_flow_is_a_clean_soft_error_not_a_panic() {
let (_d, project) = project_with_fixtures(&["completed-flow.sql"]);
let (r, code) = run(&project, "does-not-exist");
assert_eq!(code, crate::EXIT_FAILURE);
assert!(r.unwrap().human.contains("no flow matches"));
}
#[test]
fn ambiguous_flow_is_a_soft_error() {
let (_d, project) = project_with_fixtures(&["completed-flow.sql", "dead-flow.sql"]);
let (r, code) = run(&project, "01JZWY0A");
assert_eq!(code, crate::EXIT_FAILURE);
assert!(r.unwrap().human.contains("matches"));
}
#[test]
fn completed_flow_is_refused_as_nothing_to_force() {
let (_d, project) = project_with_fixtures(&["completed-flow.sql"]);
let (r, code) = run(&project, "01JZWY0A0000000000000001");
assert_eq!(code, crate::EXIT_FAILURE);
let human = r.unwrap().human;
assert!(human.contains("already completed"));
assert!(human.contains("nothing to force"));
}
#[test]
fn dead_flow_is_refused_with_e032() {
let (_d, project) = project_with_fixtures(&["dead-flow.sql"]);
let (r, code) = run(&project, "01JZWY0A0000000000000003");
assert_eq!(code, crate::EXIT_FAILURE);
let human = r.unwrap().human;
assert!(human.contains("KEEL-E032"));
assert!(human.contains("dead"));
}
#[test]
fn arming_a_failed_flow_records_the_requested_force_marker() {
let (_d, project) = project_with_fixtures(&[]);
seed_failed_cmd_flow(&project, "01FORCEME");
let (r, code) = run(&project, "01FORCEME");
assert_eq!(code, EXIT_OK);
let rendered = r.unwrap();
assert!(rendered.human.contains("01FORCEME"));
assert!(rendered.human.contains("armed"));
assert_eq!(rendered.json["armed"], true);
assert_eq!(rendered.json["flow_id"], "01FORCEME");
let conn = Connection::open(project.join(".keel/journal.db")).unwrap();
let (step_key, kind, outcome, payload_len): (String, String, String, i64) = conn
.query_row(
"SELECT step_key, kind, outcome, length(payload) FROM steps \
WHERE flow_id = '01FORCEME' AND seq = 500002",
[],
|row| Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?)),
)
.expect("the FORCE marker row exists");
assert_eq!(step_key, "cmd:force");
assert_eq!(kind, "marker");
assert_eq!(outcome, "ok");
assert!(payload_len > 0, "the armed marker carries a payload");
}
#[test]
fn substring_resolution_arms_the_uniquely_matching_flow() {
let (_d, project) = project_with_fixtures(&[]);
seed_failed_cmd_flow(&project, "01UNIQUEABCDEF");
let (r, code) = run(&project, "UNIQUEABC");
assert_eq!(code, EXIT_OK);
assert_eq!(r.unwrap().json["flow_id"], "01UNIQUEABCDEF");
}
}