onepipeline 0.28.3

Execute a task DAG over oneagentgraph and onevcs, merging their event streams into one.
Documentation
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
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
//! The command-line argument surface.
//!
//! Exactly the commands, positionals, and flags `docs/contract.md` lists —
//! parsing only. Nothing here starts, adopts, replies to, attests, stops, or
//! reports on a run; the binary parses one of these and refuses.

// llmlint: ignore-file[invalid_states_unrepresentable] every identifier here is the
// argument `docs/contract.md` spells, and this is the parsing layer only. A `RunId` or
// `NodeRef` newtype would be a public item the contract does not name, and parsing a run
// id or a `run:<id>#<node>` reference into one is the implementation the interface-only
// stage forbids (see AGENTS.md).

use std::path::PathBuf;

use clap::{Args, Parser, Subcommand};

use crate::channel::SurfaceKind;

/// What `--dag-graph` means when it names no graph: no agent graph is launched
/// at all, and deterministic code alone drives the run.
pub const DAG_GRAPH_OFF: &str = "off";

/// The planner-update pacemaker interval, in seconds, when `start` is given
/// none.
pub const DEFAULT_HEARTBEAT_INTERVAL_SECONDS: u64 = 1_800;

/// Execute a task DAG over oneagentgraph and onevcs, merging their event
/// streams into one.
#[derive(Debug, Clone, PartialEq, Eq, Parser)]
#[command(name = "onepipeline", version, about, long_about = None)]
pub struct Cli {
    /// The command to run.
    #[command(subcommand)]
    pub command: Command,
}

/// The top-level commands.
#[derive(Debug, Clone, PartialEq, Eq, Subcommand)]
#[command(rename_all = "kebab-case")]
pub enum Command {
    /// Execute a plan: drive its DAG continuously to settlement.
    Start(StartArgs),
    /// Read a plan without launching it.
    #[command(subcommand)]
    Plan(PlanCommand),
    /// Attach a fresh driver to a run whose ledger is intact.
    Adopt(AdoptArgs),
    /// The channel's server side.
    #[command(subcommand)]
    Channel(ChannelCommand),
    /// Read the next planner surface.
    Next(ReadArgs),
    /// Reply to a surface, with a verdict, graph edits, or both.
    Reply(ReplyArgs),
    /// Raise a surface to the planner.
    Surface(SurfaceArgs),
    /// Complete a ready, waiting human action.
    Attest(AttestArgs),
    /// End a run and its whole dispatch tree.
    Stop(StopArgs),
    /// List recorded runs.
    Runs(RunsArgs),
    /// A run's live state: what is driving it, and what is running.
    Status(OptionalRunArgs),
    /// Every live dispatch on this host, with its owner and load contribution.
    Host,
    /// Stream a run's merged events.
    Monitor(ReadArgs),
    /// Block until a run needs a supervisor, saying so as it waits.
    Watch(WatchArgs),
    /// Which of a session's runs has nothing watching it.
    Unwatched(UnwatchedArgs),
    /// Per-node outcomes, with each node's own evidence.
    Results(RunArgs),
    /// What each run is for, and how far it has got.
    Goals(OptionalRunArgs),
    /// A dispatched turn's tools and reasoning, from the evidence it retained.
    Transcript(TranscriptArgs),
    /// Session timing and usage.
    Telemetry(TelemetryArgs),
    /// Drive one run's engine loop in this process.
    ///
    /// Not part of the documented surface and hidden from `--help`: it is the
    /// process `start --detach` retains, because the loop that drives a run
    /// cannot outlive a launcher that is about to exit. Nothing but this
    /// crate's own launcher spells it.
    #[command(hide = true, name = crate::engine::DRIVE_VERB)]
    DriveRun(DriveRunArgs),
    /// Drive one agent graph in this process, relaying its envelopes as NDJSON.
    ///
    /// Not part of the documented surface and hidden from `--help`: it is how
    /// `start --detach` retains a driver that composes the **same**
    /// `oneagentgraph` an attached launch validates and runs with. Nothing but
    /// this crate's own launcher spells it, and it names no run — it is a graph
    /// and a task, exactly as the sibling's own `run` takes them.
    #[command(hide = true, name = crate::agentgraph::DRIVE_VERB)]
    Drive(DriveArgs),
}

