talon_core/
protocol.rs

1//! Host <-> container wire protocol.
2//!
3//! ```text
4//!   host (talon)                                container (talon-agent)
5//!   ────────────                                ───────────────────────
6//!   stdin  ── ContainerInput JSON, then EOF ──▶ read once at start
7//!   stdout ◀─ OUTPUT_START / ContainerOutput / OUTPUT_END ── (repeatable)
8//!   stderr ◀─ free-form log lines (tracing)
9//!
10//!   data/ipc/<ipc_key>/input/*.json   ──▶ /workspace/ipc/input/   (follow-ups)
11//!   data/ipc/<ipc_key>/input/_close   ──▶ end of session sentinel
12//!   data/ipc/<ipc_key>/messages/*.json ◀── /workspace/ipc/messages/ (send_message)
13//!   data/ipc/<ipc_key>/tasks/*.json    ◀── /workspace/ipc/tasks/    (task ops)
14//!   data/ipc/<ipc_key>/current_tasks.json   ──▶ snapshot (read-only for agent)
15//!   data/ipc/<ipc_key>/available_groups.json──▶ snapshot (read-only for agent)
16//! ```
17//!
18//! Every IPC file is written atomically (`<name>.json.tmp` then rename) and
19//! named `<unix_millis>-<random6>.json` so lexical order == arrival order.
20//! The host derives the *sender identity* from the directory the file was
21//! found in, never from the file contents.
22
23use serde::{Deserialize, Serialize};
24
25use crate::types::{ContainerConfig, ContextMode, ScheduleType};
26
27/// Marks the start of one JSON-encoded [`ContainerOutput`] on stdout.
28pub const OUTPUT_START_MARKER: &str = "---TALON_OUTPUT_START---";
29/// Marks the end of one JSON-encoded [`ContainerOutput`] on stdout.
30pub const OUTPUT_END_MARKER: &str = "---TALON_OUTPUT_END---";
31
32/// Name of the sentinel file that ends a container session.
33pub const IPC_CLOSE_SENTINEL: &str = "_close";
34
35/// Well-known paths inside the agent container.
36pub mod container_paths {
37    /// The group's folder (CLAUDE.md, prompts/, conversations/, logs/). cwd.
38    pub const GROUP: &str = "/workspace/group";
39    /// Shared read-only global memory (groups/global).
40    pub const GLOBAL: &str = "/workspace/global";
41    /// The project root (main group only, read-only, `.env` shadowed).
42    pub const PROJECT: &str = "/workspace/project";
43    /// Additional allowlisted mounts live under here.
44    pub const EXTRA: &str = "/workspace/extra";
45    /// IPC namespace root.
46    pub const IPC: &str = "/workspace/ipc";
47    pub const IPC_INPUT: &str = "/workspace/ipc/input";
48    pub const IPC_MESSAGES: &str = "/workspace/ipc/messages";
49    pub const IPC_TASKS: &str = "/workspace/ipc/tasks";
50    pub const CURRENT_TASKS: &str = "/workspace/ipc/current_tasks.json";
51    pub const AVAILABLE_GROUPS: &str = "/workspace/ipc/available_groups.json";
52    /// Per-lane persistent agent state (sessions/transcripts). Mounted from
53    /// `data/sessions/<session_key>/`.
54    pub const AGENT_STATE: &str = "/home/agent/.talon";
55    /// Per-group MCP server definitions.
56    pub const MCP_JSON: &str = "/workspace/group/.mcp.json";
57}
58
59/// Everything the in-container runtime needs to reach the model.
60///
61/// Delivered over **stdin only** (never env vars), so the token is not visible
62/// in `docker inspect` or `/proc/1/environ` and cannot be inherited by tool
63/// subprocesses.
64#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
65#[serde(rename_all = "camelCase")]
66pub struct LlmEndpoint {
67    /// OpenAI-compatible base URL **including** `/v1`, normally the host LLM
68    /// gateway, e.g. `http://host.docker.internal:3001/v1`.
69    pub base_url: String,
70    /// Bearer token for `base_url` (a per-run gateway token).
71    pub token: String,
72    /// Primary model name.
73    pub model: String,
74    /// Auxiliary model for compaction / extraction. Falls back to `model`.
75    #[serde(default, skip_serializing_if = "Option::is_none")]
76    pub fast_model: Option<String>,
77    /// Context window of `model`, tokens.
78    pub max_context_tokens: u32,
79    #[serde(default, skip_serializing_if = "Option::is_none")]
80    pub max_output_tokens: Option<u32>,
81    #[serde(default, skip_serializing_if = "Option::is_none")]
82    pub temperature: Option<f32>,
83    #[serde(default, skip_serializing_if = "Option::is_none")]
84    pub top_p: Option<f32>,
85    #[serde(default, skip_serializing_if = "Option::is_none")]
86    pub enable_thinking: Option<bool>,
87    #[serde(default, skip_serializing_if = "Option::is_none")]
88    pub request_timeout_secs: Option<u64>,
89}
90
91/// Written by the host to the container's stdin, then stdin is closed.
92#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
93#[serde(rename_all = "camelCase")]
94pub struct ContainerInput {
95    pub prompt: String,
96    /// Session to resume. `None` = fresh session.
97    #[serde(default, skip_serializing_if = "Option::is_none")]
98    pub session_id: Option<String>,
99    pub group_folder: String,
100    pub chat_jid: String,
101    pub is_main: bool,
102    #[serde(default)]
103    pub is_scheduled_task: bool,
104    #[serde(default, skip_serializing_if = "Option::is_none")]
105    pub assistant_name: Option<String>,
106    /// Optional pre-flight bash script for scheduled tasks. Must print
107    /// `{"wakeAgent": bool, "data"?: any}` as its last stdout line.
108    #[serde(default, skip_serializing_if = "Option::is_none")]
109    pub script: Option<String>,
110    /// Model endpoint. `None` only in tests / mock mode.
111    #[serde(default, skip_serializing_if = "Option::is_none")]
112    pub llm: Option<LlmEndpoint>,
113    /// Do not persist the session transcript (investigation lanes).
114    #[serde(default)]
115    pub ephemeral: bool,
116    /// IANA timezone of the host, for timestamps shown to the model.
117    #[serde(default, skip_serializing_if = "Option::is_none")]
118    pub timezone: Option<String>,
119}
120
121#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
122#[serde(rename_all = "lowercase")]
123pub enum OutputStatus {
124    Success,
125    Error,
126}
127
128/// Token usage for one agent turn. Field names follow the Anthropic shape the
129/// host already stores; the vLLM runtime maps
130/// `prompt_tokens - cached_tokens -> input_tokens`,
131/// `prompt_tokens_details.cached_tokens -> cache_read_input_tokens`,
132/// `completion_tokens -> output_tokens`.
133#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
134pub struct SessionUsage {
135    #[serde(default, skip_serializing_if = "Option::is_none")]
136    pub input_tokens: Option<u64>,
137    #[serde(default, skip_serializing_if = "Option::is_none")]
138    pub output_tokens: Option<u64>,
139    #[serde(default, skip_serializing_if = "Option::is_none")]
140    pub cache_creation_input_tokens: Option<u64>,
141    #[serde(default, skip_serializing_if = "Option::is_none")]
142    pub cache_read_input_tokens: Option<u64>,
143}
144
145impl SessionUsage {
146    /// Total context size of the turn (what grows when a session is resumed).
147    pub fn context_size(&self) -> u64 {
148        self.input_tokens.unwrap_or(0)
149            + self.cache_creation_input_tokens.unwrap_or(0)
150            + self.cache_read_input_tokens.unwrap_or(0)
151    }
152}
153
154/// One result block emitted by the container between the output markers.
155#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
156#[serde(rename_all = "camelCase")]
157pub struct ContainerOutput {
158    pub status: OutputStatus,
159    pub result: Option<String>,
160    #[serde(default, skip_serializing_if = "Option::is_none")]
161    pub new_session_id: Option<String>,
162    #[serde(default, skip_serializing_if = "Option::is_none")]
163    pub error: Option<String>,
164    #[serde(default, skip_serializing_if = "Option::is_none")]
165    pub usage: Option<SessionUsage>,
166    /// Success frames of agents that report it: the names of the IPC input
167    /// files (`input/*.json`) whose follow-up messages went into the turn
168    /// this frame ends. The host uses them to tell answered follow-ups from
169    /// ones that still need a run. `None` from agents that predate it.
170    #[serde(default, skip_serializing_if = "Option::is_none")]
171    pub consumed_inputs: Option<Vec<String>>,
172}
173
174impl ContainerOutput {
175    pub fn success(result: Option<String>, new_session_id: Option<String>) -> Self {
176        Self {
177            status: OutputStatus::Success,
178            result,
179            new_session_id,
180            error: None,
181            usage: None,
182            consumed_inputs: None,
183        }
184    }
185
186    pub fn error(message: impl Into<String>) -> Self {
187        Self {
188            status: OutputStatus::Error,
189            result: None,
190            new_session_id: None,
191            error: Some(message.into()),
192            usage: None,
193            consumed_inputs: None,
194        }
195    }
196
197    /// Render this output framed by the markers, ready for stdout.
198    pub fn to_framed(&self) -> String {
199        // ContainerOutput serialization cannot fail (no maps with non-string keys).
200        let json = serde_json::to_string(self).unwrap_or_else(|_| "{}".to_string());
201        format!("{OUTPUT_START_MARKER}\n{json}\n{OUTPUT_END_MARKER}\n")
202    }
203}
204
205/// Host -> container follow-up message (`input/*.json`).
206#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
207#[serde(tag = "type", rename_all = "snake_case")]
208pub enum IpcInput {
209    Message { text: String },
210}
211
212/// Container -> host outbound chat message (`messages/*.json`).
213#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
214#[serde(tag = "type", rename_all = "snake_case")]
215pub enum IpcOutbound {
216    Message {
217        #[serde(rename = "chatJid")]
218        chat_jid: String,
219        text: String,
220        #[serde(
221            rename = "groupFolder",
222            default,
223            skip_serializing_if = "Option::is_none"
224        )]
225        group_folder: Option<String>,
226        #[serde(default, skip_serializing_if = "Option::is_none")]
227        timestamp: Option<String>,
228    },
229}
230
231/// Container -> host task operation (`tasks/*.json`).
232///
233/// The host MUST authorize every variant against the *directory-derived*
234/// source group (see `talon-ipc`): non-main groups may only touch their own
235/// tasks and may never register groups.
236#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
237#[serde(tag = "type", rename_all = "snake_case")]
238#[allow(clippy::large_enum_variant)] // short-lived parse result; boxing would churn every consumer
239pub enum IpcTask {
240    ScheduleTask {
241        #[serde(rename = "taskId", default, skip_serializing_if = "Option::is_none")]
242        task_id: Option<String>,
243        prompt: String,
244        #[serde(default, skip_serializing_if = "Option::is_none")]
245        script: Option<String>,
246        schedule_type: ScheduleType,
247        schedule_value: String,
248        #[serde(default)]
249        context_mode: Option<ContextMode>,
250        #[serde(rename = "targetJid")]
251        target_jid: String,
252        #[serde(rename = "createdBy", default, skip_serializing_if = "Option::is_none")]
253        created_by: Option<String>,
254        #[serde(default, skip_serializing_if = "Option::is_none")]
255        timestamp: Option<String>,
256    },
257    PauseTask {
258        #[serde(rename = "taskId")]
259        task_id: String,
260        #[serde(default, skip_serializing_if = "Option::is_none")]
261        timestamp: Option<String>,
262    },
263    ResumeTask {
264        #[serde(rename = "taskId")]
265        task_id: String,
266        #[serde(default, skip_serializing_if = "Option::is_none")]
267        timestamp: Option<String>,
268    },
269    CancelTask {
270        #[serde(rename = "taskId")]
271        task_id: String,
272        #[serde(default, skip_serializing_if = "Option::is_none")]
273        timestamp: Option<String>,
274    },
275    UpdateTask {
276        #[serde(rename = "taskId")]
277        task_id: String,
278        #[serde(default, skip_serializing_if = "Option::is_none")]
279        prompt: Option<String>,
280        /// Empty string clears the script.
281        #[serde(default, skip_serializing_if = "Option::is_none")]
282        script: Option<String>,
283        #[serde(default, skip_serializing_if = "Option::is_none")]
284        schedule_type: Option<ScheduleType>,
285        #[serde(default, skip_serializing_if = "Option::is_none")]
286        schedule_value: Option<String>,
287        #[serde(default, skip_serializing_if = "Option::is_none")]
288        timestamp: Option<String>,
289    },
290    RefreshGroups {
291        #[serde(default, skip_serializing_if = "Option::is_none")]
292        timestamp: Option<String>,
293    },
294    RegisterGroup {
295        jid: String,
296        name: String,
297        folder: String,
298        trigger: String,
299        #[serde(
300            rename = "requiresTrigger",
301            default,
302            skip_serializing_if = "Option::is_none"
303        )]
304        requires_trigger: Option<bool>,
305        #[serde(
306            rename = "containerConfig",
307            default,
308            skip_serializing_if = "Option::is_none"
309        )]
310        container_config: Option<ContainerConfig>,
311        #[serde(default, skip_serializing_if = "Option::is_none")]
312        timestamp: Option<String>,
313    },
314}
315
316/// Result of a scheduled task's pre-flight script (last stdout line).
317#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
318#[serde(rename_all = "camelCase")]
319pub struct ScriptResult {
320    pub wake_agent: bool,
321    #[serde(default, skip_serializing_if = "Option::is_none")]
322    pub data: Option<serde_json::Value>,
323}
324
325/// Build a lexically ordered, collision-resistant IPC file name
326/// (`<unix_millis>-<random6>.json`).
327pub fn ipc_file_name() -> String {
328    let millis = chrono::Utc::now().timestamp_millis();
329    let rand = uuid::Uuid::new_v4().simple().to_string();
330    format!("{millis}-{}.json", &rand[..6])
331}
332
333#[cfg(test)]
334mod tests {
335    use super::*;
336
337    #[test]
338    fn ipc_task_roundtrip_matches_ts_shape() {
339        let raw = r#"{"type":"schedule_task","taskId":"task-1","prompt":"hi","schedule_type":"cron","schedule_value":"*/5 * * * *","context_mode":"group","targetJid":"tg:1","createdBy":"main","timestamp":"2026-01-01T00:00:00.000Z"}"#;
340        let t: IpcTask = serde_json::from_str(raw).unwrap();
341        match &t {
342            IpcTask::ScheduleTask {
343                task_id,
344                schedule_type,
345                target_jid,
346                ..
347            } => {
348                assert_eq!(task_id.as_deref(), Some("task-1"));
349                assert_eq!(*schedule_type, ScheduleType::Cron);
350                assert_eq!(target_jid, "tg:1");
351            }
352            other => panic!("unexpected {other:?}"),
353        }
354        let v = serde_json::to_value(&t).unwrap();
355        assert_eq!(v["type"], "schedule_task");
356        assert_eq!(v["targetJid"], "tg:1");
357    }
358
359    #[test]
360    fn outbound_message_shape() {
361        let raw = r#"{"type":"message","chatJid":"x","text":"t","groupFolder":"g"}"#;
362        let m: IpcOutbound = serde_json::from_str(raw).unwrap();
363        let IpcOutbound::Message { chat_jid, .. } = m;
364        assert_eq!(chat_jid, "x");
365    }
366
367    #[test]
368    fn framed_output_contains_markers() {
369        let s = ContainerOutput::success(Some("ok".into()), Some("s1".into())).to_framed();
370        assert!(s.starts_with(OUTPUT_START_MARKER));
371        assert!(s.trim_end().ends_with(OUTPUT_END_MARKER));
372        assert!(s.contains("\"newSessionId\":\"s1\""));
373    }
374
375    #[test]
376    fn usage_context_size() {
377        let u = SessionUsage {
378            input_tokens: Some(10),
379            output_tokens: Some(99),
380            cache_creation_input_tokens: None,
381            cache_read_input_tokens: Some(5),
382        };
383        assert_eq!(u.context_size(), 15);
384    }
385
386    #[test]
387    fn ipc_file_names_sort_and_differ() {
388        let a = ipc_file_name();
389        let b = ipc_file_name();
390        assert!(a.ends_with(".json"));
391        assert_ne!(a, b);
392    }
393}
talon_core/protocol.rs

talon_core/
protocol.rs
