fix(test): make twin-openai streamed message items round-trip as input (#484)

## What

Fixes the `openai_twin_*` parity-matrix failures that have been on
`main` since #449: every multi-turn scenario whose scripted response
includes text fails on its second turn with 400 `"message input items
require supported content"`.

## Root cause

Two twin behaviors collided (bisected: passes at #447, fails at #449):

1. **The twin's streaming `response.output_item.done` for message items
omitted the `content` array** (`test/twin/openai/src/sse.rs`) — it sent
only `id`/`type`/`status`/`role`, where the real API sends the completed
item in full. The openai adapter preserves message output items verbatim
(`ContentPart::Other { kind: OPENAI_MESSAGE }`) and replays them as
assistant history on the next turn — required so reasoning items keep
their "required following item" in Responses round-trips. So the replay
arrived content-less.
2. **#449 tightened the twin's input validation** to also validate
explicit `type: "message"` items (previously only type-less items were
validated as messages; anything with an explicit type was accepted
unchecked). The twin started rejecting its own round-tripped output.

The new validation caught a real infidelity in the emitter — the emit
side is what's wrong.

Nobody noticed because **CI never runs the twin e2e suites**: `rust.yml`
runs `--profile ci` without `--run-ignored`, so the parity matrix only
runs when someone invokes the e2e profile locally.

## Fix

- The streamed message `output_item.done` now carries its `output_text`
content, matching the real API and the twin's own non-streaming
`responses_json()`.
- The input validator accepts `output_text` parts on **assistant**
message items (the real API allows these; the twin's non-streaming
responses already require it for faithful replay). Non-assistant
`output_text` parts get a dedicated rejection message.

## Tests

- New contract test
`responses_stream_message_item_done_round_trips_as_input`: streams a
response, asserts the completed message item carries its `output_text`
content, and replays the item verbatim as assistant-history input,
asserting the twin accepts its own output.
- `cargo nextest run -p twin-openai` — 56 passed
- `cargo nextest run -p fabro-agent -E 'test(parity)' --run-ignored
only` — **91/91 passed** (was 7 failing)
- `cargo nextest run -p fabro-llm --run-ignored only` — 10 passed
- `cargo nextest run --workspace` — green apart from two pre-existing
env-dependent `fabro-workflow` failures that reproduce on clean `main`
in shells with provider API keys exported (unrelated; CI is green on
them because it has no such keys)
- clippy `-D warnings` / fmt — clean

Found while reviewing #481 (whose parity runs surfaced this); #481
itself is unaffected — it doesn't touch the openai adapter or the twin,
and the failures exist on its merge-base.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

## CI (separate commit, drop if unwanted)

`ci: run twin-mode e2e suites on Linux` adds a step to the existing
Linux test job running the ignored twin-mode suites for the packages
that are fully green today (`fabro-agent`, `fabro-llm`, `twin-openai`) —
104 tests, ~1s on a warm build, no secrets needed (live-only tests
self-skip in twin mode). This is what would have caught the #449
regression. The remaining ignored suites (fabro-cli twin tests,
Docker/Daytona sandbox tests, fabro-spa asset test) need their own fixes
before joining; widen the `-E` filter as they're cleaned up. Note the
step deliberately avoids the `e2e` nextest profile, since
`NEXTEST_PROFILE=e2e` implies strict mode, which fails on missing
secrets.

---------

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>

.github/workflows/rust.yml

@@ -108,6 +108,13 @@ jobs:
           cache-on-failure: true
       - uses: taiki-e/install-action@773334c0e05d7e699e4d78234494308223f3a2cf # nextest
       - run: cargo nextest run --locked --workspace --status-level slow --profile ci
+      # Twin-mode e2e suites. These are hermetic (in-process twin provider, no
+      # secrets): FABRO_TEST_MODE defaults to twin, so live-only tests
+      # self-skip. Scoped to the packages whose ignored tests are fully green
+      # in twin mode; widen as the remaining suites are fixed up for CI.
+      # Must not use the e2e nextest profile here: NEXTEST_PROFILE=e2e implies
+      # strict mode, which fails (rather than skips) live tests without keys.
+      - run: cargo nextest run --locked --workspace --status-level slow --profile ci --run-ignored only -E 'package(fabro-agent) + package(fabro-llm) + package(twin-openai)'
 
   test-macos:
     name: Test (macOS)

test/twin/openai/src/openai/models.rs

@@ -344,15 +344,16 @@ fn validate_message_input_item(item: &InputItem) -> Result<(), OpenAiError> {
         ));
     }
 
-    validate_input_content(&item.content)
+    let role = item.role.as_deref().unwrap_or_default();
+    validate_input_content(role, &item.content)
 }
 
