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}
167
168impl ContainerOutput {
169    pub fn success(result: Option<String>, new_session_id: Option<String>) -> Self {
170        Self {
171            status: OutputStatus::Success,
172            result,
173            new_session_id,
174            error: None,
175            usage: None,
176        }
177    }
178
179    pub fn error(message: impl Into<String>) -> Self {
180        Self {
181            status: OutputStatus::Error,
182            result: None,
183            new_session_id: None,
184            error: Some(message.into()),
185            usage: None,
186        }
187    }
188
189    /// Render this output framed by the markers, ready for stdout.
190    pub fn to_framed(&self) -> String {
191        // ContainerOutput serialization cannot fail (no maps with non-string keys).
192        let json = serde_json::to_string(self).unwrap_or_else(|_| "{}".to_string());
193        format!("{OUTPUT_START_MARKER}\n{json}\n{OUTPUT_END_MARKER}\n")
194    }
195}
196
197/// Host -> container follow-up message (`input/*.json`).
198#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
199#[serde(tag = "type", rename_all = "snake_case")]
200pub enum IpcInput {
201    Message { text: String },
202}
203
204/// Container -> host outbound chat message (`messages/*.json`).
205#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
206#[serde(tag = "type", rename_all = "snake_case")]
207pub enum IpcOutbound {
208    Message {
209        #[serde(rename = "chatJid")]
210        chat_jid: String,
211        text: String,
212        #[serde(
213            rename = "groupFolder",
214            default,
215            skip_serializing_if = "Option::is_none"
216        )]
217        group_folder: Option<String>,
218        #[serde(default, skip_serializing_if = "Option::is_none")]
219        timestamp: Option<String>,
220    },
221}
222
223/// Container -> host task operation (`tasks/*.json`).
224///
225/// The host MUST authorize every variant against the *directory-derived*
226/// source group (see `talon-ipc`): non-main groups may only touch their own
227/// tasks and may never register groups.
228#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
229#[serde(tag = "type", rename_all = "snake_case")]
230#[allow(clippy::large_enum_variant)] // short-lived parse result; boxing would churn every consumer
231pub enum IpcTask {
232    ScheduleTask {
233        #[serde(rename = "taskId", default, skip_serializing_if = "Option::is_none")]
234        task_id: Option<String>,
235        prompt: String,
236        #[serde(default, skip_serializing_if = "Option::is_none")]
237        script: Option<String>,
238        schedule_type: ScheduleType,
239        schedule_value: String,
240        #[serde(default)]
241        context_mode: Option<ContextMode>,
242        #[serde(rename = "targetJid")]
243        target_jid: String,
244        #[serde(rename = "createdBy", default, skip_serializing_if = "Option::is_none")]
245        created_by: Option<String>,
246        #[serde(default, skip_serializing_if = "Option::is_none")]
247        timestamp: Option<String>,
248    },
249    PauseTask {
250        #[serde(rename = "taskId")]
251        task_id: String,
252        #[serde(default, skip_serializing_if = "Option::is_none")]
253        timestamp: Option<String>,
254    },
255    ResumeTask {
256        #[serde(rename = "taskId")]
257        task_id: String,
258        #[serde(default, skip_serializing_if = "Option::is_none")]
259        timestamp: Option<String>,
260    },
261    CancelTask {
262        #[serde(rename = "taskId")]
263        task_id: String,
264        #[serde(default, skip_serializing_if = "Option::is_none")]
265        timestamp: Option<String>,
266    },
267    UpdateTask {
268        #[serde(rename = "taskId")]
269        task_id: String,
270        #[serde(default, skip_serializing_if = "Option::is_none")]
271        prompt: Option<String>,
272        /// Empty string clears the script.
273        #[serde(default, skip_serializing_if = "Option::is_none")]
274        script: Option<String>,
275        #[serde(default, skip_serializing_if = "Option::is_none")]
276        schedule_type: Option<ScheduleType>,
277        #[serde(default, skip_serializing_if = "Option::is_none")]
278        schedule_value: Option<String>,
279        #[serde(default, skip_serializing_if = "Option::is_none")]
280        timestamp: Option<String>,
281    },
282    RefreshGroups {
283        #[serde(default, skip_serializing_if = "Option::is_none")]
284        timestamp: Option<String>,
285    },
286    RegisterGroup {
287        jid: String,
288        name: String,
289        folder: String,
290        trigger: String,
291        #[serde(
292            rename = "requiresTrigger",
293            default,
294            skip_serializing_if = "Option::is_none"
295        )]
296        requires_trigger: Option<bool>,
297        #[serde(
298            rename = "containerConfig",
299            default,
300            skip_serializing_if = "Option::is_none"
301        )]
302        container_config: Option<ContainerConfig>,
303        #[serde(default, skip_serializing_if = "Option::is_none")]
304        timestamp: Option<String>,
305    },
306}
307
308/// Result of a scheduled task's pre-flight script (last stdout line).
309#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
310#[serde(rename_all = "camelCase")]
311pub struct ScriptResult {
312    pub wake_agent: bool,
313    #[serde(default, skip_serializing_if = "Option::is_none")]
314    pub data: Option<serde_json::Value>,
315}
316
317/// Build a lexically ordered, collision-resistant IPC file name
318/// (`<unix_millis>-<random6>.json`).
319pub fn ipc_file_name() -> String {
320    let millis = chrono::Utc::now().timestamp_millis();
321    let rand = uuid::Uuid::new_v4().simple().to_string();
322    format!("{millis}-{}.json", &rand[..6])
323}
324
325#[cfg(test)]
326mod tests {
327    use super::*;
328
329    #[test]
330    fn ipc_task_roundtrip_matches_ts_shape() {
331        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"}"#;
332        let t: IpcTask = serde_json::from_str(raw).unwrap();
333        match &t {
334            IpcTask::ScheduleTask {
335                task_id,
336                schedule_type,
337                target_jid,
338                ..
339            } => {
340                assert_eq!(task_id.as_deref(), Some("task-1"));
341                assert_eq!(*schedule_type, ScheduleType::Cron);
342                assert_eq!(target_jid, "tg:1");
343            }
344            other => panic!("unexpected {other:?}"),
345        }
346        let v = serde_json::to_value(&t).unwrap();
347        assert_eq!(v["type"], "schedule_task");
348        assert_eq!(v["targetJid"], "tg:1");
349    }
350
351    #[test]
352    fn outbound_message_shape() {
353        let raw = r#"{"type":"message","chatJid":"x","text":"t","groupFolder":"g"}"#;
354        let m: IpcOutbound = serde_json::from_str(raw).unwrap();
355        let IpcOutbound::Message { chat_jid, .. } = m;
356        assert_eq!(chat_jid, "x");
357    }
358
359    #[test]
360    fn framed_output_contains_markers() {
361        let s = ContainerOutput::success(Some("ok".into()), Some("s1".into())).to_framed();
362        assert!(s.starts_with(OUTPUT_START_MARKER));
363        assert!(s.trim_end().ends_with(OUTPUT_END_MARKER));
364        assert!(s.contains("\"newSessionId\":\"s1\""));
365    }
366
367    #[test]
368    fn usage_context_size() {
369        let u = SessionUsage {
370            input_tokens: Some(10),
371            output_tokens: Some(99),
372            cache_creation_input_tokens: None,
373            cache_read_input_tokens: Some(5),
374        };
375        assert_eq!(u.context_size(), 15);
376    }
377
378    #[test]
379    fn ipc_file_names_sort_and_differ() {
380        let a = ipc_file_name();
381        let b = ipc_file_name();
382        assert!(a.ends_with(".json"));
383        assert_ne!(a, b);
384    }
385}
talon_core/protocol.rs

talon_core/
protocol.rs
