chore: sync public mirror from internal

.repository-projection.json

@@ -3,11 +3,11 @@
   "projection": "deixic-code",
   "projectionSchemaVersion": 1,
   "sourceRepository": "dx-corp/mono",
-  "sourceSha": "4abc96e0aea2eaf402ea0bd8948761e7edd00b02",
+  "sourceSha": "920d7b402cd147ee686d5afc90a1ccc887f49f42",
   "destinationRepository": "dx-corp/code",
-  "priorProjectedBase": "1f8e470402d994cc0d531e99f0876071af64f5d9",
+  "priorProjectedBase": "441b133f90b75324d77935e1a54bb3c48ce460c3",
   "definitionDigest": "cb9d429542ebb0a2de9b42a7aad60d9d8696a648ceba47c30f05c0b285ca0db7",
   "toolDigest": "f6d8b2a0b6f867d1db845a086635357eed196054b6a47f0695e49fdde616d6e6",
-  "contentDigest": "5590aad944ea66c46d26e0477ecd312c00f719907e6751dcd629603a1e1ebd6d",
+  "contentDigest": "b0b525fb563e288920995306519c45d31b8847effbec7cf1bc006fc8ef4f6077",
   "publicationEligible": true
 }

packages/dex-host-rs/src/host_turn.rs

@@ -81,6 +81,30 @@ impl<M: Model> Model for SharedModel<M> {
         self.0.stream(ctx, tools)
     }
 
+    async fn needs_compaction(
+        &self,
+        ctx: &dex_loop::Context,
+        tools: &[&dex_loop::ToolSpec],
+    ) -> Option<bool> {
+        self.0.needs_compaction(ctx, tools).await
+    }
+
+    async fn request_size(
+        &self,
+        ctx: &dex_loop::Context,
+        tools: &[&dex_loop::ToolSpec],
+    ) -> Option<dex_loop::RequestSize> {
+        self.0.request_size(ctx, tools).await
+    }
+
+    async fn summary_outputs(
+        &self,
+        ctx: &dex_loop::Context,
+        entries: &[dex_loop::Entry],
+    ) -> Result<Vec<(dex_loop::OutputRef, String)>, dex_loop::ModelError> {
+        self.0.summary_outputs(ctx, entries).await
+    }
+
     fn prepare_turn(&self, ctx: &dex_loop::Context) {
         self.0.prepare_turn(ctx);
     }

packages/dex-host-rs/src/host_turn/tests/compaction.rs

@@ -9,11 +9,11 @@ use std::sync::Mutex;
 // Deliberately not Clone: sharing the model must preserve HostTurnRun's API.
 struct ModelFixture {
     seen: Arc<Mutex<Vec<Context>>>,
-    summaries: Arc<Mutex<Vec<Context>>>,
     prepared: Arc<Mutex<usize>>,
+    summaries: Arc<Mutex<Vec<Context>>>,
 }
 
