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

deepstrike_core/runtime/kernel/wire/
terminal.rs
