deepstrike_core/runtime/kernel/wire/
terminal.rs

1//! Operation terminals (spec §7.12).
2//!
3//! A terminal is the opposite of an effect: it requires no resolution, is allocated no effect id,
4//! and an operation commits **at most one** for its whole life. Both facts are expressed in the
5//! types rather than left to convention:
6//!
7//! * [`StepDisposition`] makes "a committed step publishes effects **or** a terminal" a union
8//!   instead of a `(Vec<KernelEffect>, Option<KernelTerminal>)` pair that can hold both;
9//! * [`TerminalSlot`] makes "exactly once" an API a second commit cannot pass, rather than an
10//!   invariant each of four hosts re-implements (recovery-chain distortion R-C2 #6: terminal
11//!   deduplication currently relies on host convention).
12//!
13//! The usage report is committed **inside** the terminal and nowhere else, so there is no second
14//! place a run's accounting can disagree with itself.
15
16use serde::{Deserialize, Serialize};
17
18use super::command::CancellationReason;
19use super::effect::{KernelEffect, ProviderMessage};
20use super::scalar::{NodeId, WireU64, WorkflowId};
21
22// ---------------------------------------------------------------------------------------------
23// §7.12 · the terminal union
24// ---------------------------------------------------------------------------------------------
25
26/// How an operation ended.
27#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
28#[serde(tag = "kind", rename_all = "snake_case")]
29pub enum KernelTerminal {
30    /// An agent root finished its loop.
31    Agent(AgentTerminal),
32    /// A workflow root finished its DAG. A workflow completion commits this terminal **itself** —
33    /// there is no external "complete the run" input, which is the only reason the historical
34    /// `CompleteRun` existed.
35    Workflow(WorkflowTerminal),
36    /// The operation was cancelled. Cancellation is never an effect failure: it arrives only
37    /// through `HostControl::Cancel`.
38    Cancelled(CancelledTerminal),
39    /// The kernel itself could not continue — a recovery ladder ran out, an invariant broke.
40    Failed(FailedTerminal),
41}
42
43impl KernelTerminal {
44    /// The one place a run's accounting is committed.
45    pub fn usage(&self) -> &UsageReport {
46        match self {
47            Self::Agent(terminal) => &terminal.usage,
48            Self::Workflow(terminal) => &terminal.usage,
49            Self::Cancelled(terminal) => &terminal.usage,
50            Self::Failed(terminal) => &terminal.usage,
51        }
52    }
53}
54
55#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
56#[serde(deny_unknown_fields)]
57pub struct AgentTerminal {
58    pub result: LoopResult,
59    pub usage: UsageReport,
60}
61
62#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
63#[serde(deny_unknown_fields)]
64pub struct WorkflowTerminal {
65    pub outcome: WorkflowOutcome,
66    pub usage: UsageReport,
67}
68
69#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
70#[serde(deny_unknown_fields)]
71pub struct CancelledTerminal {
72    pub reason: CancellationReason,
73    pub usage: UsageReport,
74}
75
76#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
77#[serde(deny_unknown_fields)]
78pub struct FailedTerminal {
79    pub failure: KernelFailure,
80    pub usage: UsageReport,
81}
82
83/// What an agent loop produced.
84#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
85#[serde(deny_unknown_fields)]
86pub struct LoopResult {
87    pub termination: TerminationReason,
88    #[serde(default, skip_serializing_if = "Option::is_none")]
89    pub final_message: Option<ProviderMessage>,
90    pub turns_used: u32,
91    #[serde(default, skip_serializing_if = "Option::is_none")]
92    pub pace_decision: Option<PaceDecision>,
93}
94
95#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
96#[serde(deny_unknown_fields)]
97pub struct PaceDecision {
98    pub action: PaceAction,
99    #[serde(default, skip_serializing_if = "Option::is_none")]
100    pub delay_ms: Option<WireU64>,
101    pub reason: String,
102    #[serde(default, skip_serializing_if = "Option::is_none")]
103    pub coerced_from: Option<String>,
104}
105
106#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
107#[serde(rename_all = "snake_case")]
108pub enum PaceAction {
109    Continue,
110    Sleep,
111    Stop,
112}
113
114/// Why the loop stopped.
115///
116/// `user_abort` and a generic `error` are deliberately absent: a cancellation is a
117/// [`KernelTerminal::Cancelled`] and a kernel failure is a [`KernelTerminal::Failed`]. Folding
118/// them back in here would give the same event two representations.
119#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
120#[serde(rename_all = "snake_case")]
121pub enum TerminationReason {
122    Completed,
123    MaxTurns,
124    TokenBudget,
125    Deadline,
126    /// The reactive recovery ladder for a provider context overflow ran out.
127    ContextOverflow,
128    /// Repeat fuse escalation — the agent kept re-issuing the same call. Distinct from `MaxTurns`,
129    /// which a productive run can also reach.
130    NoProgress,
131    MilestoneExceeded,
132}
133
134/// Resource accounting for one operation. No wall clock and no duration: elapsed time is a host
135/// observation, derivable from the envelope times already in the journal.
136#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
137#[serde(deny_unknown_fields)]
138pub struct UsageReport {
139    pub input_tokens: WireU64,
140    pub output_tokens: WireU64,
141    pub turns: u32,
142    #[serde(default, skip_serializing_if = "Option::is_none")]
143    pub cached_input_tokens: Option<WireU64>,
144}
145
146#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
147#[serde(deny_unknown_fields)]
148pub struct WorkflowOutcome {
149    pub workflow_id: WorkflowId,
150    pub status: WorkflowStatus,
151    #[serde(default, skip_serializing_if = "Vec::is_empty")]
152    pub completed_nodes: Vec<NodeId>,
153    #[serde(default, skip_serializing_if = "Vec::is_empty")]
154    pub failed_nodes: Vec<NodeId>,
155}
156
157#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
158#[serde(rename_all = "snake_case")]
159pub enum WorkflowStatus {
160    Completed,
161    Failed,
162    Cancelled,
163}
164
165/// A kernel-side failure, as opposed to a host effect failure. This is what a
166/// [`HostEffectFailure`](super::effect::HostEffectFailure) escalates *into* after the kernel has
167/// made its one policy decision (DEC-5) and decided it cannot continue.
168#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
169#[serde(deny_unknown_fields)]
170pub struct KernelFailure {
171    pub code: KernelFailureCode,
172    #[serde(default, skip_serializing_if = "String::is_empty")]
173    pub message: String,
174}
175
176#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
177#[serde(rename_all = "snake_case")]
178pub enum KernelFailureCode {
179    ProviderRecoveryExhausted,
180    OutputRecoveryExhausted,
181    /// A host effect failed and no recovery ladder applied.
182    HostEffectFailed,
183    ResourceExhausted,
184    InvariantViolated,
185}
186
187// ---------------------------------------------------------------------------------------------
188// exactly-once
189// ---------------------------------------------------------------------------------------------
190
191/// What a single committed step publishes.
192///
193/// A union, not a record with an optional terminal beside an effect list: "terminal and terminal
194/// observations land in the same committed step, and that step publishes no effect" is then a
195/// property of the type, not of every caller remembering to clear the other field.
196#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
197#[serde(tag = "kind", rename_all = "snake_case")]
198pub enum StepDisposition {
199    Effects(EffectsDisposition),
200    Terminal(TerminalDisposition),
201}
202
203impl StepDisposition {
204    pub fn effects(&self) -> &[KernelEffect] {
205        match self {
206            Self::Effects(disposition) => &disposition.effects,
207            Self::Terminal(_) => &[],
208        }
209    }
210
211    pub fn terminal(&self) -> Option<&KernelTerminal> {
212        match self {
213            Self::Effects(_) => None,
214            Self::Terminal(disposition) => Some(&disposition.terminal),
215        }
216    }
217
218    pub fn is_terminal(&self) -> bool {
219        matches!(self, Self::Terminal(_))
220    }
221}
222
223#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
224#[serde(deny_unknown_fields)]
225pub struct EffectsDisposition {
226    #[serde(default, skip_serializing_if = "Vec::is_empty")]
227    pub effects: Vec<KernelEffect>,
228}
229
230#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
231#[serde(deny_unknown_fields)]
232pub struct TerminalDisposition {
233    pub terminal: KernelTerminal,
234}
235
236/// The operation's single terminal slot.
237///
238/// [`TerminalSlot::commit`] succeeds once and refuses every later attempt, including an identical
239/// one. That is the exactly-once invariant of §7.12 as an API: no caller can double-commit by
240/// forgetting to check first, and a resumed operation that re-derives its terminal is told so
241/// instead of emitting a second one.
242#[derive(Debug, Clone, Default, PartialEq)]
243pub struct TerminalSlot {
244    terminal: Option<KernelTerminal>,
245}
246
247impl TerminalSlot {
248    pub fn empty() -> Self {
249        Self { terminal: None }
250    }
251
252    pub fn commit(
253        &mut self,
254        terminal: KernelTerminal,
255    ) -> Result<&KernelTerminal, Box<TerminalAlreadyCommitted>> {
256        if let Some(committed) = &self.terminal {
257            return Err(Box::new(TerminalAlreadyCommitted {
258                committed: committed.clone(),
259                rejected: terminal,
260            }));
261        }
262        Ok(self.terminal.insert(terminal))
263    }
264
265    pub fn get(&self) -> Option<&KernelTerminal> {
266        self.terminal.as_ref()
267    }
268
269    pub fn is_committed(&self) -> bool {
270        self.terminal.is_some()
271    }
272}
273
274/// A second terminal was offered for an operation that already has one.
275#[derive(Debug, Clone, PartialEq)]
276pub struct TerminalAlreadyCommitted {
277    pub committed: KernelTerminal,
278    pub rejected: KernelTerminal,
279}
280
281impl std::fmt::Display for TerminalAlreadyCommitted {
282    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
283        f.write_str("this operation has already committed a terminal")
284    }
285}
286
287impl std::error::Error for TerminalAlreadyCommitted {}
288
289#[cfg(test)]
290mod tests {
291    use std::collections::BTreeSet;
292    use std::fs;
293    use std::path::PathBuf;
294
295    use serde_json::{Value, json};
296
297    use super::super::*;
298
299    fn fixture(name: &str) -> Value {
300        let path = PathBuf::from(env!("CARGO_MANIFEST_DIR"))
301            .join("../../tests/fixtures/kernel-wire")
302            .join(name);
303        let raw = fs::read_to_string(&path)
304            .unwrap_or_else(|e| panic!("failed to read {}: {e}", path.display()));
305        serde_json::from_str(&raw).unwrap_or_else(|e| panic!("{name} is not JSON: {e}"))
306    }
307
308    fn keys(value: &Value, out: &mut BTreeSet<String>) {
309        match value {
310            Value::Object(map) => {
311                for (key, child) in map {
312                    out.insert(key.clone());
313                    keys(child, out);
314                }
315            }
316            Value::Array(items) => items.iter().for_each(|item| keys(item, out)),
317            _ => {}
318        }
319    }
320
321    fn usage() -> UsageReport {
322        UsageReport {
323            input_tokens: WireU64::new(18_402),
324            output_tokens: WireU64::new(2_117),
325            turns: 7,
326            cached_input_tokens: None,
327        }
328    }
329
330    fn samples() -> Vec<KernelTerminal> {
331        vec![
332            KernelTerminal::Agent(AgentTerminal {
333                result: LoopResult {
334                    termination: TerminationReason::Completed,
335                    final_message: None,
336                    turns_used: 7,
337                    pace_decision: None,
338                },
339                usage: usage(),
340            }),
341            KernelTerminal::Workflow(WorkflowTerminal {
342                outcome: WorkflowOutcome {
343                    workflow_id: WorkflowId::new("wf-1").unwrap(),
344                    status: WorkflowStatus::Completed,
345                    completed_nodes: vec![NodeId::new("node-a").unwrap()],
346                    failed_nodes: Vec::new(),
347                },
348                usage: usage(),
349            }),
350            KernelTerminal::Cancelled(CancelledTerminal {
351                reason: CancellationReason::User,
352                usage: usage(),
353            }),
354            KernelTerminal::Failed(FailedTerminal {
355                failure: KernelFailure {
356                    code: KernelFailureCode::ProviderRecoveryExhausted,
357                    message: "context overflow ladder exhausted".to_string(),
358                },
359                usage: usage(),
360            }),
361        ]
362    }
363
364    // -----------------------------------------------------------------------------------------
365    // shape
366    // -----------------------------------------------------------------------------------------
367
368    #[test]
369    fn a_terminal_has_four_shapes_and_every_one_commits_usage_exactly_once() {
370        let mut tags = BTreeSet::new();
371        for terminal in samples() {
372            let value = serde_json::to_value(&terminal).unwrap();
373            tags.insert(value["kind"].as_str().unwrap().to_string());
374            assert!(
375                value.get("usage").is_some(),
376                "every terminal commits the usage report: {value}"
377            );
378            // and only once — no second usage-shaped field anywhere below it
379            let mut all = BTreeSet::new();
380            keys(&value, &mut all);
381            assert!(
382                !all.contains("usage_report") && !all.contains("budget_usage"),
383                "usage travels in exactly one field: {value}"
384            );
385            let back: KernelTerminal = serde_json::from_value(value).unwrap();
386            assert_eq!(back, terminal);
387        }
388        assert_eq!(
389            tags,
390            BTreeSet::from([
391                "agent".to_string(),
392                "cancelled".to_string(),
393                "failed".to_string(),
394                "workflow".to_string(),
395            ])
396        );
397    }
398
399    #[test]
400    fn a_terminal_requires_no_resolution_and_carries_no_effect_id() {
401        for terminal in samples() {
402            let mut all = BTreeSet::new();
403            keys(&serde_json::to_value(&terminal).unwrap(), &mut all);
404            for banned in ["effect_id", "causation_input_id", "resolution"] {
405                assert!(
406                    !all.contains(banned),
407                    "a terminal is not an effect; it must not carry {banned:?}"
408                );
409            }
410        }
411    }
412
413    #[test]
414    fn a_terminal_carries_no_host_wall_clock() {
415        for terminal in samples() {
416            let mut all = BTreeSet::new();
417            keys(&serde_json::to_value(&terminal).unwrap(), &mut all);
418            for banned in [
419                "now_ms",
420                "observed_at_ms",
421                "timestamp",
422                "timestamp_ms",
423                "started_at_ms",
424                "completed_at_ms",
425                "wall_clock_ms",
426                "duration_ms",
427            ] {
428                assert!(!all.contains(banned), "terminal must not carry {banned:?}");
429            }
430        }
431    }
432
433    // -----------------------------------------------------------------------------------------
434    // exactly once (§7.12 invariant 1)
435    // -----------------------------------------------------------------------------------------
436
437    #[test]
438    fn a_terminal_slot_accepts_exactly_one_terminal() {
439        let mut slot = TerminalSlot::empty();
440        assert!(!slot.is_committed());
441        assert!(slot.get().is_none());
442
443        let first = samples().into_iter().next().unwrap();
444        slot.commit(first.clone()).expect("first terminal commits");
445        assert!(slot.is_committed());
446        assert_eq!(slot.get(), Some(&first));
447
448        // a second terminal — of any shape, including an identical one — is refused
449        for second in samples() {
450            let rejected = slot
451                .commit(second)
452                .expect_err("an operation has at most one terminal");
453            assert_eq!(rejected.committed, first);
454        }
455        assert_eq!(slot.get(), Some(&first));
456    }
457
458    #[test]
459    fn a_committed_step_publishes_effects_or_a_terminal_but_never_both() {
460        let effects = StepDisposition::Effects(EffectsDisposition {
461            effects: vec![KernelEffect {
462                effect_id: EffectId::new("op-1:step:1:effect:0").unwrap(),
463                causation_input_id: InputId::new("in-1").unwrap(),
464                effect: EffectKind::EvaluateMilestone(EvaluateMilestoneEffect {
465                    request: MilestoneRequest {
466                        contract_id: "brief-quality-primary".to_string(),
467                        phase_id: "phase-1".to_string(),
468                    },
469                }),
470            }],
471        });
472        assert!(effects.terminal().is_none());
473        assert_eq!(effects.effects().len(), 1);
474
475        let terminal = StepDisposition::Terminal(TerminalDisposition {
476            terminal: samples().into_iter().next().unwrap(),
477        });
478        assert!(terminal.terminal().is_some());
479        assert!(
480            terminal.effects().is_empty(),
481            "the terminal step publishes no effect"
482        );
483
484        // the wire shape cannot express the mixture either
485        let mixed = json!({
486            "kind": "terminal",
487            "terminal": serde_json::to_value(samples().into_iter().next().unwrap()).unwrap(),
488            "effects": [],
489        });
490        assert!(
491            serde_json::from_value::<StepDisposition>(mixed).is_err(),
492            "a step must not carry both a terminal and an effect list"
493        );
494    }
495
496    // -----------------------------------------------------------------------------------------
497    // strictness
498    // -----------------------------------------------------------------------------------------
499
500    #[test]
501    fn unknown_terminal_kinds_and_fields_are_rejected() {
502        let unknown_kind = json!({ "kind": "done", "usage": { "input_tokens": "1", "output_tokens": "1", "turns": 1 } });
503        assert!(serde_json::from_value::<KernelTerminal>(unknown_kind).is_err());
504
505        let unknown_field = json!({
506            "kind": "cancelled",
507            "reason": "user",
508            "usage": { "input_tokens": "1", "output_tokens": "1", "turns": 1 },
509            "now_ms": 1753747203000u64,
510        });
511        assert!(serde_json::from_value::<KernelTerminal>(unknown_field).is_err());
512
513        let numeric_tokens = json!({
514            "kind": "cancelled",
515            "reason": "user",
516            "usage": { "input_tokens": 1, "output_tokens": "1", "turns": 1 },
517        });
518        assert!(serde_json::from_value::<KernelTerminal>(numeric_tokens).is_err());
519    }
520
521    // -----------------------------------------------------------------------------------------
522    // goldens
523    // -----------------------------------------------------------------------------------------
524
525    #[test]
526    fn terminal_goldens_round_trip_unchanged() {
527        let mut covered = BTreeSet::new();
528        for name in [
529            "golden_terminal_agent.json",
530            "golden_terminal_workflow.json",
531            "golden_terminal_cancelled.json",
532            "golden_terminal_failed.json",
533        ] {
534            let golden = fixture(name);
535            let terminal: KernelTerminal = serde_json::from_value(golden.clone())
536                .unwrap_or_else(|e| panic!("{name} does not decode: {e}"));
537            assert_eq!(
538                serde_json::to_value(&terminal).unwrap(),
539                golden,
540                "{name}: round-trip changed the document"
541            );
542            covered.insert(golden["kind"].as_str().unwrap().to_string());
543        }
544        assert_eq!(covered.len(), 4, "one golden per terminal shape");
545    }
546}
deepstrike_core/runtime/kernel/wire/terminal.rs

deepstrike_core/runtime/kernel/wire/
terminal.rs
