subc_control/
activity.rs

1//! Additive activity ledger bodies, dispatched separately from the exhaustive
2//! client control enums by their `op` tag.
3
4use std::path::PathBuf;
5
6use serde::{Deserialize, Deserializer, Serialize};
7use serde_json::{Map, Value};
8use subc_protocol::Principal;
9
10use crate::ops;
11
12/// Advertised in HELLO_ACK and server.describe before activity operations are used.
13pub const ACTIVITY_LEDGER_CAPABILITY: &str = "activity-ledger/v1";
14
15pub const ACTIVITY_KIND_FORBIDDEN: &str = "activity_kind_forbidden";
16pub const ACTIVITY_ACTOR_UNKNOWN: &str = "activity_actor_unknown";
17pub const ACTIVITY_RATE_LIMITED: &str = "activity_rate_limited";
18pub const ACTIVITY_READ_FORBIDDEN: &str = "activity_read_forbidden";
19pub const ACTIVITY_TOO_LARGE: &str = "activity_too_large";
20pub const ACTIVITY_MALFORMED: &str = "activity_malformed";
21pub const ACTIVITY_UNAVAILABLE: &str = "activity_unavailable";
22pub const ACTIVITY_CURSOR_INCARNATION_MISMATCH: &str = "activity_cursor_incarnation_mismatch";
23pub const ACTIVITY_CURSOR_TOO_OLD: &str = "activity_cursor_too_old";
24pub const ACTIVITY_SUBSCRIBER_LAGGED: &str = "activity_subscriber_lagged";
25
26/// A live-feed position. Query continuation positions are sequence numbers only.
27#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
28pub struct ActivityCursor {
29    pub daemon_incarnation: String,
30    pub seq: u64,
31}
32
33/// A short typed pointer, never a secret value.
34#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
35#[serde(deny_unknown_fields)]
36pub struct ActivitySubject {
37    #[serde(rename = "type")]
38    pub subject_type: String,
39    pub id: String,
40}
41
42/// Verified identity stamped by the daemon, not supplied by a recording module.
43#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
44pub struct ActivityActor {
45    pub principal: Principal,
46    #[serde(default, skip_serializing_if = "Option::is_none")]
47    pub scope_owner: Option<Principal>,
48    #[serde(default, skip_serializing_if = "Option::is_none")]
49    pub agent_id: Option<String>,
50    #[serde(default, skip_serializing_if = "Option::is_none")]
51    pub run_id: Option<String>,
52    #[serde(default, skip_serializing_if = "Option::is_none")]
53    pub flow_id: Option<String>,
54    #[serde(default, skip_serializing_if = "Option::is_none")]
55    pub project_root: Option<PathBuf>,
56}
57
58/// One durably appended event. `prev` is reserved for a future hash chain and is
59/// deliberately absent from this format. Identity is (daemon_incarnation, seq),
60/// and entries are ordered by seq rather than by their wall-clock timestamps.
61#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
62pub struct ActivityEntry {
63    pub seq: u64,
64    pub at_ms: u64,
65    pub daemon_incarnation: String,
66    pub module: String,
67    pub kind: String,
68    pub summary: String,
69    #[serde(default, skip_serializing_if = "Option::is_none")]
70    pub subject: Option<ActivitySubject>,
71    pub details: Map<String, Value>,
72    pub actor: ActivityActor,
73}
74
75/// A route binding owned by the recording module, checked at this epoch.
76#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
77#[serde(deny_unknown_fields)]
78pub struct ActivityRoute {
79    pub channel: u16,
80    pub epoch: u32,
81}
82
83/// A live scope owned by the recording module, checked at this epoch.
84#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
85#[serde(deny_unknown_fields)]
86pub struct ActivityScope {
87    #[serde(rename = "ref")]
88    pub scope_ref: String,
89    pub epoch: u64,
90}
91
92/// Exactly one citation from which the daemon resolves the actor.
93#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
94#[serde(rename_all = "snake_case")]
95pub enum ActivityOn {
96    Route(ActivityRoute),
97    Scope(ActivityScope),
98}
99
100// Serde's struct `tag` adds an op during serialization but does not validate it
101// during deserialization. A private tagged enum validates the op without adding
102// variants to the public client control enums or exposing an editable op field.
103macro_rules! activity_body {
104    ($(#[$doc:meta])* $name:ident, $op:path, $wire_name:literal,
105     $(#[$wire_meta:meta])* { $( $(#[$field_meta:meta])* pub $field:ident: $ty:ty, )* }) => {
106        $(#[$doc])*
107        #[derive(Debug, Clone, Serialize, PartialEq)]
108        #[serde(tag = "op", rename = $wire_name)]
109        pub struct $name {
110            $( $(#[$field_meta])* pub $field: $ty, )*
111        }
112
113        impl $name {
114            pub const OP: &'static str = $op;
115        }
116
117        impl<'de> Deserialize<'de> for $name {
118            fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
119            where D: Deserializer<'de> {
120                #[derive(Deserialize)]
121                #[serde(tag = "op")]
122                $(#[$wire_meta])*
123                enum Wire {
124                    #[serde(rename = $wire_name)]
125                    Body { $( $(#[$field_meta])* $field: $ty, )* },
126                }
127                let Wire::Body { $( $field, )* } = Wire::deserialize(deserializer)?;
128                Ok(Self { $( $field, )* })
129            }
130        }
131    };
132}
133
134fn deserialize_details<'de, D>(deserializer: D) -> Result<Option<Map<String, Value>>, D::Error>
135where
136    D: Deserializer<'de>,
137{
138    Map::deserialize(deserializer).map(Some)
139}
140
141activity_body! {
142/// Report an event. Shape checking lives here; kind, size, ownership and rate
143/// validation remain the daemon's responsibility. There is no actor field.
144ActivityRecordRequest, ops::ACTIVITY_RECORD, "activity.record",
145#[serde(deny_unknown_fields)] {
146    pub kind: String,
147    pub summary: String,
148    #[serde(default, skip_serializing_if = "Option::is_none")]
149    pub subject: Option<ActivitySubject>,
150    #[serde(default, skip_serializing_if = "Option::is_none")]
151    #[serde(deserialize_with = "deserialize_details")]
152    pub details: Option<Map<String, Value>>,
153    #[serde(default, skip_serializing_if = "Option::is_none")]
154    pub on: Option<ActivityOn>,
155}
156
157}
158
159activity_body! {
160/// Sent only after the assigned sequence has been durably appended.
161ActivityRecordResponse, ops::ACTIVITY_RECORD, "activity.record", {
162    pub seq: u64,
163}
164
165}
166
167activity_body! {
168/// Filter a newest-first scan. `cursor` is exclusive: only lower seqs follow it.
169ActivityQueryRequest, ops::ACTIVITY_QUERY, "activity.query", {
170    #[serde(default, skip_serializing_if = "Option::is_none")]
171    pub agent: Option<String>,
172    #[serde(default, skip_serializing_if = "Option::is_none")]
173    pub project: Option<PathBuf>,
174    #[serde(default, skip_serializing_if = "Option::is_none")]
175    pub module: Option<String>,
176    #[serde(default, skip_serializing_if = "Option::is_none")]
177    pub kind: Option<String>,
178    #[serde(default, skip_serializing_if = "Option::is_none")]
179    pub since_ms: Option<u64>,
180    #[serde(default, skip_serializing_if = "Option::is_none")]
181    pub until_ms: Option<u64>,
182    pub limit: u32,
183    #[serde(default, skip_serializing_if = "Option::is_none")]
184    pub cursor: Option<u64>,
185}
186
187}
188
189activity_body! {
190/// A page plus the current incarnation's head at scan capture, even on an empty
191/// page. Subscribe from head to move from the scan to live entries without a gap.
192ActivityQueryResponse, ops::ACTIVITY_QUERY, "activity.query", {
193    pub entries: Vec<ActivityEntry>,
194    #[serde(default, skip_serializing_if = "Option::is_none")]
195    pub cursor: Option<u64>,
196    pub head: ActivityCursor,
197}
198
199}
200
201activity_body! {
202/// Replay after since and remain open for live entries. A refused or lagged feed
203/// ends with an Error frame carrying one of the activity refusal-code constants.
204ActivitySubscribeRequest, ops::ACTIVITY_SUBSCRIBE, "activity.subscribe", {
205    #[serde(default, skip_serializing_if = "Option::is_none")]
206    pub since: Option<ActivityCursor>,
207}
208
209}
210
211activity_body! {
212/// One successful append delivered on the subscription's correlation id. There
213/// is no separate acknowledgement; each response is one streamed entry.
214ActivitySubscribeResponse, ops::ACTIVITY_SUBSCRIBE, "activity.subscribe", {
215    pub entry: ActivityEntry,
216}
217
218}
subc_control/activity.rs

subc_control/
activity.rs
