feat: complete file-search streaming and API compatibility

.github/workflows/rust.yml

@@ -74,17 +74,29 @@ jobs:
         run: cargo fmt --check
 
       - name: Run clippy
-        run: cargo clippy --all-targets -- -D warnings
+        run: cargo clippy --workspace --all-targets --all-features --locked -- -D warnings
 
       - name: Run tests
-        run: cargo test
+        run: cargo test --workspace --all-features --locked
 
       - name: Test Codex and Claude launcher contracts
         run: bash scripts/tests/agentic-launchers-test.sh
 
       - name: Test agentic CLI end to end
         run: python3 scripts/tests/agentic-cli-e2e-test.py
 
+      - name: Install Python for strict SDK contracts
+        uses: actions/setup-python@a309ff8b426b58ec0e2a45f0f869d46889d02405 # v6.2.0
+        with:
+          python-version: '3.13'
+
+      - name: Verify Files, Vector Stores, and file-search SDK contracts
+        run: |
+          python3 -m venv "${RUNNER_TEMP}/file-search-sdk"
+          "${RUNNER_TEMP}/file-search-sdk/bin/pip" install 'openai==3.13.0'
+          cargo build -p agentic-server --all-features --locked
+          "${RUNNER_TEMP}/file-search-sdk/bin/python" scripts/tests/file-search-sdk-test.py target/debug/agentic-server -v
+
   postgres:
     runs-on: ubuntu-latest
     services:
@@ -129,20 +141,13 @@ jobs:
       - name: Provision pgvector extension
         run: psql "$TEST_POSTGRES_URL" -v ON_ERROR_STOP=1 -c "CREATE EXTENSION IF NOT EXISTS vector"
 
-      - name: Verify PostgreSQL schema upgrades and validation
-        run: cargo test -p agentic-server-core --lib -- --ignored --test-threads=1
-
-      - name: Verify PostgreSQL migrations and restart persistence
-        run: cargo test -p agentic-server-core --test postgres_storage_integration -- --ignored --test-threads=1
-
-      - name: Verify PostgreSQL file search persistence
-        run: cargo test -p agentic-server-core --test file_search_service postgres_file_search -- --ignored --test-threads=1
-
-      - name: Verify real pgvector indexed retrieval and transactional publication
-        run: cargo test -p agentic-server-core --test file_search_service postgres_pgvector -- --ignored --test-threads=1
+      # Every ignored core test currently requires this isolated PostgreSQL service.
+      # Keep new lifecycle/runtime targets in this gate; never add paid/network tests here.
+      - name: Verify all PostgreSQL storage, retrieval, lifecycle, and worker contracts
+        run: cargo test -p agentic-server-core --all-features --locked --lib --tests -- --ignored --test-threads=1
 
       - name: Verify bounded PDF ingestion
-        run: cargo test -p agentic-server-core --features file-search-pdf --test file_search_service pdf
+        run: cargo test -p agentic-server-core --features file-search-pdf --locked --test file_search_service pdf
 
   provider-upstream:
     runs-on: ubuntu-latest

crates/agentic-server-core/src/events/normalize.rs

@@ -102,6 +102,7 @@ fn extract_payload(event_type: SSEEventType, json: &Value) -> EventPayload {
         | SSEEventType::McpListToolsInProgress
         | SSEEventType::McpListToolsCompleted
         | SSEEventType::McpListToolsFailed
+        | SSEEventType::OutputTextAnnotationAdded
         | SSEEventType::Other => EventPayload::Raw(json.clone()),
     }
 }

crates/agentic-server-core/src/events/types.rs