/// What a plan may be asked, short of running it.
#[derive(Debug, Clone, PartialEq, Eq, Subcommand)]
#[command(rename_all = "kebab-case")]
pub enum PlanCommand {
    /// Run the engine's own plan loader, and every registered check, over one
    /// project.
    Check(PlanCheckArgs),
}

/// `onepipeline plan check`.
#[derive(Debug, Clone, PartialEq, Eq, Args)]
pub struct PlanCheckArgs {
    /// The qualified onetaskgraph project id the plan is read from,
    /// `<source>:<native>`, exactly as `start` takes it.
    pub project: String,
    /// One executable to offer the loaded plan to, repeatable and run in the
    /// order the flags are given.
    ///
    /// Resolved against the working directory this command was run from. The
    /// plan crosses its stdin as one JSON document, with
    /// `ONEPIPELINE_PLAN_CHECK_SCHEMA=1` in its environment; it answers on
    /// stdout with `{"refusals": [...]}` and exit 0. Naming none runs the
    /// loader alone.
    #[arg(long = "check", value_name = "PATH")]
    pub checks: Vec<PathBuf>,
    /// Print one JSON object rather than a line per refusal.
    #[arg(long)]
    pub json: bool,
}

/// `onepipeline start`.
#[derive(Debug, Clone, PartialEq, Eq, Args)]
pub struct StartArgs {
    /// The qualified onetaskgraph project id the plan is read from,
    /// `<source>:<native>`.
    ///
    /// Qualified by **source**, so a `local-md` project is launchable directly
    /// with no copy into a remote system first, and nothing special-cases a
    /// remote source.
    pub project: String,
    /// Stay attached, streaming the run's events and returning when it settles.
    /// The default.
    #[arg(long, conflicts_with = "detach")]
    pub attach: bool,
    /// Print the launch record and return, leaving the run unattended.
    #[arg(long)]
    pub detach: bool,
    /// The dag-scope agent graph to attach as an observer, or `off` for none.
    ///
    /// `off` is the shipped default: no agent is required to run a plan. A
    /// graph named here observes the run and authors channel surfaces; it never
    /// drives the engine.
    #[arg(long, value_name = "REF", default_value = DAG_GRAPH_OFF)]
    pub dag_graph: String,
    /// The agent graph a lifecycle node's change request body is drafted by.
    ///
    /// Naming none is the shipped default, exactly as `--dag-graph` defaults to
    /// `off`: this crate ships the flag and not the document, and a launch that
    /// names no graph opens its change requests with the body its plan states,
    /// or with none. Given here it overrides the launch config's own field.
    #[arg(long, value_name = "REF")]
    pub pr_author_graph: Option<String>,
    /// The command every op that introduces or changes a node's task is offered
    /// to before it is applied.
    ///
    /// The node crosses as JSON on its stdin; exit 0 accepts the edit and a
    /// non-zero exit refuses it, with the command's own stderr as the reason.
    /// Naming none is the shipped default and is exactly what a launch did
    /// before this flag existed. Given here it beats `ONEPIPELINE_NODE_VALIDATOR`
    /// and the launch config's own field — including when what it names is
    /// blank, which is this launch saying it has none rather than a fall-through
    /// to the rung below.
    #[arg(long, value_name = "COMMAND")]
    pub node_validator: Option<String>,
    /// The command every reply envelope carrying edits is offered to whole,
    /// after every one of its commands has passed this crate's own validation
    /// and the node validator above.
    ///
    /// One document crosses its stdin: every node the envelope introduces or
    /// changes with the op that produced each, the plan they are being edited
    /// into, and the run's goal. Exit 0 accepts the envelope and a non-zero exit
    /// refuses it whole, with the command's own stderr as the reason and the
    /// node it declared on an `objection: ID` line of that stderr named as the
    /// one it objected to. Naming
    /// none is the shipped default and is exactly what a launch did before this
    /// flag existed. Given here it beats `ONEPIPELINE_ENVELOPE_REVIEWER` and the
    /// launch config's own field — including when what it names is blank, which
    /// is this launch saying it has none rather than a fall-through to the rung
    /// below.
    #[arg(long, value_name = "COMMAND")]
    pub envelope_reviewer: Option<String>,
    /// How often the durable planner-update pacemaker comes due, in seconds.
    #[arg(long, value_name = "SECONDS", default_value_t = DEFAULT_HEARTBEAT_INTERVAL_SECONDS)]
    pub heartbeat_interval: u64,
    /// Override one dag-scope graph config field. Passed opaquely to
    /// `oneagentgraph run`, in command-line order.
    #[arg(long = "set", value_name = "PATH=VALUE")]
    pub dag_sets: Vec<String>,
    /// Override one node-scope graph config field. Passed opaquely to every
    /// node's `oneagentgraph run`, in command-line order.
    #[arg(long = "node-set", value_name = "PATH=VALUE")]
    pub node_sets: Vec<String>,
    /// Proceed even when another live session holds a targeted repository.
    #[arg(long)]
    pub acknowledge_concurrent: bool,
    /// The launch config: what this launch declares about its run, as one
    /// document. Each flag below overrides the part of it that it names.
    #[arg(long, value_name = "FILE")]
    pub launch_config: Option<PathBuf>,
    /// Keep only the events a filter admits out of every `oneagentgraph` launch
    /// this run starts, as a file path or inline JSON.
    #[arg(long, value_name = "SPEC")]
    pub filter_agentgraph: Option<String>,
    /// Keep only the events a filter admits out of every `onevcs` session this
    /// run follows, as a file path or inline JSON.
    #[arg(long, value_name = "SPEC")]
    pub filter_vcs: Option<String>,
    /// Define or override one named read-time profile, as `NAME=SPEC`.
    /// Repeatable. `planner` and `monitor` ship and are overridden by name.
    #[arg(long = "filter-profile", value_name = "NAME=SPEC")]
    pub filter_profiles: Vec<String>,
}

