From e7e4d8d7f6396c9e679d40149f8d3f4aec99839e Mon Sep 17 00:00:00 2001 From: Decebal Dobrica <352761+decebal@users.noreply.github.com> Date: Tue, 10 Nov 2026 15:19:56 +0000 Subject: [PATCH] fix(workflow): give stranded runs an exit A run left pending a commit or verification refused handoff, plan, build and resume with "reconcile before execution", and recover refused with "requires its destination-specific resume command". None of them named the command that clears the operation, so a run whose commit hook rejected the message looked permanently stuck, though checkpoint and verify both accept it. Every refusal on a pending operation now names its clearing command. A run nobody will ship also had no terminal state other than shipped, so its Chronis task stayed claimed. allstack abandon --epoch N checkpoints the active worktree, records a terminal abandoned event, releases the task when this run holds the claim under any epoch, and removes the run's worktrees under the same unexpected-edit check as ship. It is refused once publication starts. Repeating it retries the release and cleanup. --- docs/reference/cli.md | 6 +++- src/application/mod.rs | 27 +++++++++++++++- src/application/ports.rs | 1 + src/domain.rs | 30 ++++++++++++++++-- src/infrastructure/chronis.rs | 24 +++++++++++++++ src/infrastructure/runtime.rs | 3 ++ src/infrastructure/store.rs | 2 +- src/presentation.rs | 14 ++++++--- tests/application.rs | 58 +++++++++++++++++++++++++++++++++++ tests/domain.rs | 31 +++++++++++++++++++ 10 files changed, 186 insertions(+), 10 deletions(-) diff --git a/docs/reference/cli.md b/docs/reference/cli.md index b13ecc9..b1fbbf2 100644 --- a/docs/reference/cli.md +++ b/docs/reference/cli.md @@ -119,9 +119,13 @@ Inspect `status` and `history` before retrying a failed command. A pending opera | Host planning or building | Inspect the active worktree and private process evidence, then `recover`; recovery checkpoints current edits without replaying the uncertain host call | | Handoff | `recover` completes the recorded transfer and reads back Chronis ownership | | Plan recorded with `memory_pending` | `recover` retries Prime indexing without invoking the planning host again | -| Commit or verification | Repeat `verify` with the current epoch and authorized commit message; it reconciles the candidate and reruns final gates | +| Commit or verification | Repeat `verify` with the current epoch and authorized commit message; it reconciles the candidate and reruns final gates. Or `checkpoint` to keep the workspace edits and clear the commit | | Publication or later task/memory update | Repeat `ship`; destination state determines remaining work | +A refusal caused by a pending operation names the command that clears it. + +`abandon --epoch ` ends a run that will not ship, with any operation except publication pending. It checkpoints the active worktree so its edits stay in the artifact store, records the terminal `abandoned` event, releases the Chronis task when this run holds the claim, and removes the run's worktrees under the same unexpected-edit check as `ship`. Commits already made in the original checkout stay. Once `ship` has started, `abandon` is refused; finish with `ship`. Repeating `abandon` with the same epoch retries the release and cleanup. + After recovery of an uncertain planning call, planning remains necessary unless an accepted plan was recorded. Inspect preserved edits before deliberately invoking `plan` again. A command that failed after recording its result may already have advanced state; always read the destination state first. No retained checkpoint can prove that later abandoned edits were durable. Missing or corrupt objects block recovery. Chronis claimant mismatch blocks managed execution. Prime recall supplies context only; source-linked nodes cannot authorize a transition or certify a successful check. Unavailable Prime storage degrades recall while retaining authoritative run access; pending memory indexing remains explicit until reconciled. diff --git a/src/application/mod.rs b/src/application/mod.rs index ae12593..3a2be1f 100644 --- a/src/application/mod.rs +++ b/src/application/mod.rs @@ -189,7 +189,10 @@ pub async fn recover(p: &impl Platform, id: &str, epoch: u64) -> Result { return Ok(run); } Some(other) => { - anyhow::bail!("operation {other} requires its destination-specific resume command") + anyhow::bail!( + "recover cannot resolve operation {other}; {}", + crate::domain::clearing_command(other) + ) } } run.pending = None; @@ -197,6 +200,28 @@ pub async fn recover(p: &impl Platform, id: &str, epoch: u64) -> Result { Ok(run) } +/// Ends a run that will not ship. Edits are checkpointed before cleanup so they stay +/// recoverable from the artifact store; commits already in the checkout are left alone. +pub async fn abandon(p: &impl Platform, id: &str, epoch: u64) -> Result { + let mut run = p.load(id).await?; + ensure!(run.epoch == epoch, "stale host epoch"); + if run.stage != Stage::Abandoned { + run.check_epoch(epoch).map_err(anyhow::Error::msg)?; + ensure!( + run.stage != Stage::Shipping, + "publication has started; resume it with ship" + ); + run.checkpoint = p.checkpoint(&run)?; + run.stage = Stage::Abandoned; + run.pending = None; + run.transfer = None; + p.save(&mut run, "abandoned").await?; + } + p.release_task(&run)?; + p.cleanup(&run).await?; + Ok(run) +} + pub async fn verify(p: &impl Platform, id: &str, epoch: u64, message: &str) -> Result { let mut run = p.load(id).await?; run.check_epoch(epoch).map_err(anyhow::Error::msg)?; diff --git a/src/application/ports.rs b/src/application/ports.rs index a40ba6f..593f675 100644 --- a/src/application/ports.rs +++ b/src/application/ports.rs @@ -25,6 +25,7 @@ pub trait Platform { fn claim(&self, run: &Run, previous: Option<&Run>) -> Result<()>; fn claim_readback(&self, run: &Run) -> Result<()>; fn complete_task(&self, run: &Run) -> Result<()>; + fn release_task(&self, run: &Run) -> Result<()>; fn recall(&self, query: &str) -> Value; async fn remember(&self, run: &Run, phase: &str, source: &str, summary: &str) -> Result<()>; fn invoke(&self, run: &Run, prompt: &str, planning: bool) -> Result; diff --git a/src/domain.rs b/src/domain.rs index 4f9420f..58ee355 100644 --- a/src/domain.rs +++ b/src/domain.rs @@ -44,6 +44,24 @@ pub enum Stage { Verifying, Shipping, Shipped, + Abandoned, +} + +impl Stage { + pub fn terminal(self) -> bool { + matches!(self, Self::Shipped | Self::Abandoned) + } +} + +/// The command that clears a pending operation, for refusals that block on it. +pub fn clearing_command(operation: &str) -> &'static str { + match operation { + "commit" | "verification" => { + "run checkpoint to keep the workspace edits, verify to retry the commit, or abandon to end the run" + } + "publication" => "run ship to resume publication", + _ => "run recover, or abandon to end the run", + } } #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] @@ -196,6 +214,8 @@ impl Run { )) } else if self.stage == Stage::Shipped { Err("run already shipped".into()) + } else if self.stage == Stage::Abandoned { + Err("run abandoned".into()) } else { Ok(()) } @@ -204,7 +224,8 @@ impl Run { pub fn require_resolved(&self) -> Result<(), String> { match &self.pending { Some(operation) => Err(format!( - "unresolved operation {operation}; reconcile before execution" + "unresolved operation {operation}; {}", + clearing_command(operation) )), None => Ok(()), } @@ -238,6 +259,7 @@ pub fn validate_transition(previous: Option<&Run>, event: &Event) -> Result<(), "task_completed", "memory_indexed", "shipped", + "abandoned", ] .contains(&event.kind.as_str()) { @@ -276,6 +298,10 @@ pub fn validate_transition(previous: Option<&Run>, event: &Event) -> Result<(), | (Stage::Verifying, Stage::Building) | (Stage::Verifying, Stage::Shipping) | (Stage::Shipping, Stage::Shipped) + | ( + Stage::Planning | Stage::Building | Stage::Verifying, + Stage::Abandoned + ) ); if !same && !allowed { return Err("invalid stage transition".into()); @@ -294,7 +320,7 @@ pub fn validate_transition(previous: Option<&Run>, event: &Event) -> Result<(), { return Err("ownership changed without a bound handoff".into()); } - if old.stage == Stage::Shipped { + if old.stage.terminal() { return Err("terminal run cannot change".into()); } if matches!(event.run.stage, Stage::Shipping | Stage::Shipped) diff --git a/src/infrastructure/chronis.rs b/src/infrastructure/chronis.rs index 5f27146..2e07a25 100644 --- a/src/infrastructure/chronis.rs +++ b/src/infrastructure/chronis.rs @@ -206,6 +206,30 @@ impl Tasks<'_> { ); Ok(()) } + /// Releases only a claim this run holds, under any of its epochs; another claimant keeps theirs. + pub fn release(&self, run: &Run) -> Result<()> { + let Some(id) = run.task.as_deref() else { + return Ok(()); + }; + let actual = self.show(id)?; + let owner = actual + .lines() + .find_map(|s| s.strip_prefix("claimed_by:")) + .unwrap_or(""); + if owner.starts_with(&format!("allstack:{}:", run.id)) + && actual.lines().any(|s| s == "status:in-progress") + { + self.call( + &["release", id, "--reason", "Allstack run abandoned"], + Some(owner), + )?; + ensure!( + !self.show(id)?.lines().any(|s| s == "status:in-progress"), + "task release unconfirmed" + ); + } + Ok(()) + } pub fn previous_actor(run: &Run) -> String { Self::actor(run) } diff --git a/src/infrastructure/runtime.rs b/src/infrastructure/runtime.rs index a581102..b24cb5d 100644 --- a/src/infrastructure/runtime.rs +++ b/src/infrastructure/runtime.rs @@ -215,6 +215,9 @@ impl Platform for Runtime { fn complete_task(&self, run: &Run) -> Result<()> { self.tasks().complete(run) } + fn release_task(&self, run: &Run) -> Result<()> { + self.tasks().release(run) + } fn recall(&self, query: &str) -> Value { self.memory .as_ref() diff --git a/src/infrastructure/store.rs b/src/infrastructure/store.rs index e322209..a6f5c0e 100644 --- a/src/infrastructure/store.rs +++ b/src/infrastructure/store.rs @@ -111,7 +111,7 @@ impl Store { let mut active = Vec::new(); for id in ids { let run = self.load(&id).await?; - if run.stage != crate::domain::Stage::Shipped { + if !run.stage.terminal() { active.push(run); } } diff --git a/src/presentation.rs b/src/presentation.rs index bd0a46b..3a30517 100644 --- a/src/presentation.rs +++ b/src/presentation.rs @@ -6,7 +6,7 @@ use crate::{ domain::Host, infrastructure::{bootstrap, catalog, git, process, runtime::Runtime, skills}, }; -use anyhow::{Context, Result, ensure}; +use anyhow::{Context, Result}; use clap::{Parser, Subcommand}; use serde_json::{Value, json}; use std::path::PathBuf; @@ -120,6 +120,12 @@ pub enum Action { #[arg(long)] epoch: u64, }, + /// End a run that will not ship: checkpoint its edits, release its task, remove its worktrees. + Abandon { + run: String, + #[arg(long)] + epoch: u64, + }, Resume { run: String, #[arg(long)] @@ -318,6 +324,7 @@ async fn dispatch(p: &Runtime, action: Action) -> Result { Action::Handoff { run, epoch, to } => application::handoff(p, &run, epoch, to).await?, Action::Checkpoint { run, epoch } => application::checkpoint(p, &run, epoch).await?, Action::Recover { run, epoch } => application::recover(p, &run, epoch).await?, + Action::Abandon { run, epoch } => application::abandon(p, &run, epoch).await?, Action::Verify { run, epoch, @@ -341,10 +348,7 @@ async fn dispatch(p: &Runtime, action: Action) -> Result { Action::Resume { run, epoch } => { let run = p.load(&run).await?; run.check_epoch(epoch).map_err(anyhow::Error::msg)?; - ensure!( - run.pending.is_none(), - "run has unresolved operation; inspect and recover before executing" - ); + run.require_resolved().map_err(anyhow::Error::msg)?; p.claim_readback(&run)?; let request = String::from_utf8(p.object(&run.request)?)?; let plan = run diff --git a/tests/application.rs b/tests/application.rs index 6a608fb..9f122d0 100644 --- a/tests/application.rs +++ b/tests/application.rs @@ -100,6 +100,10 @@ impl Platform for Fake { ); Ok(()) } + fn release_task(&self, _: &Run) -> Result<()> { + self.calls.borrow_mut().push("release".into()); + Ok(()) + } fn recall(&self, _: &str) -> Value { json!({"authority":"context-only"}) } @@ -386,6 +390,60 @@ async fn failed_commit_can_accept_explicit_checkpoint_repair() { assert!(p.state.borrow().verification.is_some()); } +#[tokio::test] +async fn stranded_commit_names_its_exits_and_can_be_abandoned() { + let p = Fake::new(); + let run = build_flow(&p).await; + p.state.borrow_mut().pending = Some("commit".into()); + for refusal in [ + application::handoff(&p, &run.id, 3, Host::ClaudeCode) + .await + .unwrap_err(), + application::recover(&p, &run.id, 3).await.unwrap_err(), + ] { + let text = refusal.to_string(); + assert!( + ["checkpoint", "verify", "abandon"] + .iter() + .all(|c| text.contains(c)), + "{text}" + ); + } + assert!(application::abandon(&p, &run.id, 2).await.is_err()); + let abandoned = application::abandon(&p, &run.id, 3).await.unwrap(); + assert_eq!(abandoned.stage, allstack::domain::Stage::Abandoned); + assert!(abandoned.pending.is_none()); + assert!(application::build(&p, &run.id, 3).await.is_err()); + application::abandon(&p, &run.id, 3).await.unwrap(); + let calls = p.calls.borrow(); + let count = |kind: &str| calls.iter().filter(|c| *c == kind).count(); + assert_eq!(count("abandoned"), 1); + assert_eq!(count("release"), 2); + assert_eq!(count("cleanup"), 2); +} + +#[tokio::test] +async fn abandon_is_refused_once_publication_starts() { + let p = Fake::new(); + let run = build_flow(&p).await; + application::verify(&p, &run.id, 3, "feat(example): accepted change") + .await + .unwrap(); + p.fail_completion.set(true); + assert!( + application::ship(&p, &run.id, 3, "feat(example): accepted change") + .await + .is_err() + ); + let refusal = application::abandon(&p, &run.id, 3).await.unwrap_err(); + assert!(refusal.to_string().contains("ship"), "{refusal}"); + application::ship(&p, &run.id, 3, "feat(example): accepted change") + .await + .unwrap(); + assert!(application::abandon(&p, &run.id, 3).await.is_err()); + assert!(!p.calls.borrow().iter().any(|c| c == "release")); +} + #[tokio::test] async fn changed_task_claim_blocks_first_publication() { let p = Fake::new(); diff --git a/tests/domain.rs b/tests/domain.rs index 54ceb06..5ff0688 100644 --- a/tests/domain.rs +++ b/tests/domain.rs @@ -69,6 +69,37 @@ fn cannot_skip_verification_or_invent_delivery() { assert!(validate_transition(Some(&old), &event).is_err()); } +#[test] +fn only_an_unpublished_run_can_be_abandoned_and_abandonment_is_terminal() { + let event = |old: &allstack::domain::Run, stage: Stage, kind: &str| { + let mut run = old.clone(); + run.revision += 1; + run.stage = stage; + Event { + schema: 1, + operation: kind.into(), + previous: "hash".into(), + kind: kind.into(), + run, + } + }; + let mut old = common::run(); + old.revision = 1; + for stage in [Stage::Planning, Stage::Building, Stage::Verifying] { + old.stage = stage; + validate_transition(Some(&old), &event(&old, Stage::Abandoned, "abandoned")).unwrap(); + } + old.stage = Stage::Shipping; + assert!(validate_transition(Some(&old), &event(&old, Stage::Abandoned, "abandoned")).is_err()); + old.stage = Stage::Abandoned; + assert!(validate_transition(Some(&old), &event(&old, Stage::Building, "built")).is_err()); + assert!( + old.check_epoch(old.epoch) + .unwrap_err() + .contains("abandoned") + ); +} + #[test] fn policy_requires_execution_evidence_and_bounded_commands() { let mut p: Policy = toml::from_str(