@@ -125,6 +125,7 @@ pub enum SSEEventType {
     // Text content
     OutputTextDelta,
     OutputTextDone,
+    OutputTextAnnotationAdded,
     ContentPartAdded,
     ContentPartDone,
 
@@ -177,6 +178,7 @@ impl From<&str> for SSEEventType {
             "response.output_item.done" => Self::OutputItemDone,
             "response.output_text.delta" => Self::OutputTextDelta,
             "response.output_text.done" => Self::OutputTextDone,
+            "response.output_text.annotation.added" => Self::OutputTextAnnotationAdded,
             "response.content_part.added" => Self::ContentPartAdded,
             "response.content_part.done" => Self::ContentPartDone,
             "response.function_call_arguments.delta" => Self::FunctionCallArgumentsDelta,
@@ -225,6 +227,7 @@ impl TryFrom<SSEEventType> for &'static str {
             SSEEventType::OutputItemDone => Ok("response.output_item.done"),
             SSEEventType::OutputTextDelta => Ok("response.output_text.delta"),
             SSEEventType::OutputTextDone => Ok("response.output_text.done"),
+            SSEEventType::OutputTextAnnotationAdded => Ok("response.output_text.annotation.added"),
             SSEEventType::ContentPartAdded => Ok("response.content_part.added"),
             SSEEventType::ContentPartDone => Ok("response.content_part.done"),
             SSEEventType::FunctionCallArgumentsDelta => Ok("response.function_call_arguments.delta"),

crates/agentic-server-core/src/events/validate.rs

@@ -46,7 +46,7 @@ pub(crate) fn validate_frame(frame: &EventFrame) -> Result<ValidatedFrame<'_>, E
         SSEEventType::OutputItemDone => {
             validate_output_item(frame, event_name, true).map(|item| ValidatedFrame { item: Some(item) })
         }
-        SSEEventType::Other => Ok(ValidatedFrame { item: None }),
+        SSEEventType::OutputTextAnnotationAdded | SSEEventType::Other => Ok(ValidatedFrame { item: None }),
         event_type => {
             let output_index = required_output_index(frame, event_name)?;
             let item_id = validate_event_item_id(frame, event_name)?;
@@ -110,6 +110,7 @@ pub(crate) fn expected_item_type(event_type: SSEEventType) -> Option<SSEItemType
         | SSEEventType::ResponseIncomplete
         | SSEEventType::OutputItemAdded
         | SSEEventType::OutputItemDone
+        | SSEEventType::OutputTextAnnotationAdded
         | SSEEventType::Other => None,
     }
 }
@@ -276,6 +277,7 @@ fn validate_event_fields(
         | SSEEventType::McpListToolsInProgress
         | SSEEventType::McpListToolsCompleted
         | SSEEventType::McpListToolsFailed
+        | SSEEventType::OutputTextAnnotationAdded
         | SSEEventType::Other => None,
     };
     if let Some(field) = required {

crates/agentic-server-core/src/executor/accumulator/mod.rs

@@ -633,8 +633,9 @@ impl ResponseAccumulator {
             previous_response_id: previous_response_id.map(str::to_string),
             conversation_id: self.conversation_id,
             instructions: instructions.map(str::to_string),
-            tools: None,
-            tool_choice: None,
+            tools: Vec::new(),
+            tool_choice: crate::types::io::ToolChoice::default(),
+            parallel_tool_calls: false,
         }
     }
 }

crates/agentic-server-core/src/executor/engine.rs

@@ -469,9 +469,13 @@ async fn run_compaction_trigger(
         previous_response_id: ctx.original_request.previous_response_id.clone(),
         conversation_id: ctx.conversation_id.clone(),
         instructions,
-        tools: None,
-        tool_choice: None,
+        tools: ctx.enriched_request.tools.clone().unwrap_or_default(),
+        tool_choice: ctx.enriched_request.tool_choice.clone().unwrap_or_default(),
+        parallel_tool_calls: ctx.enriched_request.parallel_tool_calls.unwrap_or(false),
     };
+    for tool in &mut payload.tools {
+        tool.sanitize_for_persistence();
+    }
     ctx.inject_ids(&mut payload);
     Ok(payload)
 }
@@ -606,6 +610,9 @@ struct StreamFailureContext {
     model: String,
     previous_response_id: Option<String>,
     instructions: Option<String>,
+    tools: Vec<crate::types::tools::ResponsesTool>,
+    tool_choice: ToolChoice,
+    parallel_tool_calls: bool,
 }
 
 impl From<&RequestContext> for StreamFailureContext {
@@ -616,6 +623,19 @@ impl From<&RequestContext> for StreamFailureContext {
             model: ctx.enriched_request.model.clone(),
             previous_response_id: ctx.original_request.previous_response_id.clone(),
             instructions: ctx.original_request.instructions.clone(),
+            tools: ctx
+                .enriched_request
+                .tools
+                .clone()
+                .unwrap_or_default()
+                .into_iter()
+                .map(|mut tool| {
+                    tool.sanitize_for_persistence();
+                    tool
+                })
+                .collect(),
+            tool_choice: ctx.enriched_request.tool_choice.clone().unwrap_or_default(),
+            parallel_tool_calls: ctx.enriched_request.parallel_tool_calls.unwrap_or(false),
         }
     }
 }
@@ -639,8 +659,9 @@ impl StreamFailureContext {
             previous_response_id: self.previous_response_id.clone(),
             conversation_id: self.conversation_id.clone(),
             instructions: self.instructions.clone(),
-            tools: None,
-            tool_choice: None,
+            tools: self.tools.clone(),
+            tool_choice: self.tool_choice.clone(),
+            parallel_tool_calls: self.parallel_tool_calls,
         }
     }
 }

crates/agentic-server-core/src/executor/gateway_accumulator.rs

@@ -9,6 +9,7 @@ pub struct GatewayStreamAccumulator {
     next_sequence_number: u64,
     emitted_created: bool,
     emitted_in_progress: bool,
+    citations: super::stream_citations::StreamCitations,
 }
 
 pub(super) struct StreamEvent {
@@ -28,6 +29,7 @@ impl GatewayStreamAccumulator {
             next_sequence_number: 0,
             emitted_created: false,
             emitted_in_progress: false,
+            citations: super::stream_citations::StreamCitations::default(),
         }
     }
 
