//! Reviewer process harness, model opposition, async queue, and verdict execution.
use std::{
fs,
io::{self, Read, Write},
path::{Path, PathBuf},
process::{Command, ExitCode, Stdio},
sync::{
Arc,
atomic::{AtomicBool, AtomicU64, Ordering},
},
time::{Duration, Instant, SystemTime, UNIX_EPOCH},
};
use anyhow::Result;
use serde::{Deserialize, Serialize};
use thiserror::Error;
use crate::{
claim::{Claim, EvidenceRef},
cli::{self, Agent, ReviewScope, ReviewerHarness},
config::{self, Effort},
ledger::{
LedgerEntry, LedgerError, LedgerStore, MemorySkillClassification,
MemorySkillClassificationKind, ReviewerConfig, StructuredFinding, Verdict,
},
provenance::{self, EntireCheckpointRef},
surface,
time::unix_now,
};
pub const REVIEW_QUEUE_FILE: &str = "review-queue.jsonl";
pub const REVIEW_RUNS_DIR: &str = "runs";
const MAX_INLINE_DIFF_FILES: usize = 2;
const MAX_INLINE_DIFF_BYTES: usize = 256 * 1024;
const MAX_UNTRACKED_FILE_BYTES: u64 = 16 * 1024;
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct ReviewRequest {
pub watched_agent: Agent,
pub watched_model: String,
pub reviewer_harness: ReviewerHarness,
pub reviewer_model: String,
pub reviewer_effort: Effort,
pub allow_same_model: bool,
pub prompt: String,
}
impl ReviewRequest {
pub fn new(
watched_agent: Agent,
watched_model: impl Into<String>,
reviewer_harness: ReviewerHarness,
reviewer_model: impl Into<String>,
allow_same_model: bool,
prompt: impl Into<String>,
) -> Self {
Self {
watched_agent,
watched_model: watched_model.into(),
reviewer_harness,
reviewer_model: reviewer_model.into(),
reviewer_effort: Effort::highest(),
allow_same_model,
prompt: prompt.into(),
}
}
pub fn with_effort(mut self, effort: Effort) -> Self {
self.reviewer_effort = effort;
self
}
}
/// Resolved reviewer selection shared by `review` and `watch`. The reviewer is
/// chosen from the writer harness's adversarial pair; explicit CLI values win.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct ReviewSelection {
pub watched_agent: Agent,
pub watched_model: String,
pub reviewer_harness: ReviewerHarness,
pub reviewer_model: String,
pub reviewer_effort: Effort,
pub allow_same_model: bool,
pub strict: Option<StrictReviewConfig>,
}
impl ReviewSelection {
/// Resolve the reviewer for a writer harness from its adversarial pair,
/// applying any explicit CLI overrides. `strict` is set separately.
#[allow(clippy::too_many_arguments)]
pub fn resolve(
watched_agent: Option<Agent>,
watched_model: Option<String>,
reviewer_harness: Option<ReviewerHarness>,
reviewer_model: Option<String>,
reviewer_effort: Option<Effort>,
allow_same_model: bool,
config: &config::TruthMirrorConfig,
) -> Result<Self, ReviewerError> {
let watched_agent = match watched_agent {
Some(agent) => agent,
None => agent_from_slug(&config.default_writer)?,
};
let writer_slug = surface::agent_slug(watched_agent);
let pair = config.pair_for(writer_slug);
let harness_from_cli = reviewer_harness.is_some();
let reviewer_harness = match reviewer_harness {
Some(harness) => harness,
None => {
let slug = pair
.map(|pair| pair.reviewer.harness.as_str())
.ok_or_else(|| ReviewerError::NoPairForWriter {
writer: writer_slug.to_owned(),
})?;
harness_from_slug(slug)?
}
};
let reviewer_model = match reviewer_model {
Some(model) => model,
None => {
let pair = pair.ok_or_else(|| ReviewerError::NoPairForWriter {
writer: writer_slug.to_owned(),
})?;
// Don't pair a CLI-overridden harness with a model string meant for
// a different harness — that silently reviews with the wrong tool.
if harness_from_cli
&& !pair
.reviewer
.harness
.eq_ignore_ascii_case(harness_slug(reviewer_harness))
{
return Err(ReviewerError::OverrideNeedsModel {
role: "reviewer".to_owned(),
harness: harness_slug(reviewer_harness).to_owned(),
});
}
pair.reviewer.model.clone()
}
};
let reviewer_effort = reviewer_effort
.or_else(|| pair.map(|pair| pair.reviewer.effort))
.unwrap_or_else(Effort::highest);
Ok(Self {
watched_agent,
watched_model: watched_model.unwrap_or_default(),
reviewer_harness,
reviewer_model,
reviewer_effort,
// Either the CLI flag or config may waive model opposition.
allow_same_model: allow_same_model || config.allow_same_model,
strict: None,
})
}
/// Resolve the second-pass arbiter for the writer, preferring CLI overrides,
/// then the writer pair's `arbiter`.
pub fn resolve_arbiter(
watched_agent: Agent,
arbiter_harness: Option<ReviewerHarness>,
arbiter_model: Option<String>,
arbiter_effort: Option<Effort>,
config: &config::TruthMirrorConfig,
) -> Result<StrictReviewConfig, ReviewerError> {
let pair_arbiter = config
.pair_for(surface::agent_slug(watched_agent))
.and_then(|pair| pair.arbiter.clone());
let harness_from_cli = arbiter_harness.is_some();
let harness = match arbiter_harness {
Some(harness) => harness,
None => {
let slug = pair_arbiter
.as_ref()
.map(|arbiter| arbiter.harness.as_str())
.ok_or(ReviewerError::MissingArbiter)?;
harness_from_slug(slug)?
}
};
let model = match arbiter_model {
Some(model) => model,
None => {
let arbiter = pair_arbiter.as_ref().ok_or(ReviewerError::MissingArbiter)?;
if harness_from_cli && !arbiter.harness.eq_ignore_ascii_case(harness_slug(harness))
{
return Err(ReviewerError::OverrideNeedsModel {
role: "arbiter".to_owned(),
harness: harness_slug(harness).to_owned(),
});
}
arbiter.model.clone()
}
};
let effort = arbiter_effort
.or_else(|| pair_arbiter.as_ref().map(|arbiter| arbiter.effort))
.unwrap_or_else(Effort::highest);
Ok(StrictReviewConfig {
arbiter_harness: harness,
arbiter_model: model,
arbiter_effort: effort,
})
}
fn request_for(&self, prompt: String) -> ReviewRequest {
ReviewRequest::new(
self.watched_agent,
self.watched_model.clone(),
self.reviewer_harness,
self.reviewer_model.clone(),
self.allow_same_model,
prompt,
)
.with_effort(self.reviewer_effort)
}
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct ReviewPlan {
pub watched_agent: Agent,
pub watched_model: String,
pub reviewer_harness: ReviewerHarness,
pub reviewer_model: String,
pub allow_same_model: bool,
pub invocation: InvocationPlan,
}
impl ReviewPlan {
pub fn build(request: ReviewRequest) -> Result<Self, ReviewerError> {
validate_model_present("reviewer", &request.reviewer_model)?;
// The writer model may be unknown (pair-based selection doesn't require it);
// opposition is enforced only when the writer model is actually provided.
if request.watched_model.trim().is_empty() {
if !request.allow_same_model {
// The writer model is unknown so we cannot verify model opposition at
// runtime — the review will proceed, but warn so this doesn't go silent.
eprintln!(
"{} warning: writer model is unknown; model opposition is only pair-based \
for this review. Pass --watched-model to enforce per-model opposition; \
truth-mirror has no persistent writer-model config key yet.",
crate::messages::diagnostic_prefix()
);
}
} else if !request.allow_same_model
&& normalized_model(&request.watched_model) == normalized_model(&request.reviewer_model)
{
return Err(ReviewerError::SameModelWithoutWaiver {
watched_model: request.watched_model,
reviewer_model: request.reviewer_model,
});
}
let invocation = InvocationPlan::for_harness(
request.reviewer_harness,
&request.reviewer_model,
request.reviewer_effort,
)?;
Ok(Self {
watched_agent: request.watched_agent,
watched_model: request.watched_model,
reviewer_harness: request.reviewer_harness,
reviewer_model: request.reviewer_model,
allow_same_model: request.allow_same_model,
invocation,
})
}
pub fn run_with<R: ProcessRunner>(
&self,
prompt: &str,
runner: &R,
timeout: Option<Duration>,
) -> Result<ProcessOutput, ReviewerError> {
runner.run(&self.invocation, prompt, timeout)
}
fn reviewer_config(&self) -> ReviewerConfig {
ReviewerConfig::new(
harness_slug(self.reviewer_harness),
self.reviewer_model.clone(),
self.allow_same_model,
)
}
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct InvocationPlan {
pub program: String,
pub args: Vec<String>,
pub prompt_delivery: PromptDelivery,
}
impl InvocationPlan {
pub fn for_harness(
harness: ReviewerHarness,
model: &str,
effort: Effort,
) -> Result<Self, ReviewerError> {
validate_model_present("reviewer", model)?;
let model = model.trim();
let e = effort.as_str();
// Reasoning-effort flags verified against source: codex ReasoningEffort
// (`-c model_reasoning_effort`), pi `--thinking` (cli/args.js:249), claude
// `--effort`. Gemini/OpenCode have no effort flag, so effort is omitted.
let plan = match harness {
ReviewerHarness::Claude => Self {
program: "claude".to_owned(),
args: vec![
"--print".to_owned(),
"--model".to_owned(),
model.to_owned(),
"--effort".to_owned(),
// Claude has no `minimal`; clamp to a valid level.
effort.claude_value().to_owned(),
],
prompt_delivery: PromptDelivery::Stdin,
},
ReviewerHarness::Codex => Self {
program: "codex".to_owned(),
args: vec![
"exec".to_owned(),
"-m".to_owned(),
model.to_owned(),
"-c".to_owned(),
format!("model_reasoning_effort={e}"),
],
prompt_delivery: PromptDelivery::PositionalArgument,
},
ReviewerHarness::Pi => Self {
program: "pi".to_owned(),
args: vec![
"--model".to_owned(),
model.to_owned(),
"--thinking".to_owned(),
e.to_owned(),
// Read-only tools so the reviewer can grep the repo for the whole
// defect class (Codex `exec` / Claude `-p` have read tools already).
"--tools".to_owned(),
"read,grep,find,ls".to_owned(),
"-p".to_owned(),
],
prompt_delivery: PromptDelivery::Stdin,
},
ReviewerHarness::Gemini => Self {
program: "gemini".to_owned(),
args: vec!["-m".to_owned(), model.to_owned()],
prompt_delivery: PromptDelivery::FlagValue("-p".to_owned()),
},
ReviewerHarness::Opencode => Self {
program: "opencode".to_owned(),
args: vec!["run".to_owned(), "--model".to_owned(), model.to_owned()],
prompt_delivery: PromptDelivery::PositionalArgument,
},
ReviewerHarness::Custom => return Err(ReviewerError::UnsupportedCustomHarness),
};
Ok(plan)
}
pub fn args_for_prompt(&self, prompt: &str) -> Vec<String> {
let mut args = self.args.clone();
match &self.prompt_delivery {
PromptDelivery::Stdin => {}
PromptDelivery::PositionalArgument => args.push(prompt.to_owned()),
PromptDelivery::FlagValue(flag) => {
args.push(flag.clone());
args.push(prompt.to_owned());
}
}
args
}
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub enum PromptDelivery {
Stdin,
PositionalArgument,
FlagValue(String),
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct ProcessOutput {
pub status_code: Option<i32>,
pub stdout: String,
pub stderr: String,
}
pub trait ProcessRunner {
/// Run the reviewer invocation to completion.
///
/// `timeout` is a per-review wall-clock limit covering the WHOLE
/// invocation: stdin delivery, the review itself, the reap, and the pipe
/// drain. When the reviewer outlives it, the implementation must kill the
/// reviewer's process tree, reap the child, and return
/// [`ReviewerError::ReviewerTimeout`]. `None` means wait indefinitely.
fn run(
&self,
invocation: &InvocationPlan,
prompt: &str,
timeout: Option<Duration>,
) -> Result<ProcessOutput, ReviewerError>;
}
#[derive(Clone, Copy, Debug, Default)]
pub struct StdProcessRunner;
impl ProcessRunner for StdProcessRunner {
fn run(
&self,
invocation: &InvocationPlan,
prompt: &str,
timeout: Option<Duration>,
) -> Result<ProcessOutput, ReviewerError> {
let mut command = Command::new(&invocation.program);
command.args(invocation.args_for_prompt(prompt));
command.stdout(Stdio::piped()).stderr(Stdio::piped());
if invocation.prompt_delivery == PromptDelivery::Stdin {
command.stdin(Stdio::piped());
}
// unix: the reviewer runs as its own process-group leader, so the
// timeout kill can take the WHOLE group. A descendant that inherited
// the output pipes must not survive the reviewer and hold them open —
// 0.9.2 froze the queue exactly that way.
#[cfg(unix)]
{
use std::os::unix::process::CommandExt as _;
command.process_group(0);
}
// The deadline covers everything from spawn on — stdin delivery, the
// review itself, the reap, and the pipe drain. The first timeout fix
// started the clock only after a blocking `write_all` of the prompt
// returned, so a reviewer that never read stdin (or stalled against a
// full pipe buffer before reading) wedged the queue forever despite
// the configured timeout.
let deadline = timeout.and_then(|limit| {
Instant::now()
.checked_add(limit)
.map(|instant| (instant, limit))
});
let mut child = command.spawn().map_err(ReviewerError::Spawn)?;
// Deliver the prompt on a helper thread so a reviewer that never
// reads stdin cannot outrun the deadline: the main thread proceeds
// straight to the deadline-enforcing wait, and the kill that fires on
// timeout closes the pipe and unblocks the writer. Dropping the write
// end after a completed write preserves the EOF contract.
let mut stdin_writer = if invocation.prompt_delivery == PromptDelivery::Stdin {
let pipe = child.stdin.take().ok_or(ReviewerError::MissingStdinPipe)?;
Some(StdinWriter::spawn(pipe, prompt.as_bytes().to_vec()))
} else {
None
};
// Drain both pipes on helper threads so a chatty reviewer never
// deadlocks against a full pipe buffer while we poll for exit (this is
// what `wait_with_output` does internally — split out so a timeout can
// kill the child and still collect, or boundedly abandon, the pipes).
let mut stdout_reader = child.stdout.take().map(spawn_pipe_reader);
let mut stderr_reader = child.stderr.take().map(spawn_pipe_reader);
// Bounded drains only when a timeout was configured: `None` is the
// explicit "wait indefinitely" mode and keeps the old semantics.
let drain_grace = deadline.map(|_| PIPE_DRAIN_GRACE);
let child_pid = child.id();
let status = match wait_for_child(child, deadline) {
Ok(status) => status,
Err(error) => {
// The child was killed and reaped (or the kill surfaced an
// error). The readers hit EOF once the process tree is down;
// their partial output is discarded — a timed out review
// records the timeout, not half a verdict. Every join stays
// bounded by the grace so even a pipe-hoarding descendant
// cannot hold the queue past it.
if let Some(reader) = stdout_reader.as_mut() {
let _ = reader.try_collect(drain_grace);
}
if let Some(reader) = stderr_reader.as_mut() {
let _ = reader.try_collect(drain_grace);
}
if let Some(writer) = stdin_writer.as_mut() {
let _ = writer.try_collect(drain_grace);
}
return Err(error);
}
};
let mut prompt_write = stdin_writer
.as_mut()
.map(|writer| writer.try_collect(drain_grace));
let mut stdout = stdout_reader
.as_mut()
.map(|reader| reader.try_collect(drain_grace));
let mut stderr = stderr_reader
.as_mut()
.map(|reader| reader.try_collect(drain_grace));
if pipes_wedged(&prompt_write, &stdout, &stderr) {
// The reviewer exited but a spawned descendant inherited a pipe
// and holds it open, so the readers never saw EOF. Take the
// process tree down (best-effort; the group may already be gone)
// and give the stragglers one final grace before declaring the
// wedge — an unbounded join here re-creates the 0.9.2 freeze.
let _ = platform_tree_kill(child_pid);
if prompt_write.as_ref().is_some_and(WriteState::is_wedged) {
prompt_write = stdin_writer
.as_mut()
.map(|writer| writer.try_collect(Some(PIPE_DRAIN_GRACE)));
}
if stdout.as_ref().is_some_and(DrainState::is_wedged) {
stdout = stdout_reader
.as_mut()
.map(|reader| reader.try_collect(Some(PIPE_DRAIN_GRACE)));
}
if stderr.as_ref().is_some_and(DrainState::is_wedged) {
stderr = stderr_reader
.as_mut()
.map(|reader| reader.try_collect(Some(PIPE_DRAIN_GRACE)));
}
}
if pipes_wedged(&prompt_write, &stdout, &stderr) {
return Err(ReviewerError::ReviewerOutputWedged {
grace_secs: PIPE_DRAIN_GRACE.as_secs(),
});
}
// A prompt-delivery failure invalidates the review even when the
// reviewer went on to exit cleanly: the verdict answered a truncated
// question. Surfaced, never swallowed.
if let Some(WriteState::Finished(Err(source))) = prompt_write {
return Err(ReviewerError::WritePrompt(source));
}
Ok(ProcessOutput {
status_code: status.code(),
stdout: String::from_utf8_lossy(&stdout.and_then(drain_bytes).unwrap_or_default())
.into_owned(),
stderr: String::from_utf8_lossy(&stderr.and_then(drain_bytes).unwrap_or_default())
.into_owned(),
})
}
}
/// Result of a bounded pipe-drain collect.
enum DrainState {
/// The reader thread delivered: drained bytes, or `None` when the read
/// itself failed (treated as empty output — the process result is the
/// authoritative signal, not the captured text).
Drained(Option<Vec<u8>>),
/// The grace window expired with the pipe still open: a descendant of the
/// reviewer inherited the write end and is holding it open.
Wedged,
}
impl DrainState {
fn is_wedged(&self) -> bool {
matches!(self, Self::Wedged)
}
}
/// Extract bytes from a completed drain; a wedge is handled by the caller
/// before this point, so it degrades to empty here.
fn drain_bytes(state: DrainState) -> Option<Vec<u8>> {
match state {
DrainState::Drained(bytes) => bytes,
DrainState::Wedged => None,
}
}
/// Whether any of the three pipe collections is still wedged.
fn pipes_wedged(
prompt_write: &Option<WriteState>,
stdout: &Option<DrainState>,
stderr: &Option<DrainState>,
) -> bool {
prompt_write.as_ref().is_some_and(WriteState::is_wedged)
|| stdout.as_ref().is_some_and(DrainState::is_wedged)
|| stderr.as_ref().is_some_and(DrainState::is_wedged)
}
/// Result of a bounded stdin-writer collect.
enum WriteState {
Finished(io::Result<()>),
Wedged,
}
impl WriteState {
fn is_wedged(&self) -> bool {
matches!(self, Self::Wedged)
}
}
/// A pipe-draining thread whose output is collected through a channel, so the
/// join can be bounded by a grace window. `JoinHandle::join` alone cannot
/// time out — the 0.9.2 queue freeze was an unbounded join on a pipe whose
/// write end a surviving descendant held open.
struct PipeReader {
handle: Option<std::thread::JoinHandle<()>>,
receiver: std::sync::mpsc::Receiver<io::Result<Vec<u8>>>,
}
fn spawn_pipe_reader(mut pipe: impl Read + Send + 'static) -> PipeReader {
let (sender, receiver) = std::sync::mpsc::channel();
let handle = std::thread::spawn(move || {
let mut buffer = Vec::new();
let result = pipe.read_to_end(&mut buffer).map(|_| buffer);
let _ = sender.send(result);
});
PipeReader {
handle: Some(handle),
receiver,
}
}
impl PipeReader {
/// Collect the drained bytes. `Some(grace)` bounds the wait; `None`
/// blocks indefinitely (the explicit no-timeout mode). Once the value is
/// received the thread has finished, so the join cannot block; on an
/// expired grace the wedged thread is abandoned — bounded by process
/// lifetime, never by queue progress.
fn try_collect(&mut self, grace: Option<Duration>) -> DrainState {
use std::sync::mpsc::RecvTimeoutError;
let received = match grace {
Some(grace) => self.receiver.recv_timeout(grace),
None => self
.receiver
.recv()
.map_err(|_| RecvTimeoutError::Disconnected),
};
match received {
Ok(result) => {
if let Some(handle) = self.handle.take() {
let _ = handle.join();
}
DrainState::Drained(result.ok())
}
Err(RecvTimeoutError::Timeout | RecvTimeoutError::Disconnected) => DrainState::Wedged,
}
}
}
/// A prompt-delivery thread with the same bounded-collect contract as
/// [`PipeReader`].
struct StdinWriter {
handle: Option<std::thread::JoinHandle<()>>,
receiver: std::sync::mpsc::Receiver<io::Result<()>>,
}
impl StdinWriter {
fn spawn(mut pipe: std::process::ChildStdin, bytes: Vec<u8>) -> Self {
let (sender, receiver) = std::sync::mpsc::channel();
let handle = std::thread::spawn(move || {
let result = pipe.write_all(&bytes);
// Dropping the write end signals EOF to the reviewer.
drop(pipe);
let _ = sender.send(result);
});
Self {
handle: Some(handle),
receiver,
}
}
fn try_collect(&mut self, grace: Option<Duration>) -> WriteState {
use std::sync::mpsc::RecvTimeoutError;
let received = match grace {
Some(grace) => self.receiver.recv_timeout(grace),
None => self
.receiver
.recv()
.map_err(|_| RecvTimeoutError::Disconnected),
};
match received {
Ok(result) => {
if let Some(handle) = self.handle.take() {
let _ = handle.join();
}
WriteState::Finished(result)
}
Err(RecvTimeoutError::Timeout | RecvTimeoutError::Disconnected) => WriteState::Wedged,
}
}
}
/// How often to poll the child while waiting under a timeout. Small enough
/// that a killed reviewer is reaped promptly, large enough to stay off the
/// CPU for a 20-minute default window.
const CHILD_POLL_INTERVAL: Duration = Duration::from_millis(100);
/// How long to wait after a kill before surfacing an unreaped child as a hard
/// error. This is bounded so a process-tree kill race cannot wedge the queue
/// even if the child survives longer than the request timeout.
const KILL_REAP_GRACE: Duration = Duration::from_secs(1);
/// Grace for collecting reviewer output after the child resolves (clean exit
/// or kill + reap). Pipe drains are kernel-buffer copies that finish in
/// milliseconds once the write end closes; the bound exists so a descendant
/// holding a pipe open can never wedge the queue the way 0.9.2 did.
const PIPE_DRAIN_GRACE: Duration = Duration::from_secs(5);
/// Wait for `child`, enforcing an optional wall-clock deadline.
///
/// With no deadline this is a plain blocking wait. With a deadline the child
/// is polled; once it passes, the reviewer's whole process tree is killed and
/// the child is reaped within a bounded grace. If the direct child still does
/// not report exit after that grace, cleanup is handed to a background reaper
/// before returning [`ReviewerError::ReviewerTimeout`] so the queue hot path
/// never blocks on `wait`.
fn wait_for_child(
mut child: std::process::Child,
deadline: Option<(Instant, Duration)>,
) -> Result<std::process::ExitStatus, ReviewerError> {
let Some((deadline, limit)) = deadline else {
return child.wait().map_err(ReviewerError::Wait);
};
loop {
if let Some(status) = child.try_wait().map_err(ReviewerError::Wait)? {
return Ok(status);
}
if Instant::now() >= deadline {
return kill_and_reap(child, limit);
}
std::thread::sleep(CHILD_POLL_INTERVAL);
}
}
/// The deadline passed with the child still running: kill the reviewer's
/// process tree, reap the child, and decide the outcome. The short grace is a
/// bounded diagnostic window; if it expires, a final direct kill is attempted
/// and any remaining reap is moved off the queue hot path.
fn kill_and_reap(
child: std::process::Child,
limit: Duration,
) -> Result<std::process::ExitStatus, ReviewerError> {
let mut child = child;
kill_process_tree(&mut child)?;
let status = match wait_for_reap(&mut child, KILL_REAP_GRACE) {
Ok(status) => status,
Err(error) => return reap_after_grace(child, error, limit.as_secs()),
};
outcome_after_kill(status, limit.as_secs())
}
fn reap_after_grace(
mut child: std::process::Child,
grace_error: io::Error,
timeout_secs: u64,
) -> Result<std::process::ExitStatus, ReviewerError> {
let pid = child.id();
tracing::warn!(
pid,
%grace_error,
"reviewer kill grace expired; handing direct-child reap to background cleanup"
);
match child.kill() {
Ok(()) => {}
Err(error) if error.kind() == io::ErrorKind::InvalidInput => {}
Err(source) => {
return Err(ReviewerError::KillReviewer {
pid,
tree_error: "direct-child reap fallback".to_owned(),
source,
});
}
}
if let Some(status) = child.try_wait().map_err(ReviewerError::Wait)? {
return outcome_after_kill(status, timeout_secs);
}
let reaper = std::thread::Builder::new()
.name(format!("truth-mirror-reaper-{pid}"))
.spawn(move || {
if let Err(error) = child.wait() {
tracing::debug!(pid, %error, "background reviewer reaper stopped");
}
});
if let Err(error) = reaper {
tracing::warn!(pid, %error, "could not spawn background reviewer reaper");
}
Err(ReviewerError::ReviewerTimeout { timeout_secs })
}
fn wait_for_reap(
child: &mut std::process::Child,
grace: Duration,
) -> io::Result<std::process::ExitStatus> {
let started = Instant::now();
loop {
if let Some(status) = child.try_wait()? {
return Ok(status);
}
if started.elapsed() >= grace {
return Err(io::Error::other(format!(
"killed child did not exit within {grace:?} (pid={})",
child.id()
)));
}
std::thread::sleep(CHILD_POLL_INTERVAL);
}
}
/// Decide the outcome after the timeout kill landed.
///
/// Kill/reap race: a child that exited on its own microseconds before the
/// kill arrived still reports its REAL exit status from `wait` (unix: a
/// signal-killed process has `status.signal() == Some(..)`; a self-exited one
/// has `None`). That real outcome wins over the timeout — the review may
/// carry a valid verdict. On Windows the distinction is not observable, so
/// the timeout stands.
fn outcome_after_kill(
status: std::process::ExitStatus,
timeout_secs: u64,
) -> Result<std::process::ExitStatus, ReviewerError> {
#[cfg(unix)]
{
use std::os::unix::process::ExitStatusExt as _;
if status.signal().is_none() {
return Ok(status);
}
}
#[cfg(not(unix))]
let _ = status;
Err(ReviewerError::ReviewerTimeout { timeout_secs })
}
/// Kill the reviewer and every descendant it spawned.
///
/// unix: the child was spawned as a process-group leader, so
/// `kill -KILL -<pid>` signals the whole group — descendants cannot survive
/// to hold the output pipes open *for that group*. Descendants that escape into
/// a different process group (for example by `setsid`) are outside this signal
/// window, so this is best-effort coverage, not complete process-tree proof.
///
/// windows: `taskkill /F /T` attempts whole-tree termination. A direct kill
/// fallback still applies when taskkill cannot run. Non-unix hosts have only a
/// direct child kill fallback, so full-tree cleanup is not guaranteed there.
///
/// In every case, a genuine kill failure is surfaced, never swallowed. The
/// direct child is always reaped; process-tree cleanup remains best-effort for
/// descendants that escape the platform's tree primitive.
fn kill_process_tree(child: &mut std::process::Child) -> Result<(), ReviewerError> {
let pid = child.id();
if let Err(tree_error) = platform_tree_kill(pid) {
// The tree kill failed — most often because the process already
// exited and took the group with it. A direct kill still covers the
// child itself.
match child.kill() {
Ok(()) => {}
Err(direct_error) => {
// std reports InvalidInput when the child was already reaped:
// the kill/reap race, not a failure — the subsequent wait
// surfaces the real exit status.
if direct_error.kind() != io::ErrorKind::InvalidInput {
return Err(ReviewerError::KillReviewer {
pid,
tree_error: tree_error.to_string(),
source: direct_error,
});
}
}
}
}
Ok(())
}
#[cfg(unix)]
fn platform_tree_kill(pid: u32) -> io::Result<()> {
// Negative pid: the whole process group (the reviewer was spawned as its
// group leader, so the pgid equals the child pid).
let status = Command::new("kill")
.arg("-KILL")
.arg(format!("-{pid}"))
.stdin(Stdio::null())
.stdout(Stdio::null())
.stderr(Stdio::null())
.status()?;
if status.success() {
Ok(())
} else {
Err(io::Error::other(format!(
"kill -KILL -{pid} exited with {status}"
)))
}
}
#[cfg(windows)]
fn platform_tree_kill(pid: u32) -> io::Result<()> {
// /T takes down the whole process tree rooted at the reviewer.
let status = Command::new("taskkill")
.args(["/F", "/T", "/PID", &pid.to_string()])
.stdin(Stdio::null())
.stdout(Stdio::null())
.stderr(Stdio::null())
.status()?;
if status.success() {
Ok(())
} else {
Err(io::Error::other(format!(
"taskkill /F /T /PID {pid} exited with {status}"
)))
}
}
#[cfg(not(any(unix, windows)))]
fn platform_tree_kill(_pid: u32) -> io::Result<()> {
// Non-unix hosts have no process-tree kill primitive here, so callers can
// only attempt a best-effort direct child kill after this returns.
Err(io::Error::other(
"process-tree kill is not supported on this platform",
))
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct ReviewJob {
pub commit_sha: String,
pub claim: Claim,
pub diff: String,
/// Ground-truth constraints + recent trajectory injected into review prompts.
pub context: String,
pub request: ReviewRequest,
pub strict: Option<StrictReviewConfig>,
/// When set, this job is a petition re-review for the listed original
/// rejection SHA. The reviewer is asked "does <fix sha> materially address
/// each finding?", and a `PASS` verdict (the petition's accept — there is
/// no separate `ACCEPT` variant on the wire) transitions the original
/// rejection to `resolved`; a non-PASS verdict leaves the attempt spent
/// (escalating to `needs-human` once the counter hits the bound).
pub petition: Option<PetitionContext>,
}
/// Original rejection payload bundled into a petition re-review.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct PetitionContext {
pub original_sha: String,
pub fix_sha: String,
pub original_claim: String,
pub original_summary: String,
pub original_findings: Vec<String>,
pub original_structured_findings: Vec<StructuredFinding>,
pub original_reviewer_model: String,
pub attempts_so_far: u32,
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct StrictReviewConfig {
pub arbiter_harness: ReviewerHarness,
pub arbiter_model: String,
pub arbiter_effort: Effort,
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct ReviewExecution {
pub entries: Vec<LedgerEntry>,
}
/// If the job carries a petition context, rebuild the request with the
/// petition-specific prompt so the reviewer is asked the right question.
/// The reviewer invocation plan is otherwise unchanged; only the prompt
/// text the reviewer receives is replaced.
fn materialize_petition_prompt(mut job: ReviewJob) -> ReviewJob {
let Some(petition) = &job.petition else {
return job;
};
job.request.prompt = petition_prompt(petition, &job.claim, &job.diff, &job.context);
job
}
pub fn execute_review_job<R: ProcessRunner>(
job: ReviewJob,
runner: &R,
store: &LedgerStore,
timeout: Option<Duration>,
) -> Result<ReviewExecution, ReviewerError> {
// If this job carries a petition context, swap the prompt for the petition
// template so the reviewer sees the original claim + findings + the fix
// diff and is asked whether the fix materially addresses each finding.
let job = materialize_petition_prompt(job);
let first_plan = ReviewPlan::build(job.request.clone())?;
let first_output = first_plan.run_with(&job.request.prompt, runner, timeout)?;
ensure_process_success(&first_output)?;
let first_verdict = ParsedVerdict::parse(&first_output.stdout)?;
let first_entry = entry_from_verdict(&job, &first_plan, &first_verdict);
store.append_entry(&first_entry)?;
let mut entries = vec![first_entry];
if let Some(strict) = &job.strict
&& first_verdict.verdict == Verdict::Pass
&& first_verdict.findings.is_empty()
{
validate_strict_arbiter(&job.request, strict)?;
let strict_prompt = strict_second_pass_prompt(&job, &first_output.stdout);
let strict_request = ReviewRequest::new(
job.request.watched_agent,
job.request.watched_model.clone(),
strict.arbiter_harness,
strict.arbiter_model.clone(),
false,
strict_prompt,
)
.with_effort(strict.arbiter_effort);
let strict_plan = ReviewPlan::build(strict_request.clone())?;
let strict_output = strict_plan.run_with(&strict_request.prompt, runner, timeout)?;
ensure_process_success(&strict_output)?;
let strict_verdict = ParsedVerdict::parse(&strict_output.stdout)?;
let strict_entry = entry_from_verdict(&job, &strict_plan, &strict_verdict);
store.append_entry(&strict_entry)?;
entries.push(strict_entry);
}
// Petition transitions apply to the FINAL verdict only. Transitioning on
// the first PASS and then letting a strict arbiter REJECT would leave the
// original rejection resolved by a verdict the arbiter just overturned.
if let Some(last) = entries.last() {
apply_petition_transition(&job, last, store)?;
}
Ok(ReviewExecution { entries })
}
/// After a petition reviewer's verdict is recorded, transition the original
/// rejection: `PASS` (the petition's accept) -> `resolved` (with provenance); anything else ->
/// increment the attempt counter; on the bounded-th attempt, escalate to
/// `needs-human`. The bounded max lives in `crate::resolve::MAX_PETITION_ATTEMPTS`.
fn apply_petition_transition(
job: &ReviewJob,
entry: &LedgerEntry,
store: &LedgerStore,
) -> Result<(), ReviewerError> {
let Some(petition) = &job.petition else {
return Ok(());
};
let max_attempts = crate::resolve::MAX_PETITION_ATTEMPTS;
let next_attempts = entry.petition_attempts;
let reason = format!(
"petition review of fix {} against rejection {} (attempts {})",
petition.fix_sha, petition.original_sha, next_attempts
);
if crate::resolve::petition_accepts(entry.verdict) {
// PASS or FLAG: the fix materially addresses each finding (FLAG just
// carries non-blocking debt alongside — the debt lives on the
// petition review entry itself and surfaces via `truth-mirror debt`).
store
.append_petition_transition(
&petition.original_sha,
crate::ledger::Disposition::Resolved,
crate::ledger::ResolutionKind::Resolved,
&reason,
next_attempts,
)
.map_err(ReviewerError::Ledger)?;
} else {
{
// REJECT: the attempt is spent, stay Open unless exhausted.
if next_attempts >= max_attempts {
store
.escalate_to_needs_human_with_attempts(
&petition.original_sha,
&reason,
next_attempts,
)
.map_err(ReviewerError::Ledger)?;
} else {
store
.append_petition_transition(
&petition.original_sha,
crate::ledger::Disposition::Open,
crate::ledger::ResolutionKind::Resolved,
&reason,
next_attempts,
)
.map_err(ReviewerError::Ledger)?;
}
}
}
Ok(())
}
/// Rebuild the petition context for a queued petition item from the ledger:
/// the original rejection's claim, findings, and reviewer, plus the current
/// attempt counter. The CLI's `resolve --fixed-by` already counted this
/// attempt when it enqueued, so `attempts_so_far` INCLUDES the in-flight
/// attempt and the verdict entry carries it unchanged (no double increment).
fn petition_context_from_ledger(
original_sha: &str,
fix_sha: &str,
store: &LedgerStore,
) -> Result<PetitionContext, ReviewerError> {
let original = store.show(original_sha).map_err(ReviewerError::Ledger)?;
let attempts_so_far = store
.petition_attempts_for(original_sha)
.map_err(ReviewerError::Ledger)?;
Ok(PetitionContext {
original_sha: original_sha.to_owned(),
fix_sha: fix_sha.to_owned(),
original_claim: original.claim.clone(),
original_summary: original.summary.clone(),
original_findings: original.findings.clone(),
original_structured_findings: original.structured_findings.clone(),
original_reviewer_model: original.reviewer.model.clone(),
attempts_so_far,
})
}
/// Whether a queued petition's target can still accept a petition transition:
/// the rejection exists and is still open (or escalated to needs-human, which
/// remains petitionable). Anything else — already resolved, waived, or never
/// recorded — makes the queued petition a dead letter: the drain must consume
/// it and keep going instead of dying on `NoOpenRejection`, which is exactly
/// how 0.9.2 killed watchers when a petition was enqueued twice.
fn petition_target_is_actionable(
store: &LedgerStore,
original_sha: &str,
) -> Result<bool, ReviewerError> {
match store.show(original_sha) {
Ok(entry) => Ok(entry.is_unresolved_rejection() || entry.is_needs_human()),
Err(LedgerError::NotFound { .. }) => Ok(false),
Err(error) => Err(ReviewerError::Ledger(error)),
}
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct ParsedVerdict {
pub verdict: Verdict,
pub summary: String,
pub findings: Vec<String>,
pub structured_findings: Vec<StructuredFinding>,
pub next_steps: Vec<String>,
pub memory_skill_classification: MemorySkillClassification,
pub raw: String,
}
impl ParsedVerdict {
pub fn parse(output: &str) -> Result<Self, ReviewerError> {
// Reviewer models routinely wrap the verdict in prose and/or a
// ```json fenced block despite the prompt asking for raw JSON.
// Demanding JSON at byte 0 made the first real review error out and
// the watcher queue never drain — so parse tolerantly: raw first,
// then the fenced block, then the outermost brace span. The original
// raw output is preserved verbatim in `raw` either way.
let candidate = extract_verdict_json(output);
let mut parsed: ReviewerJsonOutput =
serde_json::from_str(candidate).map_err(|source| ReviewerError::VerdictJson {
source,
output: output.to_owned(),
})?;
if let Err(error) = parsed.validate() {
if !parsed.salvage_disallowed_pass_skill_proposal() {
return Err(error);
}
tracing::warn!(
verdict = %parsed.verdict,
"reviewer returned a PASS verdict with a disallowed memory-skill proposal; stripped the proposal and accepted the verdict"
);
parsed.validate()?;
}
let findings = parsed
.findings
.iter()
.map(StructuredFinding::display_line)
.collect();
Ok(Self {
verdict: parsed.verdict,
summary: parsed.summary,
findings,
structured_findings: parsed.findings,
next_steps: parsed.next_steps,
memory_skill_classification: parsed.memory_skill,
raw: output.to_owned(),
})
}
}
/// Pull the verdict JSON out of a reviewer reply that may wrap it in prose
/// and/or markdown fences. Every candidate is VALIDATED as JSON before it is
/// returned — a shape that fails to parse falls through to the next strategy
/// instead of poisoning the caller (leading JSON + trailing prose was exactly
/// that failure: `starts_with('{')` returned the whole text and serde died on
/// the trailing characters). Precedence:
/// 1. the whole (trimmed) output — the documented contract, zero-cost;
/// 2. the longest valid JSON value PREFIX when the output starts with `{`
/// (leading verdict, trailing prose);
/// 3. the first ```json (or bare ```) fenced block that parses;
/// 4. the first valid JSON value starting at the first `{` anywhere.
///
/// Returns a slice of `output`; the caller still surfaces the full original
/// text in parse errors and stores it verbatim as `raw`.
fn extract_verdict_json(output: &str) -> &str {
fn parses(candidate: &str) -> bool {
serde_json::from_str::<serde_json::Value>(candidate).is_ok()
}
/// The byte length of the first complete JSON value at the start of
/// `text`, if any — how we slice a verdict off leading-JSON+prose shapes.
fn json_prefix_len(text: &str) -> Option<usize> {
let mut stream = serde_json::Deserializer::from_str(text).into_iter::<serde_json::Value>();
match stream.next() {
Some(Ok(_)) => Some(stream.byte_offset()),
_ => None,
}
}
let trimmed = output.trim();
if trimmed.starts_with('{') {
if parses(trimmed) {
return trimmed;
}
if let Some(end) = json_prefix_len(trimmed) {
return trimmed[..end].trim();
}
}
// Fenced block: find ``` openers, skip the info string (e.g. `json`),
// and take the body up to the closing fence.
let mut search_from = 0;
while let Some(open_rel) = output[search_from..].find("```") {
let after_open = search_from + open_rel + 3;
let body_start = match output[after_open..].find('\n') {
Some(newline_rel) => after_open + newline_rel + 1,
None => break,
};
let Some(close_rel) = output[body_start..].find("```") else {
break;
};
let body = output[body_start..body_start + close_rel].trim();
if body.starts_with('{') && parses(body) {
return body;
}
search_from = body_start + close_rel + 3;
}
// Last resort: the first complete JSON value starting at the first `{`
// anywhere in the output (handles prose-then-JSON-then-prose).
if let Some(first) = output.find('{') {
let tail = &output[first..];
if let Some(end) = json_prefix_len(tail) {
return tail[..end].trim();
}
}
trimmed
}
#[derive(Clone, Debug, Deserialize, Eq, PartialEq)]
struct ReviewerJsonOutput {
verdict: Verdict,
summary: String,
#[serde(default)]
findings: Vec<StructuredFinding>,
#[serde(default)]
next_steps: Vec<String>,
memory_skill: MemorySkillClassification,
}
impl ReviewerJsonOutput {
/// A PASS may only propose a how-to skill. A reviewer occasionally gets the
/// verdict right while attaching a REJECT-only proposal; retain the valid
/// verdict and discard that proposal rather than throwing away the review.
fn salvage_disallowed_pass_skill_proposal(&mut self) -> bool {
if self.verdict != Verdict::Pass
|| !matches!(
self.memory_skill.kind,
MemorySkillClassificationKind::AntiPatternSkill
| MemorySkillClassificationKind::RemediationSkill
)
{
return false;
}
self.memory_skill.kind = MemorySkillClassificationKind::None;
self.memory_skill.learning_source.clear();
true
}
fn validate(&self) -> Result<(), ReviewerError> {
if self.summary.trim().is_empty() {
return Err(ReviewerError::VerdictSchema {
message: "summary must not be empty".to_owned(),
});
}
self.memory_skill
.validate_for_verdict(self.verdict)
.map_err(|message| ReviewerError::VerdictSchema { message })?;
for finding in &self.findings {
if finding.title.trim().is_empty() {
return Err(ReviewerError::VerdictSchema {
message: "finding title must not be empty".to_owned(),
});
}
if finding.body.trim().is_empty() {
return Err(ReviewerError::VerdictSchema {
message: "finding body must not be empty".to_owned(),
});
}
if finding.file.trim().is_empty() {
return Err(ReviewerError::VerdictSchema {
message: "finding file must not be empty".to_owned(),
});
}
if finding.line_start == 0 || finding.line_end == 0 {
return Err(ReviewerError::VerdictSchema {
message: "finding lines must be one-based".to_owned(),
});
}
if finding.line_end < finding.line_start {
return Err(ReviewerError::VerdictSchema {
message: "finding line_end must be greater than or equal to line_start"
.to_owned(),
});
}
if finding.confidence > 100 {
return Err(ReviewerError::VerdictSchema {
message: "finding confidence must be between 0 and 100".to_owned(),
});
}
if finding.recommendation.trim().is_empty() {
return Err(ReviewerError::VerdictSchema {
message: "finding recommendation must not be empty".to_owned(),
});
}
}
if self.verdict == Verdict::Pass && !self.findings.is_empty() {
return Err(ReviewerError::VerdictSchema {
message: "PASS verdict must not include findings".to_owned(),
});
}
if self.verdict == Verdict::Reject && self.findings.is_empty() {
return Err(ReviewerError::VerdictSchema {
message: "REJECT verdict must include at least one finding".to_owned(),
});
}
if self.verdict == Verdict::Flag && self.findings.is_empty() {
return Err(ReviewerError::VerdictSchema {
message: "FLAG verdict must include at least one finding (the debt being surfaced, in the findings array)"
.to_owned(),
});
}
Ok(())
}
}
#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "kebab-case")]
pub enum ReviewRunStatus {
Queued,
Running,
Completed,
Failed,
Cancelled,
}
impl std::fmt::Display for ReviewRunStatus {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::Queued => formatter.write_str("queued"),
Self::Running => formatter.write_str("running"),
Self::Completed => formatter.write_str("completed"),
Self::Failed => formatter.write_str("failed"),
Self::Cancelled => formatter.write_str("cancelled"),
}
}
}
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct ReviewRun {
pub id: String,
pub commit_sha: String,
pub target: String,
pub status: ReviewRunStatus,
pub phase: String,
pub ledger_entries: usize,
pub error: Option<String>,
/// PID of the process executing this run while it is `Running`. Persisted so a
/// `review status`/`cancel` can tell a live worker from a silently-dead one and
/// reap the zombie. `None` once the run reaches a terminal state (or for legacy
/// records written before pid tracking existed).
#[serde(default)]
pub worker_pid: Option<u32>,
/// OS-reported start marker of the worker process, recorded alongside the pid
/// so a REUSED pid cannot make a dead worker look alive (pid-recycling
/// defense). `None` for legacy rows or when the platform reports no marker.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub worker_start_token: Option<String>,
pub created_at_unix: u64,
pub updated_at_unix: u64,
pub started_at_unix: Option<u64>,
pub completed_at_unix: Option<u64>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub entire_checkpoint: Option<EntireCheckpointRef>,
}
#[derive(Clone, Debug, Default, Eq, PartialEq)]
pub struct ReviewRunStatusCounts {
pub queued: usize,
pub running: usize,
pub completed: usize,
pub failed: usize,
pub cancelled: usize,
pub skipped_records: usize,
}
impl ReviewRunStatusCounts {
fn add(&mut self, status: ReviewRunStatus) {
match status {
ReviewRunStatus::Queued => self.queued += 1,
ReviewRunStatus::Running => self.running += 1,
ReviewRunStatus::Completed => self.completed += 1,
ReviewRunStatus::Failed => self.failed += 1,
ReviewRunStatus::Cancelled => self.cancelled += 1,
}
}
}
#[derive(Deserialize)]
struct ReviewRunStatusRecord {
status: ReviewRunStatus,
}
impl ReviewRun {
#[cfg(test)]
fn queued(
id: impl Into<String>,
commit_sha: impl Into<String>,
target: impl Into<String>,
) -> Self {
Self::queued_with_provenance(id, commit_sha, target, None)
}
fn queued_with_provenance(
id: impl Into<String>,
commit_sha: impl Into<String>,
target: impl Into<String>,
entire_checkpoint: Option<EntireCheckpointRef>,
) -> Self {
let timestamp = unix_now();
Self {
id: id.into(),
commit_sha: commit_sha.into(),
target: target.into(),
status: ReviewRunStatus::Queued,
phase: "queued".to_owned(),
ledger_entries: 0,
error: None,
worker_pid: None,
worker_start_token: None,
created_at_unix: timestamp,
updated_at_unix: timestamp,
started_at_unix: None,
completed_at_unix: None,
entire_checkpoint,
}
}
fn mark_running(&mut self, phase: impl Into<String>) {
let timestamp = unix_now();
let identity = crate::watcher::ProcessIdentity::current();
self.status = ReviewRunStatus::Running;
self.phase = phase.into();
self.error = None;
self.worker_pid = Some(identity.pid);
self.worker_start_token =
(!identity.start_token.is_empty()).then_some(identity.start_token);
self.updated_at_unix = timestamp;
self.started_at_unix = Some(timestamp);
self.completed_at_unix = None;
}
fn mark_completed(&mut self, ledger_entries: usize) {
let timestamp = unix_now();
self.status = ReviewRunStatus::Completed;
self.phase = "completed".to_owned();
self.ledger_entries = ledger_entries;
self.error = None;
self.worker_pid = None;
self.worker_start_token = None;
self.updated_at_unix = timestamp;
self.completed_at_unix = Some(timestamp);
}
fn mark_failed(&mut self, error: impl Into<String>) {
let timestamp = unix_now();
self.status = ReviewRunStatus::Failed;
self.phase = "failed".to_owned();
self.error = Some(error.into());
self.worker_pid = None;
self.worker_start_token = None;
self.updated_at_unix = timestamp;
self.completed_at_unix = Some(timestamp);
}
fn mark_cancelled(&mut self) {
let timestamp = unix_now();
self.status = ReviewRunStatus::Cancelled;
self.phase = "cancelled".to_owned();
self.error = None;
self.worker_pid = None;
self.worker_start_token = None;
self.updated_at_unix = timestamp;
self.completed_at_unix = Some(timestamp);
}
/// Reconcile a `Running` run against worker liveness. If its recorded worker
/// identity is proven dead — pid gone, or pid reused under a different start
/// token — the worker died without recording a verdict: flip the zombie to
/// `Failed` with an explanatory reason.
///
/// The probe is TRI-STATE: [`crate::watcher::WorkerLiveness::Unknown`]
/// (the OS probe itself failed) is never treated as proof of death — the
/// row is left alone and [`ReconcileOutcome::ProbeUnknown`] is returned so
/// the caller can log the inconclusive probe. A run with no recorded pid
/// (legacy record) is likewise left untouched: only an explicit forced
/// cancel reaps it.
fn reconcile_liveness(
&mut self,
probe: impl Fn(&crate::watcher::ProcessIdentity) -> crate::watcher::WorkerLiveness,
) -> ReconcileOutcome {
if self.status != ReviewRunStatus::Running {
return ReconcileOutcome::Unchanged;
}
let Some(pid) = self.worker_pid else {
return ReconcileOutcome::Unchanged;
};
let identity = crate::watcher::ProcessIdentity {
pid,
start_token: self.worker_start_token.clone().unwrap_or_default(),
};
match probe(&identity) {
crate::watcher::WorkerLiveness::Dead => {
self.mark_failed(stale_worker_reason(pid));
ReconcileOutcome::Reaped
}
crate::watcher::WorkerLiveness::Alive => ReconcileOutcome::Unchanged,
crate::watcher::WorkerLiveness::Unknown => ReconcileOutcome::ProbeUnknown,
}
}
}
/// Outcome of reconciling one run against worker liveness.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum ReconcileOutcome {
/// The worker is proven dead; the run was flipped to `Failed`.
Reaped,
/// Nothing to do: not running, no recorded pid, or the worker is alive.
Unchanged,
/// The liveness probe could not prove the worker alive or dead. The row
/// was deliberately left alone — a probe failure is NOT proof of death.
ProbeUnknown,
}
/// Reason recorded when a running review is found abandoned by a dead worker.
fn stale_worker_reason(pid: u32) -> String {
format!("worker process {pid} exited without recording a verdict (stale run)")
}
/// Force-terminate a process with SIGKILL. Used by `review cancel --force` on a
/// worker that is still alive.
fn kill_pid(pid: u32) -> Result<(), ReviewerError> {
let status = Command::new("kill")
.arg("-KILL")
.arg(pid.to_string())
.stdout(Stdio::null())
.stderr(Stdio::null())
.status()
.map_err(ReviewerError::KillWorker)?;
if status.success() || !crate::watcher::pid_is_alive(pid) {
Ok(())
} else {
Err(ReviewerError::KillWorkerFailed { pid })
}
}
#[derive(Clone, Debug)]
pub struct ReviewRunStore {
root: PathBuf,
}
/// Directory name of the run-store mutation mutex inside the state dir.
const RUN_STORE_LOCK_DIR: &str = "review-runs.lock";
const RUN_STORE_LOCK_OWNER_FILE: &str = "review-runs.lock.owner";
const RUN_STORE_LOCK_HEARTBEAT_FILE: &str = "review-runs.lock.heartbeat";
const RUN_STORE_LOCK_LEASE_PREFIX: &str = ".review-runs.lock.lease";
/// The heartbeat is refreshed while the guard is held. A lock is stale only
/// after this many missed refresh intervals AND a process probe proves that
/// the recorded owner is dead. A fresh heartbeat never overrides a live
/// identity probe, and an Unknown probe is never reclaimed automatically.
const RUN_STORE_LOCK_HEARTBEAT_INTERVAL: Duration = Duration::from_millis(100);
const RUN_STORE_LOCK_STALE_INTERVALS: u64 = 3;
/// Total time to wait for a live contender before failing the mutation.
/// Long enough that ordinary contention (sub-millisecond critical sections)
/// never trips it, short enough to surface a genuine jam.
const RUN_STORE_LOCK_WAIT: Duration = Duration::from_secs(10);
const RUN_STORE_LOCK_POLL: Duration = Duration::from_millis(25);
static RUN_STORE_LOCK_TOKEN_COUNTER: AtomicU64 = AtomicU64::new(0);
#[derive(Clone, Debug, Deserialize, Serialize)]
struct RunStoreLockRecord {
owner: crate::watcher::ProcessIdentity,
acquisition_token: String,
}
/// Mutual-exclusion guard for run-store mutations. Implemented as an atomic
/// `create_new` + hard-link pair. The fixed lock path and the guard's unique
/// lease path are separate directory entries for the same inode, so releasing
/// an old guard can never unlink a replacement lease's unique path. Dropping
/// the guard releases the lock.
struct RunStoreLock {
path: PathBuf,
lease_path: PathBuf,
owner: crate::watcher::ProcessIdentity,
acquisition_token: String,
heartbeat_stop: Arc<AtomicBool>,
heartbeat: Option<std::thread::JoinHandle<()>>,
}
impl Drop for RunStoreLock {
fn drop(&mut self) {
self.heartbeat_stop.store(true, Ordering::Release);
if let Some(heartbeat) = self.heartbeat.take() {
let _ = heartbeat.join();
}
release_run_store_lock(
&self.path,
&self.lease_path,
&self.owner,
&self.acquisition_token,
);
}
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
enum RunStoreLockLiveness {
/// Recorded owner is still proven alive.
Alive,
/// Recorded owner is dead or replaced (pid reuse mismatch).
Dead,
/// Probe could not prove status (for example start token not readable).
/// We treat this as held and do not reclaim on this signal alone.
Unknown,
}
fn run_store_lock_owner_path(path: &Path) -> PathBuf {
path.join(RUN_STORE_LOCK_OWNER_FILE)
}
fn run_store_lock_heartbeat_path(path: &Path) -> PathBuf {
path.join(RUN_STORE_LOCK_HEARTBEAT_FILE)
}
fn run_store_lock_is_owned_by_current_process(
path: &Path,
owner: &crate::watcher::ProcessIdentity,
acquisition_token: &str,
) -> bool {
let Ok(record) = read_run_store_lock_record(path) else {
return false;
};
identity_matches_owner(&record.owner, owner) && record.acquisition_token == acquisition_token
}
fn identity_matches_owner(
recorded: &crate::watcher::ProcessIdentity,
current: &crate::watcher::ProcessIdentity,
) -> bool {
recorded.pid == current.pid
&& (recorded.start_token.is_empty()
|| current.start_token.is_empty()
|| recorded.start_token == current.start_token)
}
fn decide_run_store_lock_liveness(
recorded: &crate::watcher::ProcessIdentity,
) -> RunStoreLockLiveness {
match crate::watcher::probe_worker_identity(recorded) {
crate::watcher::WorkerLiveness::Alive => RunStoreLockLiveness::Alive,
crate::watcher::WorkerLiveness::Dead => RunStoreLockLiveness::Dead,
crate::watcher::WorkerLiveness::Unknown => RunStoreLockLiveness::Unknown,
}
}
fn lock_is_stale_by_heartbeat(path: &Path) -> bool {
let heartbeat_path = if path.is_dir() {
run_store_lock_heartbeat_path(path)
} else {
path.to_path_buf()
};
let Ok(metadata) = fs::metadata(heartbeat_path) else {
return false;
};
let Ok(modified) = metadata.modified() else {
return false;
};
match SystemTime::now().duration_since(modified) {
Ok(age) => age >= RUN_STORE_LOCK_HEARTBEAT_INTERVAL * RUN_STORE_LOCK_STALE_INTERVALS as u32,
Err(_) => false,
}
}
fn read_run_store_lock_record(path: &Path) -> Result<RunStoreLockRecord, io::Error> {
let owner_path = if path.is_dir() {
run_store_lock_owner_path(path)
} else {
path.to_path_buf()
};
let contents = fs::read_to_string(owner_path)?;
if path.is_dir() {
let owner = serde_json::from_str::<crate::watcher::ProcessIdentity>(&contents)
.map_err(io::Error::other)?;
Ok(RunStoreLockRecord {
owner,
acquisition_token: String::new(),
})
} else {
serde_json::from_str::<RunStoreLockRecord>(&contents).map_err(io::Error::other)
}
}
#[cfg(test)]
fn write_run_store_lock_owner(
lock_dir: &Path,
owner: &crate::watcher::ProcessIdentity,
) -> Result<(), ReviewerError> {
fs::create_dir_all(lock_dir).map_err(ReviewerError::RunIo)?;
let bytes = serde_json::to_vec_pretty(owner).map_err(ReviewerError::RunJson)?;
fs::write(run_store_lock_owner_path(lock_dir), bytes).map_err(ReviewerError::RunIo)?;
fs::write(run_store_lock_heartbeat_path(lock_dir), b"heartbeat")
.map_err(ReviewerError::RunIo)?;
Ok(())
}
/// Whether the lock is considered stale by owner identity and heartbeat evidence.
///
/// A persistent Unknown state is deliberately fail-closed: it is treated as
/// held until the contender's normal wait expires, which returns a timeout and
/// tells the operator to inspect the lock. There is no automatic "old enough"
/// escape hatch because that would turn an unprobeable live owner into a data
/// loss race.
fn run_store_lock_is_stale(path: &Path) -> bool {
let Ok(record) = read_run_store_lock_record(path) else {
return false;
};
matches!(
decide_run_store_lock_liveness(&record.owner),
RunStoreLockLiveness::Dead
) && lock_is_stale_by_heartbeat(path)
}
fn new_run_store_lock_token() -> String {
let nanos = SystemTime::now()
.duration_since(UNIX_EPOCH)
.map_or(0, |duration| duration.as_nanos());
let counter = RUN_STORE_LOCK_TOKEN_COUNTER.fetch_add(1, Ordering::Relaxed);
format!("{}-{nanos}-{counter}", std::process::id())
}
fn write_run_store_lock_record(
lease_path: &Path,
record: &RunStoreLockRecord,
) -> Result<(), ReviewerError> {
let bytes = serde_json::to_vec_pretty(record).map_err(ReviewerError::RunJson)?;
let mut file = fs::OpenOptions::new()
.write(true)
.truncate(true)
.open(lease_path)
.map_err(ReviewerError::RunIo)?;
file.write_all(&bytes).map_err(ReviewerError::RunIo)?;
file.write_all(b"\n").map_err(ReviewerError::RunIo)?;
file.sync_all().map_err(ReviewerError::RunIo)?;
Ok(())
}
fn refresh_run_store_lock_heartbeat(lease_path: &Path) -> io::Result<()> {
let mut file = fs::OpenOptions::new().append(true).open(lease_path)?;
// Whitespace keeps the JSON owner record parseable while updating the
// hard-linked inode's mtime. The unique lease path means this can never
// refresh a replacement lock after the fixed link changes.
file.write_all(b"\n")?;
file.sync_data()
}
fn spawn_run_store_lock_heartbeat(
lease_path: PathBuf,
stop: Arc<AtomicBool>,
) -> std::thread::JoinHandle<()> {
std::thread::spawn(move || {
while !stop.load(Ordering::Acquire) {
std::thread::sleep(RUN_STORE_LOCK_HEARTBEAT_INTERVAL);
if stop.load(Ordering::Acquire) {
break;
}
if let Err(error) = refresh_run_store_lock_heartbeat(&lease_path) {
tracing::debug!(%error, "run-store lock heartbeat refresh stopped");
break;
}
}
})
}
fn release_run_store_lock(
path: &Path,
lease_path: &Path,
owner: &crate::watcher::ProcessIdentity,
acquisition_token: &str,
) {
if run_store_lock_is_owned_by_current_process(path, owner, acquisition_token) {
let _ = fs::remove_file(path);
}
// This is our private acquisition path, never the replacement's path.
let _ = fs::remove_file(lease_path);
}
impl ReviewRunStore {
pub fn new(root: impl Into<PathBuf>) -> Self {
Self { root: root.into() }
}
pub fn runs_dir(&self) -> PathBuf {
self.root.join(REVIEW_RUNS_DIR)
}
pub fn path(&self, id: &str) -> PathBuf {
self.runs_dir().join(format!("{id}.json"))
}
/// Take the run-store mutation lock, reclaiming it from a crashed holder.
///
/// EVERY read-modify-write on a run row — worker transitions AND stale-row
/// reconciliation — happens under this lock, so a reconciler can never
/// overwrite a completion that landed between its snapshot read and its
/// write (0.9.2 clobbered `completed` rows back to `failed`).
fn lock(&self) -> Result<RunStoreLock, ReviewerError> {
fs::create_dir_all(&self.root).map_err(ReviewerError::RunIo)?;
let path = self.root.join(RUN_STORE_LOCK_DIR);
let owner = crate::watcher::ProcessIdentity::current();
let started = Instant::now();
loop {
// Handle the legacy directory-shaped lease before attempting the
// file hard-link acquisition. This also keeps a concurrent stale
// reclaim from being mistaken for an unsupported hard-link target.
if path.exists() {
if run_store_lock_is_stale(&path) {
let remove_result = if path.is_dir() {
fs::remove_dir_all(&path)
} else {
fs::remove_file(&path)
};
match remove_result {
Ok(()) => continue,
Err(error) if error.kind() == io::ErrorKind::NotFound => continue,
Err(error) => return Err(ReviewerError::RunIo(error)),
}
}
if started.elapsed() >= RUN_STORE_LOCK_WAIT {
if let Ok(record) = read_run_store_lock_record(&path)
&& matches!(
decide_run_store_lock_liveness(&record.owner),
RunStoreLockLiveness::Unknown
)
{
tracing::warn!(
"run-store lock owner remained Unknown; refusing automatic reclaim"
);
}
return Err(ReviewerError::RunStoreLockTimeout {
wait_secs: RUN_STORE_LOCK_WAIT.as_secs(),
});
}
std::thread::sleep(RUN_STORE_LOCK_POLL);
continue;
}
let acquisition_token = new_run_store_lock_token();
let lease_path = self
.root
.join(format!("{RUN_STORE_LOCK_LEASE_PREFIX}.{acquisition_token}"));
let record = RunStoreLockRecord {
owner: owner.clone(),
acquisition_token: acquisition_token.clone(),
};
match fs::OpenOptions::new()
.write(true)
.create_new(true)
.open(&lease_path)
{
Ok(_) => {
if let Err(error) = write_run_store_lock_record(&lease_path, &record) {
let _ = fs::remove_file(&lease_path);
return Err(error);
}
match fs::hard_link(&lease_path, &path) {
Ok(()) => {
let heartbeat_stop = Arc::new(AtomicBool::new(false));
let heartbeat = spawn_run_store_lock_heartbeat(
lease_path.clone(),
Arc::clone(&heartbeat_stop),
);
return Ok(RunStoreLock {
path: path.clone(),
lease_path,
owner,
acquisition_token,
heartbeat_stop,
heartbeat: Some(heartbeat),
});
}
Err(error) if error.kind() == io::ErrorKind::AlreadyExists => {
let _ = fs::remove_file(&lease_path);
}
Err(error) => {
let _ = fs::remove_file(&lease_path);
return Err(ReviewerError::RunIo(error));
}
}
}
Err(error) if error.kind() == io::ErrorKind::AlreadyExists => continue,
Err(error) => return Err(ReviewerError::RunIo(error)),
}
if started.elapsed() >= RUN_STORE_LOCK_WAIT {
if let Ok(record) = read_run_store_lock_record(&path)
&& matches!(
decide_run_store_lock_liveness(&record.owner),
RunStoreLockLiveness::Unknown
)
{
tracing::warn!(
"run-store lock owner remained Unknown; refusing automatic reclaim"
);
}
return Err(ReviewerError::RunStoreLockTimeout {
wait_secs: RUN_STORE_LOCK_WAIT.as_secs(),
});
}
std::thread::sleep(RUN_STORE_LOCK_POLL);
}
}
pub fn create_queued(
&self,
commit_sha: &str,
target: impl Into<String>,
) -> Result<ReviewRun, ReviewerError> {
let _guard = self.lock()?;
let checkpoint = entire_checkpoint_for_current_repo(commit_sha);
let run = ReviewRun::queued_with_provenance(
generate_run_id(commit_sha),
commit_sha,
target,
checkpoint,
);
self.write(&run)?;
Ok(run)
}
fn ensure_queued(
&self,
run_id: &str,
commit_sha: &str,
target: &str,
) -> Result<ReviewRun, ReviewerError> {
let _guard = self.lock()?;
match self.read(run_id) {
Ok(run) => Ok(run),
Err(ReviewerError::ReviewRunNotFound { .. }) => {
let checkpoint = entire_checkpoint_for_current_repo(commit_sha);
let run = ReviewRun::queued_with_provenance(run_id, commit_sha, target, checkpoint);
self.write(&run)?;
Ok(run)
}
Err(error) => Err(error),
}
}
pub fn read(&self, id: &str) -> Result<ReviewRun, ReviewerError> {
let path = self.path(id);
let contents = fs::read_to_string(&path).map_err(|source| match source.kind() {
io::ErrorKind::NotFound => ReviewerError::ReviewRunNotFound { id: id.to_owned() },
_ => ReviewerError::RunIo(source),
})?;
serde_json::from_str(&contents).map_err(ReviewerError::RunJson)
}
pub fn list(&self) -> Result<Vec<ReviewRun>, ReviewerError> {
let dir = self.runs_dir();
let entries = match fs::read_dir(&dir) {
Ok(entries) => entries,
Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok(Vec::new()),
Err(error) => return Err(ReviewerError::RunIo(error)),
};
let mut runs: Vec<ReviewRun> = Vec::new();
for entry in entries {
let entry = entry.map_err(ReviewerError::RunIo)?;
if entry
.path()
.extension()
.is_none_or(|extension| extension != "json")
{
continue;
}
let contents = fs::read_to_string(entry.path()).map_err(ReviewerError::RunIo)?;
runs.push(serde_json::from_str(&contents).map_err(ReviewerError::RunJson)?);
}
runs.sort_by(|left, right| {
right
.updated_at_unix
.cmp(&left.updated_at_unix)
.then_with(|| right.id.cmp(&left.id))
});
Ok(runs)
}
pub fn status_counts(&self) -> Result<ReviewRunStatusCounts, ReviewerError> {
let dir = self.runs_dir();
let entries = match fs::read_dir(&dir) {
Ok(entries) => entries,
Err(error) if error.kind() == io::ErrorKind::NotFound => {
return Ok(ReviewRunStatusCounts::default());
}
Err(error) => return Err(ReviewerError::RunIo(error)),
};
let mut counts = ReviewRunStatusCounts::default();
for entry in entries {
let Ok(entry) = entry else {
counts.skipped_records += 1;
continue;
};
let path = entry.path();
if path.extension().is_none_or(|extension| extension != "json") {
continue;
}
let Ok(contents) = fs::read_to_string(&path) else {
counts.skipped_records += 1;
continue;
};
let Ok(record) = serde_json::from_str::<ReviewRunStatusRecord>(&contents) else {
counts.skipped_records += 1;
continue;
};
counts.add(record.status);
}
Ok(counts)
}
pub fn latest_result(&self) -> Result<ReviewRun, ReviewerError> {
self.list()?
.into_iter()
.find(|run| {
matches!(
run.status,
ReviewRunStatus::Completed
| ReviewRunStatus::Failed
| ReviewRunStatus::Cancelled
)
})
.ok_or(ReviewerError::NoReviewRuns)
}
pub fn mark_running(&self, id: &str, phase: &str) -> Result<ReviewRun, ReviewerError> {
let _guard = self.lock()?;
let mut run = self.read(id)?;
run.mark_running(phase);
self.write(&run)?;
Ok(run)
}
pub fn mark_completed(
&self,
id: &str,
ledger_entries: usize,
) -> Result<ReviewRun, ReviewerError> {
let _guard = self.lock()?;
let mut run = self.read(id)?;
run.mark_completed(ledger_entries);
self.write(&run)?;
Ok(run)
}
pub fn mark_failed(
&self,
id: &str,
error: impl Into<String>,
) -> Result<ReviewRun, ReviewerError> {
let _guard = self.lock()?;
let mut run = self.read(id)?;
run.mark_failed(error);
self.write(&run)?;
Ok(run)
}
pub fn cancel_queued(&self, id: &str) -> Result<ReviewRun, ReviewerError> {
let _guard = self.lock()?;
let mut run = self.read(id)?;
if run.status != ReviewRunStatus::Queued {
return Err(ReviewerError::CannotCancelReview {
id: id.to_owned(),
status: run.status,
});
}
run.mark_cancelled();
self.write(&run)?;
Ok(run)
}
/// Read a run and, if it is a `Running` zombie (recorded worker identity is
/// proven dead), flip it to `Failed` and persist the change so `review
/// status` reports the truth instead of a run stuck `running` forever.
///
/// An inconclusive liveness probe leaves the row untouched (a probe
/// failure is not proof of death) and is logged, never silently reap.
pub fn read_reconciled(&self, id: &str) -> Result<ReviewRun, ReviewerError> {
let _guard = self.lock()?;
let mut run = self.read(id)?;
match run.reconcile_liveness(crate::watcher::probe_worker_identity) {
ReconcileOutcome::Reaped => self.write(&run)?,
ReconcileOutcome::ProbeUnknown => log_inconclusive_probe(&run),
ReconcileOutcome::Unchanged => {}
}
Ok(run)
}
/// List all runs, reconciling any `Running` zombies to `Failed` and persisting
/// the correction, so a dead worker never lingers as `running`.
pub fn list_reconciled(&self) -> Result<Vec<ReviewRun>, ReviewerError> {
let _guard = self.lock()?;
let mut runs = self.list()?;
for run in &mut runs {
match run.reconcile_liveness(crate::watcher::probe_worker_identity) {
ReconcileOutcome::Reaped => self.write(run)?,
ReconcileOutcome::ProbeUnknown => log_inconclusive_probe(run),
ReconcileOutcome::Unchanged => {}
}
}
Ok(runs)
}
/// Reap orphaned `Running` rows whose recorded worker identity is proven
/// dead, persisting each correction, and return how many rows were reaped.
///
/// `ensure-watcher` runs this before its spawn decision: after a watcher
/// is killed, its in-flight rows must stop masquerading as live work —
/// 0.9.2 left them `running` forever, which hid the fact that no live
/// watcher existed at all.
///
/// The whole read-probe-write sequence runs under the run-store lock, so
/// a worker completing concurrently can never be clobbered back to
/// `failed` by a stale snapshot. Rows whose probe is inconclusive are
/// left alone with a logged notice.
pub fn reconcile_stale_runs(&self) -> Result<usize, ReviewerError> {
let _guard = self.lock()?;
let mut reaped = 0;
for mut run in self.list()? {
match run.reconcile_liveness(crate::watcher::probe_worker_identity) {
ReconcileOutcome::Reaped => {
self.write(&run)?;
reaped += 1;
}
ReconcileOutcome::ProbeUnknown => log_inconclusive_probe(&run),
ReconcileOutcome::Unchanged => {}
}
}
Ok(reaped)
}
/// Count `Running` rows whose recorded worker identity is proven dead.
/// READ-ONLY: unlike [`Self::reconcile_stale_runs`] nothing is persisted —
/// this is the truthful view `status` reports while leaving the write to
/// the watcher lifecycle. Rows with an inconclusive probe are NOT
/// counted: an unproven death is not a stale row.
pub fn stale_running_count(&self) -> Result<usize, ReviewerError> {
let mut stale = 0;
for mut run in self.list()? {
if run.reconcile_liveness(crate::watcher::probe_worker_identity)
== ReconcileOutcome::Reaped
{
stale += 1;
}
}
Ok(stale)
}
/// Cancel a review run, tolerating a running-but-dead worker.
///
/// - `Queued` runs cancel outright.
/// - A `Running` run whose worker identity is proven dead is reaped (marked
/// `Failed` stale), so a zombie can always be cleaned up.
/// - A `Running` run with a live worker is refused unless `force`, which
/// SIGKILLs the worker and then marks the run `Cancelled`.
/// - A `Running` run whose liveness is UNPROVABLE (probe inconclusive, or a
/// legacy run with no recorded pid) is refused unless `force` — an
/// unproven death never justifies a reap.
/// - Terminal runs (completed/failed/cancelled) are refused.
pub fn cancel(&self, id: &str, force: bool) -> Result<ReviewRun, ReviewerError> {
let _guard = self.lock()?;
let mut run = self.read(id)?;
match run.status {
ReviewRunStatus::Queued => run.mark_cancelled(),
ReviewRunStatus::Running => match run.worker_pid {
Some(pid) => {
let identity = crate::watcher::ProcessIdentity {
pid,
start_token: run.worker_start_token.clone().unwrap_or_default(),
};
match crate::watcher::probe_worker_identity(&identity) {
crate::watcher::WorkerLiveness::Alive => {
if !force {
return Err(ReviewerError::ReviewRunStillAlive {
id: id.to_owned(),
pid,
});
}
kill_pid(pid)?;
run.mark_cancelled();
}
crate::watcher::WorkerLiveness::Dead => {
run.mark_failed(stale_worker_reason(pid));
}
crate::watcher::WorkerLiveness::Unknown => {
if !force {
return Err(ReviewerError::ReviewRunLivenessUnknown {
id: id.to_owned(),
});
}
run.mark_failed(
"worker liveness could not be verified; force-cancelled",
);
}
}
}
None => {
if !force {
return Err(ReviewerError::ReviewRunLivenessUnknown { id: id.to_owned() });
}
run.mark_failed("worker liveness could not be verified; force-cancelled");
}
},
terminal => {
return Err(ReviewerError::CannotCancelReview {
id: id.to_owned(),
status: terminal,
});
}
}
self.write(&run)?;
Ok(run)
}
/// Persist a run row atomically: write to a temp file, fsync, then rename
/// over the target. Same-dir renames are atomic on the supported
/// platforms, so a concurrent reader never observes a torn row.
fn write(&self, run: &ReviewRun) -> Result<(), ReviewerError> {
fs::create_dir_all(self.runs_dir()).map_err(ReviewerError::RunIo)?;
let bytes = serde_json::to_vec_pretty(run).map_err(ReviewerError::RunJson)?;
let nanos = SystemTime::now()
.duration_since(UNIX_EPOCH)
.map_or(0, |duration| duration.as_nanos());
let temp_path =
self.runs_dir()
.join(format!(".{}.{}.{}.tmp", run.id, std::process::id(), nanos));
{
let mut file = fs::File::create(&temp_path).map_err(ReviewerError::RunIo)?;
file.write_all(&bytes).map_err(ReviewerError::RunIo)?;
file.write_all(b"\n").map_err(ReviewerError::RunIo)?;
file.sync_all().map_err(ReviewerError::RunIo)?;
}
fs::rename(&temp_path, self.path(&run.id)).map_err(ReviewerError::RunIo)
}
}
/// Surface an inconclusive worker-liveness probe: the row is left alone, but
/// the uncertainty is logged, never silently ignored.
fn log_inconclusive_probe(run: &ReviewRun) {
tracing::warn!(
run_id = %run.id,
worker_pid = run.worker_pid,
"worker liveness probe inconclusive; leaving the running row untouched (a probe failure is not proof of death)"
);
}
fn entire_checkpoint_for_current_repo(commit_sha: &str) -> Option<EntireCheckpointRef> {
let repo_root = current_repo_root().unwrap_or_else(|| PathBuf::from("."));
provenance::entire_checkpoint_for_commit(&repo_root, commit_sha)
}
fn current_repo_root() -> Option<PathBuf> {
let output = Command::new("git")
.args(["rev-parse", "--show-toplevel"])
.output()
.ok()?;
output
.status
.success()
.then(|| PathBuf::from(String::from_utf8_lossy(&output.stdout).trim()))
}
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct QueuedReview {
#[serde(default)]
pub run_id: String,
pub commit_sha: String,
pub enqueued_at_unix: u64,
/// When set, this queued item is a petition re-review: `commit_sha` is the
/// FIX commit and this field names the original rejection it petitions.
/// `None` (the default, omitted on the wire) is a normal commit review, so
/// pre-existing queue lines keep parsing.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub petition_for: Option<String>,
}
#[derive(Clone, Debug)]
pub struct ReviewQueue {
root: PathBuf,
}
#[derive(Clone, Debug, Default, Eq, PartialEq)]
pub struct ReviewQueueSummary {
pub pending_count: usize,
pub oldest_enqueued_at_unix: Option<u64>,
}
impl ReviewQueueSummary {
pub fn oldest_age_secs_at(&self, now: u64) -> Option<u64> {
self.oldest_enqueued_at_unix
.map(|oldest| now.saturating_sub(oldest))
}
}
impl ReviewQueue {
pub fn new(root: impl Into<PathBuf>) -> Self {
Self { root: root.into() }
}
pub fn path(&self) -> PathBuf {
self.root.join(REVIEW_QUEUE_FILE)
}
pub fn enqueue(&self, commit_sha: impl Into<String>) -> Result<QueuedReview, ReviewerError> {
self.enqueue_item(commit_sha.into(), None)
}
/// Enqueue a petition re-review: `fix_sha` is reviewed with the petition
/// prompt, and its verdict transitions the original rejection
/// (`original_sha`). This is the queue leg of `truth-mirror resolve
/// --fixed-by` — without it the watcher would never build a petition job.
pub fn enqueue_petition(
&self,
fix_sha: impl Into<String>,
original_sha: impl Into<String>,
) -> Result<QueuedReview, ReviewerError> {
self.enqueue_item(fix_sha.into(), Some(original_sha.into()))
}
fn enqueue_item(
&self,
commit_sha: String,
petition_for: Option<String>,
) -> Result<QueuedReview, ReviewerError> {
fs::create_dir_all(&self.root).map_err(ReviewerError::QueueIo)?;
let target = if petition_for.is_some() {
"petition"
} else {
"commit"
};
let run = ReviewRunStore::new(&self.root).create_queued(&commit_sha, target)?;
let item = QueuedReview {
run_id: run.id,
commit_sha,
enqueued_at_unix: unix_now(),
petition_for,
};
let mut file = fs::OpenOptions::new()
.create(true)
.append(true)
.open(self.path())
.map_err(ReviewerError::QueueIo)?;
serde_json::to_writer(&mut file, &item).map_err(ReviewerError::QueueJson)?;
writeln!(file).map_err(ReviewerError::QueueIo)?;
Ok(item)
}
pub fn pending(&self) -> Result<Vec<QueuedReview>, ReviewerError> {
let contents = match fs::read_to_string(self.path()) {
Ok(contents) => contents,
Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok(Vec::new()),
Err(error) => return Err(ReviewerError::QueueIo(error)),
};
contents
.lines()
.filter(|line| !line.trim().is_empty())
.map(|line| serde_json::from_str(line).map_err(ReviewerError::QueueJson))
.collect()
}
pub fn summary(&self) -> Result<ReviewQueueSummary, ReviewerError> {
let pending = self.pending()?;
Ok(ReviewQueueSummary {
pending_count: pending.len(),
oldest_enqueued_at_unix: pending.iter().map(|item| item.enqueued_at_unix).min(),
})
}
/// Drop every queued item for `sha`, preserving any items appended for other
/// commits. Called after a commit is reviewed so a drain never repeats work.
pub fn remove_sha(&self, sha: &str) -> Result<(), ReviewerError> {
let remaining: Vec<QueuedReview> = self
.pending()?
.into_iter()
.filter(|item| item.commit_sha != sha)
.collect();
self.rewrite(&remaining)
}
fn rewrite(&self, items: &[QueuedReview]) -> Result<(), ReviewerError> {
if items.is_empty() {
return match fs::remove_file(self.path()) {
Ok(()) => Ok(()),
Err(error) if error.kind() == io::ErrorKind::NotFound => Ok(()),
Err(error) => Err(ReviewerError::QueueIo(error)),
};
}
let mut file = fs::File::create(self.path()).map_err(ReviewerError::QueueIo)?;
for item in items {
serde_json::to_writer(&mut file, item).map_err(ReviewerError::QueueJson)?;
writeln!(file).map_err(ReviewerError::QueueIo)?;
}
Ok(())
}
/// Drop the queued item matching BOTH the commit SHA and the petition
/// identity, preserving sibling rows for the same SHA (a fix commit's own
/// review and a petition review of it are distinct queue items).
pub fn remove_item(&self, sha: &str, petition_for: Option<&str>) -> Result<(), ReviewerError> {
let remaining: Vec<QueuedReview> = self
.pending()?
.into_iter()
.filter(|item| {
!(item.commit_sha == sha && item.petition_for.as_deref() == petition_for)
})
.collect();
self.rewrite(&remaining)
}
pub fn remove_run_id(&self, run_id: &str) -> Result<(), ReviewerError> {
let remaining: Vec<QueuedReview> = self
.pending()?
.into_iter()
.filter(|item| item.run_id != run_id)
.collect();
self.rewrite(&remaining)
}
}
/// Loads the claim and diff for a commit so the reviewer can run against it.
/// Abstracted so `drain_once` can be unit-tested without a real git repository.
pub trait MaterialLoader {
fn load(&self, sha: &str) -> Result<(Claim, String), ReviewerError>;
}
#[derive(Clone, Debug, Default)]
pub struct GitMaterialLoader {
/// Evidence-pointer patterns from config so async review parses claims the
/// same way the commit-msg gate did (a repo `jira:` pointer stays valid).
pub evidence_patterns: Vec<String>,
}
impl GitMaterialLoader {
pub fn from_config(config: &config::TruthMirrorConfig) -> Self {
Self::with_patterns(config.gates.to_policy().evidence_patterns)
}
pub fn with_patterns(evidence_patterns: Vec<String>) -> Self {
Self { evidence_patterns }
}
}
impl MaterialLoader for GitMaterialLoader {
fn load(&self, sha: &str) -> Result<(Claim, String), ReviewerError> {
let message = git_output(["show", "--format=%B", "--no-patch", sha])?;
let diff = git_output(["show", "--format=", "--patch", sha])?;
let claim = if self.evidence_patterns.is_empty() {
Claim::parse(&message)?
} else {
Claim::parse_with(&message, &self.evidence_patterns)?
};
Ok((claim, diff))
}
}
#[derive(Clone, Debug, Default, Eq, PartialEq)]
pub struct DrainReport {
pub reviewed: Vec<String>,
/// Queue items skipped after a terminal material-loading failure.
pub failed: Vec<String>,
pub ledger_entries: usize,
}
/// Review every distinct queued commit exactly once, record verdicts, and remove
/// each commit from the queue as soon as its review lands. A material-loading
/// failure is terminal for that run: it is recorded and removed so one malformed
/// claim never strands the rest of the queue. Reviewer execution failures remain
/// queued for retry.
pub fn drain_once<R: ProcessRunner, L: MaterialLoader>(
queue: &ReviewQueue,
loader: &L,
selection: &ReviewSelection,
context: &str,
runner: &R,
store: &LedgerStore,
config: &config::TruthMirrorConfig,
) -> Result<DrainReport, ReviewerError> {
let pending = queue.pending()?;
let run_store = ReviewRunStore::new(&queue.root);
// Per-review wall-clock limit from config; a wedged reviewer is killed at
// this deadline and the run recorded as failed (see the drain loop below).
let timeout = config.reviewer.timeout();
// Queue identity is (commit_sha, petition_for), NOT the bare SHA: a fix
// commit's own post-commit review and a petition review of that same fix
// (or two petitions for different rejections sharing one fix) are
// distinct jobs — deduping on SHA alone silently cancelled one of them.
let mut seen = std::collections::BTreeSet::new();
let mut order = Vec::new();
for item in &pending {
let identity = (item.commit_sha.clone(), item.petition_for.clone());
if seen.insert(identity) {
order.push(item.clone());
} else if !item.run_id.trim().is_empty() {
match run_store.read(&item.run_id) {
Ok(run) if run.status == ReviewRunStatus::Queued => {
run_store.cancel_queued(&item.run_id)?;
}
_ => {}
}
}
}
let mut report = DrainReport::default();
for item in order {
let sha = item.commit_sha;
let run_id = if item.run_id.trim().is_empty() {
generate_run_id(&sha)
} else {
item.run_id
};
let target = if item.petition_for.is_some() {
"petition"
} else {
"commit"
};
let run = run_store.ensure_queued(&run_id, &sha, target)?;
if run.status == ReviewRunStatus::Cancelled {
queue.remove_item(&sha, item.petition_for.as_deref())?;
continue;
}
run_store.mark_running(&run_id, "reviewing")?;
let (claim, diff) = match loader.load(&sha) {
Ok(material) => material,
Err(error) => {
run_store.mark_failed(&run_id, error.to_string())?;
queue.remove_item(&sha, item.petition_for.as_deref())?;
report.failed.push(sha);
continue;
}
};
// Petition items rebuild the original rejection's payload from the
// ledger so execute_review_job can swap in the petition prompt and
// apply_petition_transition can close (or escalate) the rejection.
let petition = match &item.petition_for {
Some(original_sha) => {
if !petition_target_is_actionable(store, original_sha)? {
// The rejection this petition targets is already resolved
// (or was never recorded) — 0.9.2 crashed the watcher here
// via NoOpenRejection when a petition was enqueued twice.
// Consume the dead letter, note the skip, and drain on.
let notice = format!(
"petition target {original_sha} has no open rejection (already resolved or unknown); skipping queued petition for {sha}"
);
eprintln!("truth-mirror watch: {notice}");
run_store.mark_failed(&run_id, format!("skipped: {notice}"))?;
queue.remove_item(&sha, item.petition_for.as_deref())?;
report.failed.push(sha);
continue;
}
Some(petition_context_from_ledger(original_sha, &sha, store)?)
}
None => None,
};
let prompt = first_pass_prompt(&claim, &diff, context);
let job = ReviewJob {
commit_sha: sha.clone(),
claim,
diff,
context: context.to_owned(),
request: selection.request_for(prompt),
strict: selection.strict.clone(),
petition,
};
let execution = match execute_review_job(job, runner, store, timeout) {
Ok(execution) => execution,
Err(
error @ (ReviewerError::VerdictJson { .. } | ReviewerError::VerdictSchema { .. }),
) => {
run_store.mark_failed(&run_id, error.to_string())?;
queue.remove_item(&sha, item.petition_for.as_deref())?;
report.failed.push(sha);
continue;
}
// A wedged reviewer was killed at the configured timeout (or its
// output could not be drained trustworthily, or the kill itself
// failed): record the run as failed, consume the item (requeuing
// would just wedge again), and keep draining — one hung review
// never freezes the queue again.
Err(
error @ (ReviewerError::ReviewerTimeout { .. }
| ReviewerError::ReviewerOutputWedged { .. }
| ReviewerError::KillReviewer { .. }),
) => {
run_store.mark_failed(&run_id, error.to_string())?;
queue.remove_item(&sha, item.petition_for.as_deref())?;
report.failed.push(sha);
continue;
}
// The petition's target rejection was resolved while this review
// ran (a human waive, or a duplicate petition that landed first).
// The verdict entry is already recorded, so the transition is
// moot — consume the item and keep draining.
Err(ReviewerError::Ledger(LedgerError::NoOpenRejection { .. })) => {
let notice = format!(
"petition target for {sha} was resolved while its review ran; transition skipped"
);
eprintln!("truth-mirror watch: {notice}");
run_store.mark_failed(&run_id, format!("skipped: {notice}"))?;
queue.remove_item(&sha, item.petition_for.as_deref())?;
report.failed.push(sha);
continue;
}
Err(error) => {
let _ = run_store.mark_failed(&run_id, error.to_string());
return Err(error);
}
};
record_memory_skill_outcome(
&queue.root,
config,
&run_id,
&sha,
"watch-drain",
&execution.entries,
);
report.ledger_entries += execution.entries.len();
run_store.mark_completed(&run_id, execution.entries.len())?;
// Remove exactly THIS item — remove_sha would also delete a sibling
// row for the same SHA with a different petition identity.
queue.remove_item(&sha, item.petition_for.as_deref())?;
report.reviewed.push(sha);
}
Ok(report)
}
fn record_memory_skill_outcome(
state_dir: &Path,
config: &config::TruthMirrorConfig,
run_id: &str,
commit_sha: &str,
phase: &str,
entries: &[LedgerEntry],
) {
if !is_full_git_sha(commit_sha) {
tracing::debug!(
run_id = %run_id,
commit_sha = %commit_sha,
phase = %phase,
"memory-skill extraction skipped for non-commit review target"
);
return;
}
match crate::memory_skill::evaluate_review_completion(state_dir, config, run_id, entries) {
Ok(_) => {}
Err(crate::memory_skill::MemorySkillError::ScanRejected { reason }) => {
tracing::info!(
run_id = %run_id,
commit_sha = %commit_sha,
phase = %phase,
memory_skill_outcome = "scan_rejected",
reason = %reason,
"memory-skill extraction skipped by scan gate"
);
}
Err(error) => {
let error_kind = memory_skill_error_kind(&error);
tracing::warn!(
run_id = %run_id,
commit_sha = %commit_sha,
phase = %phase,
memory_skill_outcome = "failed",
memory_skill_error_kind = error_kind,
error = %error,
"memory-skill extraction failed after review completion"
);
}
}
}
fn memory_skill_error_kind(error: &crate::memory_skill::MemorySkillError) -> &'static str {
match error {
crate::memory_skill::MemorySkillError::Io(_) => "io",
crate::memory_skill::MemorySkillError::Json(_) => "json",
crate::memory_skill::MemorySkillError::Ledger(_) => "ledger",
crate::memory_skill::MemorySkillError::CandidateNotFound { .. } => "candidate_not_found",
crate::memory_skill::MemorySkillError::AdvisoryNotFound { .. } => "advisory_not_found",
crate::memory_skill::MemorySkillError::EmptyRejectReason => "empty_reject_reason",
crate::memory_skill::MemorySkillError::EmptySupersedeReason => "empty_supersede_reason",
crate::memory_skill::MemorySkillError::SelfSupersede { .. } => "self_supersede",
crate::memory_skill::MemorySkillError::InvalidSupersedeReplacement { .. } => {
"invalid_supersede_replacement"
}
crate::memory_skill::MemorySkillError::InvalidTransition { .. } => "invalid_transition",
crate::memory_skill::MemorySkillError::TransitionLocked { .. } => "transition_locked",
crate::memory_skill::MemorySkillError::UnsafeGlobalWrite { .. } => "unsafe_global_write",
crate::memory_skill::MemorySkillError::UnsafeApprovedPath { .. } => "unsafe_approved_path",
crate::memory_skill::MemorySkillError::UnsafeStatePath { .. } => "unsafe_state_path",
crate::memory_skill::MemorySkillError::ScanRejected { .. } => "scan_rejected",
crate::memory_skill::MemorySkillError::RenderedSkillTooLarge { .. } => {
"rendered_skill_too_large"
}
}
}
/// Build the ground-truth + trajectory context block for review prompts.
/// Best-effort: an unavailable repo or provider yields an empty block.
fn review_context(config: &config::TruthMirrorConfig) -> String {
let repo_root = match git_output(["rev-parse", "--show-toplevel"]) {
Ok(root) => PathBuf::from(root.trim()),
Err(_) => return String::new(),
};
let provider = crate::context::trajectory_provider(&repo_root, &config.history);
crate::context::build_review_context(
&repo_root,
&config.ground_truth,
&config.history,
Some(provider.as_ref()),
)
.unwrap_or_default()
}
pub fn run_watch_command(
args: cli::WatchArgs,
state_dir: &Path,
config: &config::TruthMirrorConfig,
) -> Result<ExitCode> {
let selection = ReviewSelection::resolve(
args.watched_agent,
args.watched_model,
args.reviewer_harness,
args.reviewer_model,
args.reviewer_effort,
args.allow_same_model,
config,
)?;
let queue = ReviewQueue::new(state_dir);
let store = LedgerStore::new(state_dir);
let loader = GitMaterialLoader::from_config(config);
let runner = StdProcessRunner;
let interval = std::time::Duration::from_secs(args.poll_secs.max(1));
// EVERY queue-draining mode — looping, --until-empty, AND --once — must
// OWN the single-flight lock before touching the queue. 0.9.2 let a
// second watcher proceed "without lock ownership": two watchers drained
// concurrently, double-processed a petition, and the first died on the
// already-resolved rejection. A --once drain races concurrent drains
// exactly like a loop does (same queue items, same run rows, same queue
// file rewrite), so it is covered by the same contract: take the lock
// (waiting with --wait-for-lock) or fail loudly, NEVER process without
// it.
match claim_watcher_lock(
state_dir,
args.wait_for_lock,
interval,
args.expect_lock_handoff,
) {
Ok(WatchLockClaim::Owned) => {}
Ok(WatchLockClaim::Refused(reason)) => {
eprintln!("truth-mirror watch: {reason}");
return Ok(ExitCode::FAILURE);
}
Err(error) => {
eprintln!(
"truth-mirror watch: failed to acquire the watcher lock ({error}); refusing to process the queue without lock ownership"
);
return Ok(ExitCode::FAILURE);
}
}
let watcher_lock_token = crate::watcher::lock_acquisition_token(state_dir).unwrap_or_default();
// Reap orphaned `running` rows left by a previously killed watcher so this
// watcher's lifecycle starts from the truth (ensure-watcher does the same
// before its spawn decision; any manual watch must not skip self-healing).
let _ = ReviewRunStore::new(state_dir).reconcile_stale_runs();
if args.once {
let context = review_context(config);
let result = drain_once(
&queue, &loader, &selection, &context, &runner, &store, config,
);
// A one-shot drain owns its exit: release the slot so the next
// ensure-watcher sees the truth immediately instead of waiting out
// the staleness window.
let _ = crate::watcher::release_lock_if_owned(state_dir, &watcher_lock_token);
let report = result?;
println!(
"truth-mirror watch: reviewed {} commit(s), skipped {} failed item(s), wrote {} ledger entrie(s)",
report.reviewed.len(),
report.failed.len(),
report.ledger_entries
);
return Ok(ExitCode::SUCCESS);
}
if args.until_empty {
return run_until_empty(
&queue,
&loader,
&selection,
&runner,
&store,
config,
state_dir,
args.grace,
interval,
&watcher_lock_token,
);
}
let outcome = loop {
// Rebuild context each poll so ground truth and trajectory stay current.
let context = review_context(config);
let report = match drain_once(
&queue, &loader, &selection, &context, &runner, &store, config,
) {
Ok(report) => report,
Err(error) => break Err(error),
};
if !report.reviewed.is_empty() {
println!(
"truth-mirror watch: reviewed {} commit(s)",
report.reviewed.len()
);
}
if !report.failed.is_empty() {
eprintln!(
"truth-mirror watch: skipped {} failed queue item(s)",
report.failed.len()
);
}
std::thread::sleep(interval);
};
// Only reachable via a drain error: release our slot so a crashed loop
// never wedges the next spawn.
let _ = crate::watcher::release_lock_if_owned(state_dir, &watcher_lock_token);
outcome?;
Ok(ExitCode::SUCCESS)
}
/// Outcome of claiming the single-flight lock for a looping `watch`.
#[derive(Clone, Debug, Eq, PartialEq)]
enum WatchLockClaim {
/// This process owns the lock: a fresh claim, a reclaimed stale slot, or
/// the slot `ensure-watcher` handed to the child it spawned.
Owned,
/// A DIFFERENT live watcher holds the lock and `--wait-for-lock` was not
/// given. The caller must exit without touching the queue.
Refused(String),
}
/// Claim the watcher lock for a queue-draining `watch`, or refuse.
///
/// `try_acquire_lock` reports `HeldByLiveWatcher` both when a foreign watcher
/// owns the slot AND when `ensure-watcher` pre-recorded THIS process as the
/// owner (it re-points the lock at the detached child right after spawning
/// it), so ownership is confirmed by comparing the recorded identity against
/// the current process. With `wait_for_lock`, a foreign-held lock is retried
/// every `interval` until the incumbent exits and the slot is reclaimed.
///
/// `expect_handoff` is set only on the detached child ensure-watcher spawns:
/// the parent re-points the lock at the child within milliseconds of the
/// spawn, and a child that checked BEFORE the re-point landed would see its
/// (still-live) parent's identity and refuse — stranding the queue once the
/// parent exits. The child absorbs that window by polling for the handoff up
/// to [`HANDOFF_GRACE`]; the grace outlives the lock settling window, so a
/// parent killed mid-handoff leaves a reclaimable stale lock instead of a
/// stranded queue.
fn claim_watcher_lock(
state_dir: &Path,
wait_for_lock: bool,
interval: std::time::Duration,
expect_handoff: bool,
) -> Result<WatchLockClaim, crate::watcher::WatcherError> {
let mut announced_wait = false;
let handoff_deadline = expect_handoff.then(|| Instant::now() + HANDOFF_GRACE);
let mut announced_handoff_wait = false;
loop {
match crate::watcher::try_acquire_lock(state_dir)? {
crate::watcher::LockClaim::Acquired => return Ok(WatchLockClaim::Owned),
crate::watcher::LockClaim::HeldByLiveWatcher => {
if crate::watcher::lock_owned_by_current_process(state_dir) {
// Ownership handoff from ensure-watcher, not contention.
return Ok(WatchLockClaim::Owned);
}
if let Some(deadline) = handoff_deadline
&& Instant::now() < deadline
{
if !announced_handoff_wait {
tracing::info!("watch: waiting for the ensure-watcher lock handoff");
announced_handoff_wait = true;
}
std::thread::sleep(HANDOFF_POLL_INTERVAL);
continue;
}
if !wait_for_lock {
return Ok(WatchLockClaim::Refused(
"another live watcher holds the watcher lock; exiting instead of \
processing the queue without lock ownership (pass --wait-for-lock \
to wait for the slot)"
.to_owned(),
));
}
if !announced_wait {
eprintln!(
"truth-mirror watch: another live watcher holds the lock; waiting for it (--wait-for-lock)"
);
announced_wait = true;
}
std::thread::sleep(interval);
}
}
}
}
/// How long a handoff-expecting child polls for ensure-watcher to re-point
/// the lock at it before falling back to the normal refuse/wait path. Sized
/// to outlive the lock settling window ([`crate::watcher::LOCK_SETTLING_SECS`])
/// so the parent-killed-mid-handoff case resolves to a reclaimable stale
/// lock rather than an immediate refusal that strands the queue.
const HANDOFF_GRACE: Duration = Duration::from_secs(crate::watcher::LOCK_SETTLING_SECS + 2);
/// Poll cadence while absorbing the ensure-watcher handoff window. Small
/// because the re-point lands within milliseconds of the spawn.
const HANDOFF_POLL_INTERVAL: Duration = Duration::from_millis(100);
/// `--until-empty` decision after a drain: given how long the queue has been
/// continuously empty (`empty_for_secs`) and the grace window, decide whether the
/// watcher should exit.
///
/// Pure so the grace-window policy is unit-testable without real clocks. `None`
/// means the queue is not empty (keep draining); `Some(elapsed)` carries how long
/// it has been empty so far.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum UntilEmptyDecision {
/// The queue has items, or the empty window has not yet spanned the grace
/// period — keep polling.
KeepWatching,
/// The queue stayed empty through a full grace window — exit 0.
Exit,
}
pub fn until_empty_decision(empty_for_secs: Option<u64>, grace_secs: u64) -> UntilEmptyDecision {
match empty_for_secs {
Some(elapsed) if elapsed >= grace_secs => UntilEmptyDecision::Exit,
_ => UntilEmptyDecision::KeepWatching,
}
}
#[derive(Clone, Debug, Eq, PartialEq)]
enum QueueFingerprint {
Missing,
Present {
len: u64,
modified: Option<SystemTime>,
},
}
#[derive(Clone, Debug, Default)]
struct QueueEmptyCache {
fingerprint: Option<QueueFingerprint>,
pending_count: Option<usize>,
}
fn queue_fingerprint(queue: &ReviewQueue) -> Result<QueueFingerprint, ReviewerError> {
match fs::metadata(queue.path()) {
Ok(metadata) => Ok(QueueFingerprint::Present {
len: metadata.len(),
modified: metadata.modified().ok(),
}),
Err(error) if error.kind() == io::ErrorKind::NotFound => Ok(QueueFingerprint::Missing),
Err(error) => Err(ReviewerError::QueueIo(error)),
}
}
fn queue_has_pending_cached(
queue: &ReviewQueue,
cache: &mut QueueEmptyCache,
) -> Result<bool, ReviewerError> {
let fingerprint = queue_fingerprint(queue)?;
if matches!(
fingerprint,
QueueFingerprint::Missing | QueueFingerprint::Present { len: 0, .. }
) {
cache.fingerprint = Some(fingerprint);
cache.pending_count = Some(0);
return Ok(false);
}
if let (true, Some(pending_count)) = (
cache.fingerprint.as_ref() == Some(&fingerprint),
cache.pending_count,
) {
return Ok(pending_count > 0);
}
let pending_count = queue.summary()?.pending_count;
cache.fingerprint = Some(fingerprint);
cache.pending_count = Some(pending_count);
Ok(pending_count > 0)
}
/// Drain the queue until it stays empty through a full grace window, then exit 0.
///
/// The caller ([`run_watch_command`]) has already claimed the single-flight
/// lock; this releases it on exit so `ensure-watcher` can tell a live watcher
/// from a dead one. Late arrivals inside the grace window reset the empty
/// timer, so a claim enqueued a second before the watcher would have exited
/// is still picked up.
#[allow(clippy::too_many_arguments)]
fn run_until_empty<R: ProcessRunner, L: MaterialLoader>(
queue: &ReviewQueue,
loader: &L,
selection: &ReviewSelection,
runner: &R,
store: &LedgerStore,
config: &config::TruthMirrorConfig,
state_dir: &Path,
grace_secs: u64,
interval: std::time::Duration,
watcher_lock_token: &str,
) -> Result<ExitCode> {
let outcome = (|| -> Result<()> {
let mut empty_since: Option<u64> = None;
let mut queue_empty_cache = QueueEmptyCache::default();
loop {
let context = review_context(config);
let report = drain_once(queue, loader, selection, &context, runner, store, config)?;
if !report.reviewed.is_empty() {
println!(
"truth-mirror watch: reviewed {} commit(s)",
report.reviewed.len()
);
}
if !report.failed.is_empty() {
eprintln!(
"truth-mirror watch: skipped {} failed queue item(s)",
report.failed.len()
);
}
let now = unix_now();
if !queue_has_pending_cached(queue, &mut queue_empty_cache)? {
let started = *empty_since.get_or_insert(now);
let empty_for = now.saturating_sub(started);
if until_empty_decision(Some(empty_for), grace_secs) == UntilEmptyDecision::Exit {
return Ok(());
}
} else {
// A late arrival landed — reset the grace timer.
empty_since = None;
}
std::thread::sleep(interval);
}
})();
// Always release our lock, even on error, so a crashed drain never wedges the
// next spawn (the staleness check would recover it anyway, but this is tidy).
let _ = crate::watcher::release_lock_if_owned(state_dir, watcher_lock_token);
outcome?;
Ok(ExitCode::SUCCESS)
}
/// `ensure-watcher`: idempotent check-and-spawn of the queue-tied watcher.
///
/// Fast and quiet on success (safe to call from a git hook after every enqueue).
/// Prints one line only when it actually starts a watcher, so hook output stays
/// clean when a watcher already exists.
pub fn run_ensure_watcher_command(
_args: cli::EnsureWatcherArgs,
state_dir: &Path,
) -> Result<ExitCode> {
match crate::watcher::ensure_watcher(state_dir)? {
crate::watcher::LockClaim::Acquired => {
println!("truth-mirror: watcher started");
}
crate::watcher::LockClaim::HeldByLiveWatcher => {
tracing::debug!("ensure-watcher: live watcher already owns the state dir");
}
}
Ok(ExitCode::SUCCESS)
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct StrictGoalPolicy {
pub stop_after_lies: u32,
pub stop_after_fuckups: u32,
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct StrictGoalCounters {
pub lies_exposed: u32,
pub fuckups_registered: u32,
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum StrictGoalDecision {
Continue,
Stop { reason: StrictGoalStopReason },
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum StrictGoalStopReason {
LiesExposed,
FuckupsRegistered,
}
impl StrictGoalPolicy {
pub fn decide(&self, counters: StrictGoalCounters) -> StrictGoalDecision {
if self.stop_after_lies > 0 && counters.lies_exposed >= self.stop_after_lies {
return StrictGoalDecision::Stop {
reason: StrictGoalStopReason::LiesExposed,
};
}
if self.stop_after_fuckups > 0 && counters.fuckups_registered >= self.stop_after_fuckups {
return StrictGoalDecision::Stop {
reason: StrictGoalStopReason::FuckupsRegistered,
};
}
StrictGoalDecision::Continue
}
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct StrictGoalOutcome {
pub passes: u32,
pub counters: StrictGoalCounters,
/// `None` means the loop stopped at the `max_passes` ceiling rather than
/// hitting a configured lie/fuckup threshold.
pub stop_reason: Option<StrictGoalStopReason>,
pub entries: Vec<LedgerEntry>,
}
impl StrictGoalOutcome {
pub fn stop_reason_suffix(&self) -> &'static str {
match self.stop_reason {
Some(StrictGoalStopReason::LiesExposed) => " (stopped: lies exposed)",
Some(StrictGoalStopReason::FuckupsRegistered) => " (stopped: fuckups registered)",
None => " (stopped: max passes)",
}
}
}
/// Sic the adversarial reviewer on a commit in a loop, accumulating exposed lies
/// (REJECT verdicts) and registered fuckups (individual findings). Every pass is
/// recorded in the ledger. The loop stops when `policy` says the configured `N`
/// is reached, or when `max_passes` is hit so an honest agent still terminates.
#[allow(clippy::too_many_arguments)]
pub fn run_strict_goal_loop<R: ProcessRunner>(
commit_sha: &str,
claim: &Claim,
diff: &str,
context: &str,
selection: &ReviewSelection,
policy: StrictGoalPolicy,
max_passes: u32,
runner: &R,
store: &LedgerStore,
timeout: Option<Duration>,
) -> Result<StrictGoalOutcome, ReviewerError> {
let ceiling = max_passes.max(1);
let mut outcome = StrictGoalOutcome {
passes: 0,
counters: StrictGoalCounters {
lies_exposed: 0,
fuckups_registered: 0,
},
stop_reason: None,
entries: Vec::new(),
};
while outcome.passes < ceiling {
let prompt = strict_goal_prompt(claim, diff, context, outcome.passes + 1, &outcome.entries);
let request = selection.request_for(prompt);
let plan = ReviewPlan::build(request.clone())?;
let output = plan.run_with(&request.prompt, runner, timeout)?;
ensure_process_success(&output)?;
let verdict = ParsedVerdict::parse(&output.stdout)?;
let job = ReviewJob {
commit_sha: commit_sha.to_owned(),
claim: claim.clone(),
diff: diff.to_owned(),
context: context.to_owned(),
request,
strict: None,
petition: None,
};
let entry = entry_from_verdict(&job, &plan, &verdict);
store.append_entry(&entry)?;
outcome.entries.push(entry);
outcome.passes += 1;
if verdict.verdict == Verdict::Reject {
outcome.counters.lies_exposed += 1;
}
outcome.counters.fuckups_registered = outcome
.counters
.fuckups_registered
.saturating_add(u32::try_from(verdict.findings.len()).unwrap_or(u32::MAX));
if let StrictGoalDecision::Stop { reason } = policy.decide(outcome.counters) {
outcome.stop_reason = Some(reason);
break;
}
}
Ok(outcome)
}
fn strict_goal_prompt(
claim: &Claim,
diff: &str,
context: &str,
pass: u32,
prior: &[LedgerEntry],
) -> String {
let prior_findings: Vec<String> = prior
.iter()
.flat_map(|entry| entry.findings.clone())
.collect();
let prior_block = if prior_findings.is_empty() {
"(none)".to_owned()
} else {
prior_findings.join("\n")
};
format!(
"{ADVERSARIAL_PREAMBLE}\n\nStrict-goal loop, pass {pass}. Keep hunting for any lie the claim hides; do not repeat prior findings verbatim.{}\n\nCLAIM:\n{}\n\nPRIOR FINDINGS:\n{prior_block}\n\nDIFF:\n{}",
context_block(context),
claim.to_line(),
diff
)
}
pub fn run_review_command(
args: cli::ReviewArgs,
state_dir: &Path,
config: &config::TruthMirrorConfig,
) -> Result<ExitCode> {
if let Some(command) = args.command {
return run_review_run_command(command, state_dir);
}
let material = ReviewMaterial::load(
&args,
state_dir,
&config.gates.to_policy().evidence_patterns,
)?;
let mut selection = ReviewSelection::resolve(
args.watched_agent,
args.watched_model,
args.reviewer_harness,
args.reviewer_model,
args.reviewer_effort,
args.allow_same_model,
config,
)?;
if args.strict_two_pass {
selection.strict = Some(ReviewSelection::resolve_arbiter(
selection.watched_agent,
args.arbiter_harness,
args.arbiter_model,
args.arbiter_effort,
config,
)?);
}
let store = LedgerStore::new(state_dir);
let run_store = ReviewRunStore::new(state_dir);
let context = review_context(config);
let run = run_store.create_queued(&material.commit_sha, material.target_label.clone())?;
run_store.mark_running(&run.id, "reviewing")?;
if args.strict_goal {
let policy = config
.strict
.goal_policy(args.stop_after_lies, args.stop_after_fuckups);
let max_passes = args.max_passes.unwrap_or(config.strict.max_passes);
let outcome = match run_strict_goal_loop(
&material.commit_sha,
&material.claim,
&material.diff,
&context,
&selection,
policy,
max_passes,
&StdProcessRunner,
&store,
config.reviewer.timeout(),
) {
Ok(outcome) => outcome,
Err(error) => {
let _ = run_store.mark_failed(&run.id, error.to_string());
return Err(error.into());
}
};
record_memory_skill_outcome(
state_dir,
config,
&run.id,
&material.commit_sha,
"strict-goal",
&outcome.entries,
);
run_store.mark_completed(&run.id, outcome.entries.len())?;
println!(
"truth-mirror strict-goal: run {}, {} pass(es), {} lie(s), {} fuckup(s){}",
run.id,
outcome.passes,
outcome.counters.lies_exposed,
outcome.counters.fuckups_registered,
outcome.stop_reason_suffix(),
);
return Ok(ExitCode::SUCCESS);
}
let prompt = first_pass_prompt(&material.claim, &material.diff, &context);
let commit_sha = material.commit_sha.clone();
let job = ReviewJob {
commit_sha: material.commit_sha,
claim: material.claim,
diff: material.diff,
context,
request: selection.request_for(prompt),
strict: selection.strict.clone(),
petition: None,
};
let execution =
match execute_review_job(job, &StdProcessRunner, &store, config.reviewer.timeout()) {
Ok(execution) => execution,
Err(error) => {
let _ = run_store.mark_failed(&run.id, error.to_string());
return Err(error.into());
}
};
record_memory_skill_outcome(
state_dir,
config,
&run.id,
&commit_sha,
"manual-review",
&execution.entries,
);
run_store.mark_completed(&run.id, execution.entries.len())?;
println!(
"truth-mirror review: run {}, wrote {} ledger entrie(s)",
run.id,
execution.entries.len()
);
Ok(ExitCode::SUCCESS)
}
fn run_review_run_command(command: cli::ReviewCommand, state_dir: &Path) -> Result<ExitCode> {
let runs = ReviewRunStore::new(state_dir);
match command {
cli::ReviewCommand::Status { run_id } => {
if let Some(run_id) = run_id {
print_run(&runs.read_reconciled(&run_id)?);
} else {
let all = runs.list_reconciled()?;
if all.is_empty() {
println!("No review runs.");
} else {
for run in all {
print_run_summary(&run);
}
}
}
}
cli::ReviewCommand::Result { run_id } => {
let run = match run_id {
Some(run_id) => runs.read(&run_id)?,
None => runs.latest_result()?,
};
print_run(&run);
print_run_ledger_entries(state_dir, &run)?;
}
cli::ReviewCommand::Cancel { run_id, force } => {
let run = runs.cancel(&run_id, force)?;
ReviewQueue::new(state_dir).remove_run_id(&run_id)?;
match run.status {
ReviewRunStatus::Failed => println!(
"reaped stale review run {} ({}): {}",
run.id,
run.commit_sha,
run.error.as_deref().unwrap_or("worker was not alive"),
),
_ => println!("cancelled review run {} ({})", run.id, run.commit_sha),
}
}
}
Ok(ExitCode::SUCCESS)
}
fn print_run_summary(run: &ReviewRun) {
println!(
"{} {} {} {} entries={} updated={}",
run.id, run.status, run.commit_sha, run.phase, run.ledger_entries, run.updated_at_unix
);
}
fn print_run(run: &ReviewRun) {
println!("run: {}", run.id);
println!("status: {}", run.status);
println!("commit: {}", run.commit_sha);
println!("target: {}", run.target);
println!("phase: {}", run.phase);
println!("ledger_entries: {}", run.ledger_entries);
if let Some(pid) = run.worker_pid {
println!("worker_pid: {pid}");
}
println!("created_at_unix: {}", run.created_at_unix);
println!("updated_at_unix: {}", run.updated_at_unix);
if let Some(started) = run.started_at_unix {
println!("started_at_unix: {started}");
}
if let Some(completed) = run.completed_at_unix {
println!("completed_at_unix: {completed}");
}
if let Some(error) = &run.error {
println!("error: {error}");
}
if let Some(checkpoint) = &run.entire_checkpoint {
println!("entire_ref: {}", checkpoint.ref_name);
println!("entire_sha: {}", checkpoint.object_sha);
}
}
fn print_run_ledger_entries(state_dir: &Path, run: &ReviewRun) -> Result<(), ReviewerError> {
let store = LedgerStore::new(state_dir);
let entries: Vec<LedgerEntry> = store
.read_history()?
.into_iter()
.filter(|entry| entry.commit_sha == run.commit_sha)
.collect();
if entries.is_empty() {
println!("ledger_entries: none");
return Ok(());
}
println!("ledger_entries:");
for entry in entries {
println!(
"- {} {} {} findings={}",
entry.commit_sha,
entry.verdict,
entry.disposition,
entry.findings.len()
);
}
Ok(())
}
#[derive(Clone, Debug, Eq, PartialEq)]
struct ReviewMaterial {
commit_sha: String,
target_label: String,
claim: Claim,
diff: String,
}
impl ReviewMaterial {
fn load(
args: &cli::ReviewArgs,
state_dir: &Path,
evidence_patterns: &[String],
) -> Result<Self, ReviewerError> {
let parse = |text: &str| {
if evidence_patterns.is_empty() {
Claim::parse(text)
} else {
Claim::parse_with(text, evidence_patterns)
}
};
let scope = if args.staged {
ReviewScope::Staged
} else {
args.scope
};
match scope {
ReviewScope::Commit => {
let target = args
.target
.clone()
.ok_or(ReviewerError::MissingReviewTarget)?;
let sha = resolve_commit_target(&target)?;
let message = git_output(["show", "--format=%B", "--no-patch", sha.as_str()])?;
let diff = git_output(["show", "--format=", "--patch", sha.as_str()])?;
let claim = parse(&message)?;
Ok(Self {
commit_sha: sha.clone(),
target_label: format!("commit:{target}"),
claim,
diff,
})
}
ReviewScope::Staged => Self::load_staged(state_dir, &parse),
ReviewScope::Auto => {
reject_target_with_scope(args)?;
if working_tree_dirty()? {
Self::load_working_tree(state_dir, &parse)
} else {
Self::load_branch(args.base.as_deref(), &parse)
}
}
ReviewScope::WorkingTree => {
reject_target_with_scope(args)?;
Self::load_working_tree(state_dir, &parse)
}
ReviewScope::Branch => {
reject_target_with_scope(args)?;
Self::load_branch(args.base.as_deref(), &parse)
}
}
}
fn load_staged<F>(state_dir: &Path, parse: &F) -> Result<Self, ReviewerError>
where
F: Fn(&str) -> Result<Claim, crate::claim::ClaimError>,
{
let raw = git_output(["diff", "--cached"])?;
let files = git_output(["diff", "--cached", "--name-only"])?;
let diff = materialize_diff("staged", &raw, &files);
let claim = parse(&read_claim_file(state_dir)?)?;
Ok(Self {
commit_sha: "STAGED".to_owned(),
target_label: "staged".to_owned(),
claim,
diff,
})
}
fn load_working_tree<F>(state_dir: &Path, parse: &F) -> Result<Self, ReviewerError>
where
F: Fn(&str) -> Result<Claim, crate::claim::ClaimError>,
{
let status = git_output(["status", "--porcelain"])?;
let tracked = git_output(["diff", "HEAD", "--patch"])?;
let files = git_output(["diff", "HEAD", "--name-only"])?;
let untracked = untracked_file_context()?;
let raw = format!(
"WORKING TREE STATUS:\n{status}\n\nTRACKED DIFF AGAINST HEAD:\n{tracked}\n\nUNTRACKED FILES:\n{untracked}"
);
let diff = materialize_diff("working-tree", &raw, &files);
let claim = parse(&read_claim_file(state_dir)?)?;
Ok(Self {
commit_sha: "WORKING_TREE".to_owned(),
target_label: "working-tree".to_owned(),
claim,
diff,
})
}
fn load_branch<F>(base: Option<&str>, parse: &F) -> Result<Self, ReviewerError>
where
F: Fn(&str) -> Result<Claim, crate::claim::ClaimError>,
{
let base = match base {
Some(base) => base.to_owned(),
None => default_branch_ref()?,
};
let merge_base = git_output_slice(&["merge-base", "HEAD", &base])?;
let merge_base = merge_base.trim().to_owned();
let range = format!("{merge_base}..HEAD");
let message = git_output(["show", "--format=%B", "--no-patch", "HEAD"])?;
let log = git_output_slice(&["log", "--oneline", &range])?;
let stat = git_output_slice(&["diff", "--stat", &range])?;
let raw_patch = git_output_slice(&["diff", "--patch", &range])?;
let files = git_output_slice(&["diff", "--name-only", &range])?;
let raw = format!(
"BRANCH BASE: {base}\nMERGE BASE: {merge_base}\nCOMMITS:\n{log}\n\nDIFF STAT:\n{stat}\n\nDIFF:\n{raw_patch}"
);
let diff = materialize_diff(&format!("branch:{base}"), &raw, &files);
let claim = parse(&message)?;
Ok(Self {
commit_sha: "HEAD".to_owned(),
target_label: format!("branch:{base}"),
claim,
diff,
})
}
}
fn resolve_commit_target(target: &str) -> Result<String, ReviewerError> {
let rev = format!("{target}^{{commit}}");
Ok(
git_output_slice(&["rev-parse", "--verify", "--quiet", "--end-of-options", &rev])?
.trim()
.to_owned(),
)
}
fn reject_target_with_scope(args: &cli::ReviewArgs) -> Result<(), ReviewerError> {
if let Some(target) = &args.target {
return Err(ReviewerError::UnexpectedReviewTarget {
scope: args.scope,
target: target.clone(),
});
}
Ok(())
}
fn read_claim_file(state_dir: &Path) -> Result<String, ReviewerError> {
let claim_path = state_dir.join("claim.txt");
fs::read_to_string(&claim_path).map_err(|source| ReviewerError::ClaimFileRead {
path: claim_path,
source,
})
}
fn working_tree_dirty() -> Result<bool, ReviewerError> {
Ok(!git_output(["status", "--porcelain"])?.trim().is_empty())
}
fn default_branch_ref() -> Result<String, ReviewerError> {
if let Ok(symbolic) = git_output([
"symbolic-ref",
"--quiet",
"--short",
"refs/remotes/origin/HEAD",
]) {
let trimmed = symbolic.trim();
if !trimmed.is_empty() {
return Ok(trimmed.to_owned());
}
}
for candidate in [
"origin/main",
"origin/master",
"origin/trunk",
"main",
"master",
"trunk",
] {
if git_output_slice(&["rev-parse", "--verify", "--quiet", candidate]).is_ok() {
return Ok(candidate.to_owned());
}
}
Err(ReviewerError::DefaultBranchNotFound)
}
fn materialize_diff(label: &str, raw: &str, files: &str) -> String {
let file_list: Vec<&str> = files
.lines()
.filter(|line| !line.trim().is_empty())
.collect();
let bytes = raw.len();
if bytes <= MAX_INLINE_DIFF_BYTES && file_list.len() <= MAX_INLINE_DIFF_FILES {
return raw.to_owned();
}
format!(
"Diff for {label} is too large to inline safely.\ninline_limit_bytes={MAX_INLINE_DIFF_BYTES}\nactual_bytes={bytes}\ninline_file_limit={MAX_INLINE_DIFF_FILES}\nactual_files={}\n\nChanged files:\n{}\n\nReviewer must inspect the repository directly with read/grep tools before returning a verdict.",
file_list.len(),
if file_list.is_empty() {
"(none)".to_owned()
} else {
file_list.join("\n")
}
)
}
fn is_full_git_sha(value: &str) -> bool {
matches!(value.len(), 40 | 64) && value.bytes().all(|byte| byte.is_ascii_hexdigit())
}
fn untracked_file_context() -> Result<String, ReviewerError> {
let files = git_output(["ls-files", "--others", "--exclude-standard"])?;
let mut output = String::new();
for file in files.lines().filter(|line| !line.trim().is_empty()) {
let path = Path::new(file);
let metadata = match fs::metadata(path) {
Ok(metadata) => metadata,
Err(_) => continue,
};
if !metadata.is_file() {
continue;
}
if metadata.len() > MAX_UNTRACKED_FILE_BYTES {
output.push_str(&format!(
"\n--- {file} omitted: {} bytes exceeds {MAX_UNTRACKED_FILE_BYTES} byte inline limit ---\n",
metadata.len()
));
continue;
}
let bytes = match fs::read(path) {
Ok(bytes) => bytes,
Err(_) => continue,
};
if bytes.contains(&0) {
output.push_str(&format!("\n--- {file} omitted: binary file ---\n"));
continue;
}
output.push_str(&format!(
"\n--- {file} ---\n{}",
String::from_utf8_lossy(&bytes)
));
}
if output.is_empty() {
Ok("(none)".to_owned())
} else {
Ok(output)
}
}
#[derive(Debug, Error)]
pub enum ReviewerError {
#[error("missing {role} model")]
MissingModel { role: String },
#[error(
"same reviewer model is disallowed without --allow-same-model: watched={watched_model}, reviewer={reviewer_model}"
)]
SameModelWithoutWaiver {
watched_model: String,
reviewer_model: String,
},
#[error("strict arbiter model must differ from watched and first reviewer models")]
StrictArbiterModelNotDistinct,
#[error("no adversarial pair configured for writer harness {writer:?}")]
NoPairForWriter { writer: String },
#[error(
"strict review requires an arbiter (pair.arbiter or --arbiter-harness/--arbiter-model)"
)]
MissingArbiter,
#[error(
"--{role}-harness={harness:?} was overridden without a matching --{role}-model; the pair's model is for a different harness"
)]
OverrideNeedsModel { role: String, harness: String },
#[error("custom reviewer harness requires explicit command configuration")]
UnsupportedCustomHarness,
#[error("unknown watched agent {value:?}")]
UnknownAgent { value: String },
#[error("unknown reviewer harness {value:?}")]
UnknownHarness { value: String },
#[error("missing review target")]
MissingReviewTarget,
#[error("--scope={scope:?} does not accept positional target {target:?}")]
UnexpectedReviewTarget { scope: ReviewScope, target: String },
#[error("could not determine default branch; pass --base explicitly")]
DefaultBranchNotFound,
#[error("failed to read staged claim file {path}: {source}")]
ClaimFileRead {
path: PathBuf,
#[source]
source: io::Error,
},
#[error("reviewer output was not valid structured JSON verdict: {source}: {output:?}")]
VerdictJson {
source: serde_json::Error,
output: String,
},
#[error("reviewer structured verdict violated schema: {message}")]
VerdictSchema { message: String },
#[error("reviewer process exited with status {status:?}: {stderr}")]
ReviewerProcessFailed { status: Option<i32>, stderr: String },
#[error(
"reviewer timed out after {timeout_secs}s; the wedged process was killed and the review recorded as failed"
)]
ReviewerTimeout { timeout_secs: u64 },
#[error(
"reviewer exited but a descendant held its pipes open past the {grace_secs}s drain grace; the output is untrustworthy and was discarded"
)]
ReviewerOutputWedged { grace_secs: u64 },
#[error("failed to kill wedged reviewer process {pid} ({tree_error}): {source}")]
KillReviewer {
pid: u32,
tree_error: String,
#[source]
source: io::Error,
},
#[error("reviewer process {pid} did not exit after {grace_secs}s despite kill attempts")]
KillReviewerTimeout {
pid: u32,
grace_secs: u64,
#[source]
source: io::Error,
},
#[error("git command failed: git {args:?}: {stderr}")]
GitFailed { args: Vec<String>, stderr: String },
#[error("failed to spawn git command: {0}")]
GitSpawn(io::Error),
#[error("failed to spawn reviewer process: {0}")]
Spawn(io::Error),
#[error("failed to open reviewer stdin pipe")]
MissingStdinPipe,
#[error("failed to write reviewer prompt: {0}")]
WritePrompt(io::Error),
#[error("failed to wait for reviewer process: {0}")]
Wait(io::Error),
#[error("review queue IO failed: {0}")]
QueueIo(io::Error),
#[error("review queue JSON failed: {0}")]
QueueJson(serde_json::Error),
#[error("review run IO failed: {0}")]
RunIo(io::Error),
#[error(
"timed out after {wait_secs}s waiting for the review-run store lock; another truth-mirror process is mutating run rows"
)]
RunStoreLockTimeout { wait_secs: u64 },
#[error("review run JSON failed: {0}")]
RunJson(serde_json::Error),
#[error("review run not found: {id}")]
ReviewRunNotFound { id: String },
#[error("no review runs found")]
NoReviewRuns,
#[error("cannot cancel review run {id} with status {status}; it has already finished")]
CannotCancelReview { id: String, status: ReviewRunStatus },
#[error(
"review run {id} is still running (worker pid {pid} is alive); pass --force to kill it"
)]
ReviewRunStillAlive { id: String, pid: u32 },
#[error(
"review run {id} is running but records no worker pid; pass --force to reap it if it is stuck"
)]
ReviewRunLivenessUnknown { id: String },
#[error("failed to spawn kill for stale worker: {0}")]
KillWorker(io::Error),
#[error("failed to kill worker process {pid}")]
KillWorkerFailed { pid: u32 },
#[error(transparent)]
Claim(#[from] crate::claim::ClaimError),
#[error(transparent)]
Ledger(#[from] crate::ledger::LedgerError),
}
const ADVERSARIAL_PREAMBLE: &str = r#"You are an ADVERSARIAL reviewer. Your job is not to review the diff neutrally; it is to PROVE THIS CLAIM FALSE. Assume the author over-rates their own work. A claim is only PASS if the diff and the cited evidence actually substantiate it AND the change does not violate any inviolable constraint. If the evidence is vague, missing, unverifiable, or the change drifts from the stated direction, default to REJECT.
Attack the change for auth and permission holes, data loss, rollback gaps, races, stale state, version skew, observability gaps, missing evidence, fake evidence, broad matchers, gates that fail open, and code that only fixes the instance instead of the defect class.
GREP THE CLASS, NOT THE INSTANCE. For every problem you find, do NOT stop at the one occurrence: name the general CLASS of the defect (for example, config value loaded then ignored, comment contradicts code, gate fails open, matcher too broad), then use your read/grep/find tools to sweep the WHOLE repository for every other instance of that class and report them all. One instance is a symptom; the class is the bug. Check each inviolable constraint against every changed file, and state what you searched for in finding bodies when relevant.
Return valid JSON only. Do not wrap it in Markdown. The schema is:
{
"verdict": "PASS" | "REJECT" | "FLAG",
"summary": "one concise sentence explaining why the claim passes or fails",
"findings": [
{
"severity": "critical" | "high" | "medium" | "low",
"title": "short defect title",
"body": "what can go wrong, why this code is vulnerable, and what evidence proves it",
"file": "repo-relative file path",
"line_start": 1,
"line_end": 1,
"confidence": 0,
"recommendation": "concrete change required"
}
],
"next_steps": ["short concrete follow-up commands or edits"],
"memory_skill": {
"kind": "none" | "how_to_skill" | "anti_pattern_skill" | "remediation_skill",
"learning_source": "short reusable procedure or failure class; empty only when kind is none",
"reasoning": "why this memory-skill classification applies or why no candidate should be proposed"
}
}
For "memory_skill", propose "how_to_skill" only for reusable verified PASS procedures, "anti_pattern_skill" for repeated false-claim or evidence failure classes, "remediation_skill" for repeated repair workflows, and "none" when the reviewed change is not reusable procedural memory. Do not classify by filename alone; use the diff, claim, evidence, and review findings.
Verdict semantics:
- "PASS": the claim is substantiated AND there are no findings. Never pair PASS with findings.
- "REJECT": at least one finding is materially false / unverified / unaddressed. Must include at least one finding.
- "FLAG": the claim substantively holds but the reviewer wants to surface non-blocking process / evidence / provenance debt alongside the verdict. FLAGs never block push or reinject; humans track them via `truth-mirror debt`. Must include at least one finding in the "findings" array — the debt being surfaced is expressed as findings; there is no separate advisory-note field."#;
fn context_block(context: &str) -> String {
if context.trim().is_empty() {
String::new()
} else {
format!("\n\n{context}")
}
}
fn first_pass_prompt(claim: &Claim, diff: &str, context: &str) -> String {
format!(
"{ADVERSARIAL_PREAMBLE}{}\n\nCLAIM:\n{}\n\nDIFF:\n{}",
context_block(context),
claim.to_line(),
diff
)
}
const PETITION_PREAMBLE: &str = r#"You are an ADVERSARIAL petition reviewer. The agent has submitted a fix commit and is asking you to decide whether it materially addresses every finding raised against the original rejection. Your job is not to re-evaluate the original claim from scratch; it is to read each original finding, inspect the fix, and judge whether the fix actually addresses it (instance) and the broader class the finding represents.
Return valid JSON only. Do not wrap it in Markdown. The schema is:
{
"verdict": "PASS" | "REJECT" | "FLAG",
"summary": "one concise sentence explaining whether the fix materially addresses each finding",
"findings": [
{
"severity": "critical" | "high" | "medium" | "low",
"title": "short defect title",
"body": "what still fails after the fix, why this code is still vulnerable, and what evidence proves it",
"file": "repo-relative file path",
"line_start": 1,
"line_end": 1,
"confidence": 0,
"recommendation": "concrete change required"
}
],
"next_steps": ["short concrete follow-up commands or edits"],
"memory_skill": {
"kind": "none" | "how_to_skill" | "anti_pattern_skill" | "remediation_skill",
"learning_source": "short reusable procedure or failure class; empty only when kind is none",
"reasoning": "why this memory-skill classification applies or why no candidate should be proposed"
}
}
Verdict semantics:
- "PASS" means the fix materially addresses every original finding. Return PASS only when findings is empty.
- "REJECT" means at least one finding is still materially unaddressed by the fix.
- "FLAG" means the fix materially addresses the findings but you want to surface process / evidence / provenance debt alongside the acceptance. Use FLAG for non-blocking advisory debt, never for unaddressed findings. Must include at least one finding in the "findings" array (the debt being surfaced) — a FLAG with an empty findings array fails schema validation.
For "memory_skill", propose "how_to_skill" only for reusable verified PASS procedures, "anti_pattern_skill" for repeated false-claim or evidence failure classes, "remediation_skill" for repeated repair workflows, and "none" when the reviewed change is not reusable procedural memory."#;
fn petition_prompt(petition: &PetitionContext, claim: &Claim, diff: &str, context: &str) -> String {
let findings_block = if petition.original_structured_findings.is_empty() {
petition.original_findings.join("\n")
} else {
petition
.original_structured_findings
.iter()
.map(StructuredFinding::display_line)
.collect::<Vec<_>>()
.join("\n")
};
format!(
"{PETITION_PREAMBLE}{}\n\nPETITION:\n- original rejection SHA: {}\n- fix commit SHA: {}\n- petition attempts so far: {}\n- original reviewer model: {}\n\nORIGINAL CLAIM:\n{}\n\nORIGINAL SUMMARY:\n{}\n\nORIGINAL FINDINGS:\n{}\n\nFIX COMMIT DIFF:\n{}\n\nCLAIM (for ledger):\n{}",
context_block(context),
petition.original_sha,
petition.fix_sha,
petition.attempts_so_far,
petition.original_reviewer_model,
petition.original_claim,
petition.original_summary,
if findings_block.trim().is_empty() {
"(none recorded)".to_owned()
} else {
findings_block
},
diff,
claim.to_line(),
)
}
fn strict_second_pass_prompt(job: &ReviewJob, first_output: &str) -> String {
// A strict pass over a PETITION must stay on the petition question — the
// generic completeness prompt carries the fix commit's claim/diff but not
// the original rejection's findings, so the arbiter would end up judging
// a first review whose question it never saw.
if let Some(petition) = &job.petition {
let petition_question = petition_prompt(petition, &job.claim, &job.diff, &job.context);
return format!(
"{petition_question}\n\nSTRICT SECOND PASS (COMPLETENESS CRITIC): the first petition reviewer ACCEPTED that the fix materially addresses each original finding. Assume it verified some findings but not ALL of them. Re-check every original finding against the fix diff and prove the first reviewer INCOMPLETE.\n\nFIRST REVIEW:\n{first_output}"
);
}
format!(
"{ADVERSARIAL_PREAMBLE}\n\nStrict second pass (COMPLETENESS CRITIC): the first reviewer returned a CLEAN verdict. Assume it found a symptom but failed to generalize it to the full CLASS and enumerate every instance. Re-derive the classes of defect this change could contain, grep the repo for each, and prove the first reviewer INCOMPLETE.{}\n\nCLAIM:\n{}\n\nFIRST REVIEW:\n{}\n\nDIFF:\n{}",
context_block(&job.context),
job.claim.to_line(),
first_output,
job.diff
)
}
fn entry_from_verdict(job: &ReviewJob, plan: &ReviewPlan, verdict: &ParsedVerdict) -> LedgerEntry {
let mut entry = LedgerEntry::new(
job.commit_sha.clone(),
verdict.verdict,
job.claim.to_line(),
job.claim
.evidence
.iter()
.map(EvidenceRef::as_str)
.map(str::to_owned)
.collect(),
plan.reviewer_config(),
verdict.findings.clone(),
)
.with_structured_review(
verdict.summary.clone(),
verdict.structured_findings.clone(),
verdict.next_steps.clone(),
verdict.raw.clone(),
)
.with_memory_skill_classification(verdict.memory_skill_classification.clone());
if let Some(petition) = &job.petition {
entry.petition_for = Some(petition.original_sha.clone());
// `attempts_so_far` already includes the in-flight attempt — the CLI
// counted it when it enqueued the petition (see `resolve::run_petition`).
// Adding 1 here would double-count every attempt across CLI + watcher.
entry.petition_attempts = petition.attempts_so_far;
}
entry
}
fn ensure_process_success(output: &ProcessOutput) -> Result<(), ReviewerError> {
if output.status_code == Some(0) {
return Ok(());
}
Err(ReviewerError::ReviewerProcessFailed {
status: output.status_code,
stderr: output.stderr.clone(),
})
}
fn validate_strict_arbiter(
request: &ReviewRequest,
strict: &StrictReviewConfig,
) -> Result<(), ReviewerError> {
let arbiter = normalized_model(&strict.arbiter_model);
if arbiter == normalized_model(&request.watched_model)
|| arbiter == normalized_model(&request.reviewer_model)
{
return Err(ReviewerError::StrictArbiterModelNotDistinct);
}
Ok(())
}
fn validate_model_present(role: &str, model: &str) -> Result<(), ReviewerError> {
if model.trim().is_empty() {
return Err(ReviewerError::MissingModel {
role: role.to_owned(),
});
}
Ok(())
}
fn git_output<const N: usize>(args: [&str; N]) -> Result<String, ReviewerError> {
git_output_slice(&args)
}
fn git_output_slice(args: &[&str]) -> Result<String, ReviewerError> {
let output = Command::new("git")
.args(args)
.output()
.map_err(ReviewerError::GitSpawn)?;
if !output.status.success() {
return Err(ReviewerError::GitFailed {
args: args.iter().map(|arg| (*arg).to_owned()).collect(),
stderr: String::from_utf8_lossy(&output.stderr).into_owned(),
});
}
Ok(String::from_utf8_lossy(&output.stdout).into_owned())
}
fn agent_from_slug(value: &str) -> Result<Agent, ReviewerError> {
match value.trim().to_ascii_lowercase().as_str() {
"claude" => Ok(Agent::Claude),
"codex" => Ok(Agent::Codex),
"pi" => Ok(Agent::Pi),
"grok" => Ok(Agent::Grok),
_ => Err(ReviewerError::UnknownAgent {
value: value.to_owned(),
}),
}
}
fn harness_from_slug(value: &str) -> Result<ReviewerHarness, ReviewerError> {
match value.trim().to_ascii_lowercase().as_str() {
"claude" => Ok(ReviewerHarness::Claude),
"codex" => Ok(ReviewerHarness::Codex),
"pi" => Ok(ReviewerHarness::Pi),
"gemini" => Ok(ReviewerHarness::Gemini),
"opencode" => Ok(ReviewerHarness::Opencode),
"custom" => Ok(ReviewerHarness::Custom),
_ => Err(ReviewerError::UnknownHarness {
value: value.to_owned(),
}),
}
}
fn harness_slug(harness: ReviewerHarness) -> &'static str {
match harness {
ReviewerHarness::Claude => "claude",
ReviewerHarness::Codex => "codex",
ReviewerHarness::Pi => "pi",
ReviewerHarness::Gemini => "gemini",
ReviewerHarness::Opencode => "opencode",
ReviewerHarness::Custom => "custom",
}
}
/// Normalise a model identifier for opposition comparisons.
///
/// Delegates to [`crate::config::normalized_model`], which is the single
/// authoritative implementation shared by config-time validation and the
/// runtime same-model guard here (R6 dedup).
fn normalized_model(model: &str) -> String {
config::normalized_model(model)
}
fn generate_run_id(commit_sha: &str) -> String {
let nanos = SystemTime::now()
.duration_since(UNIX_EPOCH)
.map_or(0, |duration| duration.as_nanos());
let short_sha: String = commit_sha
.chars()
.filter(|character| character.is_ascii_alphanumeric())
.take(12)
.collect();
if short_sha.is_empty() {
format!("{nanos}-{}", std::process::id())
} else {
format!("{nanos}-{}-{short_sha}", std::process::id())
}
}
#[cfg(test)]
mod tests {
use std::{
cell::RefCell, collections::VecDeque, fs, path::PathBuf, process::Command, time::Duration,
};
use super::normalized_model;
use proptest::prelude::*;
use super::{
InvocationPlan, MaterialLoader, ParsedVerdict, ProcessOutput, ProcessRunner,
PromptDelivery, RUN_STORE_LOCK_DIR, ReviewJob, ReviewPlan, ReviewQueue, ReviewRequest,
ReviewRun, ReviewRunStatus, ReviewRunStore, ReviewSelection, ReviewerError,
RunStoreLockRecord, StrictGoalCounters, StrictGoalDecision, StrictGoalPolicy,
StrictGoalStopReason, StrictReviewConfig, UntilEmptyDecision, WatchLockClaim,
claim_watcher_lock, drain_once, execute_review_job, is_full_git_sha,
run_review_run_command, run_store_lock_is_stale, run_strict_goal_loop,
until_empty_decision, write_run_store_lock_owner,
};
use crate::{
claim::{Claim, EvidenceRef},
cli::{Agent, ReviewerHarness},
config::{Effort, TruthMirrorConfig},
ledger::{
FindingSeverity, LedgerStore, MemorySkillClassificationKind, StructuredFinding, Verdict,
},
watcher::{ProcessIdentity, pid_is_alive},
};
fn pass_json() -> String {
serde_json::json!({
"verdict": "PASS",
"summary": "The claim is substantiated by the diff and evidence.",
"findings": [],
"next_steps": [],
"memory_skill": {
"kind": "how_to_skill",
"learning_source": "run reusable review workflow",
"reasoning": "The passing claim describes a reusable verified procedure."
}
})
.to_string()
}
fn reject_json(title: &str) -> String {
serde_json::json!({
"verdict": "REJECT",
"summary": "The claim is not substantiated.",
"findings": [{
"severity": "high",
"title": title,
"body": "The cited evidence does not prove the claimed behavior.",
"file": "src/lib.rs",
"line_start": 1,
"line_end": 1,
"confidence": 95,
"recommendation": "Provide executable evidence that proves the claim."
}],
"next_steps": ["Run the relevant verification command."],
"memory_skill": {
"kind": "anti_pattern_skill",
"learning_source": title,
"reasoning": "The rejection identifies a reusable false-claim failure class."
}
})
.to_string()
}
#[test]
fn same_harness_different_model_is_valid() {
let request = ReviewRequest::new(
Agent::Codex,
"gpt-5.4",
ReviewerHarness::Codex,
"gpt-5.5",
false,
"review this",
);
let plan = ReviewPlan::build(request).unwrap();
assert_eq!(plan.watched_agent, Agent::Codex);
assert_eq!(plan.reviewer_harness, ReviewerHarness::Codex);
assert_eq!(plan.invocation.program, "codex");
}
#[test]
fn same_model_is_blocked_by_default() {
let request = ReviewRequest::new(
Agent::Codex,
" GPT-5.5 ",
ReviewerHarness::Claude,
"gpt-5.5",
false,
"review this",
);
let error = ReviewPlan::build(request).unwrap_err();
assert!(matches!(
error,
ReviewerError::SameModelWithoutWaiver { .. }
));
}
#[test]
fn allow_same_model_override_is_deliberate() {
let request = ReviewRequest::new(
Agent::Codex,
"gpt-5.5",
ReviewerHarness::Codex,
"gpt-5.5",
true,
"review this",
);
let plan = ReviewPlan::build(request).unwrap();
assert!(plan.allow_same_model);
assert_eq!(plan.reviewer_model, "gpt-5.5");
}
#[test]
fn provider_mapping_uses_verified_prompt_shapes_and_effort() {
let codex =
InvocationPlan::for_harness(ReviewerHarness::Codex, "gpt-5.5", Effort::Xhigh).unwrap();
assert_eq!(codex.program, "codex");
assert_eq!(
codex.args_for_prompt("prompt"),
[
"exec",
"-m",
"gpt-5.5",
"-c",
"model_reasoning_effort=xhigh",
"prompt"
]
);
let claude =
InvocationPlan::for_harness(ReviewerHarness::Claude, "opus", Effort::High).unwrap();
assert_eq!(claude.program, "claude");
assert_eq!(claude.prompt_delivery, PromptDelivery::Stdin);
assert_eq!(
claude.args_for_prompt("prompt"),
["--print", "--model", "opus", "--effort", "high"]
);
let gemini =
InvocationPlan::for_harness(ReviewerHarness::Gemini, "gemini-pro", Effort::Xhigh)
.unwrap();
assert_eq!(
gemini.args_for_prompt("prompt"),
["-m", "gemini-pro", "-p", "prompt"]
);
let pi = InvocationPlan::for_harness(ReviewerHarness::Pi, "openai/gpt-5.5", Effort::Xhigh)
.unwrap();
assert_eq!(pi.prompt_delivery, PromptDelivery::Stdin);
assert_eq!(
pi.args_for_prompt("prompt"),
[
"--model",
"openai/gpt-5.5",
"--thinking",
"xhigh",
"--tools",
"read,grep,find,ls",
"-p"
]
);
}
#[test]
fn custom_harness_requires_explicit_configuration() {
let error = InvocationPlan::for_harness(ReviewerHarness::Custom, "model", Effort::Xhigh)
.unwrap_err();
assert!(matches!(error, ReviewerError::UnsupportedCustomHarness));
}
#[test]
fn effort_maps_to_each_harness_flag() {
for effort in [
Effort::Minimal,
Effort::Low,
Effort::Medium,
Effort::High,
Effort::Xhigh,
] {
let e = effort.as_str();
let codex = InvocationPlan::for_harness(ReviewerHarness::Codex, "m", effort).unwrap();
assert!(codex.args.contains(&format!("model_reasoning_effort={e}")));
let claude = InvocationPlan::for_harness(ReviewerHarness::Claude, "m", effort).unwrap();
let claude_idx = claude.args.iter().position(|a| a == "--effort").unwrap();
// Claude has no `minimal`; it clamps to a valid level (`low`).
assert_eq!(claude.args[claude_idx + 1], effort.claude_value());
assert_ne!(claude.args[claude_idx + 1], "minimal");
let pi = InvocationPlan::for_harness(ReviewerHarness::Pi, "m", effort).unwrap();
let pi_idx = pi.args.iter().position(|a| a == "--thinking").unwrap();
assert_eq!(pi.args[pi_idx + 1], e);
}
}
#[test]
fn resolve_picks_configured_reviewer_for_every_writer() {
let config = crate::config::TruthMirrorConfig::default();
let cases = [
(Agent::Codex, ReviewerHarness::Claude, "claude-opus-4-8"),
(Agent::Claude, ReviewerHarness::Codex, "gpt-5.5"),
(Agent::Pi, ReviewerHarness::Codex, "gpt-5.5"),
(Agent::Grok, ReviewerHarness::Claude, "claude-opus-4-8"),
];
for (writer, reviewer_harness, reviewer_model) in cases {
let selection =
ReviewSelection::resolve(Some(writer), None, None, None, None, false, &config)
.unwrap();
assert_eq!(selection.reviewer_harness, reviewer_harness);
assert_eq!(selection.reviewer_model, reviewer_model);
assert_eq!(selection.reviewer_effort, Effort::Xhigh);
}
}
#[test]
fn overriding_reviewer_harness_without_model_is_rejected() {
// codex's default pair reviewer is claude; overriding harness to pi with no
// model would pair the pi harness with a claude model string.
let config = crate::config::TruthMirrorConfig::default();
let error = ReviewSelection::resolve(
Some(Agent::Codex),
None,
Some(ReviewerHarness::Pi),
None,
None,
false,
&config,
)
.unwrap_err();
assert!(matches!(error, ReviewerError::OverrideNeedsModel { .. }));
}
#[test]
fn overriding_reviewer_harness_matching_pair_is_ok() {
let config = crate::config::TruthMirrorConfig::default();
let selection = ReviewSelection::resolve(
Some(Agent::Codex),
None,
Some(ReviewerHarness::Claude),
None,
None,
false,
&config,
)
.unwrap();
assert_eq!(selection.reviewer_harness, ReviewerHarness::Claude);
assert_eq!(selection.reviewer_model, "claude-opus-4-8");
}
#[test]
fn config_allow_same_model_waives_opposition() {
let config = crate::config::TruthMirrorConfig {
allow_same_model: true,
..crate::config::TruthMirrorConfig::default()
};
let selection = ReviewSelection::resolve(
Some(Agent::Codex),
Some("gpt-5.5".to_owned()),
Some(ReviewerHarness::Codex),
Some("gpt-5.5".to_owned()),
None,
false, // CLI flag not set — the config waiver must carry it
&config,
)
.unwrap();
assert!(selection.allow_same_model);
// Same watched+reviewer model builds because the config waiver applies.
assert!(ReviewPlan::build(selection.request_for("review".to_owned())).is_ok());
}
#[test]
fn full_git_sha_accepts_sha1_and_sha256_lengths() {
assert!(is_full_git_sha(&"a".repeat(40)));
assert!(is_full_git_sha(&"b".repeat(64)));
assert!(!is_full_git_sha(&"c".repeat(39)));
assert!(!is_full_git_sha(&"g".repeat(40)));
}
#[test]
fn resolve_arbiter_uses_pair_when_cli_absent() {
let config = crate::config::TruthMirrorConfig::default();
let arbiter =
ReviewSelection::resolve_arbiter(Agent::Codex, None, None, None, &config).unwrap();
assert_eq!(arbiter.arbiter_harness, ReviewerHarness::Pi);
assert_eq!(arbiter.arbiter_effort, Effort::Xhigh);
}
#[test]
fn first_pass_prompt_is_adversarial_and_injects_context() {
let prompt = super::first_pass_prompt(
&claim(),
"THE_DIFF_BODY",
"INVIOLABLE CONSTRAINTS: never fake tests",
);
assert!(prompt.contains("PROVE THIS CLAIM FALSE"));
assert!(prompt.contains("default to REJECT"));
assert!(prompt.contains("INVIOLABLE CONSTRAINTS: never fake tests"));
assert!(prompt.contains("THE_DIFF_BODY"));
// Class-generalized review: grep the class, not the instance.
assert!(prompt.contains("GREP THE CLASS, NOT THE INSTANCE"));
assert!(prompt.contains("\"severity\""));
assert!(prompt.contains("\"recommendation\""));
assert!(prompt.contains("\"memory_skill\""));
assert!(prompt.contains("\"anti_pattern_skill\""));
}
#[test]
fn strict_second_pass_is_a_completeness_critic() {
let job = review_job(true);
let first_output = pass_json();
let prompt = super::strict_second_pass_prompt(&job, &first_output);
assert!(prompt.contains("COMPLETENESS CRITIC"));
assert!(prompt.contains("generalize"));
// Inherits the class-sweep preamble.
assert!(prompt.contains("GREP THE CLASS, NOT THE INSTANCE"));
}
#[test]
fn prompt_omits_context_block_when_empty() {
let prompt = super::first_pass_prompt(&claim(), "d", "");
// No dangling empty context header.
assert!(!prompt.contains("INVIOLABLE CONSTRAINTS"));
assert!(prompt.contains("PROVE THIS CLAIM FALSE"));
}
fn sample_petition(attempts_so_far: u32) -> super::PetitionContext {
super::PetitionContext {
original_sha: "abc123".to_owned(),
fix_sha: "def456".to_owned(),
original_claim: "CLAIM: original | verified: cargo test | evidence: tests:cargo-test"
.to_owned(),
original_summary: "Original rejection summary.".to_owned(),
original_findings: vec!["evidence too thin".to_owned()],
original_structured_findings: vec![StructuredFinding {
severity: FindingSeverity::High,
title: "thin evidence".to_owned(),
body: "Evidence pointer does not prove the claim.".to_owned(),
file: "src/lib.rs".to_owned(),
line_start: 1,
line_end: 2,
confidence: 90,
recommendation: "Add executable evidence.".to_owned(),
}],
original_reviewer_model: "claude-opus-4-1".to_owned(),
attempts_so_far,
}
}
#[test]
fn petition_prompt_includes_original_findings_fix_sha_and_attempts() {
let prompt = super::petition_prompt(&sample_petition(1), &claim(), "DIFF BODY", "");
assert!(prompt.contains("PETITION"));
assert!(prompt.contains("original rejection SHA: abc123"));
assert!(prompt.contains("fix commit SHA: def456"));
assert!(prompt.contains("petition attempts so far: 1"));
assert!(prompt.contains("Original rejection summary."));
assert!(prompt.contains("thin evidence")); // structured finding title
assert!(prompt.contains("DIFF BODY"));
assert!(prompt.contains("\"verdict\": \"PASS\" | \"REJECT\" | \"FLAG\""));
}
#[test]
fn materialize_petition_prompt_rewrites_request_prompt() {
let mut job = review_job(false);
job.petition = Some(sample_petition(2));
let original_prompt = job.request.prompt.clone();
let updated = super::materialize_petition_prompt(job);
assert_ne!(updated.request.prompt, original_prompt);
assert!(updated.request.prompt.contains("PETITION"));
assert!(updated.request.prompt.contains("fix commit SHA: def456"));
assert!(
updated
.request
.prompt
.contains("petition attempts so far: 2")
);
}
#[test]
fn materialize_petition_prompt_passthrough_when_no_petition() {
let job = review_job(false);
let original_prompt = job.request.prompt.clone();
let updated = super::materialize_petition_prompt(job);
assert_eq!(updated.request.prompt, original_prompt);
}
#[test]
fn subprocess_runner_is_mockable() {
struct MockRunner;
impl ProcessRunner for MockRunner {
fn run(
&self,
invocation: &InvocationPlan,
prompt: &str,
_timeout: Option<Duration>,
) -> Result<ProcessOutput, ReviewerError> {
assert_eq!(invocation.program, "codex");
assert_eq!(
invocation.args_for_prompt(prompt).last().unwrap(),
"review this"
);
Ok(ProcessOutput {
status_code: Some(0),
stdout: pass_json(),
stderr: String::new(),
})
}
}
let request = ReviewRequest::new(
Agent::Codex,
"gpt-5.4",
ReviewerHarness::Codex,
"gpt-5.5",
false,
"review this",
);
let plan = ReviewPlan::build(request).unwrap();
let output = plan.run_with("review this", &MockRunner, None).unwrap();
assert!(output.stdout.contains("PASS"));
}
#[cfg(unix)]
#[test]
fn std_process_runner_kills_wedged_child_at_timeout() {
// 0.9.2 field failure: `codex exec` reviewers wedged for 1–7 hours and
// froze the whole queue. The runner must kill the child at the
// configured wall-clock deadline and surface a timeout error.
let invocation = InvocationPlan {
program: "sleep".to_owned(),
args: vec!["30".to_owned()],
prompt_delivery: PromptDelivery::Stdin,
};
let runner = super::StdProcessRunner;
let started = std::time::Instant::now();
let error = runner
.run(&invocation, "", Some(Duration::from_millis(300)))
.unwrap_err();
let elapsed = started.elapsed();
assert!(
matches!(error, ReviewerError::ReviewerTimeout { .. }),
"expected a timeout error, got: {error}"
);
assert!(
elapsed < Duration::from_secs(10),
"the wedged child was not killed promptly (elapsed: {elapsed:?})"
);
}
#[cfg(unix)]
#[test]
fn std_process_runner_collects_output_without_timeout() {
// The no-timeout path keeps the old wait_with_output behavior: exit
// status plus both pipes captured.
let invocation = InvocationPlan {
program: "echo".to_owned(),
args: vec!["hello".to_owned()],
prompt_delivery: PromptDelivery::PositionalArgument,
};
let runner = super::StdProcessRunner;
let output = runner.run(&invocation, "", None).unwrap();
assert_eq!(output.status_code, Some(0));
assert_eq!(output.stdout.trim(), "hello");
}
#[cfg(unix)]
#[test]
fn std_process_runner_timeout_covers_blocked_stdin_write() {
// 0.9.2 field failure, first fix iteration: the deadline started only
// AFTER the blocking prompt `write_all` returned, so a reviewer that
// never reads stdin wedged the queue forever. `sleep` reads nothing;
// a prompt larger than any pipe buffer (64 KiB) blocks the write, and
// the deadline must still fire. (Against the old code this test hangs
// rather than fails — that hang IS the field failure.)
let invocation = InvocationPlan {
program: "sleep".to_owned(),
args: vec!["30".to_owned()],
prompt_delivery: PromptDelivery::Stdin,
};
let runner = super::StdProcessRunner;
let huge_prompt = "x".repeat(2 * 1024 * 1024);
let started = std::time::Instant::now();
let error = runner
.run(&invocation, &huge_prompt, Some(Duration::from_millis(300)))
.unwrap_err();
let elapsed = started.elapsed();
assert!(
matches!(error, ReviewerError::ReviewerTimeout { .. }),
"expected a timeout error, got: {error}"
);
assert!(
elapsed < Duration::from_secs(10),
"the deadline must cover the blocked stdin write (elapsed: {elapsed:?})"
);
}
#[cfg(unix)]
#[test]
fn std_process_runner_timeout_kills_descendants_holding_pipes() {
// 0.9.2 queue freeze, second half: the reviewer spawns a descendant
// that inherits the output pipes. Killing only the direct child left
// the descendant holding the pipes open, so the reader-thread join
// blocked until the descendant exited (~30s here — forever in the
// field). The timeout kill must take the whole process group.
let invocation = InvocationPlan {
program: "sh".to_owned(),
args: vec!["-c".to_owned(), "sleep 30 & sleep 30".to_owned()],
prompt_delivery: PromptDelivery::PositionalArgument,
};
let runner = super::StdProcessRunner;
let started = std::time::Instant::now();
let error = runner
.run(&invocation, "", Some(Duration::from_millis(300)))
.unwrap_err();
let elapsed = started.elapsed();
assert!(
matches!(error, ReviewerError::ReviewerTimeout { .. }),
"expected a timeout error, got: {error}"
);
assert!(
elapsed < Duration::from_secs(10),
"a pipe-hoarding descendant must not stall the reap (elapsed: {elapsed:?})"
);
}
#[cfg(unix)]
#[test]
fn std_process_runner_timeout_kills_spawned_descendants() {
// The descendant must not SURVIVE the timeout kill either: the
// backgrounded subshell would write its sentinel 2s in, well after
// the 300ms deadline. A live sentinel proves the tree kill missed it.
let temp = tempfile::tempdir().unwrap();
let sentinel = temp.path().join("descendant-survived");
let invocation = InvocationPlan {
program: "sh".to_owned(),
args: vec![
"-c".to_owned(),
"(sleep 2; touch \"$1\") & exec sleep 30".to_owned(),
"sentinel".to_owned(),
sentinel.display().to_string(),
],
prompt_delivery: PromptDelivery::PositionalArgument,
};
let runner = super::StdProcessRunner;
let error = runner
.run(&invocation, "", Some(Duration::from_millis(300)))
.unwrap_err();
assert!(
matches!(error, ReviewerError::ReviewerTimeout { .. }),
"expected a timeout error, got: {error}"
);
std::thread::sleep(Duration::from_millis(2500));
assert!(
!sentinel.exists(),
"the spawned descendant survived the timeout kill and wrote its sentinel"
);
}
#[cfg(unix)]
#[test]
fn outcome_after_kill_prefers_real_outcome_over_timeout() {
use std::os::unix::process::ExitStatusExt as _;
// A child that exited on its own — successfully or not — before the
// kill landed reports its REAL outcome (unix: signal() is None only
// when the process was not signal-killed).
assert!(super::outcome_after_kill(std::process::ExitStatus::from_raw(0), 1).is_ok());
assert!(super::outcome_after_kill(std::process::ExitStatus::from_raw(1 << 8), 1).is_ok());
// A signal-killed child (our SIGKILL landed) reports the timeout.
assert!(matches!(
super::outcome_after_kill(std::process::ExitStatus::from_raw(9), 7),
Err(ReviewerError::ReviewerTimeout { timeout_secs: 7 })
));
}
#[cfg(unix)]
#[test]
fn wait_for_reap_enforces_grace_and_surfaces_timeout() {
let mut child = std::process::Command::new("sh")
.arg("-c")
.arg("sleep 5")
.spawn()
.unwrap();
let started = std::time::Instant::now();
let error = super::wait_for_reap(&mut child, Duration::from_millis(100))
.expect_err("still-running child must not block forever");
let elapsed = started.elapsed();
assert_eq!(error.kind(), std::io::ErrorKind::Other);
assert!(
elapsed >= Duration::from_millis(100),
"reap timeout must wait for the configured grace; elapsed {elapsed:?}"
);
assert!(
elapsed < Duration::from_secs(1),
"reap timeout must be bounded"
);
child.kill().ok();
child.wait().ok();
}
#[cfg(unix)]
#[test]
fn reap_after_grace_returns_after_bounded_cleanup_handoff() {
let child = std::process::Command::new("sh")
.arg("-c")
.arg("sleep 5")
.spawn()
.unwrap();
let started = std::time::Instant::now();
let error = super::reap_after_grace(child, std::io::Error::other("test grace expired"), 7)
.unwrap_err();
assert!(matches!(
error,
ReviewerError::ReviewerTimeout { timeout_secs: 7 }
));
assert!(
started.elapsed() < Duration::from_secs(1),
"post-kill cleanup must not block on direct-child wait"
);
}
#[cfg(unix)]
#[test]
fn std_process_runner_surfaces_prompt_write_failure() {
// A reviewer that closes stdin immediately makes the prompt write
// fail with EPIPE. The error must surface (a truncated question
// invalidates the verdict), never be swallowed. The prompt is larger
// than any pipe buffer so the writer is still blocked when `exec
// 0<&-` closes the read end — no timing race.
let invocation = InvocationPlan {
program: "sh".to_owned(),
args: vec!["-c".to_owned(), "exec 0<&-; sleep 1".to_owned()],
prompt_delivery: PromptDelivery::Stdin,
};
let runner = super::StdProcessRunner;
let huge_prompt = "x".repeat(2 * 1024 * 1024);
let error = runner
.run(&invocation, &huge_prompt, Some(Duration::from_secs(10)))
.unwrap_err();
assert!(
matches!(error, ReviewerError::WritePrompt(_)),
"expected a prompt-write error, got: {error}"
);
}
#[test]
fn claim_watcher_lock_owns_free_slot_and_adopted_handoff() {
let temp = tempfile::tempdir().unwrap();
let state = temp.path();
// A free slot is claimed.
assert_eq!(
claim_watcher_lock(state, false, Duration::from_millis(10), false).unwrap(),
WatchLockClaim::Owned
);
let token = crate::watcher::lock_acquisition_token(state).unwrap();
crate::watcher::release_lock_if_owned(state, &token).unwrap();
// The ensure-watcher handoff: the lock already records THIS process
// (ensure-watcher re-points it at the detached child). The spawned
// watcher must adopt its own slot, not refuse it as contention.
assert_eq!(
crate::watcher::try_acquire_lock(state).unwrap(),
crate::watcher::LockClaim::Acquired
);
assert_eq!(
claim_watcher_lock(state, false, Duration::from_millis(10), false).unwrap(),
WatchLockClaim::Owned
);
let token = crate::watcher::lock_acquisition_token(state).unwrap();
crate::watcher::release_lock_if_owned(state, &token).unwrap();
}
#[test]
fn run_store_lock_is_stale_only_when_owner_is_proven_dead() {
let temp = tempfile::tempdir().unwrap();
let root = temp.path();
let lock_dir = root.join(RUN_STORE_LOCK_DIR);
fs::create_dir_all(&lock_dir).unwrap();
// Current process owner keeps the lock live even if the lock files
// themselves are otherwise old.
let owner = ProcessIdentity::current();
write_run_store_lock_owner(&lock_dir, &owner).unwrap();
assert!(
!run_store_lock_is_stale(&lock_dir),
"lock with matching owner identity should not be reclaimed"
);
// An unprobeable pid remains Unknown even after the heartbeat has gone
// stale; heartbeat age alone is never a reclaim proof.
let unknown = ProcessIdentity {
pid: 4_000_000_000,
start_token: owner.start_token.clone(),
};
write_run_store_lock_owner(&lock_dir, &unknown).unwrap();
std::thread::sleep(Duration::from_millis(350));
assert!(
!run_store_lock_is_stale(&lock_dir),
"an Unknown owner must not be reclaimed from heartbeat age alone"
);
// A real reaped child is proven dead and becomes reclaimable only after
// the configured missed-heartbeat threshold.
let dead = ProcessIdentity {
pid: reaped_pid(),
start_token: owner.start_token.clone(),
};
write_run_store_lock_owner(&lock_dir, &dead).unwrap();
std::thread::sleep(Duration::from_millis(350));
assert!(
run_store_lock_is_stale(&lock_dir),
"a proven-dead owner with no refreshed heartbeat should be reclaimable"
);
}
#[test]
fn run_store_lock_heartbeat_refreshes_while_held() {
let temp = tempfile::tempdir().unwrap();
let store = ReviewRunStore::new(temp.path());
let guard = store.lock().unwrap();
let before = fs::metadata(&guard.path).unwrap().modified().unwrap();
std::thread::sleep(Duration::from_millis(250));
let after = fs::metadata(&guard.path).unwrap().modified().unwrap();
assert!(
after > before,
"the held run-store lock heartbeat must refresh its lease mtime"
);
}
#[test]
fn run_store_lock_drop_does_not_remove_same_owner_replacement() {
let temp = tempfile::tempdir().unwrap();
let store = ReviewRunStore::new(temp.path());
let guard = store.lock().unwrap();
let replacement = RunStoreLockRecord {
owner: guard.owner.clone(),
acquisition_token: "replacement-token".to_owned(),
};
fs::write(&guard.path, serde_json::to_vec(&replacement).unwrap()).unwrap();
let lock_path = guard.path.clone();
drop(guard);
assert!(
lock_path.exists(),
"a replacement with the same process identity must survive the old guard"
);
}
#[test]
fn run_store_lock_drop_removes_only_owned_lock() {
let temp = tempfile::tempdir().unwrap();
let store = ReviewRunStore::new(temp.path());
let lock_path = store.root.join(RUN_STORE_LOCK_DIR);
drop(store.lock().unwrap());
assert!(!lock_path.exists(), "owned lock should be released on drop");
}
#[cfg(unix)]
struct KillGuard(std::process::Child);
#[cfg(unix)]
impl Drop for KillGuard {
fn drop(&mut self) {
let _ = self.0.kill();
let _ = self.0.wait();
}
}
#[cfg(unix)]
#[test]
fn claim_watcher_lock_refuses_foreign_live_owner_without_wait() {
// 0.9.2 field failure: a second watcher proceeded "without lock
// ownership" and double-processed the queue. Now a foreign-held lock
// means refusal — fast, and without disturbing the incumbent's lock.
let temp = tempfile::tempdir().unwrap();
let state = temp.path();
let sleeper = Command::new("sleep").arg("30").spawn().unwrap();
let guard = KillGuard(sleeper);
// Empty start token: liveness degrades to pid-alive, which the
// sleeper satisfies, so the lock counts as held by a live watcher.
let lock = crate::watcher::WatcherLock {
identity: crate::watcher::ProcessIdentity {
pid: guard.0.id(),
start_token: String::new(),
},
acquisition_token: "foreign-token".to_owned(),
created_at_unix: 1,
};
fs::create_dir_all(state).unwrap();
fs::write(
state.join(crate::watcher::WATCHER_LOCK_FILE),
serde_json::to_vec_pretty(&lock).unwrap(),
)
.unwrap();
let started = std::time::Instant::now();
let claim = claim_watcher_lock(state, false, Duration::from_millis(10), false).unwrap();
assert!(
matches!(claim, WatchLockClaim::Refused(_)),
"a foreign live owner must refuse the claim, got {claim:?}"
);
assert!(
started.elapsed() < Duration::from_secs(5),
"refusal must be immediate when --wait-for-lock is not given"
);
// The incumbent's lock file was neither stolen nor deleted.
assert!(state.join(crate::watcher::WATCHER_LOCK_FILE).exists());
}
#[cfg(unix)]
#[test]
fn claim_watcher_lock_absorbs_ensure_watcher_handoff_window() {
// The detached child can check the lock BEFORE ensure-watcher
// re-points it at the child's identity. With expect_handoff the child
// must wait out the window and adopt the slot once the re-point
// lands, instead of refusing and stranding the queue when the parent
// exits. Without expect_handoff the same setup refuses immediately
// (covered by claim_watcher_lock_refuses_foreign_live_owner_without_wait).
let temp = tempfile::tempdir().unwrap();
let state = temp.path().to_path_buf();
// A foreign live owner: a sleeper pid with an empty token.
let sleeper = Command::new("sleep").arg("30").spawn().unwrap();
let guard = KillGuard(sleeper);
fs::create_dir_all(&state).unwrap();
fs::write(
state.join(crate::watcher::WATCHER_LOCK_FILE),
serde_json::to_vec_pretty(&crate::watcher::WatcherLock {
identity: crate::watcher::ProcessIdentity {
pid: guard.0.id(),
start_token: String::new(),
},
acquisition_token: "handoff-token".to_owned(),
created_at_unix: crate::time::unix_now(),
})
.unwrap(),
)
.unwrap();
// Re-point the lock at THIS process after 300ms — exactly what
// ensure-watcher does right after spawn returns.
let re_point = std::thread::spawn({
let state = state.clone();
move || {
std::thread::sleep(Duration::from_millis(300));
let lock = crate::watcher::WatcherLock {
identity: crate::watcher::ProcessIdentity::current(),
acquisition_token: "handoff-token".to_owned(),
created_at_unix: crate::time::unix_now(),
};
fs::write(
state.join(crate::watcher::WATCHER_LOCK_FILE),
serde_json::to_vec_pretty(&lock).unwrap(),
)
.unwrap();
}
});
let started = std::time::Instant::now();
let claim = claim_watcher_lock(&state, false, Duration::from_millis(50), true).unwrap();
assert_eq!(claim, WatchLockClaim::Owned);
assert!(
started.elapsed() >= Duration::from_millis(250),
"the claim must have waited for the handoff (elapsed: {:?})",
started.elapsed()
);
assert!(
started.elapsed() < Duration::from_secs(5),
"the handoff must be adopted promptly once it lands"
);
re_point.join().unwrap();
crate::watcher::release_lock_if_owned(&state, "handoff-token").unwrap();
}
#[test]
fn drain_once_marks_timed_out_run_failed_and_keeps_draining() {
// A wedged reviewer must not strand the queue: the timed out item is
// recorded failed with the timeout reason and consumed, and the next
// queued commit is still reviewed in the same drain.
struct TimeoutThenPassRunner {
calls: RefCell<usize>,
observed_timeout: RefCell<Option<Duration>>,
}
impl ProcessRunner for TimeoutThenPassRunner {
fn run(
&self,
_invocation: &InvocationPlan,
_prompt: &str,
timeout: Option<Duration>,
) -> Result<ProcessOutput, ReviewerError> {
*self.observed_timeout.borrow_mut() = timeout;
let call = {
let mut calls = self.calls.borrow_mut();
*calls += 1;
*calls
};
if call == 1 {
return Err(ReviewerError::ReviewerTimeout { timeout_secs: 1200 });
}
Ok(ProcessOutput {
status_code: Some(0),
stdout: pass_json(),
stderr: String::new(),
})
}
}
let temp = tempfile::tempdir().unwrap();
let store = LedgerStore::new(temp.path());
let queue = ReviewQueue::new(temp.path());
queue.enqueue("wedged").unwrap();
queue.enqueue("healthy").unwrap();
let runner = TimeoutThenPassRunner {
calls: RefCell::new(0),
observed_timeout: RefCell::new(None),
};
let report = drain_once(
&queue,
&StaticLoader::new(),
&selection(),
"",
&runner,
&store,
&TruthMirrorConfig::default(),
)
.unwrap();
// The configured timeout (default 20 minutes) reached the runner.
assert_eq!(
*runner.observed_timeout.borrow(),
Some(Duration::from_secs(20 * 60))
);
// The wedged item was skipped as failed; the healthy one reviewed.
assert_eq!(report.failed, ["wedged"]);
assert_eq!(report.reviewed, ["healthy"]);
assert!(queue.pending().unwrap().is_empty());
let run_store = ReviewRunStore::new(temp.path());
let runs = run_store.list().unwrap();
let wedged = runs.iter().find(|run| run.commit_sha == "wedged").unwrap();
assert_eq!(wedged.status, ReviewRunStatus::Failed);
assert!(
wedged
.error
.as_deref()
.is_some_and(|error| error.contains("timed out")),
"timeout reason should be recorded, got: {:?}",
wedged.error
);
let healthy = runs.iter().find(|run| run.commit_sha == "healthy").unwrap();
assert_eq!(healthy.status, ReviewRunStatus::Completed);
}
#[test]
fn verdict_parser_extracts_rejection_findings() {
let verdict = ParsedVerdict::parse(&reject_json("missing proof")).unwrap();
assert_eq!(verdict.verdict, Verdict::Reject);
assert_eq!(verdict.structured_findings[0].title, "missing proof");
assert_eq!(verdict.structured_findings[0].confidence, 95);
assert!(verdict.findings[0].contains("missing proof"));
assert_eq!(
verdict.memory_skill_classification.learning_source,
"missing proof"
);
}
fn flag_json(title: &str) -> String {
serde_json::json!({
"verdict": "FLAG",
"summary": "Claim substantiated; flagging thin evidence.",
"findings": [{
"severity": "medium",
"title": title,
"body": "Evidence pointer is thin but the claim holds.",
"file": "src/lib.rs",
"line_start": 1,
"line_end": 1,
"confidence": 60,
"recommendation": "Tighten evidence."
}],
"next_steps": ["Add stronger evidence."],
"memory_skill": {
"kind": "none",
"learning_source": "",
"reasoning": "No reusable procedural memory to propose."
}
})
.to_string()
}
#[test]
fn verdict_parser_extracts_flag_findings() {
let verdict = ParsedVerdict::parse(&flag_json("thin evidence")).unwrap();
assert_eq!(verdict.verdict, Verdict::Flag);
assert_eq!(verdict.structured_findings[0].title, "thin evidence");
assert_eq!(verdict.structured_findings[0].confidence, 60);
assert!(verdict.findings[0].contains("thin evidence"));
}
#[test]
fn verdict_parser_rejects_flag_with_no_findings() {
let output = serde_json::json!({
"verdict": "FLAG",
"summary": "Claim substantiated but flagging.",
"findings": [],
"next_steps": [],
"memory_skill": {
"kind": "none",
"learning_source": "",
"reasoning": "no reusable memory"
}
});
let error = ParsedVerdict::parse(&output.to_string()).unwrap_err();
assert!(matches!(error, ReviewerError::VerdictSchema { .. }));
}
#[test]
fn verdict_parser_accepts_lowercase_flag_via_schema() {
// Reviewer harnesses vary in case; serde rename_all = "UPPERCASE"
// canonicalises, but reviewers may emit lowercase. Confirm the
// existing parser does NOT silently coerce: lowercase should fail
// (the schema is the source of truth for new entries).
let output = serde_json::json!({
"verdict": "flag",
"summary": "Claim substantiated but flagging.",
"findings": [{
"severity": "low",
"title": "x",
"body": "y",
"file": "src/lib.rs",
"line_start": 1,
"line_end": 1,
"confidence": 50,
"recommendation": "z"
}],
"next_steps": [],
"memory_skill": {
"kind": "none",
"learning_source": "",
"reasoning": "none"
}
});
let error = ParsedVerdict::parse(&output.to_string()).unwrap_err();
assert!(matches!(error, ReviewerError::VerdictJson { .. }));
}
#[test]
fn verdict_parser_rejects_missing_memory_skill_classification() {
let output = serde_json::json!({
"verdict": "PASS",
"summary": "The claim is substantiated.",
"findings": [],
"next_steps": []
});
let error = ParsedVerdict::parse(&output.to_string()).unwrap_err();
assert!(matches!(error, ReviewerError::VerdictJson { .. }));
}
#[test]
fn verdict_parser_rejects_unrecoverable_memory_skill_classification() {
let mut output: serde_json::Value =
serde_json::from_str(&reject_json("missing proof")).unwrap();
output["memory_skill"]["kind"] = serde_json::json!("how_to_skill");
let error = ParsedVerdict::parse(&output.to_string()).unwrap_err();
assert!(matches!(error, ReviewerError::VerdictSchema { .. }));
}
#[test]
fn verdict_parser_accepts_normalized_float_confidence() {
let mut output: serde_json::Value =
serde_json::from_str(&reject_json("missing proof")).unwrap();
output["findings"][0]["confidence"] = serde_json::json!(0.95);
let verdict = ParsedVerdict::parse(&output.to_string()).unwrap();
assert_eq!(verdict.structured_findings[0].confidence, 95);
}
#[test]
fn verdict_parser_rejects_legacy_line_protocol() {
let error =
ParsedVerdict::parse("VERDICT: REJECT\nFINDINGS:\n- missing proof\n").unwrap_err();
assert!(matches!(error, ReviewerError::VerdictJson { .. }));
}
#[test]
fn large_diff_materialization_falls_back_to_file_summary() {
let files = "a.rs\nb.rs\nc.rs\n";
let materialized = super::materialize_diff("branch:main", "tiny diff", files);
assert!(materialized.contains("too large to inline safely"));
assert!(materialized.contains("actual_files=3"));
assert!(materialized.contains("a.rs\nb.rs\nc.rs"));
assert!(materialized.contains("inspect the repository directly"));
}
#[test]
fn review_queue_schedules_commits_without_running_models() {
let temp = tempfile::tempdir().unwrap();
let queue = ReviewQueue::new(temp.path());
queue.enqueue("abc123").unwrap();
let pending = queue.pending().unwrap();
assert_eq!(pending.len(), 1);
assert_eq!(pending[0].commit_sha, "abc123");
assert!(!pending[0].run_id.is_empty());
let run = ReviewRunStore::new(temp.path())
.read(&pending[0].run_id)
.unwrap();
assert_eq!(run.commit_sha, "abc123");
assert_eq!(run.status, ReviewRunStatus::Queued);
assert_eq!(run.entire_checkpoint, None);
}
#[test]
fn review_queue_summary_counts_pending_and_oldest() {
let temp = tempfile::tempdir().unwrap();
let queue = ReviewQueue::new(temp.path());
std::fs::write(
queue.path(),
"{\"commit_sha\":\"abc123\",\"enqueued_at_unix\":30}\n{\"commit_sha\":\"def456\",\"enqueued_at_unix\":10}\n",
)
.unwrap();
let summary = queue.summary().unwrap();
assert_eq!(summary.pending_count, 2);
assert_eq!(summary.oldest_enqueued_at_unix, Some(10));
assert_eq!(summary.oldest_age_secs_at(42), Some(32));
}
#[test]
fn review_run_serializes_optional_entire_checkpoint() {
let run = ReviewRun::queued_with_provenance(
"run-id",
"abcdef123456",
"commit",
Some(crate::provenance::EntireCheckpointRef {
ref_name: "entire/session-abcdef".to_owned(),
object_sha: "123456".to_owned(),
}),
);
let json = serde_json::to_string(&run).unwrap();
assert!(json.contains("entire_checkpoint"));
assert!(json.contains("entire/session-abcdef"));
}
#[test]
fn review_run_omits_absent_optional_entire_checkpoint() {
let run = ReviewRun::queued("run-id", "abcdef123456", "commit");
let json = serde_json::to_string(&run).unwrap();
assert!(!json.contains("entire_checkpoint"));
}
#[test]
fn review_run_status_counts_read_only_status_fields() {
let temp = tempfile::tempdir().unwrap();
let store = ReviewRunStore::new(temp.path());
let queued = store.create_queued("abc123", "commit").unwrap();
let mut failed = ReviewRun::queued("failed-run", "def456", "commit");
failed.mark_failed("boom");
store.write(&failed).unwrap();
fs::write(store.path("malformed-run"), "{not-json\n").unwrap();
let counts = store.status_counts().unwrap();
assert_eq!(counts.queued, 1);
assert_eq!(counts.failed, 1);
assert_eq!(counts.running, 0);
assert_eq!(counts.skipped_records, 1);
assert_eq!(queued.status, ReviewRunStatus::Queued);
}
#[test]
fn review_cancel_marks_queued_run_and_removes_queue_item() {
let temp = tempfile::tempdir().unwrap();
let queue = ReviewQueue::new(temp.path());
let queued = queue.enqueue("abc123").unwrap();
run_review_run_command(
crate::cli::ReviewCommand::Cancel {
run_id: queued.run_id.clone(),
force: false,
},
temp.path(),
)
.unwrap();
assert!(queue.pending().unwrap().is_empty());
let run = ReviewRunStore::new(temp.path())
.read(&queued.run_id)
.unwrap();
assert_eq!(run.status, ReviewRunStatus::Cancelled);
}
/// A pid guaranteed to be dead: spawn a trivial process, reap it, return its pid.
fn reaped_pid() -> u32 {
let mut child = Command::new("true").spawn().expect("spawn `true`");
let pid = child.id();
child.wait().expect("reap `true`");
pid
}
/// Persist a run pinned to `Running` with a specific worker pid, bypassing the
/// normal lifecycle so liveness handling can be exercised deterministically.
fn write_running_run(store: &ReviewRunStore, worker_pid: Option<u32>) -> ReviewRun {
let mut run = store.create_queued("abc123", "commit").unwrap();
run.status = ReviewRunStatus::Running;
run.phase = "reviewing".to_owned();
run.worker_pid = worker_pid;
store.write(&run).unwrap();
run
}
#[test]
fn pid_liveness_probe_tracks_real_processes() {
assert!(pid_is_alive(std::process::id()));
assert!(!pid_is_alive(reaped_pid()));
assert_eq!(
crate::watcher::probe_pid_liveness(std::process::id()),
crate::watcher::PidLiveness::Alive
);
assert_eq!(
crate::watcher::probe_pid_liveness(reaped_pid()),
crate::watcher::PidLiveness::Dead
);
}
#[test]
fn reconcile_liveness_only_reaps_dead_running_runs() {
use crate::watcher::WorkerLiveness;
let mut queued = ReviewRun::queued("id", "abc123", "commit");
// A queued run is never touched, even if the probe reports dead.
assert_eq!(
queued.reconcile_liveness(|_| WorkerLiveness::Dead),
super::ReconcileOutcome::Unchanged
);
assert_eq!(queued.status, ReviewRunStatus::Queued);
queued.mark_running("reviewing");
// Running + alive stays running.
assert_eq!(
queued.reconcile_liveness(|_| WorkerLiveness::Alive),
super::ReconcileOutcome::Unchanged
);
assert_eq!(queued.status, ReviewRunStatus::Running);
// Running + dead flips to failed with a stale-worker reason.
assert_eq!(
queued.reconcile_liveness(|_| WorkerLiveness::Dead),
super::ReconcileOutcome::Reaped
);
assert_eq!(queued.status, ReviewRunStatus::Failed);
assert!(queued.error.as_deref().unwrap().contains("stale run"));
assert!(queued.worker_pid.is_none());
// A legacy running run with no recorded pid is left alone.
let mut legacy = ReviewRun::queued("id2", "def456", "commit");
legacy.status = ReviewRunStatus::Running;
legacy.worker_pid = None;
assert_eq!(
legacy.reconcile_liveness(|_| WorkerLiveness::Dead),
super::ReconcileOutcome::Unchanged
);
assert_eq!(legacy.status, ReviewRunStatus::Running);
}
#[test]
fn reconcile_liveness_probe_unknown_leaves_row_untouched() {
// 0.9.2 treated ANY probe failure as proof of death and failed live
// runs. An inconclusive probe must leave the row alone.
let mut run = ReviewRun::queued("id", "abc123", "commit");
run.status = ReviewRunStatus::Running;
run.worker_pid = Some(424242);
assert_eq!(
run.reconcile_liveness(|_| crate::watcher::WorkerLiveness::Unknown),
super::ReconcileOutcome::ProbeUnknown
);
assert_eq!(run.status, ReviewRunStatus::Running);
assert!(run.error.is_none());
}
#[cfg(unix)]
#[test]
fn reconcile_liveness_detects_pid_reuse_via_start_token() {
// The recorded worker identity is (pid, start token): a live pid
// under a DIFFERENT token means the worker died and the OS recycled
// its pid — proven dead, reap it. A matching token means genuinely
// alive.
let mut run = ReviewRun::queued("id", "abc123", "commit");
run.status = ReviewRunStatus::Running;
run.worker_pid = Some(std::process::id());
let identity = crate::watcher::ProcessIdentity::current();
run.worker_start_token = Some(identity.start_token.clone());
assert_eq!(
run.reconcile_liveness(crate::watcher::probe_worker_identity),
super::ReconcileOutcome::Unchanged
);
assert_eq!(run.status, ReviewRunStatus::Running);
run.worker_start_token = Some("a different process's start marker".to_owned());
assert_eq!(
run.reconcile_liveness(crate::watcher::probe_worker_identity),
super::ReconcileOutcome::Reaped
);
assert_eq!(run.status, ReviewRunStatus::Failed);
}
#[test]
fn mark_running_records_worker_start_identity() {
let temp = tempfile::tempdir().unwrap();
let store = ReviewRunStore::new(temp.path());
let run = store.create_queued("abc123", "commit").unwrap();
let running = store.mark_running(&run.id, "reviewing").unwrap();
assert_eq!(running.worker_pid, Some(std::process::id()));
#[cfg(unix)]
assert!(
running.worker_start_token.is_some(),
"unix platforms report a start marker via ps"
);
// Terminal transitions clear the identity again.
let completed = store.mark_completed(&run.id, 1).unwrap();
assert_eq!(completed.worker_pid, None);
assert_eq!(completed.worker_start_token, None);
}
#[test]
fn run_store_lock_excludes_concurrent_mutations() {
// The mutation lock is real mutual exclusion: a second acquirer waits
// until the first guard drops.
let temp = tempfile::tempdir().unwrap();
let store = ReviewRunStore::new(temp.path());
let guard = store.lock().unwrap();
let (sender, receiver) = std::sync::mpsc::channel();
let contender = std::thread::spawn({
let store = store.clone();
move || {
let _guard = store.lock().unwrap();
sender.send(()).unwrap();
}
});
assert!(
receiver.recv_timeout(Duration::from_millis(200)).is_err(),
"the contender acquired the lock while it was still held"
);
drop(guard);
receiver
.recv_timeout(Duration::from_secs(5))
.expect("the contender did not acquire the released lock");
contender.join().unwrap();
}
#[test]
fn reconcile_never_clobbers_a_concurrently_completed_run() {
// 0.9.2 race: the reconciler snapshot-reads a Running row, the worker
// completes it, then the reconciler writes its stale Failed snapshot
// over the Completed row. Under the run-store lock the read-probe-
// write is atomic in both orders, so the worker's truth always wins.
for _ in 0..25 {
let temp = tempfile::tempdir().unwrap();
let store = ReviewRunStore::new(temp.path());
let run = write_running_run(&store, Some(reaped_pid()));
let completer = std::thread::spawn({
let store = store.clone();
let run_id = run.id.clone();
move || store.mark_completed(&run_id, 1)
});
let reconciler = std::thread::spawn({
let store = store.clone();
move || store.reconcile_stale_runs()
});
completer.join().unwrap().unwrap();
reconciler.join().unwrap().unwrap();
// If the reconciler reaped first the completion lands on top
// (Failed → Completed); if the completion landed first the
// reconciler skips the non-Running row. Either way: Completed.
let final_run = store.read(&run.id).unwrap();
assert_eq!(
final_run.status,
ReviewRunStatus::Completed,
"a completed run must never be clobbered back to failed"
);
}
}
#[test]
fn review_status_reaps_running_run_with_dead_worker_and_persists() {
let temp = tempfile::tempdir().unwrap();
let store = ReviewRunStore::new(temp.path());
let run = write_running_run(&store, Some(reaped_pid()));
let reconciled = store.read_reconciled(&run.id).unwrap();
assert_eq!(reconciled.status, ReviewRunStatus::Failed);
assert!(reconciled.error.as_deref().unwrap().contains("stale run"));
// The correction is persisted, not just displayed.
assert_eq!(store.read(&run.id).unwrap().status, ReviewRunStatus::Failed);
// list_reconciled agrees.
let listed = store.list_reconciled().unwrap();
assert_eq!(listed.len(), 1);
assert_eq!(listed[0].status, ReviewRunStatus::Failed);
}
#[test]
fn review_status_leaves_running_run_with_live_worker() {
let temp = tempfile::tempdir().unwrap();
let store = ReviewRunStore::new(temp.path());
let run = write_running_run(&store, Some(std::process::id()));
let reconciled = store.read_reconciled(&run.id).unwrap();
assert_eq!(reconciled.status, ReviewRunStatus::Running);
}
#[test]
fn cancel_reaps_running_run_with_dead_worker_without_force() {
let temp = tempfile::tempdir().unwrap();
let store = ReviewRunStore::new(temp.path());
let run = write_running_run(&store, Some(reaped_pid()));
let cancelled = store.cancel(&run.id, false).unwrap();
assert_eq!(cancelled.status, ReviewRunStatus::Failed);
assert!(cancelled.error.as_deref().unwrap().contains("stale run"));
}
#[test]
fn cancel_refuses_live_running_run_without_force() {
let temp = tempfile::tempdir().unwrap();
let store = ReviewRunStore::new(temp.path());
let run = write_running_run(&store, Some(std::process::id()));
let error = store.cancel(&run.id, false).unwrap_err();
assert!(matches!(error, ReviewerError::ReviewRunStillAlive { .. }));
// The run is untouched.
assert_eq!(
store.read(&run.id).unwrap().status,
ReviewRunStatus::Running
);
}
#[test]
fn cancel_force_kills_live_worker_and_cancels() {
let temp = tempfile::tempdir().unwrap();
let store = ReviewRunStore::new(temp.path());
let mut child = Command::new("sleep")
.arg("30")
.spawn()
.expect("spawn sleep");
let pid = child.id();
let run = write_running_run(&store, Some(pid));
let cancelled = store.cancel(&run.id, true).unwrap();
assert_eq!(cancelled.status, ReviewRunStatus::Cancelled);
// Reap the killed child, then confirm it is truly gone.
let _ = child.wait();
assert!(!pid_is_alive(pid));
}
#[test]
fn cancel_legacy_running_run_requires_force_then_reaps() {
let temp = tempfile::tempdir().unwrap();
let store = ReviewRunStore::new(temp.path());
let run = write_running_run(&store, None);
let error = store.cancel(&run.id, false).unwrap_err();
assert!(matches!(
error,
ReviewerError::ReviewRunLivenessUnknown { .. }
));
let reaped = store.cancel(&run.id, true).unwrap();
assert_eq!(reaped.status, ReviewRunStatus::Failed);
}
#[test]
fn cancel_refuses_already_terminal_run() {
let temp = tempfile::tempdir().unwrap();
let store = ReviewRunStore::new(temp.path());
let run = store.create_queued("abc123", "commit").unwrap();
store.mark_completed(&run.id, 0).unwrap();
let error = store.cancel(&run.id, true).unwrap_err();
assert!(matches!(
error,
ReviewerError::CannotCancelReview {
status: ReviewRunStatus::Completed,
..
}
));
}
#[test]
fn execute_review_records_reject_verdict() {
let temp = tempfile::tempdir().unwrap();
let store = LedgerStore::new(temp.path());
let job = review_job(false);
let runner = SequenceRunner::new([reject_json("unsupported")]);
let execution = execute_review_job(job, &runner, &store, None).unwrap();
assert_eq!(execution.entries.len(), 1);
assert_eq!(execution.entries[0].verdict, Verdict::Reject);
assert_eq!(
execution.entries[0].structured_findings[0].title,
"unsupported"
);
assert!(
execution.entries[0]
.raw_reviewer_output
.contains("\"REJECT\"")
);
assert_eq!(store.unresolved_rejections().unwrap().len(), 1);
}
#[test]
fn strict_two_pass_records_both_clean_passes() {
let temp = tempfile::tempdir().unwrap();
let store = LedgerStore::new(temp.path());
let job = review_job(true);
let runner = SequenceRunner::new([pass_json(), pass_json()]);
let execution = execute_review_job(job, &runner, &store, None).unwrap();
assert_eq!(execution.entries.len(), 2);
assert_eq!(store.read_history().unwrap().len(), 2);
assert_eq!(execution.entries[0].reviewer.model, "gpt-5.5");
assert_eq!(execution.entries[1].reviewer.model, "claude-opus-4-8");
}
#[test]
fn strict_arbiter_model_must_be_third_model() {
let temp = tempfile::tempdir().unwrap();
let store = LedgerStore::new(temp.path());
let mut job = review_job(true);
job.strict.as_mut().unwrap().arbiter_model = "gpt-5.5".to_owned();
let runner = SequenceRunner::new([pass_json()]);
let error = execute_review_job(job, &runner, &store, None).unwrap_err();
assert!(matches!(
error,
ReviewerError::StrictArbiterModelNotDistinct
));
}
#[test]
fn strict_goal_policy_stops_at_configured_lie_or_fuckup_count() {
let policy = StrictGoalPolicy {
stop_after_lies: 2,
stop_after_fuckups: 3,
};
assert_eq!(
policy.decide(StrictGoalCounters {
lies_exposed: 1,
fuckups_registered: 2
}),
StrictGoalDecision::Continue
);
assert_eq!(
policy.decide(StrictGoalCounters {
lies_exposed: 2,
fuckups_registered: 0
}),
StrictGoalDecision::Stop {
reason: StrictGoalStopReason::LiesExposed
}
);
assert_eq!(
policy.decide(StrictGoalCounters {
lies_exposed: 0,
fuckups_registered: 3
}),
StrictGoalDecision::Stop {
reason: StrictGoalStopReason::FuckupsRegistered
}
);
}
#[test]
fn drain_once_reviews_each_commit_once_and_clears_queue() {
let temp = tempfile::tempdir().unwrap();
let store = LedgerStore::new(temp.path());
let queue = ReviewQueue::new(temp.path());
queue.enqueue("abc123").unwrap();
queue.enqueue("abc123").unwrap(); // duplicate SHA reviewed only once
queue.enqueue("def456").unwrap();
let loader = StaticLoader::new();
let runner = SequenceRunner::new([reject_json("unsupported"), pass_json()]);
let selection = selection();
let report = drain_once(
&queue,
&loader,
&selection,
"",
&runner,
&store,
&TruthMirrorConfig::default(),
)
.unwrap();
assert_eq!(report.reviewed, ["abc123", "def456"]);
assert_eq!(report.ledger_entries, 2);
assert!(queue.pending().unwrap().is_empty());
assert_eq!(store.read_history().unwrap().len(), 2);
assert_eq!(store.unresolved_rejections().unwrap().len(), 1);
let runs = ReviewRunStore::new(temp.path()).list().unwrap();
assert_eq!(runs.len(), 3);
assert_eq!(
runs.iter()
.filter(|run| run.status == ReviewRunStatus::Completed)
.count(),
2
);
assert_eq!(
runs.iter()
.filter(|run| run.status == ReviewRunStatus::Cancelled)
.count(),
1
);
}
#[test]
fn drain_once_skips_a_poisoned_claim_and_reviews_surrounding_items() {
let temp = tempfile::tempdir().unwrap();
let store = LedgerStore::new(temp.path());
let queue = ReviewQueue::new(temp.path());
queue.enqueue("good-one").unwrap();
queue.enqueue("poisoned").unwrap();
queue.enqueue("good-two").unwrap();
let loader = PoisonedLoader;
let runner = ConstRunner::new(pass_json());
let report = drain_once(
&queue,
&loader,
&selection(),
"",
&runner,
&store,
&TruthMirrorConfig::default(),
)
.unwrap();
assert_eq!(report.reviewed, ["good-one", "good-two"]);
assert_eq!(report.failed, ["poisoned"]);
assert!(queue.pending().unwrap().is_empty());
assert_eq!(store.read_history().unwrap().len(), 2);
let poisoned_run = ReviewRunStore::new(temp.path())
.list()
.unwrap()
.into_iter()
.find(|run| run.commit_sha == "poisoned")
.unwrap();
assert_eq!(poisoned_run.status, ReviewRunStatus::Failed);
assert!(
poisoned_run
.error
.as_deref()
.is_some_and(|error| error.contains("missing evidence pointer"))
);
}
#[test]
fn drain_once_skips_a_verdict_schema_failure_and_reviews_surrounding_items() {
let temp = tempfile::tempdir().unwrap();
let store = LedgerStore::new(temp.path());
let queue = ReviewQueue::new(temp.path());
queue.enqueue("good-one").unwrap();
queue.enqueue("bad-verdict").unwrap();
queue.enqueue("good-two").unwrap();
let invalid_verdict = serde_json::json!({
"verdict": "PASS",
"summary": "",
"findings": [],
"next_steps": [],
"memory_skill": {
"kind": "none",
"learning_source": "",
"reasoning": "No reusable learning was found."
}
})
.to_string();
let runner = SequenceRunner::new([pass_json(), invalid_verdict, pass_json()]);
let report = drain_once(
&queue,
&StaticLoader::new(),
&selection(),
"",
&runner,
&store,
&TruthMirrorConfig::default(),
)
.unwrap();
assert_eq!(report.reviewed, ["good-one", "good-two"]);
assert_eq!(report.failed, ["bad-verdict"]);
assert!(queue.pending().unwrap().is_empty());
assert_eq!(store.read_history().unwrap().len(), 2);
let failed_run = ReviewRunStore::new(temp.path())
.list()
.unwrap()
.into_iter()
.find(|run| run.commit_sha == "bad-verdict")
.unwrap();
assert_eq!(failed_run.status, ReviewRunStatus::Failed);
assert!(
failed_run
.error
.as_deref()
.is_some_and(|error| error.contains("summary must not be empty"))
);
}
#[test]
fn pass_with_disallowed_skill_proposal_is_salvaged_without_a_proposal() {
let output = serde_json::json!({
"verdict": "PASS",
"summary": "The claim is substantiated by the diff and evidence.",
"findings": [],
"next_steps": [],
"memory_skill": {
"kind": "anti_pattern_skill",
"learning_source": "avoid this unrelated pattern",
"reasoning": "The reviewer attached the wrong proposal kind."
}
})
.to_string();
let parsed = ParsedVerdict::parse(&output).unwrap();
assert_eq!(parsed.verdict, Verdict::Pass);
assert_eq!(
parsed.memory_skill_classification.kind,
MemorySkillClassificationKind::None
);
assert!(
parsed
.memory_skill_classification
.learning_source
.is_empty()
);
}
#[test]
fn drain_once_completes_when_memory_skill_extraction_fails() {
let temp = tempfile::tempdir().unwrap();
let store = LedgerStore::new(temp.path());
let queue = ReviewQueue::new(temp.path());
let sha = "a".repeat(40);
queue.enqueue(&sha).unwrap();
let loader = StaticLoader::new();
let runner = ConstRunner::new(pass_json());
let mut config = TruthMirrorConfig::default();
config.skills.enabled = true;
config.memory_skill.enabled = true;
config.memory_skill.signals.min_occurrences = 1;
config.memory_skill.scan.blocked_patterns = vec!["Capture a verified procedure".to_owned()];
let report =
drain_once(&queue, &loader, &selection(), "", &runner, &store, &config).unwrap();
assert_eq!(report.reviewed, [sha]);
assert_eq!(report.ledger_entries, 1);
assert!(queue.pending().unwrap().is_empty());
assert_eq!(store.read_history().unwrap().len(), 1);
let candidate_dir =
crate::memory_skill::MemorySkillStore::new(temp.path(), &config.memory_skill)
.unwrap()
.candidate_dir()
.to_path_buf();
assert!(!candidate_dir.exists());
}
#[test]
fn drain_once_is_a_noop_on_empty_queue() {
let temp = tempfile::tempdir().unwrap();
let store = LedgerStore::new(temp.path());
let queue = ReviewQueue::new(temp.path());
let loader = StaticLoader::new();
let runner = ConstRunner::new(pass_json());
let report = drain_once(
&queue,
&loader,
&selection(),
"",
&runner,
&store,
&TruthMirrorConfig::default(),
)
.unwrap();
assert!(report.reviewed.is_empty());
assert_eq!(report.ledger_entries, 0);
assert_eq!(store.read_history().unwrap().len(), 0);
}
#[test]
fn until_empty_keeps_watching_while_queue_has_items() {
// A non-empty queue is reported as `None` elapsed → keep watching.
assert_eq!(
until_empty_decision(None, 60),
UntilEmptyDecision::KeepWatching
);
}
#[test]
fn until_empty_keeps_watching_inside_grace_window() {
// Empty for less than the grace window → keep watching for late arrivals.
assert_eq!(
until_empty_decision(Some(0), 60),
UntilEmptyDecision::KeepWatching
);
assert_eq!(
until_empty_decision(Some(59), 60),
UntilEmptyDecision::KeepWatching
);
}
#[test]
fn until_empty_exits_once_grace_window_elapses() {
// Empty through a full grace window (>= grace) → exit.
assert_eq!(until_empty_decision(Some(60), 60), UntilEmptyDecision::Exit);
assert_eq!(
until_empty_decision(Some(120), 60),
UntilEmptyDecision::Exit
);
}
#[test]
fn until_empty_with_zero_grace_exits_immediately_when_empty() {
assert_eq!(until_empty_decision(Some(0), 0), UntilEmptyDecision::Exit);
// Still keeps watching while items remain, even with zero grace.
assert_eq!(
until_empty_decision(None, 0),
UntilEmptyDecision::KeepWatching
);
}
#[test]
fn strict_goal_loop_stops_at_configured_lie_count() {
let temp = tempfile::tempdir().unwrap();
let store = LedgerStore::new(temp.path());
let policy = StrictGoalPolicy {
stop_after_lies: 1,
stop_after_fuckups: 0,
};
let runner = SequenceRunner::new([reject_json("lie")]);
let outcome = run_strict_goal_loop(
"abc123",
&claim(),
"diff",
"",
&selection(),
policy,
5,
&runner,
&store,
None,
)
.unwrap();
assert_eq!(outcome.passes, 1);
assert_eq!(outcome.counters.lies_exposed, 1);
assert_eq!(outcome.stop_reason, Some(StrictGoalStopReason::LiesExposed));
assert_eq!(store.read_history().unwrap().len(), 1);
}
#[test]
fn strict_goal_loop_terminates_at_max_passes_for_honest_agent() {
let temp = tempfile::tempdir().unwrap();
let store = LedgerStore::new(temp.path());
let policy = StrictGoalPolicy {
stop_after_lies: 2,
stop_after_fuckups: 5,
};
let runner = ConstRunner::new(pass_json());
let outcome = run_strict_goal_loop(
"abc123",
&claim(),
"diff",
"",
&selection(),
policy,
3,
&runner,
&store,
None,
)
.unwrap();
assert_eq!(outcome.passes, 3);
assert_eq!(outcome.counters.lies_exposed, 0);
assert_eq!(outcome.stop_reason, None);
assert_eq!(store.read_history().unwrap().len(), 3);
}
#[test]
fn strict_goal_loop_stops_when_fuckups_accumulate() {
let temp = tempfile::tempdir().unwrap();
let store = LedgerStore::new(temp.path());
let policy = StrictGoalPolicy {
stop_after_lies: 0,
stop_after_fuckups: 2,
};
// Each structured finding registers one fuckup; two passes hit N=2.
let runner = ConstRunner::new(reject_json("nit"));
let outcome = run_strict_goal_loop(
"abc123",
&claim(),
"diff",
"",
&selection(),
policy,
10,
&runner,
&store,
None,
)
.unwrap();
assert_eq!(outcome.passes, 2);
assert_eq!(outcome.counters.lies_exposed, 2);
assert_eq!(outcome.counters.fuckups_registered, 2);
assert_eq!(
outcome.stop_reason,
Some(StrictGoalStopReason::FuckupsRegistered)
);
}
proptest! {
#[test]
fn strict_goal_loop_never_exceeds_max_passes(max in 1u32..6) {
let temp = tempfile::tempdir().unwrap();
let store = LedgerStore::new(temp.path());
// Both thresholds disabled: only the ceiling can stop the loop.
let policy = StrictGoalPolicy { stop_after_lies: 0, stop_after_fuckups: 0 };
let runner = ConstRunner::new(pass_json());
let outcome = run_strict_goal_loop(
"abc123", &claim(), "diff", "", &selection(), policy, max, &runner, &store, None,
)
.unwrap();
prop_assert!(outcome.passes <= max);
prop_assert_eq!(outcome.passes, max);
prop_assert!(outcome.stop_reason.is_none());
}
}
proptest! {
#[test]
fn model_opposition_is_enforced_for_arbitrary_models(
watched in "[A-Za-z0-9._/-]{1,32}",
reviewer in "[A-Za-z0-9._/-]{1,32}",
) {
let request = ReviewRequest::new(
Agent::Codex,
watched.clone(),
ReviewerHarness::Codex,
reviewer.clone(),
false,
"review this",
);
let result = ReviewPlan::build(request);
// Use the same normalisation that ReviewPlan::build uses so that
// provider-prefixed models (e.g. "openai-codex/gpt-5.5" vs "gpt-5.5")
// are treated as the same model after the B2 fix.
if normalized_model(watched.trim()) == normalized_model(reviewer.trim()) {
let blocked = matches!(result, Err(ReviewerError::SameModelWithoutWaiver { .. }));
prop_assert!(blocked);
} else {
prop_assert!(result.is_ok());
}
}
}
fn claim() -> Claim {
Claim::new(
"add review",
"cargo test",
vec![EvidenceRef::parse("tests:cargo-test").unwrap()],
)
.unwrap()
}
fn selection() -> ReviewSelection {
ReviewSelection {
watched_agent: Agent::Codex,
watched_model: "gpt-5.4".to_owned(),
reviewer_harness: ReviewerHarness::Codex,
reviewer_model: "gpt-5.5".to_owned(),
reviewer_effort: Effort::Xhigh,
allow_same_model: false,
strict: None,
}
}
struct StaticLoader {
claim: Claim,
diff: String,
}
impl StaticLoader {
fn new() -> Self {
Self {
claim: claim(),
diff: "diff --git a/src/lib.rs b/src/lib.rs".to_owned(),
}
}
}
impl MaterialLoader for StaticLoader {
fn load(&self, _sha: &str) -> Result<(Claim, String), ReviewerError> {
Ok((self.claim.clone(), self.diff.clone()))
}
}
struct PoisonedLoader;
impl MaterialLoader for PoisonedLoader {
fn load(&self, sha: &str) -> Result<(Claim, String), ReviewerError> {
if sha == "poisoned" {
return Err(crate::claim::ClaimError::MissingEvidence.into());
}
StaticLoader::new().load(sha)
}
}
struct ConstRunner {
output: String,
}
impl ConstRunner {
fn new(output: impl Into<String>) -> Self {
Self {
output: output.into(),
}
}
}
impl ProcessRunner for ConstRunner {
fn run(
&self,
_invocation: &InvocationPlan,
_prompt: &str,
_timeout: Option<Duration>,
) -> Result<ProcessOutput, ReviewerError> {
Ok(ProcessOutput {
status_code: Some(0),
stdout: self.output.clone(),
stderr: String::new(),
})
}
}
fn review_job(strict: bool) -> ReviewJob {
let claim = claim();
ReviewJob {
commit_sha: "abc123".to_owned(),
diff: "diff --git a/src/lib.rs b/src/lib.rs".to_owned(),
context: String::new(),
request: ReviewRequest::new(
Agent::Codex,
"gpt-5.4",
ReviewerHarness::Codex,
"gpt-5.5",
false,
"review this",
),
claim,
strict: strict.then_some(StrictReviewConfig {
arbiter_harness: ReviewerHarness::Claude,
arbiter_model: "claude-opus-4-8".to_owned(),
arbiter_effort: Effort::Xhigh,
}),
petition: None,
}
}
struct SequenceRunner {
outputs: RefCell<VecDeque<String>>,
}
impl SequenceRunner {
fn new<I, S>(outputs: I) -> Self
where
I: IntoIterator<Item = S>,
S: Into<String>,
{
Self {
outputs: RefCell::new(outputs.into_iter().map(Into::into).collect()),
}
}
}
impl ProcessRunner for SequenceRunner {
fn run(
&self,
_invocation: &InvocationPlan,
_prompt: &str,
_timeout: Option<Duration>,
) -> Result<ProcessOutput, ReviewerError> {
let stdout = self.outputs.borrow_mut().pop_front().unwrap();
Ok(ProcessOutput {
status_code: Some(0),
stdout,
stderr: String::new(),
})
}
}
#[test]
fn extract_verdict_json_handles_raw_fenced_and_prose_wrapped_output() {
// Raw JSON (the documented contract) passes through untouched.
let raw = r#"{"verdict":"PASS"}"#;
assert_eq!(super::extract_verdict_json(raw), raw);
// Prose around a ```json fence: the fenced body is extracted.
let fenced = "Here is my verdict:\n```json\n{\"verdict\":\"PASS\"}\n```\nthanks";
assert_eq!(
super::extract_verdict_json(fenced),
"{\"verdict\":\"PASS\"}"
);
// Bare ``` fence without an info string.
let bare = "verdict below\n```\n{\"a\":1}\n```";
assert_eq!(super::extract_verdict_json(bare), "{\"a\":1}");
// A non-JSON fence first, the JSON fence second: skips to the JSON one.
let two = "```\nnot json\n```\nthen\n```json\n{\"b\":2}\n```";
assert_eq!(super::extract_verdict_json(two), "{\"b\":2}");
// No fence at all: the outermost brace span.
let prose = "The verdict is {\"a\": {\"b\":2}} as required.";
assert_eq!(super::extract_verdict_json(prose), "{\"a\": {\"b\":2}}");
// Nothing JSON-shaped: falls back to the trimmed output so serde's
// error carries the real text.
assert_eq!(super::extract_verdict_json(" nope "), "nope");
}
#[test]
fn parsed_verdict_accepts_markdown_fenced_reviewer_output() {
// The dead-watcher regression: reviewer models wrap the verdict in
// prose + a fenced block; 0.8.0 demanded JSON at byte 0 and the first
// real review errored out, so the queue never drained.
let fenced = format!(
"The fix looks good. Verdict follows.\n\n```json\n{}\n```\n",
pass_json()
);
let parsed = ParsedVerdict::parse(&fenced).unwrap();
assert_eq!(parsed.verdict, Verdict::Pass);
// `raw` preserves the reviewer's original output verbatim.
assert_eq!(parsed.raw, fenced);
}
#[test]
fn queued_review_lines_without_petition_field_still_parse() {
let legacy = r#"{"run_id":"r1","commit_sha":"abc","enqueued_at_unix":1}"#;
let item: super::QueuedReview = serde_json::from_str(legacy).unwrap();
assert_eq!(item.petition_for, None);
}
#[test]
fn enqueue_petition_round_trips_petition_for() {
let temp = tempfile::tempdir().unwrap();
let queue = ReviewQueue::new(temp.path());
queue.enqueue("normal1").unwrap();
queue.enqueue_petition("fix456", "orig123").unwrap();
let pending = queue.pending().unwrap();
assert_eq!(pending.len(), 2);
assert_eq!(pending[0].petition_for, None);
assert_eq!(pending[1].commit_sha, "fix456");
assert_eq!(pending[1].petition_for.as_deref(), Some("orig123"));
}
#[test]
fn drain_once_runs_petition_jobs_and_transitions_the_original_rejection() {
use crate::ledger::{Disposition, LedgerEntry, ResolutionKind, ReviewerConfig};
let temp = tempfile::tempdir().unwrap();
let store = LedgerStore::new(temp.path());
let queue = ReviewQueue::new(temp.path());
// The original rejection, plus the CLI's attempt bookkeeping
// (`resolve --fixed-by` counts the attempt when it enqueues).
let rejected = LedgerEntry::new(
"orig123",
Verdict::Reject,
"CLAIM: original | verified: cargo test | evidence: tests:cargo-test",
vec!["tests:cargo-test".to_owned()],
ReviewerConfig::new("claude", "claude-opus-4-1", false),
vec!["evidence too thin".to_owned()],
);
store.append_entry(&rejected).unwrap();
store
.append_petition_transition(
"orig123",
Disposition::Open,
ResolutionKind::Resolved,
"petition review enqueued: fix=fix456, attempts=1/2",
1,
)
.unwrap();
queue.enqueue_petition("fix456", "orig123").unwrap();
let loader = StaticLoader::new();
let runner = ConstRunner::new(pass_json());
let report = drain_once(
&queue,
&loader,
&selection(),
"",
&runner,
&store,
&TruthMirrorConfig::default(),
)
.unwrap();
assert_eq!(report.reviewed, ["fix456"]);
// The petition review entry lands under the FIX sha, tagged with the
// original, carrying the CLI's counter unchanged (no double increment).
let review = store.show("fix456").unwrap();
assert_eq!(review.petition_for.as_deref(), Some("orig123"));
assert_eq!(review.petition_attempts, 1);
// apply_petition_transition resolved the original rejection.
let original = store.show("orig123").unwrap();
assert_eq!(original.disposition, Disposition::Resolved);
assert!(queue.pending().unwrap().is_empty());
assert!(store.blocking_rejections().unwrap().is_empty());
}
#[test]
fn extract_verdict_json_takes_leading_json_before_trailing_prose() {
// Round-2 review finding: leading JSON + trailing prose used to be
// returned whole (starts_with '{'), and serde died on the trailer.
let leading = "{\"verdict\":\"PASS\"} \n\nHope this helps!";
assert_eq!(
super::extract_verdict_json(leading),
"{\"verdict\":\"PASS\"}"
);
// Prose on both sides without fences: first complete JSON value wins
// even when the trailing prose contains a stray closing brace.
let both = "verdict: {\"a\":1} trailing prose with a stray } brace";
assert_eq!(super::extract_verdict_json(both), "{\"a\":1}");
}
#[test]
fn drain_once_reviews_normal_and_petition_items_for_the_same_fix_sha() {
use crate::ledger::{Disposition, LedgerEntry, ResolutionKind, ReviewerConfig};
let temp = tempfile::tempdir().unwrap();
let store = LedgerStore::new(temp.path());
let queue = ReviewQueue::new(temp.path());
let rejected = LedgerEntry::new(
"orig123",
Verdict::Reject,
"CLAIM: original | verified: cargo test | evidence: tests:cargo-test",
vec!["tests:cargo-test".to_owned()],
ReviewerConfig::new("claude", "claude-opus-4-1", false),
vec!["evidence too thin".to_owned()],
);
store.append_entry(&rejected).unwrap();
store
.append_petition_transition(
"orig123",
Disposition::Open,
ResolutionKind::Resolved,
"petition review enqueued: fix=fix456, attempts=1/2",
1,
)
.unwrap();
// The SAME fix commit is queued twice with different identities: its
// own post-commit review AND a petition review. Dedup by bare SHA
// used to cancel one of them silently.
queue.enqueue("fix456").unwrap();
queue.enqueue_petition("fix456", "orig123").unwrap();
let loader = StaticLoader::new();
let runner = ConstRunner::new(pass_json());
let report = drain_once(
&queue,
&loader,
&selection(),
"",
&runner,
&store,
&TruthMirrorConfig::default(),
)
.unwrap();
assert_eq!(report.reviewed, ["fix456", "fix456"]);
assert!(queue.pending().unwrap().is_empty());
// Both entries exist in history: the fix commit's own PASS and the
// petition review tagged with the original rejection.
let history = store.read_history().unwrap();
let fix_entries: Vec<_> = history
.iter()
.filter(|entry| entry.commit_sha == "fix456")
.collect();
assert_eq!(fix_entries.len(), 2);
assert!(fix_entries.iter().any(|entry| entry.petition_for.is_none()));
assert!(
fix_entries
.iter()
.any(|entry| entry.petition_for.as_deref() == Some("orig123"))
);
// And the original rejection resolved via the petition leg.
assert_eq!(
store.show("orig123").unwrap().disposition,
crate::ledger::Disposition::Resolved
);
}
#[test]
fn flag_petition_verdict_resolves_the_original_rejection() {
use crate::ledger::{Disposition, LedgerEntry, ResolutionKind, ReviewerConfig};
let temp = tempfile::tempdir().unwrap();
let store = LedgerStore::new(temp.path());
let queue = ReviewQueue::new(temp.path());
let rejected = LedgerEntry::new(
"orig123",
Verdict::Reject,
"CLAIM: original | verified: cargo test | evidence: tests:cargo-test",
vec!["tests:cargo-test".to_owned()],
ReviewerConfig::new("claude", "claude-opus-4-1", false),
vec!["evidence too thin".to_owned()],
);
store.append_entry(&rejected).unwrap();
store
.append_petition_transition(
"orig123",
Disposition::Open,
ResolutionKind::Resolved,
"petition review enqueued: fix=fix456, attempts=1/2",
1,
)
.unwrap();
queue.enqueue_petition("fix456", "orig123").unwrap();
// FLAG = the fix materially addresses the findings + non-blocking
// debt (per the petition prompt) — it must ACCEPT, with the debt
// recorded on the petition review entry itself.
let loader = StaticLoader::new();
let runner = ConstRunner::new(flag_json("evidence could be tighter"));
drain_once(
&queue,
&loader,
&selection(),
"",
&runner,
&store,
&TruthMirrorConfig::default(),
)
.unwrap();
assert_eq!(
store.show("orig123").unwrap().disposition,
crate::ledger::Disposition::Resolved
);
let history = store.read_history().unwrap();
let petition_entry = history
.iter()
.find(|entry| {
entry.petition_for.as_deref() == Some("orig123") && entry.commit_sha == "fix456"
})
.unwrap();
assert_eq!(petition_entry.verdict, Verdict::Flag);
}
#[test]
fn reconcile_stale_runs_reaps_dead_worker_rows_and_counts_live_view() {
// 0.9.2 field failure: harness-killed watchers left orphaned
// `running` rows behind and nothing ever reconciled them.
let temp = tempfile::tempdir().unwrap();
let store = ReviewRunStore::new(temp.path());
// An orphaned running row whose recorded worker pid is dead (a real
// reaped child: `kill -0` provably reports ESRCH for it).
let mut orphan = ReviewRun::queued("run-orphan", "abc123", "commit");
orphan.status = ReviewRunStatus::Running;
orphan.worker_pid = Some(reaped_pid());
store.write(&orphan).unwrap();
// A genuinely live running row (this test process).
let mut live = ReviewRun::queued("run-live", "def456", "commit");
live.mark_running("reviewing");
store.write(&live).unwrap();
// The read-only view counts the stale row without persisting.
assert_eq!(store.stale_running_count().unwrap(), 1);
assert_eq!(
store.read("run-orphan").unwrap().status,
ReviewRunStatus::Running
);
// Reconciliation reaps exactly the orphan, persisted.
assert_eq!(store.reconcile_stale_runs().unwrap(), 1);
let orphan = store.read("run-orphan").unwrap();
assert_eq!(orphan.status, ReviewRunStatus::Failed);
assert!(
orphan
.error
.as_deref()
.is_some_and(|error| error.contains("stale run")),
"stale reason should be recorded, got: {:?}",
orphan.error
);
assert_eq!(
store.read("run-live").unwrap().status,
ReviewRunStatus::Running
);
assert_eq!(store.stale_running_count().unwrap(), 0);
}
#[test]
fn drain_once_skips_petition_whose_target_is_already_resolved() {
// 0.9.2 field crash: a petition enqueued twice (short-sha + full-sha
// rows for one rejection) let the second copy reach the drain after
// the first had resolved it — NoOpenRejection killed the watcher.
// The dead petition must be consumed with a logged skip while the
// rest of the queue keeps draining.
use crate::ledger::{Disposition, LedgerEntry, ResolutionKind, ReviewerConfig};
let temp = tempfile::tempdir().unwrap();
let store = LedgerStore::new(temp.path());
let queue = ReviewQueue::new(temp.path());
let rejected = LedgerEntry::new(
"orig123",
Verdict::Reject,
"CLAIM: original | verified: cargo test | evidence: tests:cargo-test",
vec!["tests:cargo-test".to_owned()],
ReviewerConfig::new("claude", "claude-opus-4-1", false),
vec!["evidence too thin".to_owned()],
);
store.append_entry(&rejected).unwrap();
// Resolved up front: the queued petition is now a dead letter.
store
.append_petition_transition(
"orig123",
Disposition::Resolved,
ResolutionKind::Resolved,
"already fixed elsewhere",
1,
)
.unwrap();
queue.enqueue_petition("fix456", "orig123").unwrap();
queue.enqueue("later999").unwrap();
let loader = StaticLoader::new();
let runner = ConstRunner::new(pass_json());
let report = drain_once(
&queue,
&loader,
&selection(),
"",
&runner,
&store,
&TruthMirrorConfig::default(),
)
.unwrap();
// The dead petition was consumed WITHOUT invoking the reviewer; the
// next queued commit still drained.
assert_eq!(report.failed, ["fix456"]);
assert_eq!(report.reviewed, ["later999"]);
assert!(queue.pending().unwrap().is_empty());
// The skipped run records the reason.
let run_store = ReviewRunStore::new(temp.path());
let runs = run_store.list().unwrap();
let skipped = runs.iter().find(|run| run.commit_sha == "fix456").unwrap();
assert_eq!(skipped.status, ReviewRunStatus::Failed);
assert!(
skipped
.error
.as_deref()
.is_some_and(|error| error.contains("no open rejection")),
"skip reason should be recorded, got: {:?}",
skipped.error
);
// No ledger entry exists for the dead petition (reviewer never ran).
assert!(
store
.read_history()
.unwrap()
.iter()
.all(|entry| entry.commit_sha != "fix456")
);
}
#[test]
fn drain_once_survives_petition_target_resolved_mid_review() {
// The race window between the drain's actionability check and the
// transition: the rejection is open when the review starts but gets
// resolved while the reviewer runs (a human waive, or a duplicate
// petition landing first). 0.9.2 propagated NoOpenRejection out of
// the drain and killed the watcher; now it is skip-and-continue.
use crate::ledger::{Disposition, LedgerEntry, ResolutionKind, ReviewerConfig};
struct ResolvingRunner {
state_dir: PathBuf,
}
impl ProcessRunner for ResolvingRunner {
fn run(
&self,
_invocation: &InvocationPlan,
_prompt: &str,
_timeout: Option<Duration>,
) -> Result<ProcessOutput, ReviewerError> {
// Mid-review resolution: the verdict will be recorded, but
// the transition target is gone by the time it lands.
LedgerStore::new(&self.state_dir)
.append_petition_transition(
"orig123",
Disposition::Resolved,
ResolutionKind::Resolved,
"raced resolve while the petition review ran",
1,
)
.unwrap();
Ok(ProcessOutput {
status_code: Some(0),
stdout: pass_json(),
stderr: String::new(),
})
}
}
let temp = tempfile::tempdir().unwrap();
let store = LedgerStore::new(temp.path());
let queue = ReviewQueue::new(temp.path());
let rejected = LedgerEntry::new(
"orig123",
Verdict::Reject,
"CLAIM: original | verified: cargo test | evidence: tests:cargo-test",
vec!["tests:cargo-test".to_owned()],
ReviewerConfig::new("claude", "claude-opus-4-1", false),
vec!["evidence too thin".to_owned()],
);
store.append_entry(&rejected).unwrap();
store
.append_petition_transition(
"orig123",
Disposition::Open,
ResolutionKind::Resolved,
"petition review enqueued: fix=fix456, attempts=1/2",
1,
)
.unwrap();
queue.enqueue_petition("fix456", "orig123").unwrap();
let runner = ResolvingRunner {
state_dir: temp.path().to_path_buf(),
};
let report = drain_once(
&queue,
&StaticLoader::new(),
&selection(),
"",
&runner,
&store,
&TruthMirrorConfig::default(),
)
.unwrap();
// The watcher lived: the item was consumed as a logged skip.
assert_eq!(report.failed, ["fix456"]);
assert!(queue.pending().unwrap().is_empty());
assert_eq!(
store.show("orig123").unwrap().disposition,
Disposition::Resolved
);
}
}