/// `onepipeline adopt`.
///
/// The same attach/detach pair [`StartArgs`] has, with the same default and the
/// same meaning: attached, this process drives the run it took over; detached,
/// the driver it retains does, and the launcher returns once that driver has
/// claimed the run.
#[derive(Debug, Clone, PartialEq, Eq, Args)]
pub struct AdoptArgs {
    /// The run id.
    pub run: String,
    /// Stay attached, driving the adopted run and returning when it settles.
    /// The default.
    #[arg(long, conflicts_with = "detach")]
    pub attach: bool,
    /// Print the launch record and return, leaving the fresh driver unattended.
    #[arg(long)]
    pub detach: bool,
}

/// The flag that tells a retained driver it is taking a run over rather than
/// driving one nothing has driven yet.
///
/// Named here, beside the argument it parses, because the launcher spells it on
/// a command line: a spelling only one side of that knew could drift.
pub(crate) const ADOPT_FLAG: &str = "adopt";

/// `onepipeline drive-run` — the retained driver of a detached launch.
///
/// The run it drives, and whether it is **adopting** it: the bookkeeping an
/// adoption does belongs under the ownership lock, and this is the process that
/// takes that lock, so a detaching adoption hands the work here rather than
/// doing it on behalf of a driver that does not exist yet.
#[derive(Debug, Clone, PartialEq, Eq, Args)]
pub struct DriveRunArgs {
    /// The run id.
    pub run: String,
    /// Take the run over from the driver that had it, recording the adoption
    /// under the lock this process is the one to hold.
    #[arg(long = ADOPT_FLAG)]
    pub adopt: bool,
}

