Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
23 changes: 23 additions & 0 deletions src/acp_child/protocol.rs
Original file line number Diff line number Diff line change
Expand Up @@ -424,6 +424,29 @@ mod tests {
assert_eq!(state, Foreground::Idle(Some(v2::StopReason::Refusal)));
}

#[test]
fn foreground_running_then_pre_start_error_completes_without_accepting_stale_idle() {
let update = |state, reason: Option<&str>| {
let mut value = json!({"sessionId": "child", "update": {
"sessionUpdate": "state_update", "state": state,
}});
if let Some(reason) = reason {
value["update"]["stopReason"] = json!(reason);
}
foreground(&value).unwrap().unwrap()
};
let mut state = Foreground::Waiting;
assert!(!state.advance(update("idle", Some("_error"))));
assert!(state.advance(update("running", None)));
assert!(state.advance(update("idle", Some("_error"))));
let Foreground::Idle(reason) = state.clone() else {
panic!("accepted prompt must settle");
};
assert!(completion(reason).is_err());
assert!(!state.advance(update("idle", Some("end_turn"))));
assert!(!state.advance(update("running", None)));
}

#[test]
fn recognizes_idle_without_treating_running_as_completion() {
let update = |state| json!({"sessionId": "child", "update": {"sessionUpdate": "state_update", "state": state}});
Expand Down
129 changes: 127 additions & 2 deletions src/protocols/acp/activity.rs
Original file line number Diff line number Diff line change
Expand Up @@ -41,8 +41,9 @@ struct Activity {
}

/// Session-owned lifecycle instrument shared by the actor and its observer.
/// Admission alone is silent; TurnStarted allocates the activity identity. All
/// logical turns drained by an execution share that identity until settlement.
/// Ordinary admission is silent; TurnStarted allocates the activity identity.
/// Accepted v2 prompts explicitly start an interval before publication or execution.
/// All logical turns drained by an execution share that identity until settlement.
/// Projection runs synchronously outside the state lock. An in-flight claim
/// prevents another projection from overtaking it, including callback reentry.
/// The actor must flush final content/diagnostics before settling to Idle.
Expand Down Expand Up @@ -75,6 +76,66 @@ impl SessionActivity {
Ok(())
}

/// The serialized actor checks this before accepting another prompt.
pub(super) fn check_prompt_admission(&self) -> Result<(), AcpRuntimeError> {
let state = self
.state
.lock()
.map_err(|_| AcpRuntimeError::Loop("session activity is poisoned".into()))?;
if state.state != State::Idle || state.executing || state.projecting {
return Err(AcpRuntimeError::Loop(
"session activity is unavailable".into(),
));
}
Ok(())
}

/// An accepted asynchronous prompt owes a terminal even if no logical turn starts.
/// Called inside `execute`, before any fallible prompt publication or model work.
pub(super) fn start_prompt(&self) -> Result<(), AcpRuntimeError> {
let mut state = self
.state
.lock()
.map_err(|_| AcpRuntimeError::Loop("session activity is poisoned".into()))?;
if state.state != State::Idle
|| !state.executing
|| state.projecting
|| state.origin != ExecutionOrigin::Prompt
{
return Err(AcpRuntimeError::Loop(
"session activity is unavailable".into(),
));
}
let Some(id) = state.next_id.checked_add(1) else {
state.state = State::Unavailable;
return Err(AcpRuntimeError::Loop(
"session activity identity exhausted".into(),
));
};
let transition = Transition {
id,
origin: ExecutionOrigin::Prompt,
active: true,
reason: FinishReason::Completed,
error: None,
};
state.current = Some(transition.clone());
state.next_id = id;
state.state = State::Running;
state.projecting = true;
drop(state);
let result = self.project_transition(transition);
if result.is_err() {
// Running was not admitted. An uncorrelated Idle cannot complete the
// parent's waiting route, and replaying Running is not transport-safe.
// Isolate this interval; the actor rejects future prompts before acceptance.
if let Ok(mut state) = self.state.lock() {
state.state = State::Unavailable;
}
}
result
}

pub(super) fn observe(&self, event: &AgentEvent) {
let Ok(mut state) = self.state.lock() else {
return;
Expand Down Expand Up @@ -404,6 +465,70 @@ mod tests {
assert!(idle.error.as_ref().unwrap().contains("terminal error"));
}

#[tokio::test]
async fn accepted_prompt_interval_survives_pre_start_error_and_rejects_duplicate_start() {
let transitions = Arc::new(Mutex::new(Vec::new()));
let output = transitions.clone();
let activity = SessionActivity::new(move |transition| {
output.lock().unwrap().push(transition);
Ok(())
});
assert!(activity.start_prompt().is_err());
for reason in [FinishReason::Error, FinishReason::Cancelled] {
let result = activity
.execute(
ExecutionOrigin::Prompt,
async {
activity.start_prompt()?;
assert!(activity.start_prompt().is_err());
if reason == FinishReason::Error {
Err(AcpRuntimeError::Loop("before TurnStarted".into()))
} else {
Ok(reason.clone())
}
},
|reason| Some(reason.clone()),
)
.await;
assert_eq!(result.is_err(), reason == FinishReason::Error);
activity.settle(None, None).unwrap();
}
let events = transitions.lock().unwrap();
assert_eq!(events.len(), 4);
for (index, pair) in events.as_chunks::<2>().0.iter().enumerate() {
assert!(pair[0].active);
assert!(!pair[1].active);
assert_eq!(pair[0].id, pair[1].id);
assert_eq!(pair[0].id, index as u64 + 1);
}
assert_eq!(events[1].reason, FinishReason::Error);
assert_eq!(events[3].reason, FinishReason::Cancelled);
}

#[tokio::test]
async fn abandoned_accepted_prompt_isolates_without_replaying_terminal() {
let transitions = Arc::new(Mutex::new(Vec::new()));
let output = transitions.clone();
let activity = SessionActivity::new(move |transition| {
output.lock().unwrap().push(transition);
Ok(())
});
let mut execution = Box::pin(activity.execute(
ExecutionOrigin::Prompt,
async {
activity.start_prompt()?;
std::future::pending::<Result<(), AcpRuntimeError>>().await
},
|_| None,
));
assert!(futures_util::poll!(execution.as_mut()).is_pending());
drop(execution);
assert!(activity.start_prompt().is_err());
assert!(activity.settle(None, None).is_err());
assert!(!activity.state.is_poisoned());
assert_eq!(transitions.lock().unwrap().len(), 1);
}

#[tokio::test]
async fn finalization_normalizes_cancellation_and_never_skips_failed_drain() {
let diagnostic = Arc::new(Mutex::new(Vec::new()));
Expand Down
Loading
Loading