agent_graph_mcp/
operator.rs

1//! Daemon-owned Unix operator service. The client never opens SQLite.
2use crate::{
3    evidence::digest,
4    operator_auth::{
5        peer_credentials, validate_window, AuthenticatedOperator, OperatorAction, PeerCredentials,
6    },
7    operator_ipc::{validate, OperatorFrame, OperatorResponse, PROTOCOL},
8    store::{
9        OperatorRetentionError, OperatorRetentionRequest, OperatorRetentionResult, PersistentStore,
10    },
11};
12use chrono::{DateTime, Utc};
13use serde_json::{json, Value};
14use std::{collections::BTreeSet, sync::Arc};
15use tokio::{
16    io::{AsyncReadExt, AsyncWriteExt},
17    net::UnixStream,
18};
19
20const MAX_FRAME: usize = 1024 * 1024;
21
22#[derive(Clone)]
23pub struct OperatorService {
24    store: PersistentStore,
25    allowed_uids: Arc<BTreeSet<u32>>,
26    daemon_instance_id: String,
27}
28
29impl OperatorService {
30    pub fn new(
31        store: PersistentStore,
32        allowed_uids: BTreeSet<u32>,
33        daemon_instance_id: String,
34    ) -> Self {
35        Self {
36            store,
37            allowed_uids: Arc::new(allowed_uids),
38            daemon_instance_id,
39        }
40    }
41
42    pub fn handle(&self, peer: PeerCredentials, frame: OperatorFrame) -> OperatorResponse {
43        let error = |code: &str| OperatorResponse {
44            protocol: PROTOCOL.into(),
45            ok: false,
46            error_code: Some(code.into()),
47            receipt_id: None,
48        };
49        if let Err(code) = validate(&frame) {
50            return error(code);
51        }
52        if frame.resource_kind != "graph" {
53            return error("OPERATOR_RESOURCE_UNSUPPORTED");
54        }
55        if !self.allowed_uids.contains(&peer.uid) {
56            return error("OPERATOR_PEER_UNAUTHORIZED");
57        }
58        let issued_at = match DateTime::parse_from_rfc3339(&frame.issued_at) {
59            Ok(v) => v.with_timezone(&Utc),
60            Err(_) => return error("OPERATOR_TIMESTAMP_INVALID"),
61        };
62        let expires_at = match DateTime::parse_from_rfc3339(&frame.expires_at) {
63            Ok(v) => v.with_timezone(&Utc),
64            Err(_) => return error("OPERATOR_TIMESTAMP_INVALID"),
65        };
66        let operator = AuthenticatedOperator {
67            uid: peer.uid,
68            gid: peer.gid,
69            action: frame.action,
70            resource_kind: frame.resource_kind.clone(),
71            resource_id: frame.resource_id.clone(),
72            expected_state_digest: frame.expected_state_digest.clone(),
73            nonce: frame.nonce.clone(),
74            issued_at,
75            expires_at,
76        };
77        if let Err(code) = validate_window(&operator, Utc::now()) {
78            return error(code);
79        }
80        if !matches!(
81            frame.action,
82            OperatorAction::SetGraphRetention
83                | OperatorAction::ApproveGraphDeletion
84                | OperatorAction::DeleteGraph
85                | OperatorAction::ClearExecutionLineage
86                | OperatorAction::PurgeGraph
87                | OperatorAction::PromoteTemplate
88                | OperatorAction::Migrate
89                | OperatorAction::Install
90        ) {
91            return error("OPERATOR_ACTION_UNSUPPORTED");
92        }
93        let material: Value = match frame.decision_material.as_deref() {
94            Some(raw) => match serde_json::from_str(raw) {
95                Ok(v) => v,
96                Err(_) => return error("OPERATOR_DECISION_INVALID"),
97            },
98            None => json!({}),
99        };
100        let request_digest = digest(&json!({
101            "protocol": frame.protocol, "request_id": frame.request_id, "action": frame.action,
102            "resource_kind": frame.resource_kind, "resource_id": frame.resource_id,
103            "expected_state_digest": frame.expected_state_digest, "nonce": frame.nonce,
104            "issued_at": frame.issued_at, "expires_at": frame.expires_at,
105            "decision_material": material, "peer_uid": peer.uid, "peer_gid": peer.gid,
106        }));
107        let request = OperatorRetentionRequest {
108            request_digest,
109            action: frame.action,
110            graph_id: frame.resource_id,
111            expected_state_digest: frame.expected_state_digest,
112            nonce: frame.nonce,
113            operator_uid: peer.uid,
114            daemon_instance_id: self.daemon_instance_id.clone(),
115            issued_at: operator.issued_at.to_rfc3339(),
116            expires_at: operator.expires_at.to_rfc3339(),
117            state: material
118                .get("state")
119                .and_then(Value::as_str)
120                .map(str::to_owned),
121            reason: material
122                .get("reason")
123                .and_then(Value::as_str)
124                .map(str::to_owned),
125            review_after: material
126                .get("review_after")
127                .and_then(Value::as_str)
128                .map(str::to_owned),
129        };
130        match self.store.apply_operator_retention(&request) {
131            Ok(OperatorRetentionResult::Applied { receipt_id })
132            | Ok(OperatorRetentionResult::Replayed { receipt_id }) => OperatorResponse {
133                protocol: PROTOCOL.into(),
134                ok: true,
135                error_code: None,
136                receipt_id: Some(receipt_id),
137            },
138            Err(err) => error(match err {
139                OperatorRetentionError::StaleState => "AUTHORIZATION_STATE_STALE",
140                OperatorRetentionError::NonceReplayed => "AUTHORIZATION_NONCE_REPLAYED",
141                OperatorRetentionError::Tombstoned => "GRAPH_TOMBSTONED",
142                OperatorRetentionError::Referenced => "GRAPH_REFERENCED",
143                OperatorRetentionError::ReferencedBySubgraph => "GRAPH_REFERENCED_BY_SUBGRAPH",
144                OperatorRetentionError::InvalidState => "RETENTION_STATE_INVALID",
145                OperatorRetentionError::InvalidTransition => "RETENTION_TRANSITION_INVALID",
146                OperatorRetentionError::NotFound => "GRAPH_NOT_FOUND",
147                OperatorRetentionError::InvalidAction => "OPERATOR_ACTION_UNSUPPORTED",
148                OperatorRetentionError::Persistence => "OPERATOR_PERSISTENCE_FAILED",
149            }),
150        }
151    }
152}
153
154pub async fn serve_connection(stream: UnixStream, service: OperatorService) -> std::io::Result<()> {
155    let peer = peer_credentials(&stream).await?;
156    let (mut rx, mut tx) = stream.into_split();
157    loop {
158        let mut header = [0u8; 4];
159        if rx.read_exact(&mut header).await.is_err() {
160            return Ok(());
161        }
162        let len = u32::from_be_bytes(header) as usize;
163        if len == 0 || len > MAX_FRAME {
164            return Ok(());
165        }
166        let mut payload = vec![0u8; len];
167        if rx.read_exact(&mut payload).await.is_err() {
168            return Ok(());
169        }
170        let response = match serde_json::from_slice::<OperatorFrame>(&payload) {
171            Ok(frame) => service.handle(peer.clone(), frame),
172            Err(_) => OperatorResponse {
173                protocol: PROTOCOL.into(),
174                ok: false,
175                error_code: Some("OPERATOR_FRAME_INVALID".into()),
176                receipt_id: None,
177            },
178        };
179        let body = serde_json::to_vec(&response).map_err(std::io::Error::other)?;
180        tx.write_all(&(body.len() as u32).to_be_bytes()).await?;
181        tx.write_all(&body).await?;
182        tx.flush().await?;
183    }
184}
agent_graph_mcp/operator.rs

agent_graph_mcp/
operator.rs
