kanade_shared/
subject.rs

1pub const COMMANDS_ALL: &str = "commands.all";
2
3pub fn commands_group(name: &str) -> String {
4    format!("commands.group.{name}")
5}
6
7pub fn commands_pc(pc_id: &str) -> String {
8    format!("commands.pc.{pc_id}")
9}
10
11// `commands_exec` (subject `commands.exec.<job_id>`) was removed in
12// v0.22.1. The STREAM_EXEC stream now catches the existing
13// `commands.{all,group.X,pc.Y}` subjects directly, so the dedicated
14// per-exec subject isn't needed any more. See
15// `kanade-agent::command_replay` for how reconnecting agents catch
16// up on missed messages.
17
18pub fn results(request_id: &str) -> String {
19    format!("results.{request_id}")
20}
21
22pub fn heartbeat(pc_id: &str) -> String {
23    format!("heartbeat.{pc_id}")
24}
25
26/// `kill.<exec_id>` — Spec §2.6 Layer 3 abort signal. The exec_id is
27/// the deployment / scheduler-fire UUID (formerly named `job_id`
28/// pre-v0.29; renamed for accuracy — every `Command.exec_id` is a
29/// per-deploy UUID, not a job-catalog id).
30pub fn kill(exec_id: &str) -> String {
31    format!("kill.{exec_id}")
32}
33
34pub fn inventory(pc_id: &str, category: &str) -> String {
35    format!("inventory.{pc_id}.{category}")
36}
37
38/// `events.started.<exec_id>.<pc_id>` — v0.30 / PR α' lifecycle
39/// event published by the agent just before spawning a script's
40/// child process. Lets the backend project an in-flight row into
41/// `execution_results` (with `finished_at = NULL`) so the SPA
42/// Activity table can show running rows alongside finished ones.
43/// Backend subscribes via [`EVENTS_STARTED_FILTER`].
44pub fn events_started(exec_id: &str, pc_id: &str) -> String {
45    format!("events.started.{exec_id}.{pc_id}")
46}
47
48/// Wildcard the backend events projector consumes on STREAM_EVENTS.
49/// Narrow (`events.started.>`) rather than the whole `events.>` so
50/// future event types can carry their own filters without rerouting
51/// the started subset.
52pub const EVENTS_STARTED_FILTER: &str = "events.started.>";
53
54pub const INVENTORY_HW: &str = "hw";
55pub const INVENTORY_SW: &str = "sw";
56pub const INVENTORY_NET: &str = "net";
57
58/// `logs.fetch.<pc_id>` — request/reply: operator (or backend) sends
59/// a `LogsRequest`; the addressed agent replies with the tail of its
60/// local log file. On-demand only, no stream.
61pub fn logs_fetch(pc_id: &str) -> String {
62    format!("logs.fetch.{pc_id}")
63}
64
65/// `agents.<pc_id>.ping` — v0.38 / #133 request/reply for the
66/// active "ping" round-trip. The agent answers with a fresh
67/// `Heartbeat` on demand instead of the backend waiting up to ~30 s
68/// for the next periodic heartbeat tick to land. Distinct subject
69/// from `heartbeat.<pc_id>` so the periodic publisher is unaffected
70/// and old agents that don't subscribe simply time the request out.
71pub fn ping(pc_id: &str) -> String {
72    format!("agents.{pc_id}.ping")
73}
74
75// v0.14: subject::inventory_request was retired alongside the
76// hardcoded inventory loop. On-demand collection now goes through
77// the normal exec path (`kanade exec configs/jobs/inventory-
78// hw.yaml`) — Command + ExecResult + the inventory-fact projector
79// give operators the same effect with no extra subject.
80
81#[cfg(test)]
82mod tests {
83    use super::*;
84
85    #[test]
86    fn commands_all_constant() {
87        assert_eq!(COMMANDS_ALL, "commands.all");
88    }
89
90    #[test]
91    fn commands_group_formats_name() {
92        assert_eq!(commands_group("canary"), "commands.group.canary");
93        assert_eq!(commands_group("wave1"), "commands.group.wave1");
94    }
95
96    #[test]
97    fn commands_pc_formats_id() {
98        assert_eq!(commands_pc("minipc"), "commands.pc.minipc");
99        assert_eq!(commands_pc("PC1234"), "commands.pc.PC1234");
100    }
101
102    #[test]
103    fn results_formats_request_id() {
104        assert_eq!(results("req-1"), "results.req-1");
105    }
106
107    #[test]
108    fn heartbeat_formats_pc_id() {
109        assert_eq!(heartbeat("minipc"), "heartbeat.minipc");
110    }
111
112    #[test]
113    fn kill_formats_exec_id() {
114        assert_eq!(kill("exec-uuid-1"), "kill.exec-uuid-1");
115    }
116
117    #[test]
118    fn logs_fetch_formats_pc_id() {
119        assert_eq!(logs_fetch("minipc"), "logs.fetch.minipc");
120    }
121
122    #[test]
123    fn ping_formats_pc_id() {
124        assert_eq!(ping("minipc"), "agents.minipc.ping");
125    }
126
127    #[test]
128    fn events_started_formats_exec_id_and_pc_id() {
129        assert_eq!(
130            events_started("exec-uuid-1", "minipc"),
131            "events.started.exec-uuid-1.minipc",
132        );
133    }
134
135    #[test]
136    fn events_started_filter_is_narrow_wildcard() {
137        assert_eq!(EVENTS_STARTED_FILTER, "events.started.>");
138    }
139
140    #[test]
141    fn inventory_formats_pc_id_and_category() {
142        assert_eq!(inventory("minipc", "hw"), "inventory.minipc.hw");
143        assert_eq!(inventory("minipc", INVENTORY_HW), "inventory.minipc.hw");
144        assert_eq!(inventory("minipc", INVENTORY_SW), "inventory.minipc.sw");
145        assert_eq!(inventory("minipc", INVENTORY_NET), "inventory.minipc.net");
146    }
147}
kanade_shared/subject.rs

kanade_shared/
subject.rs
