use std::path::Path;
use serde::Serialize;
use toml_edit::Value;
use crate::diff::{ChangeHunk, PolicyOp, PolicyPath, propose};
use crate::render::to_json;
use crate::{EXIT_USAGE, Rendered, evidence};
#[derive(Debug, Serialize)]
struct AddReport {
added: bool,
already_designated: bool,
changes: Vec<ChangeHunk>,
entrypoint: String,
patch: String,
written: bool,
}
pub fn run(project: &Path, entrypoint: &str, diff_only: bool) -> Rendered {
let full = match normalize_entrypoint(entrypoint) {
Ok(f) => f,
Err(e) => return usage_error(&e),
};
let toml_path = evidence::keel_toml(project);
let current = if toml_path.exists() {
match std::fs::read_to_string(&toml_path) {
Ok(t) => Some(t),
Err(e) => return usage_error(&format!("could not read {}: {e}", toml_path.display())),
}
} else {
None
};
let existing = match existing_entrypoints(current.as_deref()) {
Ok(list) => list,
Err(e) => return usage_error(&e),
};
if existing.iter().any(|e| e == &full) {
let human = format!(
"keel \u{25b8} {full} is already designated in [flows] entrypoints \u{2014} nothing to do."
);
return Rendered::ok(
human,
to_json(&AddReport {
added: false,
already_designated: true,
changes: Vec::new(),
entrypoint: full,
patch: String::new(),
written: false,
}),
);
}
let mut array = toml_edit::Array::new();
for e in &existing {
array.push(e.as_str());
}
array.push(full.as_str());
let ops = [PolicyOp::Set {
path: PolicyPath::new(["flows", "entrypoints"]),
value: Value::Array(array),
}];
let proposal = match propose(current.as_deref(), &ops) {
Ok(p) => p,
Err(e) => return usage_error(&e.to_string()),
};
let written = !diff_only;
if written && let Err(e) = std::fs::write(&toml_path, &proposal.new_text) {
return usage_error(&format!("could not write {}: {e}", toml_path.display()));
}
let human = if diff_only {
format!(
"keel \u{25b8} would designate {full} as a durable flow (keel.toml not written).\n\
\napply with `git apply` (or `patch -p1`):\n\n{}",
proposal.patch
)
} else {
format!(
"keel \u{25b8} designated {full} as a durable flow in keel.toml.\n \
Run it with `keel run <script>`; kill it mid-run and the next `keel run` resumes.\n \
Inspect with `keel flows` / `keel trace <flow>`."
)
};
Rendered::ok(
human,
to_json(&AddReport {
added: true,
already_designated: false,
changes: proposal.changes,
entrypoint: full,
patch: proposal.patch,
written,
}),
)
}
fn normalize_entrypoint(raw: &str) -> Result<String, String> {
let raw = raw.trim();
if raw.is_empty() {
return Err(no_parse(raw, "it is empty"));
}
if raw.chars().any(char::is_whitespace) {
return Err(no_parse(raw, "it contains whitespace"));
}
for prefix in ["py:", "rs:", "ts:"] {
if let Some(rest) = raw.strip_prefix(prefix) {
if rest.is_empty() {
return Err(no_parse(raw, "nothing follows the language prefix"));
}
return Ok(raw.to_owned());
}
}
if is_python_shape(raw) {
return Ok(format!("py:{raw}"));
}
if is_ts_shape(raw) {
return Ok(format!("ts:{raw}"));
}
Err(no_parse(
raw,
"it is neither `module.path:function` (Python) nor `path/file.ts#function` (JS/TS)",
))
}
fn no_parse(raw: &str, why: &str) -> String {
format!(
"{raw:?} is not a flow entrypoint ref: {why}. \
Use `py:module.path:function`, `ts:path/file.ts#function`, or a bare form \
from `keel flows suggest` (e.g. `pipeline.ingest:main`)."
)
}
fn is_python_shape(s: &str) -> bool {
let Some((module, function)) = s.rsplit_once(':') else {
return false;
};
!module.is_empty()
&& module.split('.').all(is_py_ident)
&& is_py_ident(function)
&& !s.contains('#')
&& !s.contains('/')
}
fn is_py_ident(s: &str) -> bool {
let mut chars = s.chars();
chars
.next()
.is_some_and(|c| c.is_ascii_alphabetic() || c == '_')
&& chars.all(|c| c.is_ascii_alphanumeric() || c == '_')
}
fn is_ts_shape(s: &str) -> bool {
let Some((file, function)) = s.rsplit_once('#') else {
return false;
};
let has_js_ext = [".js", ".mjs", ".cjs", ".ts", ".mts", ".cts", ".jsx", ".tsx"]
.iter()
.any(|ext| file.ends_with(ext));
has_js_ext && !function.is_empty()
}
pub(crate) fn existing_entrypoints(current: Option<&str>) -> Result<Vec<String>, String> {
let Some(text) = current else {
return Ok(Vec::new());
};
let value: toml::Value = text
.parse()
.map_err(|e: toml::de::Error| format!("keel.toml is not valid TOML: {e}"))?;
let Some(flows) = value.get("flows") else {
return Ok(Vec::new());
};
let Some(flows) = flows.as_table() else {
return Err(
"keel.toml's `flows` key is not a table; fix it before adding entrypoints \
(expected `[flows]` with `entrypoints = [\"py:…\"]`)."
.to_owned(),
);
};
let Some(entrypoints) = flows.get("entrypoints") else {
return Ok(Vec::new());
};
let Some(list) = entrypoints.as_array() else {
return Err(
"keel.toml's `flows.entrypoints` is not an array; fix it before adding entrypoints \
(expected `entrypoints = [\"py:…\"]`)."
.to_owned(),
);
};
list.iter()
.map(|v| {
v.as_str().map(str::to_owned).ok_or_else(|| {
"keel.toml's `flows.entrypoints` contains a non-string entry; fix it before \
adding entrypoints."
.to_owned()
})
})
.collect()
}
fn usage_error(message: &str) -> Rendered {
#[derive(Serialize)]
struct ErrReport<'a> {
code: &'static str,
error: &'a str,
}
Rendered {
human: format!("keel \u{25b8} KEEL-E001: {message}"),
json: to_json(&ErrReport {
code: "KEEL-E001",
error: message,
}),
exit: EXIT_USAGE,
to_stderr: true,
}
}
#[cfg(test)]
mod tests {
use super::*;
fn project() -> tempfile::TempDir {
tempfile::TempDir::new().unwrap()
}
fn read_toml(dir: &tempfile::TempDir) -> String {
std::fs::read_to_string(dir.path().join("keel.toml")).unwrap()
}
#[test]
fn bare_python_shape_infers_py() {
assert_eq!(
normalize_entrypoint("pipeline.ingest:main").unwrap(),
"py:pipeline.ingest:main"
);
assert_eq!(normalize_entrypoint("app:run").unwrap(), "py:app:run");
}
#[test]
fn bare_ts_shape_infers_ts() {
assert_eq!(
normalize_entrypoint("jobs/nightly.ts#run").unwrap(),
"ts:jobs/nightly.ts#run"
);
assert_eq!(
normalize_entrypoint("app.mjs#fetchData").unwrap(),
"ts:app.mjs#fetchData"
);
}
#[test]
fn explicit_prefixes_pass_through() {
for full in ["py:pipeline.ingest:main", "ts:jobs/n.ts#run", "rs:crate::x"] {
assert_eq!(normalize_entrypoint(full).unwrap(), full);
}
}
#[test]
fn garbage_refs_are_rejected_with_guidance() {
for bad in ["", " ", "just-a-name", "py:", "has space:fn", "file.txt#x"] {
let err = normalize_entrypoint(bad).unwrap_err();
assert!(err.contains("keel flows suggest"), "{err}");
}
}
#[test]
fn creates_keel_toml_and_flows_table_when_absent() {
let dir = project();
let r = run(dir.path(), "pipeline.ingest:main", false);
assert_eq!(r.exit, crate::EXIT_OK);
assert_eq!(r.json["added"], true);
assert_eq!(r.json["written"], true);
assert_eq!(r.json["entrypoint"], "py:pipeline.ingest:main");
let text = read_toml(&dir);
assert_eq!(
text,
"[flows]\nentrypoints = [\"py:pipeline.ingest:main\"]\n"
);
let patch = r.json["patch"].as_str().unwrap();
assert!(
patch.starts_with("--- /dev/null\n+++ b/keel.toml\n"),
"{patch}"
);
assert_eq!(crate::diff::apply_unified("", patch).unwrap(), text);
}
#[test]
fn appends_to_existing_policy_preserving_user_bytes() {
let dir = project();
let existing = "\
# my tuning \u{2014} keep me
[target.\"api.example.com\"] # seen in: app.py:4
timeout = \"9s\" # tuned by us
";
std::fs::write(dir.path().join("keel.toml"), existing).unwrap();
let r = run(dir.path(), "py:pipeline.ingest:main", false);
assert_eq!(r.exit, crate::EXIT_OK);
let text = read_toml(&dir);
assert!(
text.starts_with(existing),
"untouched bytes survive: {text}"
);
assert!(text.contains("[flows]\nentrypoints = [\"py:pipeline.ingest:main\"]\n"));
let patch = r.json["patch"].as_str().unwrap();
assert_eq!(crate::diff::apply_unified(existing, patch).unwrap(), text);
assert_eq!(r.json["changes"][0]["path"], "flows.entrypoints");
}
#[test]
fn appends_to_an_existing_entrypoints_array_in_order() {
let dir = project();
std::fs::write(
dir.path().join("keel.toml"),
"[flows]\nentrypoints = [\"py:a.b:c\"]\n",
)
.unwrap();
let r = run(dir.path(), "jobs/nightly.ts#run", false);
assert_eq!(r.exit, crate::EXIT_OK);
assert_eq!(
read_toml(&dir),
"[flows]\nentrypoints = [\"py:a.b:c\", \"ts:jobs/nightly.ts#run\"]\n"
);
}
#[test]
fn rerun_is_an_exact_noop() {
let dir = project();
let r1 = run(dir.path(), "pipeline.ingest:main", false);
assert_eq!(r1.json["added"], true);
let before = read_toml(&dir);
for spelling in ["pipeline.ingest:main", "py:pipeline.ingest:main"] {
let r = run(dir.path(), spelling, false);
assert_eq!(r.exit, crate::EXIT_OK);
assert_eq!(r.json["added"], false);
assert_eq!(r.json["already_designated"], true);
assert_eq!(r.json["written"], false);
assert_eq!(r.json["patch"], "");
}
assert_eq!(read_toml(&dir), before, "file byte-identical after re-runs");
}
#[test]
fn diff_flag_previews_without_writing() {
let dir = project();
let r = run(dir.path(), "pipeline.ingest:main", true);
assert_eq!(r.exit, crate::EXIT_OK);
assert_eq!(r.json["written"], false);
assert!(
!dir.path().join("keel.toml").exists(),
"--diff never writes"
);
assert!(r.human.contains("apply with `git apply`"));
assert!(!r.json["patch"].as_str().unwrap().is_empty());
}
#[test]
fn invalid_toml_is_a_usage_error() {
let dir = project();
std::fs::write(dir.path().join("keel.toml"), "not [valid\n").unwrap();
let r = run(dir.path(), "pipeline.ingest:main", false);
assert_eq!(r.exit, EXIT_USAGE);
assert!(r.to_stderr);
assert!(r.human.contains("KEEL-E001"));
assert!(r.human.contains("not valid TOML"));
}
#[test]
fn non_array_entrypoints_is_refused_not_overwritten() {
let dir = project();
let original = "[flows]\nentrypoints = \"py:a:b\"\n";
std::fs::write(dir.path().join("keel.toml"), original).unwrap();
let r = run(dir.path(), "pipeline.ingest:main", false);
assert_eq!(r.exit, EXIT_USAGE);
assert!(r.human.contains("not an array"));
assert_eq!(read_toml(&dir), original, "nothing was destroyed");
}
#[test]
fn invalid_ref_is_a_usage_error_before_any_io() {
let dir = project();
let r = run(dir.path(), "not a ref", false);
assert_eq!(r.exit, EXIT_USAGE);
assert_eq!(r.json["code"], "KEEL-E001");
assert!(!dir.path().join("keel.toml").exists());
}
}