mj_core/relay/
protocol.rs

1//! Wire protocol for the durable ACP relay: request/response envelopes,
2//! error shapes, and newline-delimited JSON framing. This module is pure
3//! serde plus byte-oriented framing; it has no filesystem or state-machine
4//! concerns of its own.
5
6use std::io::{BufRead, Write};
7
8use anyhow::{Context, Result, anyhow, bail};
9use serde::{Deserialize, Serialize};
10
11use crate::elicitation::ElicitationResponse;
12use crate::project_memory::ProjectMemorySnapshot;
13
14use super::snapshot::{RelayCommand, RelayEvent, RelayOperationalState};
15use super::{MAX_FRAME_BYTES, RELAY_MIN_PROTOCOL_VERSION, RELAY_PROTOCOL_VERSION};
16
17#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
18#[serde(deny_unknown_fields)]
19pub struct RelayVersionRange {
20    pub min: u32,
21    pub max: u32,
22}
23
24impl RelayVersionRange {
25    pub const CURRENT: Self = Self {
26        min: RELAY_MIN_PROTOCOL_VERSION,
27        max: RELAY_PROTOCOL_VERSION,
28    };
29
30    pub const fn contains(self, version: u32) -> bool {
31        self.min <= version && version <= self.max
32    }
33
34    pub fn negotiate(self, peer: Self) -> Option<u32> {
35        let minimum = self.min.max(peer.min);
36        let maximum = self.max.min(peer.max);
37        (minimum <= maximum).then_some(maximum)
38    }
39}
40
41/// A request on the new controller-to-relay boundary. ACP payloads remain ACP
42/// payloads; only durability and queue-control operations are Hel-specific.
43#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
44#[serde(
45    tag = "method",
46    content = "params",
47    rename_all = "snake_case",
48    deny_unknown_fields
49)]
50pub enum RelayRequest {
51    Hello {
52        controller_version: String,
53        supported: RelayVersionRange,
54    },
55    Attach {
56        after_ordinal: u64,
57        after_digest: String,
58    },
59    Acknowledge {
60        through_ordinal: u64,
61        through_digest: String,
62    },
63    Submit {
64        command_id: String,
65        command: RelayCommand,
66    },
67    Status,
68    AttachmentPresent {
69        reference: crate::attachment::AttachmentRef,
70    },
71    InstallAttachment {
72        reference: crate::attachment::AttachmentRef,
73        data: String,
74    },
75    ReadAttachment {
76        reference: crate::attachment::AttachmentRef,
77    },
78    /// Add hidden background context attached to the next real prompt.
79    /// This mutates only the relay-private snapshot and is never projected as
80    /// conversation history.
81    InstallPromptContext {
82        text: String,
83    },
84    /// Read the session-private memory replica and the baseline it was seeded
85    /// from. Connection-only: memory content never enters the relay journal.
86    ProjectMemorySnapshot,
87    /// Install a controller-reconciled tree into both the replica and its
88    /// baseline for the next three-way synchronization.
89    InstallProjectMemorySnapshot {
90        snapshot: ProjectMemorySnapshot,
91    },
92    /// Report non-secret metadata for this session's harness credentials.
93    /// The runtime handles credential requests on the connection and never
94    /// passes them through the durable relay.
95    CredentialState,
96    /// Read this session's harness credential file as base64. The payload is
97    /// connection-only and must never enter relay state or observations.
98    ReadCredentials,
99    /// Install a base64-encoded credential file into this session's harness
100    /// home. The destination path is fixed by the worker launch config.
101    InstallCredentials {
102        data: String,
103    },
104    /// Report non-secret metadata for this session's synced skills trees.
105    /// Handled on the connection like credential requests; the durable relay
106    /// never sees them.
107    SkillsState,
108    /// Replace this session's synced skills trees with a base64-encoded
109    /// `skills` archive. The destination directories are fixed by the
110    /// worker launch config and the harness skills whitelist.
111    InstallSkills {
112        data: String,
113    },
114    /// Report whether this worker has a synchronized GitHub CLI token and its
115    /// non-secret fingerprint. This request is connection-only.
116    GithubTokenState,
117    /// Install the controller's current GitHub CLI token into worker-private
118    /// runtime storage. The token never enters durable relay state.
119    InstallGithubToken {
120        data: String,
121    },
122    /// Remove the worker's synchronized GitHub CLI token.
123    RemoveGithubToken,
124    /// Resolve one in-flight form without journaling its answer.
125    RespondElicitation {
126        elicitation_id: String,
127        response: ElicitationResponse,
128    },
129    /// Stop one task from the current process-local background-work level.
130    /// The opaque id must come from `RelayOperationalState.background_commands`.
131    StopBackgroundTask {
132        background_task_id: String,
133    },
134    /// Fetch controller work queued by the parent session's private MCP
135    /// socket. Connection-only: request payloads do not enter chat history.
136    SubagentRequests,
137    /// Acknowledge a completed MCP request and cache its bounded answer for
138    /// subsequent tool calls and restart recovery.
139    CompleteSubagentRequest {
140        result: crate::subagent::SubagentToolResult,
141    },
142    /// Drive the second-opinion reviewer that runs beside this session.
143    ///
144    /// The reviewer is a sidecar, not a session: it shares this worker's
145    /// target and working directory and owns nothing else. Its own durable
146    /// relay answers the attach, acknowledge, submit and status requests
147    /// nested here, so the reviewer's conversation is journaled and replayed
148    /// the same way the primary's is.
149    Reviewer {
150        /// Which reviewing agent this is for. Absent means the default role,
151        /// which is the one plan review uses; a turn review in the extended
152        /// tier also names its supervisor, its intent analyst, and each
153        /// specialist lane. An older controller sends no role, and an older
154        /// worker ignores one, so the field is additive in both directions.
155        #[serde(default, skip_serializing_if = "Option::is_none")]
156        role: Option<String>,
157        request: ReviewerRequest,
158    },
159}
160
161/// What a controller asks of the second-opinion reviewer sidecar.
162#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
163#[serde(tag = "action", content = "params", rename_all = "snake_case")]
164pub enum ReviewerRequest {
165    /// Start the reviewer, or report the running one when `config` matches it.
166    /// The reviewer's profile must already be staged under the worker root.
167    Start {
168        config: Box<crate::worker_launch::ReviewerLaunchConfig>,
169    },
170    /// Replay the reviewer's journal from a cursor, as `Attach` does for the
171    /// primary.
172    Attach {
173        after_ordinal: u64,
174        after_digest: String,
175    },
176    Acknowledge {
177        through_ordinal: u64,
178        through_digest: String,
179    },
180    Submit {
181        command_id: String,
182        command: RelayCommand,
183    },
184    Status,
185    /// Answer a form the reviewer's harness is waiting on.
186    ///
187    /// A reviewer that asks for permission and is never answered stalls the
188    /// whole review, so its forms travel the same connection-only path the
189    /// primary's do.
190    RespondElicitation {
191        elicitation_id: String,
192        response: ElicitationResponse,
193    },
194    /// Cancel any turn in flight and stop the reviewer's process group,
195    /// keeping its staged profile, native session and journal for next time.
196    Pause,
197    /// Report what changed in every workspace repository since `baselines`.
198    ///
199    /// A baseline is a Git tree id recorded by an earlier capture, keyed by
200    /// repository root. When that baseline is unavailable, the worker uses
201    /// the baseline pinned before its primary harness started. If neither
202    /// tree is available, coverage starts at this capture rather than
203    /// presenting the whole repository as this turn's work. Capture never
204    /// touches the repository's index or working tree.
205    CaptureDelta {
206        baselines: std::collections::BTreeMap<std::path::PathBuf, String>,
207    },
208    /// Record `trees` as the new review baselines, pinning each so a later
209    /// `git gc` cannot collect it.
210    AdvanceBaseline {
211        trees: std::collections::BTreeMap<std::path::PathBuf, String>,
212    },
213    /// Run Bifrost's one-shot semantic diff analysis over captured trees and
214    /// return the changed-callable packet the review prompts embed.
215    AnalyzeDelta {
216        repositories: Vec<AnalyzeDeltaRepository>,
217    },
218    /// Collect the specialist lanes the review supervisor asked for through
219    /// its MCP tool since the last time the controller asked. This request is
220    /// answered by the sidecar itself rather than by any one role.
221    TakeLaneDispatches,
222}
223
224/// One repository's captured endpoints for the Bifrost analysis pre-pass.
225#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
226#[serde(deny_unknown_fields)]
227pub struct AnalyzeDeltaRepository {
228    pub root: std::path::PathBuf,
229    /// Absent for a repository with no recorded baseline, which the worker
230    /// resolves to that repository's empty tree.
231    #[serde(default, skip_serializing_if = "Option::is_none")]
232    pub baseline_tree: Option<String>,
233    pub current_tree: String,
234}
235
236/// What one repository contributed to a cumulative review delta.
237#[derive(Debug, Clone, PartialEq, Eq, Default, Serialize, Deserialize)]
238#[serde(deny_unknown_fields)]
239pub struct RepoDelta {
240    pub root: std::path::PathBuf,
241    pub baseline_tree: Option<String>,
242    pub current_tree: String,
243    /// Unified diff, bounded worker-side; an empty patch means this repository
244    /// has nothing to review.
245    pub patch: String,
246    /// Human-readable file and line totals, computed from the untruncated
247    /// patch so bounding cannot make a change look smaller than it is.
248    pub diffstat: String,
249    pub changed_lines: usize,
250}
251
252impl ReviewerRequest {
253    pub const fn action_name(&self) -> &'static str {
254        match self {
255            Self::Start { .. } => "reviewer_start",
256            Self::Attach { .. } => "reviewer_attach",
257            Self::Acknowledge { .. } => "reviewer_acknowledge",
258            Self::Submit { .. } => "reviewer_submit",
259            Self::Status => "reviewer_status",
260            Self::RespondElicitation { .. } => "reviewer_respond_elicitation",
261            Self::Pause => "reviewer_pause",
262            Self::CaptureDelta { .. } => "reviewer_capture_delta",
263            Self::AdvanceBaseline { .. } => "reviewer_advance_baseline",
264            Self::AnalyzeDelta { .. } => "reviewer_analyze_delta",
265            Self::TakeLaneDispatches => "reviewer_take_lane_dispatches",
266        }
267    }
268}
269
270impl RelayRequest {
271    pub const fn method_name(&self) -> &'static str {
272        match self {
273            Self::Hello { .. } => "hello",
274            Self::Attach { .. } => "attach",
275            Self::Acknowledge { .. } => "acknowledge",
276            Self::Submit { .. } => "submit",
277            Self::Status => "status",
278            Self::InstallPromptContext { .. } => "install_prompt_context",
279            Self::ProjectMemorySnapshot => "project_memory_snapshot",
280            Self::InstallProjectMemorySnapshot { .. } => "install_project_memory_snapshot",
281            Self::AttachmentPresent { .. } => "attachment_present",
282            Self::InstallAttachment { .. } => "install_attachment",
283            Self::ReadAttachment { .. } => "read_attachment",
284            Self::CredentialState => "credential_state",
285            Self::ReadCredentials => "read_credentials",
286            Self::InstallCredentials { .. } => "install_credentials",
287            Self::SkillsState => "skills_state",
288            Self::InstallSkills { .. } => "install_skills",
289            Self::GithubTokenState => "github_token_state",
290            Self::InstallGithubToken { .. } => "install_github_token",
291            Self::RemoveGithubToken => "remove_github_token",
292            Self::RespondElicitation { .. } => "respond_elicitation",
293            Self::StopBackgroundTask { .. } => "stop_background_task",
294            Self::SubagentRequests => "subagent_requests",
295            Self::CompleteSubagentRequest { .. } => "complete_subagent_request",
296            Self::Reviewer { request, .. } => request.action_name(),
297        }
298    }
299
300    /// Oldest protocol that understands this method or command payload. Form
301    /// answers landed in protocol 2, hidden context in 3, project-memory sync
302    /// in 4, user shell commands in 5, the reviewer sidecar in 6, and the
303    /// non-steering turn cancellation in 7.
304    pub fn minimum_protocol(&self) -> u32 {
305        match self {
306            Self::AttachmentPresent { .. }
307            | Self::InstallAttachment { .. }
308            | Self::ReadAttachment { .. } => 8,
309            Self::StopBackgroundTask { .. } => 9,
310            Self::SubagentRequests | Self::CompleteSubagentRequest { .. } => 12,
311            Self::RespondElicitation { .. } => 2,
312            Self::InstallPromptContext { .. } => 3,
313            Self::ProjectMemorySnapshot | Self::InstallProjectMemorySnapshot { .. } => 4,
314            Self::Submit { command, .. } => command.minimum_protocol(),
315            Self::Reviewer { .. } => 6,
316            _ => RELAY_MIN_PROTOCOL_VERSION,
317        }
318    }
319
320    pub fn supported_at(&self, protocol_version: u32) -> bool {
321        RelayVersionRange::CURRENT.contains(protocol_version)
322            && protocol_version >= self.minimum_protocol()
323    }
324}
325
326pub fn incompatible_request_protocol(protocol_version: u32) -> RelayResponseBody {
327    relay_error(
328        RelayErrorCode::IncompatibleProtocol,
329        format!(
330            "request uses protocol {protocol_version}, relay supports protocol {}-{}",
331            RELAY_MIN_PROTOCOL_VERSION, RELAY_PROTOCOL_VERSION
332        ),
333        false,
334        None,
335    )
336}
337
338pub fn incompatible_request_protocol_response(
339    request_id: String,
340    protocol_version: u32,
341) -> RelayResponseEnvelope {
342    RelayResponseEnvelope {
343        request_id,
344        protocol_version,
345        body: incompatible_request_protocol(protocol_version),
346    }
347}
348
349#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
350#[serde(deny_unknown_fields)]
351pub struct RelayRequestEnvelope {
352    pub request_id: String,
353    pub protocol_version: u32,
354    pub request: RelayRequest,
355}
356
357#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
358pub struct RelayResponseEnvelope {
359    pub request_id: String,
360    pub protocol_version: u32,
361    #[serde(flatten)]
362    pub body: RelayResponseBody,
363}
364
365#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
366#[serde(tag = "result", rename_all = "snake_case")]
367// This is a short-lived wire DTO. Boxing every successful response would add
368// an allocation without reducing retained relay state.
369#[allow(clippy::large_enum_variant)]
370pub enum RelayResponseBody {
371    Ok { payload: RelayResponsePayload },
372    Error { error: RelayProtocolError },
373}
374
375#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
376#[serde(tag = "type", content = "data", rename_all = "snake_case")]
377pub enum RelayResponsePayload {
378    Hello {
379        negotiated: u32,
380        relay_version: String,
381        session_id: String,
382        /// Content address of the worker executable that answered. The crate
383        /// version cannot tell two builds apart, so this is what a controller
384        /// compares against the binary it would install. Absent from a worker
385        /// built before the field existed, which counts as outdated.
386        #[serde(default, skip_serializing_if = "Option::is_none")]
387        worker_build: Option<String>,
388    },
389    Attached {
390        state: RelayOperationalState,
391        events: Vec<RelayEvent>,
392        through_ordinal: u64,
393        through_digest: String,
394    },
395    Acknowledged {
396        through_ordinal: u64,
397        through_digest: String,
398    },
399    Accepted {
400        command_id: String,
401        ordinal: u64,
402    },
403    Status(RelayOperationalState),
404    AttachmentPresent {
405        present: bool,
406    },
407    AttachmentInstalled,
408    AttachmentData {
409        data: String,
410    },
411    PromptContextInstalled,
412    ProjectMemorySnapshot {
413        baseline: ProjectMemorySnapshot,
414        replica: ProjectMemorySnapshot,
415    },
416    ProjectMemorySnapshotInstalled,
417    /// Fingerprint and freshness of a session's harness credentials. Neither
418    /// value is secret.
419    CredentialState {
420        present: bool,
421        fingerprint: String,
422        freshness_epoch_ms: Option<i64>,
423    },
424    /// Base64 of a session's credential file. Sent only on the connection
425    /// socket, never recorded.
426    Credentials {
427        data: String,
428    },
429    /// Fingerprint of a session's synced skills trees. Not secret.
430    SkillsState {
431        present: bool,
432        fingerprint: String,
433    },
434    /// Presence and fingerprint of the worker-private GitHub CLI token.
435    GithubTokenState {
436        present: bool,
437        fingerprint: String,
438    },
439    ElicitationResolved {
440        elicitation_id: String,
441    },
442    BackgroundTaskStopRequested {
443        background_task_id: String,
444    },
445    SubagentRequests {
446        requests: Vec<crate::subagent::SubagentToolRequest>,
447        results: Vec<crate::subagent::SubagentToolResult>,
448    },
449    SubagentRequestCompleted,
450    /// The reviewer sidecar is running under the requested configuration.
451    ReviewerStarted {
452        #[serde(default, skip_serializing_if = "Option::is_none")]
453        native_session_id: Option<String>,
454        /// What the reviewer's harness advertises right now, which is what the
455        /// waterfall offers the user.
456        config_options: Vec<agent_client_protocol::schema::v1::SessionConfigOption>,
457        /// Whether this call reused an already-running reviewer.
458        reused: bool,
459        state: Box<RelayOperationalState>,
460    },
461    /// The reviewer's process group has been stopped; its files remain.
462    ReviewerPaused,
463    /// What every workspace repository changed since the stored baselines.
464    ReviewDelta {
465        repositories: Vec<RepoDelta>,
466    },
467    /// The review baselines now name the trees the controller sent.
468    ReviewBaselineAdvanced,
469    /// Bifrost's changed-callable packet for the captured trees.
470    ReviewChangedFunctions {
471        packet: String,
472    },
473    /// Specialist lanes the review supervisor asked for.
474    LaneDispatches {
475        requests: Vec<crate::review::lanes::ReviewSubagentRequest>,
476    },
477}
478
479#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
480pub struct RelayProtocolError {
481    pub code: RelayErrorCode,
482    pub message: String,
483    pub retryable: bool,
484    #[serde(default, skip_serializing_if = "Option::is_none")]
485    pub detail: Option<RelayErrorDetail>,
486}
487
488#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
489#[serde(rename_all = "snake_case")]
490pub enum RelayErrorCode {
491    IncompatibleProtocol,
492    InvalidRequest,
493    InvalidState,
494    Desynchronized,
495    Internal,
496}
497
498#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
499#[serde(tag = "kind", rename_all = "snake_case")]
500pub enum RelayErrorDetail {
501    Desynchronized {
502        requested_after: u64,
503        requested_digest: String,
504        earliest_available: u64,
505        earliest_digest: String,
506        latest: u64,
507        latest_digest: String,
508    },
509}
510
511pub fn relay_protocol_error(
512    code: RelayErrorCode,
513    message: impl Into<String>,
514    retryable: bool,
515    detail: Option<RelayErrorDetail>,
516) -> RelayProtocolError {
517    RelayProtocolError {
518        code,
519        message: message.into(),
520        retryable,
521        detail,
522    }
523}
524
525pub fn relay_error(
526    code: RelayErrorCode,
527    message: impl Into<String>,
528    retryable: bool,
529    detail: Option<RelayErrorDetail>,
530) -> RelayResponseBody {
531    RelayResponseBody::Error {
532        error: relay_protocol_error(code, message, retryable, detail),
533    }
534}
535
536pub fn unsupported_relay_method_response(
537    request_id: String,
538    protocol_version: u32,
539    method: String,
540) -> RelayResponseEnvelope {
541    RelayResponseEnvelope {
542        request_id,
543        protocol_version,
544        body: relay_error(
545            RelayErrorCode::InvalidRequest,
546            format!("relay does not support method {method:?}"),
547            false,
548            None,
549        ),
550    }
551}
552
553pub fn invalid_relay_request_response(
554    request_id: String,
555    protocol_version: u32,
556    message: String,
557) -> RelayResponseEnvelope {
558    RelayResponseEnvelope {
559        request_id,
560        protocol_version,
561        body: relay_error(RelayErrorCode::InvalidRequest, message, false, None),
562    }
563}
564
565pub fn read_relay_frame(reader: &mut impl BufRead) -> Result<Option<RelayRequestEnvelope>> {
566    let mut bytes = Vec::new();
567    let (read, _) = read_bounded_line(reader, &mut bytes, MAX_FRAME_BYTES)
568        .context("read relay protocol frame")?;
569    if read == 0 {
570        return Ok(None);
571    }
572    if bytes.last() == Some(&b'\r') {
573        bytes.pop();
574    }
575    if bytes.is_empty() {
576        bail!("empty relay protocol frame");
577    }
578    serde_json::from_slice(&bytes)
579        .context("parse relay protocol request")
580        .map(Some)
581}
582
583pub fn write_relay_frame(writer: &mut impl Write, response: &RelayResponseEnvelope) -> Result<()> {
584    serde_json::to_writer(&mut *writer, response)?;
585    writer.write_all(b"\n")?;
586    writer.flush()?;
587    Ok(())
588}
589
590pub fn read_bounded_line(
591    reader: &mut impl BufRead,
592    line: &mut Vec<u8>,
593    maximum_bytes: usize,
594) -> Result<(usize, bool)> {
595    line.clear();
596    let mut consumed_total = 0_usize;
597    loop {
598        let available = reader.fill_buf()?;
599        if available.is_empty() {
600            return Ok((consumed_total, false));
601        }
602        let newline = available.iter().position(|byte| *byte == b'\n');
603        let content_bytes = newline.unwrap_or(available.len());
604        let next_len = line
605            .len()
606            .checked_add(content_bytes)
607            .ok_or_else(|| anyhow!("relay journal line length overflow"))?;
608        super::snapshot::ensure_byte_budget(next_len, maximum_bytes, "relay journal event")?;
609        line.extend_from_slice(&available[..content_bytes]);
610        let consumed = content_bytes + usize::from(newline.is_some());
611        reader.consume(consumed);
612        consumed_total = consumed_total
613            .checked_add(consumed)
614            .ok_or_else(|| anyhow!("relay journal length overflow"))?;
615        if newline.is_some() {
616            return Ok((consumed_total, true));
617        }
618    }
619}
mj_core/relay/protocol.rs

mj_core/relay/
protocol.rs