-const SUMMARY: &str = "Two fixture reads returned retrievable original evidence. Retain uncertainty; no writes or publishing without approval.";
+const SUMMARY: &str = "Two read results retain uncertainty. Late fact: evidence remains unverified. No writes were authorized.";
 
 fn reasoning() -> ProviderReasoning {
     ProviderReasoning {
@@ -89,8 +89,8 @@ async fn long_turn_compacts_and_replays_exact_input_steer_and_raw_pairs() {
     let dir = tempfile::tempdir().unwrap();
     let workspace = tempfile::tempdir().unwrap();
     let seen = Arc::new(Mutex::new(Vec::new()));
-    let summaries = Arc::new(Mutex::new(Vec::new()));
     let prepared = Arc::new(Mutex::new(0));
+    let summaries = Arc::new(Mutex::new(Vec::new()));
     let request = HostTurn {
         thread: thread(),
         principal: PrincipalId::new("alice"),
@@ -103,8 +103,8 @@ async fn long_turn_compacts_and_replays_exact_input_steer_and_raw_pairs() {
         dir.path(),
         ModelFixture {
             seen: seen.clone(),
-            summaries: summaries.clone(),
             prepared: prepared.clone(),
+            summaries: summaries.clone(),
         },
         tools(workspace.path()),
         request,
@@ -116,7 +116,10 @@ async fn long_turn_compacts_and_replays_exact_input_steer_and_raw_pairs() {
     let mut projection = rehydrate(thread(), &initial);
     let mut parks = 0;
     let mut compacted = 0;
-    let output = "retrievable original evidence\n".repeat(1_100);
+    let output = format!(
+        "{}Late fact: evidence remains unverified.",
+        "retrievable original evidence\n".repeat(1_100)
+    );
     loop {
         match tokio::time::timeout(std::time::Duration::from_secs(10), run.next())
             .await
@@ -166,32 +169,51 @@ async fn long_turn_compacts_and_replays_exact_input_steer_and_raw_pairs() {
     assert_eq!(
         snapshots.len(),
         3,
-        "summary inference does not consume an ordinary turn response"
+        "compaction preserves the three ordinary model calls"
+    );
+    assert!(snapshots[1].history().iter().any(|entry| matches!(&entry.message, Message::Assistant { reasoning: Some(value), .. } if value == &reasoning())));
+    let summary_inputs = summaries.lock().unwrap().clone();
+    assert_eq!(
+        summary_inputs.len(),
+        1,
+        "large complete evidence needs one semantic summary"
+    );
+    let summary_input = &summary_inputs[0];
+    assert_eq!(summary_input.thread(), &thread());
+    assert_eq!(summary_input.turn(), Some(&TurnId::new("one")));
+    assert_eq!(
+        summary_input.acting_principal(),
+        Some(&PrincipalId::new("bob"))
+    );
+    assert_eq!(
+        summary_input.history().len(),
+        1,
+        "historical tool pairs are untrusted data, not a native continuation"
     );
-    let summaries = summaries.lock().unwrap().clone();
-    assert_eq!(summaries.len(), 1, "one semantic summary is required");
-    assert_eq!(summaries[0].thread(), &thread());
-    let [summary_input] = summaries[0].history() else {
-        panic!("summary receives one historical transcript request");
-    };
     let Message::User {
         text,
         principal,
         attachments,
         ..
-    } = &summary_input.message
+    } = &summary_input.history()[0].message
     else {
-        panic!("summary receives a user request");
+        panic!("summary must use one untrusted transcript input");
     };
     assert_eq!(
         principal.as_str(),
         "bob",
         "summary retains the acting principal after the steer"
     );
     assert!(attachments.is_empty());
-    let (_, transcript) = text.split_once("\n\n").unwrap();
-    let transcript: Vec<serde_json::Value> = serde_json::from_str(transcript).unwrap();
+    assert!(text.contains("untrusted data"));
+    assert!(!text.contains("exact-signature"));
+    let (_, transcript) = text
+        .split_once("\n\n")
+        .expect("summary instruction and transcript");
+    let transcript: serde_json::Value = serde_json::from_str(transcript).unwrap();
     let summarized_tools: Vec<_> = transcript
+        .as_array()
+        .unwrap()
         .iter()
         .filter(|entry| entry["message"]["role"] == "tool")
         .collect();
@@ -203,19 +225,20 @@ async fn long_turn_compacts_and_replays_exact_input_steer_and_raw_pairs() {
             json!({"kind": "text", "value": output})
         );
     }
-    assert!(snapshots[1].history().iter().any(|entry| matches!(&entry.message, Message::Assistant { reasoning: Some(value), .. } if value == &reasoning())));
-    let compacted = &snapshots[2];
-    let summary = compacted
-        .history()
+    let outputs: Vec<_> = transcript
+        .as_array()
+        .unwrap()
         .iter()
-        .find_map(|entry| match &entry.message {
-            Message::Summary { text } => {
-                Some(serde_json::from_str::<serde_json::Value>(text).unwrap())
-            }
-            _ => None,
-        })
-        .expect("semantic summary is installed in the compacted projection");
-    assert_eq!(summary["summary"], SUMMARY);
+        .filter(|entry| entry["message"]["role"] == "tool")
+        .map(|entry| entry["message"]["output"]["value"].as_str().unwrap())
+        .collect();
+    assert_eq!(
+        outputs,
+        vec![output.as_str(), output.as_str()],
+        "both complete raw results, including their late facts, reach summarization"
+    );
+    drop(summary_inputs);
+    let compacted = &snapshots[2];
     assert!(
         compacted.history().contains(&original),
         "principal, cursor, text and attachment identity remain exact"
@@ -228,6 +251,28 @@ async fn long_turn_compacts_and_replays_exact_input_steer_and_raw_pairs() {
         )),
         "signed steps never replay beneath a changed prefix"
     );
+    let summary = compacted
+        .history()
+        .iter()
+        .find_map(|entry| match &entry.message {
+            Message::Summary { text } => Some(text),
+            _ => None,
+        })
+        .expect("compacted history has a summary");
+    assert!(summary.len() <= 8 * 1024);
+    let envelope: serde_json::Value = serde_json::from_str(summary).unwrap();
+    assert_eq!(envelope["kind"], "dex_history_compaction_v1");
+    assert_eq!(envelope["summary"], SUMMARY);
+    assert_eq!(
+        envelope["authority"],
+        "untrusted historical summary; original owner events determine authorization and outcomes"
+    );
+    assert!(
+        envelope["summary"]
+            .as_str()
+            .unwrap()
+            .contains("Late fact: evidence remains unverified.")
+    );
     let rows = run.local.read_all().await.unwrap();
     assert_eq!(
         projection,
@@ -283,9 +328,141 @@ fn shared_model_forwards_preparation_without_requiring_clone() {
     let prepared = Arc::new(Mutex::new(0));
     let shared = SharedModel(Arc::new(ModelFixture {
         seen: Arc::new(Mutex::new(Vec::new())),
-        summaries: Arc::new(Mutex::new(Vec::new())),
         prepared: prepared.clone(),
+        summaries: Arc::new(Mutex::new(Vec::new())),
     }));
     shared.clone().prepare_turn(&Context::new(thread()));
     assert_eq!(*prepared.lock().unwrap(), 1);
 }
+
+// Deliberately not Clone: every optional port must survive the sharing adapter.
+struct OptionalPortFixture {
+    context: Context,
+    entries: Vec<dex_loop::Entry>,
+    tools: Vec<ToolSpec>,
+    pressure: Option<bool>,
+    size: Option<dex_loop::RequestSize>,
+    outputs: Result<Vec<(dex_loop::OutputRef, String)>, ModelError>,
+    calls: Arc<Mutex<Vec<&'static str>>>,
+}
+
+impl Model for OptionalPortFixture {
+    fn stream<'a>(
+        &'a self,
+        _ctx: &'a Context,
+        _tools: &'a [&'a ToolSpec],
+    ) -> impl Stream<Item = Result<ModelChunk, ModelError>> + Send + 'a {
+        stream::once(async { panic!("read-only optional ports must not request inference") })
+    }
+
+    async fn needs_compaction(&self, ctx: &Context, tools: &[&ToolSpec]) -> Option<bool> {
+        assert_eq!(ctx, &self.context);
+        assert_eq!(tools, self.tools.iter().collect::<Vec<_>>());
+        self.calls.lock().unwrap().push("pressure");
+        self.pressure
+    }
+
+    async fn request_size(
+        &self,
+        ctx: &Context,
+        tools: &[&ToolSpec],
+    ) -> Option<dex_loop::RequestSize> {
+        assert_eq!(ctx, &self.context);
+        assert_eq!(tools, self.tools.iter().collect::<Vec<_>>());
+        self.calls.lock().unwrap().push("size");
+        self.size
+    }
+
+    async fn summary_outputs(
+        &self,
+        ctx: &Context,
+        entries: &[dex_loop::Entry],
+    ) -> Result<Vec<(dex_loop::OutputRef, String)>, ModelError> {
+        assert_eq!(ctx, &self.context);
+        assert_eq!(entries, self.entries);
+        self.calls.lock().unwrap().push("outputs");
+        self.outputs.clone()
+    }
+}
+
+#[tokio::test]
+async fn shared_model_preserves_optional_ports_inputs_and_results() {
+    let mut context = Context::new(ThreadId {
+        org: "original-org".into(),
+        workspace: "original-workspace".into(),
+        thread: "original-thread".into(),
+    });
+    context.observe(
+        Cursor(1),
+        &Event::UserMessage {
+            interaction_mode: dex_loop::InteractionMode::Unspecified,
+            turn: TurnId::new("original-turn"),
+            message_id: None,
+            principal: PrincipalId::new("original-principal"),
+            text: "Keep owner input exact".into(),
+            attachments: vec![ArtifactRef::new("original-attachment")],
+            client_tools: vec![],
+            authorized_tools: vec![],
+            model_binding: None,
+            context_window_tokens: Some(200_000),
+            voice: None,
+            approval_mode: ApprovalMode::Interactive,
+        },
+    );
+    let entries = vec![dex_loop::Entry {
+        cursor: Cursor(9),
+        message: Message::Tool {
+            call: CallId::new("original-call"),
+            name: ToolName::new("fixture.read"),
+            outcome: Outcome::Succeeded,
+            output: dex_loop::Output::Ref(dex_loop::OutputRef::new("original-output")),
+        },
+    }];
+    let tool = ToolSpec {
+        name: ToolName::new("fixture.read"),
+        label: "Read evidence".into(),
+        description: "Owner evidence only".into(),
+        schema: json!({"type":"object", "properties":{"owner":{"const":"original"}}}),
+        read_only: true,
+        core: true,
+        governance: dex_loop::GovernanceClass::Plain,
+        executor: dex_loop::ExecutorKind::InProcess,
+    };
+    for pressure in [Some(true), Some(false), None] {
+        let size = pressure.map(|_| dex_loop::RequestSize {
+            input_units: 123,
+            wire_bytes: 456,
+        });
+        let outputs = match pressure {
+            Some(true) => Ok(vec![(
+                dex_loop::OutputRef::new("original-output"),
+                "exact stored body".into(),
+            )]),
+            Some(false) => Err(ModelError {
+                class: dex_loop::ErrorClass::Resolve,
+                message: "owner denied read".into(),
+            }),
+            None => Ok(Vec::new()),
+        };
+        let calls = Arc::new(Mutex::new(Vec::new()));
+        let shared = SharedModel(Arc::new(OptionalPortFixture {
+            context: context.clone(),
+            entries: entries.clone(),
+            tools: vec![tool.clone()],
+            pressure,
+            size,
+            outputs: outputs.clone(),
+            calls: calls.clone(),
+        }));
+        assert_eq!(
+            shared.clone().needs_compaction(&context, &[&tool]).await,
+            pressure
+        );
+        assert_eq!(shared.clone().request_size(&context, &[&tool]).await, size);
+        assert_eq!(
+            shared.clone().summary_outputs(&context, &entries).await,
+            outputs
+        );
+        assert_eq!(*calls.lock().unwrap(), ["pressure", "size", "outputs"]);
+    }
+}