put applied_step back

This commit is contained in:
Lifei Zhou 2026-10-09 15:14:16 +11:00
parent 76338f0a10
commit 860f52bd88
5 changed files with 16 additions and 9 deletions

View file

@ -26,8 +26,9 @@ is a function of the persisted conversation, not of in-memory loop state.
boundaries. Their schemas are sent to inference and matching calls are
dispatched against the current session.
- **`StateMachine<'a, S, E>`** — holds `Vec<Step<..>>` and a `CancellationToken`.
`step()` runs one pass and reports in `StepResult::next` whether to
`Continue`, `Yield` to the client, or stop because the run was `Cancelled`;
`step()` runs one pass and returns a `StepResult` whose `applied_step` names
the step that applied (`None` if none did) and whose `next` says whether to
`Continue`, `Yield` to the client, or `Cancel` because the run was cancelled;
`apply()` writes effects back; `run()` loops until `next` is not `Continue`.
- **`ConversationEffect`** — the default effect type: `AppendMessage`,
`ReplaceConversation`, `PatchToolRequestMeta`, `SetMessageVisibility`. Bring

View file

@ -61,6 +61,7 @@ pub struct StateMachine<'a, S, E = ConversationEffect> {
fn empty_result<E>(next: Next) -> StepResult<E> {
StepResult {
effects: Vec::new(),
applied_step: None,
next,
}
}
@ -125,7 +126,7 @@ where
for (index, step) in self.steps.iter().enumerate() {
let name = step.operation().name();
if self.cancel.is_cancelled() {
return Ok(empty_result(Next::Cancelled));
return Ok(empty_result(Next::Cancel));
}
let execution = async {
match step {
@ -165,7 +166,7 @@ where
Some(Err(error)) if !self.cancel.is_cancelled() => return Err(error),
_ => {
*self.interrupted_step.lock().unwrap() = Some(index);
return Ok(empty_result(Next::Cancelled));
return Ok(empty_result(Next::Cancel));
}
};
@ -173,11 +174,12 @@ where
OperationResult::NotApplicable => {}
OperationResult::Applied(mut result) => {
tracing::debug!(step = name, "applied step");
result.applied_step = Some(name);
for effect in &mut result.effects {
effect.ensure_message_ids();
}
if self.cancel.is_cancelled() {
result.next = Next::Cancelled;
result.next = Next::Cancel;
}
return Ok(result);
}
@ -217,7 +219,7 @@ where
.await?;
match result.next {
Next::Continue => {}
Next::Yield | Next::Cancelled => break,
Next::Yield | Next::Cancel => break,
}
}
self.finalize(runtime, session_id, emit).await

View file

@ -163,11 +163,12 @@ pub trait Inference<S, E: MaybeSend + 'static = ConversationEffect>: Operation<S
pub enum Next {
Continue,
Yield,
Cancelled,
Cancel,
}
pub struct StepResult<E = ConversationEffect> {
pub effects: Vec<E>,
pub applied_step: Option<&'static str>,
pub next: Next,
}
@ -183,6 +184,7 @@ pub fn not_applicable<E>() -> Result<OperationResult<E>> {
pub fn applied<E>(effects: impl IntoIterator<Item = E>) -> Result<OperationResult<E>> {
Ok(OperationResult::Applied(StepResult {
effects: effects.into_iter().collect(),
applied_step: None,
next: Next::Continue,
}))
}
@ -190,6 +192,7 @@ pub fn applied<E>(effects: impl IntoIterator<Item = E>) -> Result<OperationResul
pub fn yielded<E>() -> Result<OperationResult<E>> {
Ok(OperationResult::Applied(StepResult {
effects: Vec::new(),
applied_step: None,
next: Next::Yield,
}))
}
@ -197,6 +200,7 @@ pub fn yielded<E>() -> Result<OperationResult<E>> {
pub fn yielded_with<E>(effects: impl IntoIterator<Item = E>) -> Result<OperationResult<E>> {
Ok(OperationResult::Applied(StepResult {
effects: effects.into_iter().collect(),
applied_step: None,
next: Next::Yield,
}))
}

View file

@ -470,7 +470,7 @@ async fn cancellation_interrupts_inference_discovery() {
}
cancel.cancel();
assert_eq!(step.await.unwrap().unwrap().next, Next::Cancelled);
assert_eq!(step.await.unwrap().unwrap().next, Next::Cancel);
}
#[tokio::test]

View file

@ -455,7 +455,7 @@ impl TestPipeline {
&emit,
)
.await?;
if !result.effects.is_empty() {
if result.applied_step.is_some() {
applied_steps += 1;
}
while let Ok(event) = rx.try_recv() {