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

agent_first_http/shared/
afdata.rs
