fix(acp): 拒绝 Codex 计划时可以留下修改意见，不再直接结束会话

src-tauri/src/acp/lifecycle.rs

@@ -335,6 +335,11 @@ pub(crate) async fn handle_event(
             };
             let (conversation_id, last_text, claimed) = {
                 let mut snap = state_arc.write().await;
+                // Request-linked, immutable sequence identity: the next prompt
+                // may already have finished before this worker sees the event.
+                if snap.plan_revisions.is_boundary(envelope.seq) {
+                    return Ok(());
+                }
                 // Re-entrant for this worker: a retry of this same envelope
                 // claims again and goes on. Only a CLI run's fallback that
                 // already wrote this turn's status makes it skip.
@@ -3884,6 +3889,86 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn handle_event_plan_revision_boundary_never_enters_complete_call() {
+        let db = test_helpers::fresh_in_memory_db().await;
+        let folder = test_helpers::seed_folder(&db, "/tmp/plan-revision-boundary").await;
+        let conv = conversation_service::create(&db.conn, folder, AgentType::Codex, None, None)
+            .await
+            .unwrap();
+        let mgr = ConnectionManager::new();
+        let connection = fake_connection_with_state("revision-child", Some(conv.id));
+        {
+            let mut state = connection.state.write().await;
+            state
+                .plan_revisions
+                .accept("review-1".into(), "Revise first".into());
+            state.plan_revisions.bind_boundary(10);
+            state.plan_revisions.take_next().unwrap();
+            state.plan_revisions.finish(
+                "review-1",
+                crate::acp::plan_revision::RevisionStatus::Delivered,
+            );
+            state
+                .plan_revisions
+                .accept("review-2".into(), "Revise again".into());
+            state.plan_revisions.bind_boundary(20);
+            state.plan_revisions.abort_pending();
+            // Lifecycle is deliberately delayed until both the next turn and
+            // its cleanup have run. Neither can erase the boundary identity.
+        }
+        mgr.connections
+            .lock()
+            .await
+            .insert("revision-child".into(), connection);
+        let (broker, driver) = stage_pending_delegation("revision-child", conv.id).await;
+        for seq in [10, 20] {
+            let event = EventEnvelope {
+                seq,
+                connection_id: "revision-child".into(),
+                payload: AcpEvent::TurnComplete {
+                    session_id: "same-session".into(),
+                    stop_reason: "end_turn".into(),
+                    agent_type: "codex".into(),
+                },
+            };
+            handle_event(&db.conn, &mgr, &event, Some(&broker))
+                .await
+                .unwrap();
+            assert_eq!(
+                read_row_status(&db, conv.id).await,
+                ConversationStatus::InProgress
+            );
+            assert_eq!(
+                broker.pending_count().await,
+                1,
+                "complete_call must not run at a revision boundary"
+            );
+            assert!(!driver.is_finished());
+        }
+        let event = EventEnvelope {
+            seq: 30,
+            connection_id: "revision-child".into(),
+            payload: AcpEvent::TurnComplete {
+                session_id: "same-session".into(),
+                stop_reason: "rejected".into(),
+                agent_type: "codex".into(),
+            },
+        };
+        handle_event(&db.conn, &mgr, &event, Some(&broker))
+            .await
+            .unwrap();
+        assert_eq!(
+            read_row_status(&db, conv.id).await,
+            ConversationStatus::Cancelled
+        );
+        assert_eq!(broker.pending_count().await, 0);
+        assert!(matches!(
+            driver.await.unwrap(),
+            DelegationOutcome::Err { .. }
+        ));
+    }
+
     /// A turn whose status a CLI run's fallback already wrote is skipped
     /// whole — a late lifecycle write must not overwrite what the user did to
     /// the row after the fallback — while the worker's own retries of the same

src-tauri/src/acp/manager.rs

@@ -2168,6 +2168,41 @@ impl ConnectionManager {
         Ok(())
     }
 
+    /// Atomically claim a Codex review and await the loop's durable-in-session receipt.
+    pub async fn request_plan_changes(
+        &self,
+        conn_id: &str,
+        request_id: &str,
+        feedback: &str,
+    ) -> Result<crate::acp::types::PlanRevisionReceipt, crate::app_error::AppCommandError> {
+        use crate::acp::plan_revision::PlanRevisions;
+        use crate::app_error::AppCommandError;
+
+        let feedback = PlanRevisions::normalize(feedback)?;
+        let cmd_tx = {
+            let connections = self.connections.lock().await;
+            connections
+                .get(conn_id)
+                .ok_or_else(|| AppCommandError::not_found("connection_not_found"))?
+                .cmd_tx
+                .clone()
+        };
+        let (reply, result) = tokio::sync::oneshot::channel();
+        cmd_tx
+            .send(ConnectionCommand::RequestPlanChanges {
+                request_id: request_id.to_string(),
+                feedback,
+                reply,
+            })
+            .await
+            .map_err(|_| AppCommandError::task_execution_failed("plan_revision_send_failed"))?;
+        // Enqueue success is NOT a claim. The loop returns only after saving
+        // the feedback and resolving the permission in the same serial path.
+        result
+            .await
+            .map_err(|_| AppCommandError::task_execution_failed("plan_revision_send_failed"))?
+    }
+
     /// Answer a permission card as the person looking at it.
     pub async fn respond_permission(
         &self,

src-tauri/src/acp/mod.rs

@@ -40,6 +40,7 @@ pub mod opencode_catalog;
 pub mod opencode_launch;
 pub mod opencode_plugins;
 pub mod plan_approval;
+pub mod plan_revision;
 pub mod preflight;
 pub mod prompt_hydration;
 pub mod provider_models;

src-tauri/src/acp/plan_revision.rs

@@ -0,0 +1,134 @@
+//! Connection-scoped Codex plan revisions. Receipts and physical boundary
+//! identities survive subsequent turns; lifecycle must never consult a live flag.
+
+use std::collections::BTreeMap;
+
+use crate::app_error::AppCommandError;
+use serde::{Deserialize, Serialize};
+
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+#[serde(rename_all = "camelCase")]
+pub struct PlanRevisionReceipt {
+    pub request_id: String,
+    pub status: String,
+}
+
+#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
+#[serde(rename_all = "snake_case")]
+pub enum RevisionStatus {
+    Accepted,
+    Sending,
+    Delivered,
+    Failed,
+    Aborted,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct PlanRevision {
+    pub request_id: String,
+    pub feedback: String,
+    pub status: RevisionStatus,
+    /// Assigned once, before publishing that physical TurnComplete. Never
+    /// cleared by a new prompt, cancellation, or a failed subsequent delivery.
+    pub boundary_seq: Option<u64>,
+}
+
+#[derive(Debug, Default, Clone)]
+pub struct PlanRevisions {
+    records: BTreeMap<String, PlanRevision>,
+}
+
+impl PlanRevisions {
+    pub fn normalize(feedback: &str) -> Result<String, AppCommandError> {
+        let feedback = feedback.trim();
+        if feedback.is_empty() {
+            return Err(AppCommandError::invalid_input("plan_feedback_empty"));
+        }
+        if feedback.chars().count() > crate::acp::plan_approval::MAX_FEEDBACK_CHARS {
+            return Err(AppCommandError::invalid_input("plan_feedback_too_long"));
+        }
+        Ok(feedback.to_string())
+    }
+
+    pub fn receipt(request_id: &str) -> PlanRevisionReceipt {
+        PlanRevisionReceipt {
+            request_id: request_id.to_string(),
+            status: "accepted".to_string(),
+        }
+    }
+
+    pub fn retry(
+        &self,
+        request_id: &str,
+        feedback: &str,
+    ) -> Result<Option<PlanRevisionReceipt>, AppCommandError> {
+        match self.records.get(request_id) {
+            Some(record) if record.feedback == feedback => Ok(Some(Self::receipt(request_id))),
+            Some(_) => Err(AppCommandError::already_exists(
+                "plan_review_already_decided",
+            )),
+            None => Ok(None),
+        }
+    }
+
+    pub fn accept(&mut self, request_id: String, feedback: String) {
+        self.records.insert(
+            request_id.clone(),
+            PlanRevision {
+                request_id,
+                feedback,
+                status: RevisionStatus::Accepted,
+                boundary_seq: None,
+            },
+        );
+    }
+
+    pub fn has_accepted(&self) -> bool {
+        self.records
+            .values()
+            .any(|r| r.status == RevisionStatus::Accepted)
+    }
+
+    pub fn bind_boundary(&mut self, seq: u64) {
+        for record in self.records.values_mut() {
+            if record.status == RevisionStatus::Accepted && record.boundary_seq.is_none() {
+                record.boundary_seq = Some(seq);
+            }
+        }
+    }
+
+    pub fn is_boundary(&self, seq: u64) -> bool {
+        self.records.values().any(|r| r.boundary_seq == Some(seq))
+    }
+
+    pub fn take_next(&mut self) -> Option<PlanRevision> {
+        let record = self
+            .records
+            .values_mut()
+            .find(|r| r.status == RevisionStatus::Accepted && r.boundary_seq.is_some())?;
+        record.status = RevisionStatus::Sending;
+        Some(record.clone())
+    }
+
+    pub fn finish(&mut self, request_id: &str, status: RevisionStatus) {
+        if let Some(record) = self.records.get_mut(request_id) {
+            record.status = status;
+        }
+    }
+
+    pub fn abort_pending(&mut self) {
+        for record in self.records.values_mut() {
+            if matches!(
+                record.status,
+                RevisionStatus::Accepted | RevisionStatus::Sending
+            ) {
+                record.status = RevisionStatus::Aborted;
+            }
+        }
+    }
+
+    pub fn snapshot(&self) -> Vec<PlanRevision> {
+        self.records.values().cloned().collect()
+    }
+}

src-tauri/src/acp/plan_revision_permission.rs

@@ -0,0 +1,94 @@
+//! Atomic permission claims and explicit failures for accepted revisions.
+
+use super::*;
+use crate::acp::plan_revision::PlanRevisions;
+
+pub(super) async fn request_plan_changes(
+    perms: &PendingPermissions,
+    state: &Arc<RwLock<SessionState>>,
+    emitter: &EventEmitter,
+    request_id: String,
+    feedback: String,
+) -> Result<PlanRevisionReceipt, AppCommandError> {
+    let feedback = PlanRevisions::normalize(&feedback)?;
+    // Queue -> state is the permission subsystem's lock order. Persist the
+    // claim BEFORE releasing Codex's responder; even an immediate end_turn
+    // cannot overtake this command on the serial conversation loop.
+    let mut queue = perms.lock().await;
+    {
+        let mut s = state.write().await;
+        if let Some(receipt) = s.plan_revisions.retry(&request_id, &feedback)? {
+            return Ok(receipt);
+        }
+        if queue.decided.contains(&request_id) {
+            return Err(AppCommandError::already_exists(
+                "plan_review_already_decided",
+            ));
+        }
+        let parked = queue
+            .responders
+            .get(&request_id)
+            .filter(|p| !p.responder.is_withdrawn())
+            .ok_or_else(|| AppCommandError::not_found("plan_review_not_pending"))?;
+        if s.agent_type != AgentType::Codex
+            || !parked.codex_plan_review
+            || !parked.options.iter().any(|o| o.option_id == "revise_plan")
+        {
+            return Err(AppCommandError::invalid_input("not_codex_plan_review"));
+        }
+        if !s.turn_in_flight || s.plan_revisions.has_accepted() {
+            return Err(AppCommandError::not_found("plan_review_not_pending"));
+        }
+        s.plan_revisions.accept(request_id.clone(), feedback);
+    }
+    resolve_permission_locked(
+        &mut queue,
+        state,
+        emitter,
+        request_id.clone(),
+        "revise_plan".into(),
+        PermissionAnswerOrigin::User,
+    )
+    .await;
+    Ok(PlanRevisions::receipt(&request_id))
+}
+
+pub(super) async fn fail_plan_revisions(
+    state: &Arc<RwLock<SessionState>>,
+    emitter: &EventEmitter,
+    agent_type: AgentType,
+    reason: &str,
+) {
+    let failed = {
+        let mut s = state.write().await;
+        let pending: Vec<_> = s
+            .plan_revisions
+            .snapshot()
+            .into_iter()
+            .filter(|r| matches!(r.status, RevisionStatus::Accepted | RevisionStatus::Sending))
+            .map(|r| r.request_id)
+            .collect();
+        for id in &pending {
+            s.plan_revisions.finish(id, RevisionStatus::Failed);
+        }
+        pending
+    };
+    for request_id in failed {
+        let error =
+            AppCommandError::task_execution_failed("plan_revision_send_failed").with_detail(reason);
+        emit_with_state(
+            state,
+            emitter,
+            AcpEvent::Error {
+                message: format!("plan_revision_send_failed: {request_id}"),
+                agent_type: agent_type.to_string(),
+                code: Some("plan_revision_send_failed".into()),
+                details: Some(
+                    serde_json::json!({ "requestId": request_id, "error": error }).to_string(),
+                ),
+                terminal: false,
+            },
+        )
+        .await;
+    }
+}