1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
//! What `lev ps` and the control socket see: one row per run, live from the
//! world rather than re-read from disk.
//!
//! [`wait_reason`](WorldHost::wait_reason) is the interesting part. A status of
//! `Waiting` says nothing about whether a person is needed, which is what issue
//! #184 was about, so the row carries why.
//!
//! Retention lives here too, and deliberately: a finished run stays listable for
//! a while after it ends, so how long the list keeps one is a property of the
//! list rather than of the health bookkeeping it used to sit beside.
use super::*;
impl WorldHost {
/// Why `entity` is [`AgentStatus::Waiting`], read off the markers the engine
/// already maintains. `None` when the agent is not waiting, or when it is
/// waiting for a reason nothing has claimed.
///
/// Order matters. A taint-gate block and a stage checkpoint each open a hub
/// request of their own, so both also carry [`AwaitingInteraction`]; asking
/// the specific markers first is what keeps them from all reporting as a
/// generic prompt.
pub fn wait_reason(&self, agent: crate::world::AgentId) -> Option<WaitReason> {
let world = self.world.world();
// An id from another world names a different agent here, which would
// report that one's wait reason as this run's.
let entity = agent.resolve_in(world)?;
let state = world.get::<AgentState>(entity)?;
if state.status != AgentStatus::Waiting {
return None;
}
if world
.get::<crate::gate_prompt::AwaitingGatePrompt>(entity)
.is_some()
{
return Some(WaitReason::TaintGate);
}
if world
.get::<crate::interaction_points::AwaitingInteractionPoint>(entity)
.is_some()
{
return Some(WaitReason::InteractionPoint);
}
if let Some(fanout) = world.get::<crate::fanout::FanOutWaiting>(entity) {
return Some(WaitReason::FanOutWorkers {
outstanding: fanout.outstanding(),
});
}
if world
.get::<crate::pipeline::WaitingForChildren>(entity)
.is_some()
{
let outstanding = world
.get::<SubAgentChildren>(entity)
.map(|c| {
c.children
.iter()
.filter(|&&child| {
world
.get::<AgentState>(child)
.is_some_and(|s| !crate::pipeline::is_terminal_status(&s.status))
})
.count()
})
.unwrap_or(0);
return Some(WaitReason::Children { outstanding });
}
if world.get::<AwaitingInteraction>(entity).is_some() {
// The hub is keyed by agent id, and one agent can only be parked on
// one prompt at a time, so the first match is the one blocking it.
let kind = self
.interactions
.pending()
.into_iter()
.find(|(agent_id, _)| *agent_id == state.agent_id)
.map(|(_, req)| req.kind);
return Some(match kind {
Some(leviath_core::interaction::InteractionKind::ToolApproval) => {
WaitReason::ToolApproval
}
_ => WaitReason::UserPrompt,
});
}
None
}
/// One listing row for a run, read off the live world.
///
/// Shared by [`Self::list`] and by the unload path in [`Self::emit_events`],
/// so a run's last row is built exactly the way every row before it was.
/// Takes the state rather than looking it up because the unload path already
/// holds one, and a `None` it could never return would be a branch nothing
/// can reach.
pub(super) fn entry_for(
&self,
run_id: &str,
entity: Entity,
state: &AgentState,
) -> RunListEntry {
let world = self.world.world();
let metadata = world.get::<RunMetadata>(entity);
let has_output = world
.get::<crate::persistence::FinalOutput>(entity)
.is_some();
RunListEntry {
run_id: run_id.to_string(),
title: metadata.and_then(|m| m.title.clone()),
status: state.status.clone(),
wait_reason: self.wait_reason(crate::world::AgentId::in_world(world, entity)),
stage: state.current_stage.clone(),
stage_index: world
.get::<crate::pipeline::StageCursor>(entity)
.map(|c| c.index),
num_stages: metadata.map(|m| m.num_stages),
iteration: state.iteration,
tool_calls: world.get::<TokenTotals>(entity).map_or(0, |t| t.tool_calls),
last_progress_at: world
.get::<crate::pipeline::PersistWatermark>(entity)
.and_then(|w| w.last_progress_at()),
unattended: metadata.is_some_and(|m| m.unattended),
empty_output: world
.get::<crate::persistence::RunOutcomeFlags>(entity)
.is_some_and(|f| {
// `produced_output` lives on the component only after a
// persist tick fills it, so it is answered from the live
// entity here. Without this, a researcher that submitted a
// perfectly good answer still read `complete (no output)`
// in `lev ps` while `meta.json` said otherwise - the exact
// drift between the two surfaces that one shared
// `is_empty_output` exists to prevent.
let mut flags = f.0.clone();
flags.produced_output = has_output;
crate::persistence::is_empty_output(&state.status, &flags)
}),
read_paths: metadata.and_then(|m| m.read_paths),
has_final_output: has_output,
}
}
/// List every known live run with the context an operator needs to read its
/// status: why it is waiting, where it is, and when it last moved.
pub(super) fn list(&self) -> Vec<RunListEntry> {
let world = self.world.world();
self.by_run_id
.iter()
.filter_map(|(run_id, &agent)| {
let state = world.get::<AgentState>(agent.entity())?;
Some(self.entry_for(run_id, agent.entity(), state))
})
// Parked (paused, paged-out) runs are still the daemon's runs; an
// operator must not lose sight of one just because it left memory.
.chain(self.parked.values().cloned())
.collect()
}
/// The runs unloaded recently enough to still be reported, oldest first.
///
/// Kept apart from [`Self::list`] rather than folded into it because
/// "running now" and "finished a moment ago" are different questions, and
/// two callers already depend on the first one: `lev daemon status` counts
/// the hosted agents, and the dashboard uses the listing to decide which
/// runs the daemon still holds.
pub(super) fn finished(&self) -> Vec<RunListEntry> {
self.finished
.iter()
.map(|(_, entry)| entry.clone())
.collect()
}
/// How long a run stays in the listing after the daemon unloads it. `0`
/// keeps none, which is how the listing behaved before issue #205. Served
/// from `[limits] finished_retention_secs`.
pub fn set_finished_retention_secs(&mut self, secs: u64) {
self.finished_retention_secs = secs;
}
/// Keep `entry` in the listing as a run that finished at `at`.
///
/// One row per run: an id already held is replaced rather than duplicated,
/// so however often a run is unloaded it is reported once.
///
/// `last_progress_at` is filled in from `at` when the run never persisted a
/// snapshot. That is not a guess. A run that died on its first inference has
/// no watermark to read, and the listing would show its age as `-` - which
/// is the one thing an operator or a scheduler most wants to know about a
/// run that failed instantly. For a run being unloaded, the unload is the
/// last thing that happened to it.
pub(super) fn record_finished(&mut self, mut entry: RunListEntry, at: i64) {
if self.finished_retention_secs == 0 {
return;
}
entry.last_progress_at.get_or_insert(at);
self.finished
.retain(|(_, held)| held.run_id != entry.run_id);
self.finished.push_back((at, entry));
while self.finished.len() > MAX_RETAINED_FINISHED {
self.finished.pop_front();
}
}
/// Drop unloaded runs that have outlived the retention window.
///
/// `now` is passed in rather than read here so a test can age the buffer
/// without sleeping through the window, the same reason `lev ps`'s
/// `format_runs` takes it. Called once per [`Self::emit_events`], which the
/// serve loop runs before it handles any control op, so a listing never has
/// to prune on the way out.
pub(super) fn prune_finished(&mut self, now: i64) {
let window = self.finished_retention_secs as i64;
while let Some(&(at, _)) = self.finished.front() {
if now.saturating_sub(at) <= window {
break;
}
self.finished.pop_front();
}
}
}