@@ -50,10 +52,20 @@ impl GatewayStreamAccumulator {
         rebase_output_index(&mut frame.wire, output_offset);
     }
 
+    pub(super) fn citation_frames(&mut self, frame: &EventFrame, offset: usize) -> ExecutorResult<Vec<EventFrame>> {
+        self.citations.before_done(frame, offset)
+    }
+
     pub(crate) fn terminal_response_chunk(&mut self, payload: &ResponsePayload) -> ExecutorResult<String> {
         let mut frame = terminal_response_frame(payload)?;
+        let mut chunk = String::new();
+        for mut annotation in self.citation_frames(&frame, 0)? {
+            self.stamp_event(&mut annotation, 0);
+            chunk.push_str(&checked_stream_event(&annotation)?);
+        }
         self.stamp_event(&mut frame, 0);
-        checked_stream_event(&frame)
+        chunk.push_str(&checked_stream_event(&frame)?);
+        Ok(chunk)
     }
 
     #[cfg(test)]

crates/agentic-server-core/src/executor/mod.rs

@@ -21,6 +21,7 @@ pub mod gateway_accumulator;
 mod pending_calls;
 mod pipeline;
 mod response_budget;
+mod stream_citations;
 mod upstream;
 
 pub use compaction::compact_response;

crates/agentic-server-core/src/executor/pipeline.rs

@@ -25,6 +25,7 @@ pub(super) struct StreamPayload {
 /// Lives for the response, preserving gateway event numbering across inference rounds.
 pub(super) struct AgentPipeline {
     pub(super) request: RequestContext,
+    response_tool_choice: crate::types::io::ToolChoice,
     tool_search_state: Option<ToolSearchState>,
     delivery: StreamDelivery,
     round: Option<RoundIngestion>,
@@ -37,13 +38,29 @@ impl AgentPipeline {
         sender: Option<Sender<StreamEvent>>,
     ) -> Self {
         Self {
+            response_tool_choice: request.enriched_request.tool_choice.clone().unwrap_or_default(),
             request,
             tool_search_state,
             delivery: StreamDelivery::new(sender),
             round: None,
         }
     }
 
+    pub(super) fn response_tool_choice(&self) -> &crate::types::io::ToolChoice {
+        &self.response_tool_choice
+    }
+
+    pub(super) fn response_tools(&self) -> Vec<crate::types::tools::ResponsesTool> {
+        let mut tools = self.tool_search_state().filter(|state| state.is_active()).map_or_else(
+            || self.request.enriched_request.tools.clone().unwrap_or_default(),
+            ToolSearchState::public_response_tools,
+        );
+        for tool in &mut tools {
+            tool.sanitize_for_persistence();
+        }
+        tools
+    }
+
     pub(super) fn tool_search_state(&self) -> Option<&ToolSearchState> {
         self.tool_search_state.as_ref()
     }
@@ -114,6 +131,9 @@ impl AgentPipeline {
             self.request.original_request.previous_response_id.as_deref(),
             self.request.original_request.instructions.as_deref(),
         )?;
+        payload.tools = self.response_tools();
+        payload.tool_choice = self.response_tool_choice.clone();
+        payload.parallel_tool_calls = self.request.enriched_request.parallel_tool_calls.unwrap_or(false);
         self.request.inject_ids(&mut payload);
         Ok(payload)
     }

crates/agentic-server-core/src/executor/pipeline/delivery.rs

@@ -148,7 +148,27 @@ fn should_defer_stream_event(frame: &EventFrame, defer_from_output_index: Option
 }
 
 async fn emit_stream_frame(frame: &mut EventFrame, emit_ctx: &mut StreamEmitContext<'_>) -> ExecutorResult<bool> {
+    // Completed grounded text supplies authoritative offsets and annotation order.
+    if frame.event_type == SSEEventType::OutputTextAnnotationAdded
+        && frame
+            .wire
+            .rest
+            .get("annotation")
+            .and_then(|annotation| annotation.get("type"))
+            .and_then(Value::as_str)
+            == Some("file_citation")
+    {
+        return Ok(false);
+    }
     apply_context_response_ids(&mut frame.wire, emit_ctx.request);
+    for mut annotation in emit_ctx.accumulator.citation_frames(frame, emit_ctx.output_offset)? {
+        if emit_ctx
+            .accumulator
+            .process_event(&mut annotation, emit_ctx.output_offset)
+        {
+            emit_sse_frame(emit_ctx.sender, &annotation).await?;
+        }
+    }
     let emitted = emit_ctx.accumulator.process_event(frame, emit_ctx.output_offset);
     if emitted {
         emit_sse_frame(emit_ctx.sender, frame).await?;