agent_first_http/shared/
afdata.rs

1//! AFDATA protocol adapter.
2//!
3//! Every `afhttp` command emits one AFDATA protocol-v1 event per invocation:
4//! a single-line `{"kind":"result","result":...}` or `{"kind":"error",
5//! "error":...}` JSON value followed by a newline.
6
7use serde::Serialize;
8use serde::de::DeserializeOwned;
9use std::io::Write;
10use std::sync::OnceLock;
11
12use crate::shared::error::Error;
13
14static OUTPUT_TO: OnceLock<agent_first_data::OutputTo> = OnceLock::new();
15
16/// Install the process-wide AFDATA stream selector for this run.
17///
18/// Which selector that is comes from the resolved invocation's output plan —
19/// the CLI registry has already decided which destinations the matched shape
20/// admits and which one applies — so nothing here re-reads argv.
21pub fn install_output_to(selector: agent_first_data::OutputTo) -> Result<(), Error> {
22    OUTPUT_TO.set(selector).map_err(|_| {
23        Error::new(
24            crate::shared::error::ErrorCode::InternalError,
25            "AFDATA output routing was initialized more than once",
26        )
27    })
28}
29
30#[must_use]
31pub fn output_to() -> agent_first_data::OutputTo {
32    OUTPUT_TO
33        .get()
34        .copied()
35        .unwrap_or(agent_first_data::OutputTo::Split)
36}
37
38/// Emit a typed result event to `writer` and write it as one line of JSON followed
39/// by a newline. The command-specific `code` remains inside `result`.
40///
41/// Redacts AFDATA `_secret` fields by default and never panics on well-formed
42/// input — but we still funnel through this single seam so `print_stdout` /
43/// `print_stderr` stay clippy-denied at crate level.
44pub fn emit<W: Write, T: Serialize>(writer: &mut W, code: &str, payload: &T) -> Result<(), Error> {
45    emit_inner(writer, code, payload, RedactionMode::Default)
46}
47
48/// Emit a payload without AFDATA redaction. Use only for commands that require
49/// an explicit reveal flag and whose payload contains no unrelated secrets.
50pub fn emit_unredacted<W: Write, T: Serialize>(
51    writer: &mut W,
52    code: &str,
53    payload: &T,
54) -> Result<(), Error> {
55    emit_inner(writer, code, payload, RedactionMode::None)
56}
57
58/// Emit a result through the process-wide `--output-to` route.
59pub fn emit_process<T: Serialize>(code: &str, payload: &T) -> Result<(), Error> {
60    emit_process_inner(code, payload, RedactionMode::Default)
61}
62
63/// Emit a result with redaction disabled through the process-wide route.
64///
65/// This is reserved for explicit reveal commands whose payload contains no
66/// unrelated secrets.
67pub fn emit_process_unredacted<T: Serialize>(code: &str, payload: &T) -> Result<(), Error> {
68    emit_process_inner(code, payload, RedactionMode::None)
69}
70
71/// Emit a non-terminal progress event through the process-wide route.
72///
73/// A command that blocks on a person has something to say before it can say
74/// how the run ended. Progress is the only event kind that may precede the
75/// terminal one, and on a split route it lands on the diagnostic stream, so a
76/// caller reading only `result` is unaffected.
77pub fn emit_process_progress<T: Serialize>(code: &str, payload: &T) -> Result<(), Error> {
78    let value = prepare_payload(code, payload)?;
79    let event = agent_first_data::json_progress(value).build();
80    let mut emitter = agent_first_data::CliEmitter::from_output_to_with(
81        output_to(),
82        agent_first_data::OutputFormat::Json,
83        output_options(RedactionMode::Default),
84    )
85    .with_strict_protocol();
86    emitter.emit(event).map_err(|error| {
87        Error::new(
88            crate::shared::error::ErrorCode::InternalError,
89            error.to_string(),
90        )
91    })
92}
93
94/// Emit an otherwise-redacted result while intentionally revealing only the
95/// short-lived takeover capability field. The explicit `panel` and
96/// `fetch --takeover` operations are the only callers.
97pub fn emit_process_revealing_takeover<T: Serialize>(code: &str, payload: &T) -> Result<(), Error> {
98    let value = prepare_revealed_takeover_payload(code, payload)?;
99    emit_process_value(value, RedactionMode::None)
100}
101
102#[derive(Clone, Copy)]
103enum RedactionMode {
104    Default,
105    None,
106}
107
108fn emit_inner<W: Write, T: Serialize>(
109    writer: &mut W,
110    code: &str,
111    payload: &T,
112    redaction: RedactionMode,
113) -> Result<(), Error> {
114    let value = prepare_payload(code, payload)?;
115
116    let options = output_options(redaction);
117    let mut emitter = agent_first_data::CliEmitter::with_options(
118        writer,
119        agent_first_data::OutputFormat::Json,
120        options,
121    )
122    .with_strict_protocol();
123    emitter.emit_result(value).map_err(|err| {
124        Error::new(
125            crate::shared::error::ErrorCode::InternalError,
126            err.to_string(),
127        )
128    })?;
129    Ok(())
130}
131
132fn emit_process_inner<T: Serialize>(
133    code: &str,
134    payload: &T,
135    redaction: RedactionMode,
136) -> Result<(), Error> {
137    emit_process_value(prepare_payload(code, payload)?, redaction)
138}
139
140fn emit_process_value(value: serde_json::Value, redaction: RedactionMode) -> Result<(), Error> {
141    let mut emitter = agent_first_data::CliEmitter::from_output_to_with(
142        output_to(),
143        agent_first_data::OutputFormat::Json,
144        output_options(redaction),
145    )
146    .with_strict_protocol();
147    emitter.emit_result(value).map_err(|err| {
148        Error::new(
149            crate::shared::error::ErrorCode::InternalError,
150            err.to_string(),
151        )
152    })
153}
154
155fn output_options(redaction: RedactionMode) -> agent_first_data::OutputOptions {
156    match redaction {
157        RedactionMode::Default => agent_first_data::OutputOptions {
158            redaction: agent_first_data::Redactor::new(),
159            style: agent_first_data::PlainStyle::Raw,
160        },
161        RedactionMode::None => agent_first_data::OutputOptions {
162            redaction: agent_first_data::Redactor::new()
163                .policy(agent_first_data::RedactionPolicy::Off),
164            style: agent_first_data::PlainStyle::Raw,
165        },
166    }
167}
168
169fn prepare_payload<T: Serialize>(code: &str, payload: &T) -> Result<serde_json::Value, Error> {
170    let value = serde_json::to_value(payload).map_err(|e| {
171        Error::new(
172            crate::shared::error::ErrorCode::InternalError,
173            format!("AFDATA: failed to serialize payload: {e}"),
174        )
175    })?;
176    wrap_payload(code, value).map(|value| crate::shared::redact::redact_url_fields(&value))
177}
178
179fn prepare_revealed_takeover_payload<T: Serialize>(
180    code: &str,
181    payload: &T,
182) -> Result<serde_json::Value, Error> {
183    let original = prepare_payload(code, payload)?;
184    let mut redacted = agent_first_data::Redactor::new().value(&original);
185    restore_named_field(&original, &mut redacted, "takeover_url_secret");
186    Ok(redacted)
187}
188
189fn restore_named_field(
190    original: &serde_json::Value,
191    redacted: &mut serde_json::Value,
192    field_name: &str,
193) {
194    match (original, redacted) {
195        (serde_json::Value::Object(original), serde_json::Value::Object(redacted)) => {
196            for (key, original_value) in original {
197                let Some(redacted_value) = redacted.get_mut(key) else {
198                    continue;
199                };
200                if key == field_name {
201                    *redacted_value = original_value.clone();
202                } else {
203                    restore_named_field(original_value, redacted_value, field_name);
204                }
205            }
206        }
207        (serde_json::Value::Array(original), serde_json::Value::Array(redacted)) => {
208            for (original, redacted) in original.iter().zip(redacted.iter_mut()) {
209                restore_named_field(original, redacted, field_name);
210            }
211        }
212        _ => {}
213    }
214}
215
216fn wrap_payload(code: &str, value: serde_json::Value) -> Result<serde_json::Value, Error> {
217    let serde_json::Value::Object(mut map) = value else {
218        return Err(Error::new(
219            crate::shared::error::ErrorCode::InternalError,
220            "AFDATA result payload must serialize to a JSON object",
221        ));
222    };
223
224    map.insert("code".into(), serde_json::Value::String(code.to_string()));
225    Ok(serde_json::Value::Object(map))
226}
227
228/// Convenience: emit an SDK-built AFDATA error event.
229pub fn emit_error<W: Write>(writer: &mut W, err: &Error) -> Result<(), Error> {
230    let mut emitter = agent_first_data::CliEmitter::with_options(
231        writer,
232        agent_first_data::OutputFormat::Json,
233        output_options(RedactionMode::Default),
234    )
235    .with_strict_protocol();
236    let event = agent_first_data::json_error(err.error_code.as_str(), &err.detail)
237        .retryable_if(err.retryable)
238        .build()
239        .map_err(|err| {
240            Error::new(
241                crate::shared::error::ErrorCode::InternalError,
242                err.to_string(),
243            )
244        })?;
245    emitter.emit(event).map_err(|emit_err| {
246        Error::new(
247            crate::shared::error::ErrorCode::InternalError,
248            emit_err.to_string(),
249        )
250    })
251}
252
253/// Emit an SDK-built AFDATA error event through `--output-to`.
254pub fn emit_process_error(err: &Error) -> Result<(), Error> {
255    let event = agent_first_data::json_error(err.error_code.as_str(), &err.detail)
256        .retryable_if(err.retryable)
257        .build()
258        .map_err(|error| {
259            Error::new(
260                crate::shared::error::ErrorCode::InternalError,
261                error.to_string(),
262            )
263        })?;
264    let mut emitter = agent_first_data::CliEmitter::from_output_to_with(
265        output_to(),
266        agent_first_data::OutputFormat::Json,
267        output_options(RedactionMode::Default),
268    )
269    .with_strict_protocol();
270    emitter.emit(event).map_err(|error| {
271        Error::new(
272            crate::shared::error::ErrorCode::InternalError,
273            error.to_string(),
274        )
275    })
276}
277
278/// Emit an error event with caller-owned extension fields and trace.
279pub fn emit_error_with<W: Write>(
280    writer: &mut W,
281    code: &str,
282    message: &str,
283    fields: serde_json::Value,
284    trace: serde_json::Value,
285) -> Result<(), Error> {
286    let mut emitter = agent_first_data::CliEmitter::with_options(
287        writer,
288        agent_first_data::OutputFormat::Json,
289        output_options(RedactionMode::Default),
290    )
291    .with_strict_protocol();
292    let retryable = fields
293        .get("retryable")
294        .and_then(serde_json::Value::as_bool)
295        .unwrap_or(false);
296    let fields = crate::shared::redact::redact_url_fields(&fields);
297    let trace = crate::shared::redact::redact_url_fields(&trace);
298    let fields = match fields {
299        serde_json::Value::Object(mut fields) => {
300            fields.remove("retryable");
301            serde_json::Value::Object(fields)
302        }
303        other => other,
304    };
305    let event = agent_first_data::json_error(code, message)
306        .retryable_if(retryable)
307        .fields(fields)
308        .trace(trace)
309        .build()
310        .map_err(|err| {
311            Error::new(
312                crate::shared::error::ErrorCode::InternalError,
313                err.to_string(),
314            )
315        })?;
316    emitter.emit(event).map_err(|err| {
317        Error::new(
318            crate::shared::error::ErrorCode::InternalError,
319            err.to_string(),
320        )
321    })
322}
323
324/// Emit an error with extension fields and trace through `--output-to`.
325pub fn emit_process_error_with(
326    code: &str,
327    message: &str,
328    fields: serde_json::Value,
329    trace: serde_json::Value,
330) -> Result<(), Error> {
331    let retryable = fields
332        .get("retryable")
333        .and_then(serde_json::Value::as_bool)
334        .unwrap_or(false);
335    let fields = crate::shared::redact::redact_url_fields(&fields);
336    let trace = crate::shared::redact::redact_url_fields(&trace);
337    let fields = match fields {
338        serde_json::Value::Object(mut fields) => {
339            fields.remove("retryable");
340            serde_json::Value::Object(fields)
341        }
342        other => other,
343    };
344    let event = agent_first_data::json_error(code, message)
345        .retryable_if(retryable)
346        .fields(fields)
347        .trace(trace)
348        .build()
349        .map_err(|error| {
350            Error::new(
351                crate::shared::error::ErrorCode::InternalError,
352                error.to_string(),
353            )
354        })?;
355    let mut emitter = agent_first_data::CliEmitter::from_output_to_with(
356        output_to(),
357        agent_first_data::OutputFormat::Json,
358        output_options(RedactionMode::Default),
359    )
360    .with_strict_protocol();
361    emitter.emit(event).map_err(|error| {
362        Error::new(
363            crate::shared::error::ErrorCode::InternalError,
364            error.to_string(),
365        )
366    })
367}
368
369/// Build a strict AFDATA error event for HTTP response bodies.
370pub fn error_value(code: &str, message: &str, retryable: bool) -> serde_json::Value {
371    let value: serde_json::Value = agent_first_data::json_error(code, message)
372        .retryable_if(retryable)
373        .build()
374        .map(Into::into)
375        .unwrap_or_else(|_| serde_json::json!({}));
376    crate::shared::redact::redact_value(&value)
377}
378
379/// Build a strict AFDATA result event for HTTP response bodies.
380pub fn result_value(code: &str, mut payload: serde_json::Value) -> serde_json::Value {
381    let payload = match &mut payload {
382        serde_json::Value::Object(fields) => {
383            fields
384                .entry("code".to_string())
385                .or_insert_with(|| serde_json::Value::String(code.to_string()));
386            payload
387        }
388        _ => serde_json::json!({"code": code, "value": payload}),
389    };
390    let value: serde_json::Value = agent_first_data::json_result(payload).build().into();
391    crate::shared::redact::redact_value(&value)
392}
393
394/// Build a strict HTTP result while intentionally revealing only
395/// `takeover_url_secret`. Used by the authenticated handoff-minting endpoint.
396pub fn result_value_revealing_takeover(
397    code: &str,
398    mut payload: serde_json::Value,
399) -> serde_json::Value {
400    let payload = match &mut payload {
401        serde_json::Value::Object(fields) => {
402            fields
403                .entry("code".to_string())
404                .or_insert_with(|| serde_json::Value::String(code.to_string()));
405            payload
406        }
407        _ => serde_json::json!({"code": code, "value": payload}),
408    };
409    let original: serde_json::Value = agent_first_data::json_result(payload).build().into();
410    let original = crate::shared::redact::redact_url_fields(&original);
411    let mut redacted = agent_first_data::Redactor::new().value(&original);
412    restore_named_field(&original, &mut redacted, "takeover_url_secret");
413    redacted
414}
415
416pub fn decode_result<T: DeserializeOwned>(bytes: &[u8]) -> Result<T, Error> {
417    let text = std::str::from_utf8(bytes).map_err(|error| {
418        Error::new(
419            crate::shared::error::ErrorCode::InternalError,
420            format!("decode AFDATA event: {error}"),
421        )
422    })?;
423    match agent_first_data::decode_protocol_event(text) {
424        Ok(agent_first_data::DecodedEvent::Result(result)) => serde_json::from_value(result.result)
425            .map_err(|error| {
426                Error::new(
427                    crate::shared::error::ErrorCode::InternalError,
428                    format!("decode AFDATA result payload: {error}"),
429                )
430            }),
431        Ok(_) => Err(Error::new(
432            crate::shared::error::ErrorCode::InternalError,
433            "expected AFDATA result event",
434        )),
435        Err(error) => Err(Error::new(
436            crate::shared::error::ErrorCode::InternalError,
437            format!("invalid AFDATA event: {error}"),
438        )),
439    }
440}
441
442pub fn decode_error(bytes: &[u8]) -> Result<Error, Error> {
443    let text = std::str::from_utf8(bytes).map_err(|error| {
444        Error::new(
445            crate::shared::error::ErrorCode::InternalError,
446            format!("decode AFDATA error event: {error}"),
447        )
448    })?;
449    match agent_first_data::decode_protocol_event(text) {
450        Ok(agent_first_data::DecodedEvent::Error(error)) => {
451            let code = serde_json::from_value(serde_json::Value::String(error.code))
452                .unwrap_or(crate::shared::error::ErrorCode::InternalError);
453            Ok(Error::new(code, error.message).with_retryable(error.retryable))
454        }
455        Ok(_) => Err(Error::new(
456            crate::shared::error::ErrorCode::InternalError,
457            "expected AFDATA error event",
458        )),
459        Err(error) => Err(Error::new(
460            crate::shared::error::ErrorCode::InternalError,
461            format!("invalid AFDATA error event: {error}"),
462        )),
463    }
464}
465
466#[cfg(test)]
467mod tests {
468    use super::*;
469
470    #[derive(Serialize)]
471    struct HealthPayload {
472        status: &'static str,
473        uptime_s: u64,
474    }
475
476    #[test]
477    fn json_result_event_is_single_line_with_code_field() {
478        let mut buf = Vec::new();
479        let payload = HealthPayload {
480            status: "ok",
481            uptime_s: 42,
482        };
483        emit(&mut buf, "health", &payload).unwrap();
484        let s = String::from_utf8(buf).unwrap_or_default();
485        assert!(s.ends_with('\n'));
486        let trimmed = s.trim_end();
487        let parsed: serde_json::Value = serde_json::from_str(trimmed).unwrap();
488        assert_eq!(parsed["kind"], "result");
489        assert_eq!(parsed["result"]["code"], "health");
490        assert_eq!(parsed["result"]["status"], "ok");
491        assert_eq!(parsed["result"]["uptime_s"], 42);
492        assert_eq!(trimmed.lines().count(), 1);
493    }
494
495    #[test]
496    fn error_event_uses_error_code_tag() {
497        let mut buf = Vec::new();
498        let err = Error::new(
499            crate::shared::error::ErrorCode::NavigationTimeout,
500            "no load",
501        );
502        emit_error(&mut buf, &err).unwrap();
503        let parsed: serde_json::Value =
504            serde_json::from_slice(&buf).unwrap_or(serde_json::Value::Null);
505        assert_eq!(parsed["kind"], "error");
506        assert_eq!(parsed["error"]["code"], "navigation_timeout");
507        assert_eq!(parsed["error"]["message"], "no load");
508        assert_eq!(parsed["error"]["retryable"], true);
509    }
510
511    #[test]
512    fn error_extension_fields_are_flattened_into_error_payload() {
513        let mut buf = Vec::new();
514        emit_error_with(
515            &mut buf,
516            "navigation_timeout",
517            "no load",
518            serde_json::json!({
519                "retryable": true,
520                "stage": "capture_text",
521                "details": "scalar detail remains an explicitly named field"
522            }),
523            serde_json::json!({"duration_ms": 10}),
524        )
525        .unwrap();
526        let parsed: serde_json::Value = serde_json::from_slice(&buf).unwrap();
527        assert_eq!(parsed["error"]["stage"], "capture_text");
528        assert_eq!(parsed["error"]["retryable"], true);
529        assert_eq!(
530            parsed["error"]["details"],
531            "scalar detail remains an explicitly named field"
532        );
533        assert!(parsed["error"].get("fields").is_none());
534    }
535
536    #[derive(Serialize)]
537    struct SecretPayload {
538        token_secret: &'static str,
539    }
540
541    #[test]
542    fn afdata_event_redacts_secret_fields() {
543        let mut buf = Vec::new();
544        emit(
545            &mut buf,
546            "container_status",
547            &SecretPayload {
548                token_secret: "supersecret",
549            },
550        )
551        .unwrap();
552        let parsed: serde_json::Value = serde_json::from_slice(&buf).unwrap();
553        assert_eq!(parsed["result"]["token_secret"], "***");
554    }
555
556    #[test]
557    fn http_result_redacts_common_url_query_credentials() {
558        let value = result_value(
559            "fetch",
560            serde_json::json!({
561                "request_url": "https://user:pass@example.test/?token=abc&safe=ok"
562            }),
563        );
564        assert_eq!(
565            value["result"]["request_url"],
566            "https://user:***@example.test/?token=***&safe=ok"
567        );
568    }
569
570    #[test]
571    fn takeover_result_reveals_only_the_explicit_capability() {
572        let value = result_value_revealing_takeover(
573            "takeover_handoff",
574            serde_json::json!({
575                "takeover_url_secret":
576                    "https://example.test/takeover?handoff_secret=short-lived",
577                "host_token_secret": "long-lived",
578                "request_url": "https://example.test/?token=also-secret"
579            }),
580        );
581        assert_eq!(
582            value["result"]["takeover_url_secret"],
583            "https://example.test/takeover?handoff_secret=short-lived"
584        );
585        assert_eq!(value["result"]["host_token_secret"], "***");
586        assert_eq!(
587            value["result"]["request_url"],
588            "https://example.test/?token=***"
589        );
590    }
591
592    #[test]
593    fn envelope_uses_sdk_result_payload_without_nested_envelope() {
594        let mut buf = Vec::new();
595        let payload = serde_json::json!({"code": -32000, "message": "cdp error"});
596        emit(&mut buf, "cdp", &payload).unwrap();
597        let parsed: serde_json::Value = serde_json::from_slice(&buf).unwrap();
598        assert_eq!(parsed["kind"], "result");
599        assert_eq!(parsed["result"]["code"], "cdp");
600        assert_eq!(parsed["result"]["code"], "cdp");
601        assert_eq!(parsed["result"]["message"], "cdp error");
602        assert!(parsed["result"].get("result").is_none());
603    }
604}
agent_first_http/shared/afdata.rs

agent_first_http/shared/
afdata.rs