/// A read verb that shapes its event view through a filter profile.
///
/// Naming neither reads through the shipped [`DEFAULT_PROFILE`] — the planner's
/// view, which is what these verbs are for.
///
/// [`DEFAULT_PROFILE`]: crate::filter::DEFAULT_PROFILE
#[derive(Debug, Clone, PartialEq, Eq, Args)]
pub struct ReadArgs {
    /// The run id.
    pub run: String,
    /// The profile to read through: a name this run has, or a filter spec as a
    /// file path or inline JSON.
    #[arg(long, value_name = "NAME|SPEC", conflicts_with = "all")]
    pub filter: Option<String>,
    /// Read every event in the store, through no profile at all.
    #[arg(long)]
    pub all: bool,
}

/// How long a `watch` waits before giving up, when it is given none.
///
/// One supervisory turn: long enough that a watch is worth making rather than a
/// poll, short enough that a caller which meant to look and move on is not held
/// for the length of the run.
pub const DEFAULT_WATCH_TIMEOUT_SECONDS: u64 = 300;

/// How often a `watch` says it is still there while nothing is happening, when
/// it is given none.
pub const DEFAULT_WATCH_TICK_SECONDS: u64 = 30;

/// The cursor-token prefix a `watch` prints and reads back.
///
/// Versioned because the token is a promise to a *later* invocation, which may
/// be a different build: a token this build cannot place is refused by name
/// rather than resumed from as though it meant a byte count.
pub const WATCH_CURSOR_VERSION: &str = "1";

/// The word a caller spells [`WatchTimeout::Unbounded`] with.
///
/// A word rather than a number, and deliberately not `0`: that value's published
/// meaning is to read the run once and return, so spelling "no bound" with it
/// would have made one value mean both the shortest wait there is and the
/// longest.
pub const WATCH_TIMEOUT_UNBOUNDED: &str = "none";

/// How long a `watch` waits before giving up.
///
/// The two spellings answer two different questions, which is why they are not
/// one number. A caller that wants a look at the run and not a wait asks for `0`;
/// a supervisor that wants to be woken when something worth knowing happens asks
/// for [`WATCH_TIMEOUT_UNBOUNDED`] and is not woken by a clock at all. Every
/// number in between bounds the wait as it always did.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum WatchTimeout {
    /// Give up after this many seconds. `0` reads once and returns.
    Bounded(u64),
    /// Never give up on the clock: the wait ends on a condition or not at all.
    Unbounded,
}

impl std::fmt::Display for WatchTimeout {
    fn fmt(&self, out: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::Bounded(seconds) => write!(out, "{seconds}"),
            Self::Unbounded => out.write_str(WATCH_TIMEOUT_UNBOUNDED),
        }
    }
}

impl std::str::FromStr for WatchTimeout {
    type Err = String;

    fn from_str(text: &str) -> std::result::Result<Self, Self::Err> {
        if text == WATCH_TIMEOUT_UNBOUNDED {
            return Ok(Self::Unbounded);
        }
        text.parse().map(Self::Bounded).map_err(|_| {
            format!(
                "'{text}' is not a wait this verb can take: a wait is a number of seconds, \
                 where `0` reads the run once and returns, or `{WATCH_TIMEOUT_UNBOUNDED}`, \
                 which does not bound the wait at all"
            )
        })
    }
}

/// Every condition `--until` accepts: how a caller spells it, and what this
/// build reads it as.
///
/// **One table, read by the parser and by the refusal both**, so the message a
/// mistyped condition gets cannot omit a condition this build accepts — the
/// accepted set *is* this list, rather than a second copy of it standing beside
/// a match. The README and the divergence record are reconciled against the
/// spellings here too, so what a supervisor is told to type is what the parser
/// reads.
const CONDITIONS: [(&str, WatchUntil); 5] = [
    ("settled", WatchUntil::Settled),
    ("surface", WatchUntil::Surface),
    ("nothing-driving", WatchUntil::NothingDriving),
    ("node-settled", WatchUntil::NodeSettled),
    // The one entry that is a **shape** rather than a word: what follows the
    // prefix is a node id the caller supplies, so the parser reads this row by
    // its prefix and builds the condition from the text after it. The value
    // beside it is never returned.
    (WATCH_NODE_CONDITION_SHAPE, WatchUntil::Node(String::new())),
];

