agentmux_srv/backend/blockcontroller/shell/
translation.rs

1// Copyright 2025-2026, AgentMux Corp.
2// SPDX-License-Identifier: Apache-2.0
3
4//! Agent-event extraction and translation (Phase 1.5 PR 1). Feeds agent-pane
5//! stdout through the Claude stream-json translator and publishes `AgentEvent`s.
6
7use crate::backend::mps;
8
9/// Maximum size of the per-block line buffer used by the agent-event
10/// translation path. Past this, the buffer is reset (a producer that
11/// never emits a newline can't grow it unboundedly). 1 MiB is far
12/// beyond any plausible stream-json frame size.
13pub(super) const AGENT_LINE_BUFFER_CAP: usize = 1024 * 1024;
14
15/// Append `chunk` to `line_buf`, drain complete lines, JSON-parse
16/// each, and run successful parses through `translator`. Returns the
17/// events the translator emitted. Caller is responsible for
18/// publishing them.
19///
20/// Buffers RAW BYTES (not lossy-decoded strings) so a multi-byte
21/// UTF-8 character split across two PTY reads decodes cleanly when
22/// the complete line arrives. Decoding each chunk lossily would
23/// insert U+FFFD into the middle of words for non-ASCII content
24/// (CJK, emoji, accented chars), silently corrupting
25/// `AssistantText`/`Done.response` while the parallel raw-byte MPS
26/// path stays correct — drone consumers would have no way to
27/// recover. Reagent P1 + codex P2 on PR #833.
28///
29/// Pure function — split out from `accumulate_and_translate` so the
30/// line-buffering + JSON-fast-reject + translator-call logic is
31/// unit-testable without spinning up a broker.
32pub(super) fn extract_agent_events(
33    line_buf: &mut Vec<u8>,
34    chunk: &[u8],
35    translator: &mut crate::agents::translator::claude::ClaudeTranslator,
36) -> Vec<crate::agents::types::AgentEvent> {
37    use crate::agents::translator::Translator as _;
38    let mut out: Vec<crate::agents::types::AgentEvent> = Vec::new();
39    line_buf.extend_from_slice(chunk);
40    if line_buf.len() > AGENT_LINE_BUFFER_CAP {
41        // No newline in a megabyte — definitely not stream-json.
42        // Reset to keep memory bounded.
43        line_buf.clear();
44        return out;
45    }
46    while let Some(nl) = line_buf.iter().position(|&b| b == b'\n') {
47        let line_bytes: Vec<u8> = line_buf.drain(..=nl).collect();
48        // Decode the COMPLETE line as lossy UTF-8 — split codepoints
49        // are now fully present, so lossy-vs-strict only matters for
50        // genuinely malformed bytes which would also fail strict.
51        let line = String::from_utf8_lossy(&line_bytes);
52        let trimmed = line.trim_end_matches(['\n', '\r']);
53        if !trimmed.starts_with('{') {
54            // Fast-reject: stream-json frames are JSON objects.
55            // Skips ANSI escapes, prompts, blank lines, etc.
56            continue;
57        }
58        let Ok(frame) = serde_json::from_str::<serde_json::Value>(trimmed) else {
59            continue;
60        };
61        out.extend(translator.translate(frame));
62    }
63    out
64}
65
66/// Publish the events `extract_agent_events` produced on the MPS
67/// scope `agent_event:<block_id>`. Phase 1.5 PR 1 hook for agent
68/// panes. Called only when `is_agent` is true at spawn time (see
69/// read-task closure in `start()`).
70pub(super) fn accumulate_and_translate(
71    broker: &mps::Broker,
72    block_id: &str,
73    line_buf: &mut Vec<u8>,
74    chunk: &[u8],
75    translator: &mut crate::agents::translator::claude::ClaudeTranslator,
76) {
77    for event in extract_agent_events(line_buf, chunk, translator) {
78        broker.publish_event_for::<crate::backend::mps_events::AgentEventStream>(&block_id, &event, vec![], 0);
79    }
80}
agentmux_srv/backend/blockcontroller/shell/translation.rs

agentmux_srv/backend/blockcontroller/shell/
translation.rs