-fn validate_input_content(content: &InputContent) -> Result<(), OpenAiError> {
+fn validate_input_content(role: &str, content: &InputContent) -> Result<(), OpenAiError> {
     match content {
         InputContent::Text(_) => Ok(()),
         InputContent::Parts(parts) if !parts.is_empty() => {
             for part in parts {
-                validate_input_content_part(part)?;
+                validate_input_content_part(role, part)?;
             }
             Ok(())
         }
@@ -363,9 +364,12 @@ fn validate_input_content(content: &InputContent) -> Result<(), OpenAiError> {
     }
 }
 
-fn validate_input_content_part(part: &ContentPart) -> Result<(), OpenAiError> {
+fn validate_input_content_part(role: &str, part: &ContentPart) -> Result<(), OpenAiError> {
     match part.kind.as_str() {
         "input_text" | "text" if part.text.as_deref().is_some() => Ok(()),
+        // Assistant history items are replayed with their original output
+        // parts; the real API accepts output_text on assistant messages.
+        "output_text" if role == "assistant" && part.text.as_deref().is_some() => Ok(()),
         "input_image"
             if part
                 .image_url
@@ -378,6 +382,14 @@ fn validate_input_content_part(part: &ContentPart) -> Result<(), OpenAiError> {
             "input",
             "text input parts require text",
         )),
+        "output_text" if role == "assistant" => Err(OpenAiError::invalid_request(
+            "input",
+            "text input parts require text",
+        )),
+        "output_text" => Err(OpenAiError::invalid_request(
+            "input",
+            "output_text parts are only valid on assistant messages",
+        )),
         "input_image" => Err(OpenAiError::invalid_request(
             "input",
             "image input parts require a supported image_url",

test/twin/openai/src/sse.rs

@@ -149,6 +149,9 @@ pub fn responses_sse_response(plan: &ResponsePlan, transport: TransportOptions)
             ));
         }
 
+        // The completed item carries its full content, like the real API.
+        // Adapters round-trip this item verbatim into the next request's
+        // input, so omitting content here produces an invalid replay.
         events.push(sse_event(
             "response.output_item.done",
             &json!({
@@ -158,6 +161,10 @@ pub fn responses_sse_response(plan: &ResponsePlan, transport: TransportOptions)
                     "type": "message",
                     "status": "completed",
                     "role": "assistant",
+                    "content": [{
+                        "type": "output_text",
+                        "text": message_text,
+                    }],
                 },
                 "output_index": next_output_index,
             }),

test/twin/openai/tests/responses_contract.rs

@@ -225,6 +225,66 @@ async fn responses_reject_malformed_image_input() {
     assert_eq!(body["error"]["param"], "input");
 }
 
+#[tokio::test]
+async fn responses_stream_message_item_done_round_trips_as_input() {
+    let server = common::spawn_server().await.expect("server should start");
+
+    let (status, chunks) = server
+        .post_responses_stream(json!({
+            "model": "gpt-test",
+            "input": "stream this request",
+            "stream": true
+        }))
+        .await;
+    assert_eq!(status, 200);
+
+    let joined = chunks.join("");
+    let transcript = common::parse_sse_transcript(joined.as_bytes()).expect("valid sse");
+    let message_item = transcript
+        .events
+        .iter()
+        .filter(|event| event.event.as_deref() == Some("response.output_item.done"))
+        .filter_map(|event| serde_json::from_str::<serde_json::Value>(&event.data).ok())
+        .map(|payload| payload["item"].clone())
+        .find(|item| item["type"] == "message")
+        .expect("stream should emit a completed message item");
+
+    // The completed item carries its full content, like the real API.
+    let content = message_item["content"]
+        .as_array()
+        .expect("completed message item should include content");
+    assert_eq!(content.len(), 1);
+    assert_eq!(content[0]["type"], "output_text");
+    assert!(
+        content[0]["text"]
+            .as_str()
+            .is_some_and(|text| !text.is_empty())
+    );
+
+    // Adapters replay the completed item verbatim as assistant history on the
+    // next turn, so the twin must accept its own streamed output as input.
+    let response = server
+        .post_responses(json!({
+            "model": "gpt-test",
+            "input": [
+                {
+                    "type": "message",
+                    "role": "user",
+                    "content": [{"type": "input_text", "text": "stream this request"}]
+                },
+                message_item,
+                {
+                    "type": "message",
+                    "role": "user",
+                    "content": [{"type": "input_text", "text": "and again"}]
+                },
+            ],
+            "stream": false
+        }))
+        .await;
+    assert_eq!(response.status(), 200);
+}
+
 #[tokio::test]
 async fn responses_stream_emits_expected_sse_sequence() {
     let server = common::spawn_server().await.expect("server should start");