/// How a caller names one node to return on, with the id it stands in for.
pub const WATCH_NODE_CONDITION_SHAPE: &str = "node=<ID>";

/// What that shape stands in for: the id follows this.
const NODE_ID_PLACEHOLDER: &str = "<ID>";

/// Every condition `--until` accepts, spelled as a caller types it.
///
/// Derived from the one table above rather than restated, so a spelling this
/// build reads is a spelling it names.
pub fn watch_conditions() -> [&'static str; 5] {
    CONDITIONS.map(|(spelling, _)| spelling)
}

/// What ends a `watch`, as a caller names it.
///
/// **Repeatable, and additive to what the verb always returns on.** A run that
/// settles `complete` and a run nothing is driving end every wait whether or not
/// they were asked for, because a wait that could outlive the run it watches is
/// the unbounded silence this verb exists to end — so [`Settled`](Self::Settled)
/// and [`NothingDriving`](Self::NothingDriving) name conditions rather than
/// switch them on, and what `--until settled` *adds* is nothing, which is why it
/// still means "do not return on a blocking surface".
///
/// [`Surface`](Self::Surface) is spelled for the surface rather than for a
/// "decision", which in this crate is the wider fact `status` reports: a ready
/// human action is a decision point too, and a value that claimed to cover both
/// while returning on one of them would be the prose-shaped promise this verb
/// exists to replace.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub enum WatchUntil {
    /// Return on a blocking surface waiting to be answered. The default, and on
    /// its own it is the pair a supervisor answers: this or the run finishing.
    #[default]
    Surface,
    /// Return when the run finishes. Named on its own it adds nothing, so it is
    /// still how a caller says a blocking surface should be reported and waited
    /// through rather than returned on.
    Settled,
    /// Return when nothing is driving the run. Always returned on, named here so
    /// a caller can spell the whole vocabulary.
    NothingDriving,
    /// Return when any node of the run settles.
    NodeSettled,
    /// Return when this node of the run settles. Validated against the run's own
    /// graph when the command is invoked.
    Node(String),
}

impl std::fmt::Display for WatchUntil {
    fn fmt(&self, out: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::Surface => out.write_str("surface"),
            Self::Settled => out.write_str("settled"),
            Self::NothingDriving => out.write_str("nothing-driving"),
            Self::NodeSettled => out.write_str("node-settled"),
            Self::Node(node) => write!(
                out,
                "{}{node}",
                WATCH_NODE_CONDITION_SHAPE
                    .strip_suffix(NODE_ID_PLACEHOLDER)
                    .unwrap_or(WATCH_NODE_CONDITION_SHAPE)
            ),
        }
    }
}

impl std::str::FromStr for WatchUntil {
    type Err = String;

    /// Read one condition, or refuse it naming the vocabulary.
    ///
    /// The refusal is made by the parser, so it happens as the command line is
    /// read: nothing is streamed and nothing waits behind a condition this verb
    /// does not have. What it cannot judge here is a condition that is *spelled*
    /// right and names a node this run does not hold — that one needs the run's
    /// graph, and `src/watch.rs` refuses it before it blocks.
    fn from_str(text: &str) -> std::result::Result<Self, Self::Err> {
        for (spelling, condition) in CONDITIONS {
            match spelling.strip_suffix(NODE_ID_PLACEHOLDER) {
                // A row that stands for a shape matches on its prefix, and the
                // id is what the caller wrote after it. An empty one names no
                // node, so it falls through to the refusal rather than becoming
                // a condition about a node with no name.
                Some(prefix) => {
                    if let Some(node) = text.strip_prefix(prefix).filter(|node| !node.is_empty()) {
                        return Ok(Self::Node(node.to_string()));
                    }
                }
                None if text == spelling => return Ok(condition),
                None => {}
            }
        }
        Err(format!(
            "'{text}' is not a condition this verb returns on; it returns on {}",
            watch_conditions().join(", ")
        ))
    }
}

