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    emit_process_progress_value(prepare_payload(code, payload)?, RedactionMode::Default)
79}
80
81/// Emit an otherwise-redacted result while intentionally revealing only the
82/// short-lived takeover capability field. The explicit `panel` and
83/// `fetch --takeover` operations are the only callers.
84pub fn emit_process_revealing_takeover<T: Serialize>(code: &str, payload: &T) -> Result<(), Error> {
85    let value = prepare_revealed_takeover_payload(code, payload)?;
86    emit_process_value(value, RedactionMode::None)
87}
88
89fn emit_process_progress_value(
90    value: serde_json::Value,
91    redaction: RedactionMode,
92) -> Result<(), Error> {
93    let event = agent_first_data::json_progress(value).build();
94    let mut emitter = agent_first_data::CliEmitter::from_output_to_with(
95        output_to(),
96        agent_first_data::OutputFormat::Json,
97        output_options(redaction),
98    )
99    .with_strict_protocol();
100    emitter.emit(event).map_err(|error| {
101        Error::new(
102            crate::shared::error::ErrorCode::InternalError,
103            error.to_string(),
104        )
105    })
106}
107
108#[derive(Clone, Copy)]
109enum RedactionMode {
110    Default,
111    None,
112}
113
114fn emit_inner<W: Write, T: Serialize>(
115    writer: &mut W,
116    code: &str,
117    payload: &T,
118    redaction: RedactionMode,
119) -> Result<(), Error> {
120    let value = prepare_payload(code, payload)?;
121
122    let options = output_options(redaction);
123    let mut emitter = agent_first_data::CliEmitter::with_options(
124        writer,
125        agent_first_data::OutputFormat::Json,
126        options,
127    )
128    .with_strict_protocol();
129    emitter.emit_result(value).map_err(|err| {
130        Error::new(
131            crate::shared::error::ErrorCode::InternalError,
132            err.to_string(),
133        )
134    })?;
135    Ok(())
136}
137
138fn emit_process_inner<T: Serialize>(
139    code: &str,
140    payload: &T,
141    redaction: RedactionMode,
142) -> Result<(), Error> {
143    emit_process_value(prepare_payload(code, payload)?, redaction)
144}
145
146fn emit_process_value(value: serde_json::Value, redaction: RedactionMode) -> Result<(), Error> {
147    let mut emitter = agent_first_data::CliEmitter::from_output_to_with(
148        output_to(),
149        agent_first_data::OutputFormat::Json,
150        output_options(redaction),
151    )
152    .with_strict_protocol();
153    emitter.emit_result(value).map_err(|err| {
154        Error::new(
155            crate::shared::error::ErrorCode::InternalError,
156            err.to_string(),
157        )
158    })
159}
160
161fn output_options(redaction: RedactionMode) -> agent_first_data::OutputOptions {
162    match redaction {
163        RedactionMode::Default => agent_first_data::OutputOptions {
164            redaction: agent_first_data::Redactor::new(),
165            style: agent_first_data::PlainStyle::Raw,
166        },
167        RedactionMode::None => agent_first_data::OutputOptions {
168            redaction: agent_first_data::Redactor::new()
169                .policy(agent_first_data::RedactionPolicy::Off),
170            style: agent_first_data::PlainStyle::Raw,
171        },
172    }
173}
174
175fn prepare_payload<T: Serialize>(code: &str, payload: &T) -> Result<serde_json::Value, Error> {
176    let value = serde_json::to_value(payload).map_err(|e| {
177        Error::new(
178            crate::shared::error::ErrorCode::InternalError,
179            format!("AFDATA: failed to serialize payload: {e}"),
180        )
181    })?;
182    wrap_payload(code, value).map(|value| crate::shared::redact::redact_url_fields(&value))
183}
184
185fn prepare_revealed_takeover_payload<T: Serialize>(
186    code: &str,
187    payload: &T,
188) -> Result<serde_json::Value, Error> {
189    let original = prepare_payload(code, payload)?;
190    let mut redacted = agent_first_data::Redactor::new().value(&original);
191    restore_named_field(&original, &mut redacted, "takeover_url_secret");
192    Ok(redacted)
193}
194
195fn restore_named_field(
196    original: &serde_json::Value,
197    redacted: &mut serde_json::Value,
198    field_name: &str,
199) {
200    match (original, redacted) {
201        (serde_json::Value::Object(original), serde_json::Value::Object(redacted)) => {
202            for (key, original_value) in original {
203                let Some(redacted_value) = redacted.get_mut(key) else {
204                    continue;
205                };
206                if key == field_name {
207                    *redacted_value = original_value.clone();
208                } else {
209                    restore_named_field(original_value, redacted_value, field_name);
210                }
211            }
212        }
213        (serde_json::Value::Array(original), serde_json::Value::Array(redacted)) => {
214            for (original, redacted) in original.iter().zip(redacted.iter_mut()) {
215                restore_named_field(original, redacted, field_name);
216            }
217        }
218        _ => {}
219    }
220}
221
222fn wrap_payload(code: &str, value: serde_json::Value) -> Result<serde_json::Value, Error> {
223    let serde_json::Value::Object(mut map) = value else {
224        return Err(Error::new(
225            crate::shared::error::ErrorCode::InternalError,
226            "AFDATA result payload must serialize to a JSON object",
227        ));
228    };
229
230    map.insert("code".into(), serde_json::Value::String(code.to_string()));
231    Ok(serde_json::Value::Object(map))
232}
233
234/// Convenience: emit an SDK-built AFDATA error event.
235pub fn emit_error<W: Write>(writer: &mut W, err: &Error) -> Result<(), Error> {
236    let mut emitter = agent_first_data::CliEmitter::with_options(
237        writer,
238        agent_first_data::OutputFormat::Json,
239        output_options(RedactionMode::Default),
240    )
241    .with_strict_protocol();
242    let event = agent_first_data::json_error(err.error_code.as_str(), &err.detail)
243        .retryable_if(err.retryable)
244        .build()
245        .map_err(|err| {
246            Error::new(
247                crate::shared::error::ErrorCode::InternalError,
248                err.to_string(),
249            )
250        })?;
251    emitter.emit(event).map_err(|emit_err| {
252        Error::new(
253            crate::shared::error::ErrorCode::InternalError,
254            emit_err.to_string(),
255        )
256    })
257}
258
259/// Emit an SDK-built AFDATA error event through `--output-to`.
260pub fn emit_process_error(err: &Error) -> Result<(), Error> {
261    let event = agent_first_data::json_error(err.error_code.as_str(), &err.detail)
262        .retryable_if(err.retryable)
263        .build()
264        .map_err(|error| {
265            Error::new(
266                crate::shared::error::ErrorCode::InternalError,
267                error.to_string(),
268            )
269        })?;
270    let mut emitter = agent_first_data::CliEmitter::from_output_to_with(
271        output_to(),
272        agent_first_data::OutputFormat::Json,
273        output_options(RedactionMode::Default),
274    )
275    .with_strict_protocol();
276    emitter.emit(event).map_err(|error| {
277        Error::new(
278            crate::shared::error::ErrorCode::InternalError,
279            error.to_string(),
280        )
281    })
282}
283
284/// Emit an error event with caller-owned extension fields and trace.
285pub fn emit_error_with<W: Write>(
286    writer: &mut W,
287    code: &str,
288    message: &str,
289    fields: serde_json::Value,
290    trace: serde_json::Value,
291) -> Result<(), Error> {
292    let mut emitter = agent_first_data::CliEmitter::with_options(
293        writer,
294        agent_first_data::OutputFormat::Json,
295        output_options(RedactionMode::Default),
296    )
297    .with_strict_protocol();
298    let retryable = fields
299        .get("retryable")
300        .and_then(serde_json::Value::as_bool)
301        .unwrap_or(false);
302    let fields = crate::shared::redact::redact_url_fields(&fields);
303    let trace = crate::shared::redact::redact_url_fields(&trace);
304    let fields = match fields {
305        serde_json::Value::Object(mut fields) => {
306            fields.remove("retryable");
307            serde_json::Value::Object(fields)
308        }
309        other => other,
310    };
311    let event = agent_first_data::json_error(code, message)
312        .retryable_if(retryable)
313        .fields(fields)
314        .trace(trace)
315        .build()
316        .map_err(|err| {
317            Error::new(
318                crate::shared::error::ErrorCode::InternalError,
319                err.to_string(),
320            )
321        })?;
322    emitter.emit(event).map_err(|err| {
323        Error::new(
324            crate::shared::error::ErrorCode::InternalError,
325            err.to_string(),
326        )
327    })
328}
329
330/// Emit an error with extension fields and trace through `--output-to`.
331pub fn emit_process_error_with(
332    code: &str,
333    message: &str,
334    fields: serde_json::Value,
335    trace: serde_json::Value,
336) -> Result<(), Error> {
337    let retryable = fields
338        .get("retryable")
339        .and_then(serde_json::Value::as_bool)
340        .unwrap_or(false);
341    let fields = crate::shared::redact::redact_url_fields(&fields);
342    let trace = crate::shared::redact::redact_url_fields(&trace);
343    let fields = match fields {
344        serde_json::Value::Object(mut fields) => {
345            fields.remove("retryable");
346            serde_json::Value::Object(fields)
347        }
348        other => other,
349    };
350    let event = agent_first_data::json_error(code, message)
351        .retryable_if(retryable)
352        .fields(fields)
353        .trace(trace)
354        .build()
355        .map_err(|error| {
356            Error::new(
357                crate::shared::error::ErrorCode::InternalError,
358                error.to_string(),
359            )
360        })?;
361    let mut emitter = agent_first_data::CliEmitter::from_output_to_with(
362        output_to(),
363        agent_first_data::OutputFormat::Json,
364        output_options(RedactionMode::Default),
365    )
366    .with_strict_protocol();
367    emitter.emit(event).map_err(|error| {
368        Error::new(
369            crate::shared::error::ErrorCode::InternalError,
370            error.to_string(),
371        )
372    })
373}
374
375/// Build a strict AFDATA error event for HTTP response bodies.
376pub fn error_value(code: &str, message: &str, retryable: bool) -> serde_json::Value {
377    let value: serde_json::Value = agent_first_data::json_error(code, message)
378        .retryable_if(retryable)
379        .build()
380        .map(Into::into)
381        .unwrap_or_else(|_| serde_json::json!({}));
382    crate::shared::redact::redact_value(&value)
383}
384
385/// Build a strict AFDATA result event for HTTP response bodies.
386pub fn result_value(code: &str, mut payload: serde_json::Value) -> serde_json::Value {
387    let payload = match &mut payload {
388        serde_json::Value::Object(fields) => {
389            fields
390                .entry("code".to_string())
391                .or_insert_with(|| serde_json::Value::String(code.to_string()));
392            payload
393        }
394        _ => serde_json::json!({"code": code, "value": payload}),
395    };
396    let value: serde_json::Value = agent_first_data::json_result(payload).build().into();
397    crate::shared::redact::redact_value(&value)
398}
399
400/// Build a strict HTTP result while intentionally revealing only
401/// `takeover_url_secret`. Used by the authenticated handoff-minting endpoint.
402pub fn result_value_revealing_takeover(
403    code: &str,
404    mut payload: serde_json::Value,
405) -> serde_json::Value {
406    let payload = match &mut payload {
407        serde_json::Value::Object(fields) => {
408            fields
409                .entry("code".to_string())
410                .or_insert_with(|| serde_json::Value::String(code.to_string()));
411            payload
412        }
413        _ => serde_json::json!({"code": code, "value": payload}),
414    };
415    let original: serde_json::Value = agent_first_data::json_result(payload).build().into();
416    let original = crate::shared::redact::redact_url_fields(&original);
417    let mut redacted = agent_first_data::Redactor::new().value(&original);
418    restore_named_field(&original, &mut redacted, "takeover_url_secret");
419    redacted
420}
421
422pub fn decode_result<T: DeserializeOwned>(bytes: &[u8]) -> Result<T, Error> {
423    let text = std::str::from_utf8(bytes).map_err(|error| {
424        Error::new(
425            crate::shared::error::ErrorCode::InternalError,
426            format!("decode AFDATA event: {error}"),
427        )
428    })?;
429    match agent_first_data::decode_protocol_event(text) {
430        Ok(agent_first_data::DecodedEvent::Result(result)) => serde_json::from_value(result.result)
431            .map_err(|error| {
432                Error::new(
433                    crate::shared::error::ErrorCode::InternalError,
434                    format!("decode AFDATA result payload: {error}"),
435                )
436            }),
437        Ok(_) => Err(Error::new(
438            crate::shared::error::ErrorCode::InternalError,
439            "expected AFDATA result event",
440        )),
441        Err(error) => Err(Error::new(
442            crate::shared::error::ErrorCode::InternalError,
443            format!("invalid AFDATA event: {error}"),
444        )),
445    }
446}
447
448pub fn decode_error(bytes: &[u8]) -> Result<Error, Error> {
449    let text = std::str::from_utf8(bytes).map_err(|error| {
450        Error::new(
451            crate::shared::error::ErrorCode::InternalError,
452            format!("decode AFDATA error event: {error}"),
453        )
454    })?;
455    match agent_first_data::decode_protocol_event(text) {
456        Ok(agent_first_data::DecodedEvent::Error(error)) => {
457            let code = serde_json::from_value(serde_json::Value::String(error.code))
458                .unwrap_or(crate::shared::error::ErrorCode::InternalError);
459            Ok(Error::new(code, error.message).with_retryable(error.retryable))
460        }
461        Ok(_) => Err(Error::new(
462            crate::shared::error::ErrorCode::InternalError,
463            "expected AFDATA error event",
464        )),
465        Err(error) => Err(Error::new(
466            crate::shared::error::ErrorCode::InternalError,
467            format!("invalid AFDATA error event: {error}"),
468        )),
469    }
470}
471
472#[cfg(test)]
473mod tests {
474    use super::*;
475
476    #[derive(Serialize)]
477    struct HealthPayload {
478        status: &'static str,
479        uptime_s: u64,
480    }
481
482    #[test]
483    fn json_result_event_is_single_line_with_code_field() {
484        let mut buf = Vec::new();
485        let payload = HealthPayload {
486            status: "ok",
487            uptime_s: 42,
488        };
489        emit(&mut buf, "health", &payload).unwrap();
490        let s = String::from_utf8(buf).unwrap_or_default();
491        assert!(s.ends_with('\n'));
492        let trimmed = s.trim_end();
493        let parsed: serde_json::Value = serde_json::from_str(trimmed).unwrap();
494        assert_eq!(parsed["kind"], "result");
495        assert_eq!(parsed["result"]["code"], "health");
496        assert_eq!(parsed["result"]["status"], "ok");
497        assert_eq!(parsed["result"]["uptime_s"], 42);
498        assert_eq!(trimmed.lines().count(), 1);
499    }
500
501    #[test]
502    fn error_event_uses_error_code_tag() {
503        let mut buf = Vec::new();
504        let err = Error::new(
505            crate::shared::error::ErrorCode::NavigationTimeout,
506            "no load",
507        );
508        emit_error(&mut buf, &err).unwrap();
509        let parsed: serde_json::Value =
510            serde_json::from_slice(&buf).unwrap_or(serde_json::Value::Null);
511        assert_eq!(parsed["kind"], "error");
512        assert_eq!(parsed["error"]["code"], "navigation_timeout");
513        assert_eq!(parsed["error"]["message"], "no load");
514        assert_eq!(parsed["error"]["retryable"], true);
515    }
516
517    #[test]
518    fn error_extension_fields_are_flattened_into_error_payload() {
519        let mut buf = Vec::new();
520        emit_error_with(
521            &mut buf,
522            "navigation_timeout",
523            "no load",
524            serde_json::json!({
525                "retryable": true,
526                "stage": "capture_text",
527                "details": "scalar detail remains an explicitly named field"
528            }),
529            serde_json::json!({"duration_ms": 10}),
530        )
531        .unwrap();
532        let parsed: serde_json::Value = serde_json::from_slice(&buf).unwrap();
533        assert_eq!(parsed["error"]["stage"], "capture_text");
534        assert_eq!(parsed["error"]["retryable"], true);
535        assert_eq!(
536            parsed["error"]["details"],
537            "scalar detail remains an explicitly named field"
538        );
539        assert!(parsed["error"].get("fields").is_none());
540    }
541
542    #[derive(Serialize)]
543    struct SecretPayload {
544        token_secret: &'static str,
545    }
546
547    #[test]
548    fn afdata_event_redacts_secret_fields() {
549        let mut buf = Vec::new();
550        emit(
551            &mut buf,
552            "container_status",
553            &SecretPayload {
554                token_secret: "supersecret",
555            },
556        )
557        .unwrap();
558        let parsed: serde_json::Value = serde_json::from_slice(&buf).unwrap();
559        assert_eq!(parsed["result"]["token_secret"], "***");
560    }
561
562    #[test]
563    fn http_result_redacts_common_url_query_credentials() {
564        let value = result_value(
565            "fetch",
566            serde_json::json!({
567                "request_url": "https://user:pass@example.test/?token=abc&safe=ok"
568            }),
569        );
570        assert_eq!(
571            value["result"]["request_url"],
572            "https://user:***@example.test/?token=***&safe=ok"
573        );
574    }
575
576    #[test]
577    fn takeover_result_reveals_only_the_explicit_capability() {
578        let value = result_value_revealing_takeover(
579            "takeover_handoff",
580            serde_json::json!({
581                "takeover_url_secret":
582                    "https://example.test/takeover?handoff_secret=short-lived",
583                "host_token_secret": "long-lived",
584                "request_url": "https://example.test/?token=also-secret"
585            }),
586        );
587        assert_eq!(
588            value["result"]["takeover_url_secret"],
589            "https://example.test/takeover?handoff_secret=short-lived"
590        );
591        assert_eq!(value["result"]["host_token_secret"], "***");
592        assert_eq!(
593            value["result"]["request_url"],
594            "https://example.test/?token=***"
595        );
596    }
597
598    #[test]
599    fn envelope_uses_sdk_result_payload_without_nested_envelope() {
600        let mut buf = Vec::new();
601        let payload = serde_json::json!({"code": -32000, "message": "cdp error"});
602        emit(&mut buf, "cdp", &payload).unwrap();
603        let parsed: serde_json::Value = serde_json::from_slice(&buf).unwrap();
604        assert_eq!(parsed["kind"], "result");
605        assert_eq!(parsed["result"]["code"], "cdp");
606        assert_eq!(parsed["result"]["code"], "cdp");
607        assert_eq!(parsed["result"]["message"], "cdp error");
608        assert!(parsed["result"].get("result").is_none());
609    }
610}
agent_first_http/shared/afdata.rs

agent_first_http/shared/
afdata.rs
