Harden instruction lease flow and add trace replay protocol tests

AGENTS.md

@@ -19,3 +19,9 @@ Use Rally when protocol adherence matters more than free-form collaboration.
 - Keep workflow-specific instruction/policy text in workflow crates.
 - Keep wrapper templates workflow-agnostic.
 - If you add/rename a run command, update wrapper-facing docs (`README.md`, `docs/skill-install-run.md`) in the same change.
+
+## Protocol Testing
+
+- Keep protocol tests small and fixture-driven in `tests/trace_replay.rs`.
+- When changing `next`/lease/state semantics, add or update replay fixtures for happy path plus common misuse cases (missing/stale/wrong instruction IDs).
+- Run `cargo test -q --test trace_replay` after protocol-level changes.

crates/rally-workflow-consult/src/lib.rs

@@ -322,14 +322,19 @@ pub fn status(state: &SessionState) -> Result<Vec<String>> {
     ];
 
     if workflow_state.require_read_ack {
-        lines.push(format!("Read acknowledgments complete: {read_complete}/{total}"));
+        lines.push(format!(
+            "Read acknowledgments complete: {read_complete}/{total}"
+        ));
     } else {
         lines.push("Read acknowledgments complete: disabled".to_string());
     }
 
     let pending_ready = pending_ready_agents(&workflow_state);
     if !pending_ready.is_empty() {
-        lines.push(format!("Pending ready agents: {}", pending_ready.join(", ")));
+        lines.push(format!(
+            "Pending ready agents: {}",
+            pending_ready.join(", ")
+        ));
     }
 
     if workflow_state.require_read_ack {
@@ -481,7 +486,12 @@ fn unread_peers_for_agent(workflow_state: &ConsultWorkflowState, agent: &str) ->
         .collect()
 }
 
-fn post_action(state: &mut SessionState, session_dir: &Path, agent: &str, args: &Value) -> Result<String> {
+fn post_action(
+    state: &mut SessionState,
+    session_dir: &Path,
+    agent: &str,
+    args: &Value,
+) -> Result<String> {
     if state.phase != SessionPhase::Implement {
         bail!(
             "action 'post' is only valid during phase Implement (current: {:?})",
@@ -532,7 +542,10 @@ fn ready_action(state: &mut SessionState, agent: &str) -> Result<String> {
         .get_mut(agent)
         .expect("agent should exist after membership check");
     if !agent_state.posted {
-        bail!("action 'ready' requires a prior 'post' action for '{}'", agent);
+        bail!(
+            "action 'ready' requires a prior 'post' action for '{}'",
+            agent
+        );
     }
     if agent_state.ready {
         bail!("action 'ready' already completed for '{}'", agent);
@@ -568,7 +581,10 @@ fn read_action(state: &mut SessionState, agent: &str, args: &Value) -> Result<St
     let mut workflow_state = read_workflow_state(state)?;
     ensure_agent_membership(&mut workflow_state, agent);
     if !workflow_state.agents.contains_key(&peer) {
-        bail!("action 'read' peer '{}' is not a registered consult agent", peer);
+        bail!(
+            "action 'read' peer '{}' is not a registered consult agent",
+            peer
+        );
     }
 
     let all_ready = workflow_state.agents.values().all(|a| a.ready);
@@ -587,14 +603,21 @@ fn read_action(state: &mut SessionState, agent: &str, args: &Value) -> Result<St
         bail!("action 'read' requires '{}' to post first", agent);
     }
     if agent_state.read_peers.get(&peer).copied().unwrap_or(false) {
-        bail!("action 'read' already recorded peer '{}' for '{}'", peer, agent);
+        bail!(
+            "action 'read' already recorded peer '{}' for '{}'",
+            peer,
+            agent
+        );
     }
 
     agent_state.read_peers.insert(peer.clone(), true);
     let unread = unread_peers_for_agent(&workflow_state, agent);
     write_workflow_state(state, &workflow_state)?;
     if unread.is_empty() {
-        Ok(format!("recorded read for '{}'; all peer reads are complete", peer))
+        Ok(format!(
+            "recorded read for '{}'; all peer reads are complete",
+            peer
+        ))
     } else {
         Ok(format!(
             "recorded read for '{}'; remaining unread peers: {}",
@@ -604,7 +627,11 @@ fn read_action(state: &mut SessionState, agent: &str, args: &Value) -> Result<St
     }
 }
 
-fn normalize_and_validate_post_path(session_dir: &Path, agent: &str, raw_path: &str) -> Result<String> {
+fn normalize_and_validate_post_path(
+    session_dir: &Path,
+    agent: &str,
+    raw_path: &str,
+) -> Result<String> {
     let path = Path::new(raw_path);
     if path.as_os_str().is_empty() {
         bail!("action 'post' requires a non-empty path");
@@ -1012,10 +1039,7 @@ mod tests {
             &json!({ "path": "findings/bob/finding.md" }),
         )
         .expect_err("post should reject cross-agent paths");
-        assert!(
-            err.to_string()
-                .contains("must be inside 'findings/alice'")
-        );
+        assert!(err.to_string().contains("must be inside 'findings/alice'"));
 
         fs::remove_dir_all(&base)?;
         Ok(())

crates/rally-workflow-plan/src/lib.rs

@@ -2087,9 +2087,10 @@ mod tests {
         let err = policy
             .allow_agree(&state, 1, "agent-a")
             .expect_err("three-agent sessions should still require two cross-agent challenges");
-        assert!(err
-            .to_string()
-            .contains("at least two cross-agent challenges"));
+        assert!(
+            err.to_string()
+                .contains("at least two cross-agent challenges")
+        );
 
         let mut workflow_state = read_workflow_state(&state)?;
         let issue = workflow_state

justfile

@@ -6,6 +6,7 @@ nix-check:
 
 pre-merge: nix-check
     cargo test --workspace --all-targets
+    cargo test --test event_audit_checker
     @echo "pre-merge complete"
 
 # Verify /rally wrapper works across all locally-installed coding agents

src/cli.rs

@@ -157,6 +157,11 @@ pub struct NextArgs {
     pub agent: String,
     #[arg(long, help = "Seconds to wait before returning exit code 1")]
     pub timeout: Option<u64>,
+    #[arg(
+        long,
+        help = "Emit a single JSON envelope instead of plain text instruction output"
+    )]
+    pub json: bool,
 }
 
 #[derive(Args, Debug)]
@@ -165,6 +170,16 @@ pub struct DoneArgs {
     pub session: String,
     #[arg(long = "as")]
     pub agent: String,
+    #[arg(
+        long,
+        help = "Instruction id from `rally next --json` to acknowledge before completing done"
+    )]
+    pub instruction_id: Option<String>,
+    #[arg(
+        long,
+        help = "Require --instruction-id and reject stale/missing instruction acknowledgments"
+    )]
+    pub strict_instruction_id: bool,
 }
 
 #[derive(Args, Debug)]
@@ -185,6 +200,16 @@ pub struct FileIssueArgs {
     pub question: Option<String>,
     #[arg(long)]
     pub context: Option<String>,
+    #[arg(
+        long,
+        help = "Instruction id from `rally next --json` to acknowledge before action execution"
+    )]
+    pub instruction_id: Option<String>,
+    #[arg(
+        long,
+        help = "Require --instruction-id and reject stale/missing instruction acknowledgments"
+    )]
+    pub strict_instruction_id: bool,
 }
 
 #[derive(Args, Debug)]