/// The streaming verb's own run and profile selection, plus the four things a
/// supervisor needs to write no wake loop at all: a bound on the wait — or none
/// — a heartbeat so silence and death are tellable apart, a cursor to resume
/// from, and the conditions it returns on.
#[derive(Debug, Clone, PartialEq, Eq, Args)]
pub struct WatchArgs {
    /// The run, and the profile its event view is shaped through — exactly as
    /// `monitor` takes them.
    #[command(flatten)]
    pub read: ReadArgs,
    /// How long to wait before giving up, in seconds. `0` reads once and
    /// returns; `none` does not bound the wait at all.
    #[arg(long, value_name = "SECONDS|none", default_value_t = WatchTimeout::Bounded(DEFAULT_WATCH_TIMEOUT_SECONDS))]
    pub timeout: WatchTimeout,
    /// How long a silence may last before this stream says it is still there,
    /// in seconds. `0` turns the heartbeat off.
    ///
    /// **The stream's** interval, and deliberately not spelled
    /// `--heartbeat-interval`: that one is `start`'s, and it sets how often the
    /// pacemaker agent surfaces a planner update. The two are different clocks
    /// on different verbs, and neither verb accepts the other's flag.
    #[arg(long, value_name = "SECONDS", default_value_t = DEFAULT_WATCH_TICK_SECONDS)]
    pub tick_interval: u64,
    /// Resume from the cursor an earlier watch printed, emitting nothing it
    /// already did.
    #[arg(long, value_name = "CURSOR")]
    pub cursor: Option<String>,
    /// What ends the wait, beside the run finishing and nothing driving it.
    /// Repeatable: the wait returns on the first of them that fires, and says
    /// which one did.
    #[arg(long, value_name = "CONDITION", default_values_t = [WatchUntil::Surface])]
    pub until: Vec<WatchUntil>,
}

/// `onepipeline unwatched`.
///
/// The session is taken as an option **as well as** from the environment, and
/// that is the whole of the argument surface. Its consumer is a hook that is
/// handed the session it must ask about on standard input, while the environment
/// it runs in carries somebody else's — so a verb that could only read the
/// environment would answer confidently about the wrong session.
#[derive(Debug, Clone, PartialEq, Eq, Args)]
pub struct UnwatchedArgs {
    /// The launching session whose runs to ask about. Omitted, the session
    /// `ONEPIPELINE_LAUNCHER_SESSION` names.
    #[arg(long, value_name = "ID")]
    pub session: Option<String>,
}

/// `onepipeline drive` — the retained driver a detached launch starts.
///
/// The arguments a graph run needs and no more, spelled as `oneagentgraph run`
/// spells them: this is the same launch, made by this build's own copy of that
/// library rather than by whichever one the host has installed.
#[derive(Debug, Clone, PartialEq, Eq, Args)]
pub struct DriveArgs {
    /// The agent-graph config to run.
    pub graph: String,
    /// The task prose every member without its own is given.
    #[arg(long, value_name = "TEXT")]
    pub task: String,
    /// The directory the graph's members work in.
    #[arg(long, value_name = "DIR")]
    pub dir: PathBuf,
    /// One `k=v` label stamped on every envelope, repeatable.
    #[arg(long = "label", value_name = "KEY=VALUE")]
    pub labels: Vec<String>,
    /// One opaque graph-config override, repeatable, applied in order.
    #[arg(long = "set", value_name = "PATH=VALUE")]
    pub sets: Vec<String>,
    /// The source filter this launch relays through, inline as JSON. Spelled as
    /// `oneagentgraph run` spells it, because an overridden binary is what
    /// receives it.
    #[arg(long, value_name = "SPEC")]
    pub event_filter: Option<String>,
    /// Hold this process open until the graph's own run record carries the
    /// ending it stamps *after* announcing its settlement.
    ///
    /// What the launcher of an observer passes, because it reads that record
    /// rather than the envelopes this process relays — the engine's `Ending`
    /// is where that is written down. Named in plain code rather than linked,
    /// because the engine is private and this struct is public, and rustdoc
    /// refuses that link under this repository's denied warnings. Spelled by
    /// `retained_command` and by nobody else, so it is not an operator-facing
    /// flag; it is on this hidden verb's argv because that argv is how a
    /// retained launch is told anything.
    #[arg(long)]
    pub await_ending: bool,
}

