use crate::agent_result;
use crate::agents::{AgentAdapter, ClaudeAgent};
use crate::git::hermetic_command;
use crate::monitor::{self, CloseRule};
use crate::phase_id::PhaseId;
use serde::{Deserialize, Serialize};
use std::io::{BufRead, BufReader, Write};
use std::path::{Path, PathBuf};
use std::process::Stdio;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::mpsc;
use std::time::{Duration, Instant};
use tracing::warn;
pub const TOKEN_PREFIX: &str = "DEVFLOW_DELIVERY_CANARY_";
const CAPTURE_FILE: &str = "delivery-canary.jsonl";
static TOKEN_SEQ: AtomicU64 = AtomicU64::new(0);
pub fn declare_token() -> String {
use std::hash::{Hash, Hasher};
let nanos = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_nanos())
.unwrap_or(0);
let seq = TOKEN_SEQ.fetch_add(1, Ordering::Relaxed);
let mut hasher = std::collections::hash_map::DefaultHasher::new();
nanos.hash(&mut hasher);
std::process::id().hash(&mut hasher);
seq.hash(&mut hasher);
format!("{TOKEN_PREFIX}{:016x}", hasher.finish())
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum CanaryOutcome {
Confirmed,
Absent,
Unverified(String),
}
pub trait CanaryLauncher {
fn run(&self, prompt: &str, capture: &Path) -> Result<(), String>;
}
pub fn canary_capture_path(capture_dir: &Path) -> PathBuf {
capture_dir.join(CAPTURE_FILE)
}
pub fn canary_prompt(token: &str) -> String {
format!(
"DevFlow startup check of Claude Code's background-task notification path. \
Do exactly the following and nothing else — do not read, create or modify any file, \
and do not run any command.\n\
\n\
1. Dispatch ONE background task whose entire job is to reply with the word `ok`.\n\
2. Wait for that task's completion notification to arrive. Do not finish before it does.\n\
3. In the turn that follows that notification, end your message with these two lines, \
each on its own line and exactly as written:\n\
\n\
{token}\n\
DEVFLOW_RESULT: {{\"status\":\"success\"}}\n\
\n\
The first line is a single-use token supplied by DevFlow. Reproduce it character for \
character; do not shorten, summarise, quote or comment on it."
)
}
pub fn run_delivery_canary<L: CanaryLauncher>(launcher: &L, capture_dir: &Path) -> CanaryOutcome {
let token = declare_token();
let capture = canary_capture_path(capture_dir);
if let Err(err) = crate::workflow::ensure_devflow_dir(capture_dir) {
return CanaryOutcome::Unverified(format!(
"could not prepare the canary capture directory {}: {err}",
capture_dir.display()
));
}
if let Err(reason) = launcher.run(&canary_prompt(&token), &capture) {
return CanaryOutcome::Unverified(reason);
}
let text = match std::fs::read(&capture) {
Ok(bytes) => String::from_utf8_lossy(&bytes).into_owned(),
Err(err) => {
return CanaryOutcome::Unverified(format!(
"the canary ran but its capture {} could not be read: {err}",
capture.display()
));
}
};
if agent_result::token_reported_in_capture(&text, &token) {
CanaryOutcome::Confirmed
} else {
CanaryOutcome::Absent
}
}
const CANARY_IDLE_SECS: u64 = 120;
const CANARY_DEADLINE_SECS: u64 = 300;
const CANARY_REAP_GRACE_SECS: u64 = 10;
const REAP_POLL: Duration = Duration::from_millis(100);
pub struct ClaudeCanaryLauncher {
pub workdir: PathBuf,
}
impl CanaryLauncher for ClaudeCanaryLauncher {
fn run(&self, prompt: &str, capture: &Path) -> Result<(), String> {
let (program, args) = ClaudeAgent.exec_command(PhaseId::new(0), prompt, &[]);
let mut capture_file = std::fs::File::create(capture).map_err(|err| {
format!(
"could not create the canary capture {}: {err}",
capture.display()
)
})?;
let mut child = hermetic_command(program, &self.workdir)
.args(&args)
.stdin(Stdio::piped())
.stdout(Stdio::piped())
.stderr(Stdio::null())
.spawn()
.map_err(|err| format!("could not run `{program}`: {err}"))?;
let mut child_stdin = child
.stdin
.take()
.ok_or_else(|| "the canary child exposed no stdin pipe".to_string())?;
let child_stdout = child
.stdout
.take()
.ok_or_else(|| "the canary child exposed no stdout pipe".to_string())?;
let (close_tx, close_rx) = mpsc::channel::<()>();
let turn = monitor::user_turn_line(prompt);
let writer = std::thread::spawn(move || {
let wrote = child_stdin
.write_all(turn.as_bytes())
.and_then(|()| child_stdin.write_all(b"\n"))
.and_then(|()| child_stdin.flush());
if let Err(err) = wrote {
warn!("could not write the canary's user turn to the child's stdin: {err}");
return;
}
let _ = close_rx.recv();
drop(child_stdin);
});
let (line_tx, line_rx) = mpsc::channel::<String>();
let reader = std::thread::spawn(move || {
for line in BufReader::new(child_stdout).lines() {
let Ok(line) = line else {
break;
};
if let Err(err) = writeln!(capture_file, "{line}") {
warn!("could not append to the canary capture: {err}");
}
let _ = capture_file.flush();
if line_tx.send(line).is_err() {
break;
}
}
});
let mut rule = CloseRule::default();
let mut close_signalled = false;
let idle = Duration::from_secs(CANARY_IDLE_SECS);
let deadline = Instant::now() + Duration::from_secs(CANARY_DEADLINE_SECS);
loop {
let remaining = deadline.saturating_duration_since(Instant::now());
if remaining.is_zero() {
break;
}
match line_rx.recv_timeout(idle.min(remaining)) {
Ok(line) => {
if close_signalled {
continue;
}
rule.observe(&line);
if rule.should_close() {
let _ = close_tx.send(());
close_signalled = true;
}
}
Err(mpsc::RecvTimeoutError::Disconnected | mpsc::RecvTimeoutError::Timeout) => {
break;
}
}
}
drop(close_tx);
reap(&mut child);
let _ = writer.join();
let _ = reader.join();
Ok(())
}
}
fn reap(child: &mut std::process::Child) {
let deadline = Instant::now() + Duration::from_secs(CANARY_REAP_GRACE_SECS);
loop {
match child.try_wait() {
Ok(Some(_)) => return,
Ok(None) => {}
Err(err) => {
warn!("could not poll the canary child: {err}");
return;
}
}
if Instant::now() >= deadline {
break;
}
std::thread::sleep(REAP_POLL);
}
let _ = child.kill();
let _ = child.wait();
}
pub fn claude_cli_version() -> Option<String> {
let output = std::process::Command::new("claude")
.arg("--version")
.output()
.ok()?;
if !output.status.success() {
return None;
}
let version = String::from_utf8_lossy(&output.stdout).trim().to_string();
(!version.is_empty()).then_some(version)
}
#[cfg(test)]
mod tests {
use super::*;
const INIT_LINE: &str = r#"{"type":"system","subtype":"init","cwd":"/tmp/work","session_id":"s-1","claude_code_version":"2.1.220","uuid":"u-init"}"#;
fn echoed_prompt_line(prompt: &str) -> String {
serde_json::json!({
"type": "user",
"message": { "role": "user", "content": prompt },
"session_id": "s-1",
"uuid": "u-echo",
})
.to_string()
}
fn top_level_result_line(text: &str) -> String {
serde_json::json!({
"type": "result",
"subtype": "success",
"is_error": false,
"num_turns": 3,
"stop_reason": "end_turn",
"session_id": "s-1",
"uuid": "u-res",
"result": text,
})
.to_string()
}
fn subagent_result_line(text: &str) -> String {
serde_json::json!({
"type": "result",
"subtype": "success",
"is_error": false,
"session_id": "s-1",
"uuid": "u-sub",
"parent_tool_use_id": "toolu_01CanarySubagent",
"result": text,
})
.to_string()
}
fn token_in(prompt: &str) -> String {
let start = prompt
.find(TOKEN_PREFIX)
.expect("the canary prompt must carry the declared token");
let rest = &prompt[start + TOKEN_PREFIX.len()..];
let suffix: String = rest.chars().take_while(char::is_ascii_hexdigit).collect();
assert!(
!suffix.is_empty(),
"the token in the prompt must have a body after its prefix"
);
format!("{TOKEN_PREFIX}{suffix}")
}
struct CannedLauncher<F: Fn(&str) -> Vec<String>> {
lines: F,
}
impl<F: Fn(&str) -> Vec<String>> CanaryLauncher for CannedLauncher<F> {
fn run(&self, prompt: &str, capture: &Path) -> Result<(), String> {
let token = token_in(prompt);
let body = (self.lines)(&token).join("\n");
std::fs::write(capture, format!("{body}\n")).map_err(|err| err.to_string())?;
Ok(())
}
}
struct FailingLauncher(&'static str);
impl CanaryLauncher for FailingLauncher {
fn run(&self, _prompt: &str, _capture: &Path) -> Result<(), String> {
Err(self.0.to_string())
}
}
#[test]
fn canary_confirmed_when_token_returns_in_a_top_level_result() {
let dir = tempfile::tempdir().unwrap();
let launcher = CannedLauncher {
lines: |token| {
vec![
INIT_LINE.to_string(),
top_level_result_line(&format!(
"The background task finished.\n{token}\nDEVFLOW_RESULT: {{\"status\":\"success\"}}"
)),
]
},
};
let outcome = run_delivery_canary(&launcher, dir.path());
assert_eq!(
outcome,
CanaryOutcome::Confirmed,
"a token inside a top-level result is the whole point of the guard"
);
}
#[test]
fn canary_absent_when_token_appears_only_as_a_prompt_echo() {
let dir = tempfile::tempdir().unwrap();
let launcher = CannedLauncher {
lines: |token| {
vec![
INIT_LINE.to_string(),
echoed_prompt_line(&canary_prompt(token)),
top_level_result_line("I could not dispatch a background task."),
]
},
};
let outcome = run_delivery_canary(&launcher, dir.path());
let capture = std::fs::read_to_string(canary_capture_path(dir.path())).unwrap();
let carrying: Vec<serde_json::Value> = capture
.lines()
.filter(|line| line.contains(TOKEN_PREFIX))
.map(|line| serde_json::from_str(line).expect("fixture lines are JSON"))
.collect();
assert!(
!carrying.is_empty(),
"fixture must actually contain the echoed token"
);
assert!(
carrying.iter().all(|event| event["type"] == "user"),
"fixture must place the echoed token ONLY inside a `user` event — \
if any result event carries it, this test is not exercising the echo case"
);
assert_eq!(
outcome,
CanaryOutcome::Absent,
"an echoed token must never satisfy the guard (30-05's false positive)"
);
}
#[test]
fn canary_absent_when_token_appears_only_in_a_non_top_level_event() {
let dir = tempfile::tempdir().unwrap();
let launcher = CannedLauncher {
lines: |token| {
vec![
INIT_LINE.to_string(),
subagent_result_line(&format!("child reporting: {token}")),
top_level_result_line("Done."),
]
},
};
let outcome = run_delivery_canary(&launcher, dir.path());
let capture = std::fs::read_to_string(canary_capture_path(dir.path())).unwrap();
assert!(
capture.contains(TOKEN_PREFIX),
"fixture must actually contain the token inside the subagent result"
);
assert!(
capture.contains("parent_tool_use_id"),
"fixture must actually mark that result as subagent-authored"
);
assert_eq!(
outcome,
CanaryOutcome::Absent,
"a subagent-authored result must not certify orchestrator-level delivery"
);
}
#[test]
fn canary_unverified_when_the_launcher_fails() {
let dir = tempfile::tempdir().unwrap();
let outcome = run_delivery_canary(
&FailingLauncher("could not run `claude`: No such file or directory (os error 2)"),
dir.path(),
);
match outcome {
CanaryOutcome::Unverified(reason) => {
assert!(
reason.contains("No such file or directory"),
"the reason the guard could not run must survive into the outcome, \
got: {reason}"
);
}
other => panic!("a launcher failure must be Unverified, not {other:?}"),
}
}
#[test]
fn declared_tokens_differ_between_runs() {
let first = declare_token();
let second = declare_token();
assert_ne!(
first, second,
"each canary run must declare its own token, or a stale capture could satisfy it"
);
assert!(
first.starts_with(TOKEN_PREFIX) && second.starts_with(TOKEN_PREFIX),
"both tokens must carry the greppable prefix"
);
}
}