@@ -195,6 +220,16 @@ pub struct ChallengeArgs {
     pub agent: String,
     #[arg(long)]
     pub issue: u32,
+    #[arg(
+        long,
+        help = "Instruction id from `rally next --json` to acknowledge before action execution"
+    )]
+    pub instruction_id: Option<String>,
+    #[arg(
+        long,
+        help = "Require --instruction-id and reject stale/missing instruction acknowledgments"
+    )]
+    pub strict_instruction_id: bool,
 }
 
 #[derive(Args, Debug)]
@@ -205,6 +240,16 @@ pub struct AgreeArgs {
     pub agent: String,
     #[arg(long)]
     pub issue: u32,
+    #[arg(
+        long,
+        help = "Instruction id from `rally next --json` to acknowledge before action execution"
+    )]
+    pub instruction_id: Option<String>,
+    #[arg(
+        long,
+        help = "Require --instruction-id and reject stale/missing instruction acknowledgments"
+    )]
+    pub strict_instruction_id: bool,
 }
 
 #[derive(Args, Debug)]
@@ -213,6 +258,16 @@ pub struct CheckpointArgs {
     pub session: String,
     #[arg(long = "as")]
     pub agent: String,
+    #[arg(
+        long,
+        help = "Instruction id from `rally next --json` to acknowledge before action execution"
+    )]
+    pub instruction_id: Option<String>,
+    #[arg(
+        long,
+        help = "Require --instruction-id and reject stale/missing instruction acknowledgments"
+    )]
+    pub strict_instruction_id: bool,
 }
 
 #[derive(Clone, Copy, Debug, Eq, PartialEq, ValueEnum)]
@@ -232,6 +287,16 @@ pub struct ReviewArgs {
     pub verdict: ReviewVerdictArg,
     #[arg(long)]
     pub message: Option<String>,
+    #[arg(
+        long,
+        help = "Instruction id from `rally next --json` to acknowledge before action execution"
+    )]
+    pub instruction_id: Option<String>,
+    #[arg(
+        long,
+        help = "Require --instruction-id and reject stale/missing instruction acknowledgments"
+    )]
+    pub strict_instruction_id: bool,
 }
 
 #[derive(Args, Debug)]
@@ -312,6 +377,16 @@ pub struct WorkflowActionArgs {
     pub args_json: Option<String>,
     #[arg(long, help = "Path to a JSON file containing action arguments")]
     pub args_file: Option<PathBuf>,
+    #[arg(
+        long,
+        help = "Instruction id from `rally next --json` to acknowledge before action execution"
+    )]
+    pub instruction_id: Option<String>,
+    #[arg(
+        long,
+        help = "Require --instruction-id and reject stale/missing instruction acknowledgments"
+    )]
+    pub strict_instruction_id: bool,
 }
 
 #[derive(Args, Debug)]

src/command_surface.rs

@@ -16,7 +16,7 @@ use crate::{
     },
     commands,
     state::{SessionHandle, SessionType},
-    watch,
+    watch::{self, NextOutputMode},
 };
 
 const RALLY_DEFAULT_BIN: &str = "rally";
@@ -94,6 +94,7 @@ pub fn run(args: &SkillRunArgs, registry: &WorkflowRegistry) -> Result<i32> {
                     &session_name,
                     &actual_name,
                     Some(DEFAULT_RUN_TIMEOUT_SECS),
+                    NextOutputMode::Text,
                     registry,
                 )?;
                 if code == 1 {
@@ -164,6 +165,7 @@ pub fn run(args: &SkillRunArgs, registry: &WorkflowRegistry) -> Result<i32> {
         &resolved.session,
         &actual_name,
         Some(DEFAULT_RUN_TIMEOUT_SECS),
+        NextOutputMode::Text,
         registry,
     )?;
     if code == 1 {