/// The channel's server side.
#[derive(Debug, Clone, PartialEq, Eq, Subcommand)]
#[command(rename_all = "kebab-case")]
pub enum ChannelCommand {
    /// Serve the channel as an observer member's judge-side command provider.
    Serve(RunArgs),
}

/// A command that names one run and nothing else.
#[derive(Debug, Clone, PartialEq, Eq, Args)]
pub struct RunArgs {
    /// The run id.
    pub run: String,
}

/// A view that defaults to every run when given none.
#[derive(Debug, Clone, PartialEq, Eq, Args)]
pub struct OptionalRunArgs {
    /// The run id. Omitted, the view covers every run.
    pub run: Option<String>,
}

/// `onepipeline reply`.
#[derive(Debug, Clone, PartialEq, Eq, Args)]
pub struct ReplyArgs {
    /// The run id.
    pub run: String,
    /// The reply envelope. Omitted, it is read from stdin.
    pub file: Option<PathBuf>,
}

/// `onepipeline surface`.
///
/// The message body arrives the way [`ReplyArgs`]'s envelope does — from a file,
/// or from stdin when none is named — so agent-authored prose never has to pass
/// through a shell. Divergence 38 records why. `--message` still works, and is
/// refused beside a file.
#[derive(Debug, Clone, PartialEq, Eq, Args)]
pub struct SurfaceArgs {
    /// The run id.
    pub run: String,
    /// The file the surface's text is read from. Omitted, it is read from
    /// stdin — unless `--message` carried it.
    #[arg(conflicts_with = "message")]
    pub file: Option<PathBuf>,
    /// What the surface is asking about.
    #[arg(long, value_enum)]
    pub kind: SurfaceKind,
    /// The surface's text, inline. Prefer the file or the stdin form: whatever
    /// is written here is read by a shell first.
    #[arg(long, value_name = "TEXT")]
    pub message: Option<String>,
}

/// `onepipeline attest`.
#[derive(Debug, Clone, PartialEq, Eq, Args)]
pub struct AttestArgs {
    /// The run id.
    pub run: String,
    /// The human action's reference.
    pub reference: String,
}

/// `onepipeline stop`.
#[derive(Debug, Clone, PartialEq, Eq, Args)]
pub struct StopArgs {
    /// The run id.
    pub run: String,
    /// Stop a run this session does not own. The owner is named either way.
    #[arg(long)]
    pub force: bool,
}

/// `onepipeline runs`.
#[derive(Debug, Clone, PartialEq, Eq, Args)]
pub struct RunsArgs {
    /// List only the runs this session launched.
    #[arg(long)]
    pub mine: bool,
}

/// `onepipeline transcript`.
#[derive(Debug, Clone, PartialEq, Eq, Args)]
pub struct TranscriptArgs {
    /// The run id.
    pub run: String,
    /// The node whose transcript to read. Omitted, every node that dispatched.
    pub node: Option<String>,
}

/// `onepipeline telemetry`.
#[derive(Debug, Clone, PartialEq, Eq, Args)]
pub struct TelemetryArgs {
    /// The run id. Omitted, the view covers every run.
    pub run: Option<String>,
    /// Break the wall clock down into buckets that sum exactly.
    #[arg(long)]
    pub breakdown: bool,
}