Merge pull request 'Desk reads: fork_too_large (list only) replaces DeskForkTooLarge' (#39) from eureka-substrate/fork-too-large into main

crates/huginn-daemon/tests/desk_door.rs

@@ -569,3 +569,217 @@ fn a_busy_pass_is_not_followed_by_the_idle_sleep() {
     let gap = answered[4] - answered[3];
     assert!(gap < Duration::from_millis(300), "the second batch waited {gap:?}: the loop slept after a busy pass ({answered:?})");
 }
+
+// ---- fork_too_large over the real socket ----
+
+fn forks_page(who: &str, cursor: Option<&str>) -> DeskRequest {
+    let mut value = serde_json::to_value(forks_request(who, &[who])).unwrap();
+    value["Forks"]["cursor"] = cursor.map_or(serde_json::Value::Null, |text| serde_json::json!(text));
+    serde_json::from_value(value).unwrap()
+}
+
+fn fork_read(who: &str, fork_id: &str) -> DeskRequest {
+    DeskRequest::Fork { mind: INSTANCE.into(), subject: short(who), deciders: vec![short(who)], fork_id: fork_id.into() }
+}
+
+/// The wire spelling of a not-stored refusal's code, and its text.
+fn refusal_of(response: &DeskResponse) -> Option<(String, String)> {
+    let DeskResponse::NotStored(refusal) = response else { return None };
+    Some((serde_json::to_value(refusal.code).unwrap().as_str().unwrap().to_string(), refusal.error.clone()))
+}
+
+/// Three forks, the middle one's record grown over the socket until its card alone is past the page
+/// budget. The list names it and its order (`fork_too_large`), the cursor `c<order>` walks on to the
+/// older fork and ends at order 0, a READ of it never returns that code (an entry while it fits the
+/// response bound, the door's named response-too-large past it), and `authority` on a read is still
+/// refused.
+#[test]
+fn a_fork_too_large_to_page_is_named_by_the_list_and_never_by_the_read() {
+    let _serial = serial();
+    assert_ne!(my_uid(), 0, "the door refuses uid 0, so this test cannot run as root");
+    let (root, run) = (tempfile::tempdir().unwrap(), tempfile::tempdir().unwrap());
+    let hash = seed(root.path());
+    {
+        let mut mind = Mind::open(root.path(), &INSTANCE.into()).unwrap();
+        let documents = ["Q2", "Q3"]
+            .into_iter()
+            .map(|label| PipelineDocument::Question(PipelineQuestion {
+                campaign: "eureka-state".into(),
+                label: label.into(),
+                title: Title("Who owns the state?".into()),
+                question: "Who owns the state?".into(),
+                options: vec![
+                    QuestionOption { label: "A".into(), text: "the organ".into() },
+                    QuestionOption { label: "B".into(), text: "the client".into() },
+                ],
+                recommended: "A".into(),
+                depends: vec![],
+                raised_in: None,
+                asked_on: Date("2026-09-16".into()),
+            }))
+            .collect();
+        let provenance = PipelineProvenance { faculty: Faculty::Hands, agent: short("claude"), session: short("session-2"), tool: short("admit"), imported_from: None };
+        let outcome = mind.admit(PipelineAdmissionBatch { instance: INSTANCE.into(), provenance, documents }, Utc.with_ymd_and_hms(2026, 9, 16, 12, 5, 0).unwrap());
+        assert!(matches!(outcome, PipelineAdmissionOutcome::Committed { .. }), "{outcome:?}");
+    }
+    let (desk, log) = (run.path().join("desk.sock"), run.path().join("daemon.log"));
+    let _daemon = spawn(root.path(), &desk, my_uid(), &log);
+    let (older, big, newer) = (FORK, "eureka-state:question:Q2", "eureka-state:question:Q3");
+
+    // Notes of about 300 bytes, until the big fork's list page is the refusal.
+    let note = "a long reason, ".repeat(20);
+    let mut sent = 0_u32;
+    let refused = loop {
+        let listed = ask(&desk, "m-g", &forks_page("x", Some("c2")));
+        if refusal_of(&listed).is_some() {
+            break listed;
+        }
+        assert!(matches!(listed, DeskResponse::Forks { .. }), "{listed:?}");
+        sent += 1;
+        assert!(sent < 400, "the record does not grow");
+        let mut grow = answer("x", &hash, &format!("00000000-0000-4000-8000-{sent:012}"));
+        if let DeskRequest::Answer { request, .. } = &mut grow {
+            request.fork_id = big.into();
+            request.answer = Some(ForkAnswer { choice: None, note: Some(note.clone()) });
+        }
+        assert!(matches!(ask(&desk, "m-w", &grow), DeskResponse::Answered { .. }));
+    };
+
+    // (1) The list names the fork and its order, and nothing else is new in the text.
+    let (code, text) = refusal_of(&refused).unwrap();
+    assert_eq!(code, "fork_too_large", "{refused:?}");
+    assert!(text.contains(big) && text.contains("order 1"), "{text}");
+
+    // (2) The cursor c<order> resumes to the older fork, and order 0 is the end. The newest fork pages
+    // alone ahead of it; the big one is named from the first page too.
+    let first = ask(&desk, "m-p0", &forks_page("x", None));
+    let DeskResponse::Forks { list, .. } = first else { panic!("{first:?}") };
+    assert_eq!((list.forks.iter().map(|e| e.card.fork_id.as_str().to_string()).collect::<Vec<_>>(), list.next_cursor.map(|c| c.order())), (vec![newer.to_string()], Some(2)));
+    let rest = ask(&desk, "m-p1", &forks_page("x", Some("c1")));
+    let DeskResponse::Forks { list, .. } = rest else { panic!("{rest:?}") };
+    assert_eq!((list.forks.iter().map(|e| e.card.fork_id.as_str().to_string()).collect::<Vec<_>>(), list.next_cursor.is_none()), (vec![older.to_string()], true));
+    assert_eq!(list.forks[0].card.order, 0);
+
+    // (3) A read of the same fork: its whole entry while it fits the response bound.
+    let read = ask(&desk, "m-r", &fork_read("x", big));
+    let DeskResponse::Fork { entry, .. } = &read else { panic!("a read of the oversize fork is its entry, not a refusal: {read:?}") };
+    assert_eq!((entry.card.fork_id.as_str(), entry.card.order), (big, 1));
+    // Past DESK_MAX_RESPONSE_BYTES the read is the door's named response-too-large refusal, still not fork_too_large.
+    let mut more = 0_u32;
+    let too_big = loop {
+        let read = ask(&desk, "m-r", &fork_read("x", big));
+        if matches!(read, DeskResponse::Refused(_)) {
+            break read;
+        }
+        assert!(matches!(read, DeskResponse::Fork { .. }), "{read:?}");
+        more += 1;
+        assert!(more < 400, "the record does not grow");
+        let mut grow = answer("x", &hash, &format!("00000000-0000-4000-8000-{:012}", 1000 + more));
+        if let DeskRequest::Answer { request, .. } = &mut grow {
+            request.fork_id = big.into();
+            request.answer = Some(ForkAnswer { choice: None, note: Some(note.clone()) });
+        }
+        assert!(matches!(ask(&desk, "m-w", &grow), DeskResponse::Answered { .. }));
+    };
+    assert!(matches!(too_big, DeskResponse::Refused(huginn_mind::MindRefusal::ResponseTooLarge { .. })), "{too_big:?}");
+    assert!(refusal_of(&ask(&desk, "m-g", &forks_page("x", Some("c2")))).is_some_and(|(code, _)| code == "fork_too_large"));
+
+    // (4) `authority` on a read is still refused whole (the contract's request_invalid is the door's payload-not-a-request).
+    for (operation, request) in [("forks", forks_request("x", &["x"])), ("fork", fork_read("x", big))] {
+        let mut value = serde_json::to_value(&request).unwrap();
+        value[if operation == "forks" { "Forks" } else { "Fork" }]["authority"] = serde_json::json!("Operator");
+        let message = CultNetMessage::OperationRequest {
+            message_id: "m-ra".into(),
+            service_id: huginn_mind::desk_wire::DESK_SERVICE_ID.into(),
+            operation: operation.into(),
+            payload_schema: huginn_mind::desk_wire::DESK_REQUEST_SCHEMA.into(),
+            payload_encoding: "messagepack-base64".into(),
+            payload: base64::Engine::encode(&base64::engine::general_purpose::STANDARD, rmp_serde::to_vec_named(&value).unwrap()),
+            source_runtime_id: None,
+            target_runtime_id: None,
+        };
+        let reply = exchange(&desk, &framed(&message)).expect("the door replied");
+        assert_eq!(decode_desk_response(&reply).unwrap().1.unwrap_err().code, "payload-not-a-request", "{operation}");
+    }
+}
+
+/// Q1 (order 0), Q2, Q3 held; the daemon up. Returns what the tests need.
+fn three_forks() -> (tempfile::TempDir, tempfile::TempDir, String, PathBuf, Daemon) {
+    let (root, run) = (tempfile::tempdir().unwrap(), tempfile::tempdir().unwrap());
+    let hash = seed(root.path());
+    {
+        let mut mind = Mind::open(root.path(), &INSTANCE.into()).unwrap();
+        let documents = ["Q2", "Q3"]
+            .into_iter()
+            .map(|label| PipelineDocument::Question(PipelineQuestion {
+                campaign: "eureka-state".into(),
+                label: label.into(),
+                title: Title("Who owns the state?".into()),
+                question: "Who owns the state?".into(),
+                options: vec![
+                    QuestionOption { label: "A".into(), text: "the organ".into() },
+                    QuestionOption { label: "B".into(), text: "the client".into() },
+                ],
+                recommended: "A".into(),
+                depends: vec![],
+                raised_in: None,
+                asked_on: Date("2026-09-16".into()),
+            }))
+            .collect();
+        let provenance = PipelineProvenance { faculty: Faculty::Hands, agent: short("claude"), session: short("session-2"), tool: short("admit"), imported_from: None };
+        let outcome = mind.admit(PipelineAdmissionBatch { instance: INSTANCE.into(), provenance, documents }, Utc.with_ymd_and_hms(2026, 9, 16, 12, 5, 0).unwrap());
+        assert!(matches!(outcome, PipelineAdmissionOutcome::Committed { .. }), "{outcome:?}");
+    }
+    let (desk, log) = (run.path().join("desk.sock"), run.path().join("daemon.log"));
+    let daemon = spawn(root.path(), &desk, my_uid(), &log);
+    (root, run, hash, desk, daemon)
+}
+
+/// Grows `fork`'s record over the socket until the list page at `cursor` is the refusal; returns it.
+fn grown_until_refused(desk: &Path, hash: &str, fork: &str, cursor: Option<&str>) -> DeskResponse {
+    let note = "a long reason, ".repeat(20);
+    for sent in 1..400_u32 {
+        let listed = ask(desk, "m-g", &forks_page("x", cursor));
+        if refusal_of(&listed).is_some() {
+            return listed;
+        }
+        assert!(matches!(listed, DeskResponse::Forks { .. }), "{listed:?}");
+        let mut grow = answer("x", hash, &format!("00000000-0000-4000-8000-{sent:012}"));
+        if let DeskRequest::Answer { request, .. } = &mut grow {
+            request.fork_id = fork.into();
+            request.answer = Some(ForkAnswer { choice: None, note: Some(note.clone()) });
+        }
+        assert!(matches!(ask(desk, "m-w", &grow), DeskResponse::Answered { .. }));
+    }
+    panic!("the record does not grow")
+}
+
+fn page_ids(response: DeskResponse) -> (Vec<String>, Option<u64>) {
+    let DeskResponse::Forks { list, .. } = response else { panic!("{response:?}") };
+    (list.forks.iter().map(|e| e.card.fork_id.as_str().to_string()).collect(), list.next_cursor.map(|c| c.order()))
+}
+
+/// The oldest fork (order 0) oversize: the list names it and its order, and says it is the end, not a cursor.
+#[test]
+fn the_oldest_fork_too_large_says_the_list_ends_and_names_no_cursor() {
+    let _serial = serial();
+    let (_root, _run, hash, desk, _daemon) = three_forks();
+    let refused = grown_until_refused(&desk, &hash, FORK, Some("c1"));
+    let (code, text) = refusal_of(&refused).unwrap();
+    assert_eq!(code, "fork_too_large");
+    assert!(text.contains(FORK) && text.contains("order 0") && !text.contains("c0"), "{text}");
+}
+
+/// The newest fork (order 2) oversize: the first page is the refusal, and c2 resumes to the older forks.
+#[test]
+fn the_newest_fork_too_large_is_named_and_its_cursor_resumes_to_the_older_forks() {
+    let _serial = serial();
+    let (_root, _run, hash, desk, _daemon) = three_forks();
+    let newest = "eureka-state:question:Q3";
+    let refused = grown_until_refused(&desk, &hash, newest, None);
+    let (code, text) = refusal_of(&refused).unwrap();
+    assert_eq!(code, "fork_too_large");
+    assert!(text.contains(newest) && text.contains("order 2") && text.contains("c2"), "{text}");
+    let rest = page_ids(ask(&desk, "m-p", &forks_page("x", Some("c2"))));
+    assert_eq!(rest, (vec!["eureka-state:question:Q2".to_string(), FORK.to_string()], None));
+}

crates/huginn-mind/src/desk_wire.rs

@@ -152,13 +152,14 @@ pub struct ForkDeskAnswer {
     pub request_id: String,
 }
 
-/// The three requests Huginn refuses without storing, by code.
+/// The refusals Huginn sends without storing, by code. `ForkTooLarge` is the list's alone.
 #[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
 #[serde(rename_all = "snake_case")]
 pub enum ForkRefusalCode {
     ForkUnknown,
     BodyInvalid,
     RequestIdConflict,
+    ForkTooLarge,
 }
 
 /// `gamecult.fork.refusal.v0`. `error` is a fixed sentence, never an echo of what was sent;
@@ -382,6 +383,13 @@ fn fork_unknown() -> ForkRefusal {
     fork_refusal(ForkRefusalCode::ForkUnknown, "fork_unknown: no such fork".into())
 }
 
+/// A list page whose first entry alone is past the page budget: names that fork and its `order`. A
+/// client continues with the cursor `c<order>`; at order 0 there is no such cursor and the list ends.
+fn fork_too_large(fork: &str, order: u64) -> ForkRefusal {
+    let onward = if order == 0 { "this is the end of the list".to_string() } else { format!("continue with cursor c{order}") };
+    fork_refusal(ForkRefusalCode::ForkTooLarge, format!("fork_too_large: fork {fork} at order {order} is past the page budget by itself; {onward}"))
+}
+
 fn body_invalid(field: &str) -> ForkRefusal {
     fork_refusal(ForkRefusalCode::BodyInvalid, format!("body_invalid: the body does not satisfy gamecult.fork.answer_request.v0 ({field})"))
 }

crates/huginn-mind/src/desk_wire/reads.rs

@@ -326,7 +326,7 @@ impl<S: MindStore> Mind<S> {
     /// A page holds at most `limit` entries (never more than 50) and stops short only when the next
     /// entry would push the encoded response past `DESK_PAGE_BUDGET` (the one `DESK_MAX_RESPONSE_BYTES`
     /// less the envelope margin); counts and stamp are unaffected. A page whose FIRST entry alone is past
-    /// the budget is `MindRefusal::DeskForkTooLarge`, naming that fork and its order: the caller goes on
+    /// the budget is the `fork_too_large` refusal, naming that fork and its order: the caller goes on
     /// with the cursor `c<order>`, which returns the forks below it, so no older fork is unreachable.
     /// The fork itself is still its whole entry on `desk_fork`. Nothing is dropped silently.
     ///
@@ -384,7 +384,7 @@ impl<S: MindStore> Mind<S> {
             if bytes > DESK_PAGE_BUDGET {
                 let list = held(&mut response);
                 if list.forks.len() == 1 {
-                    return Ok(DeskResponse::Refused(MindRefusal::DeskForkTooLarge { fork: key.to_string(), order: *order, bytes: bytes as u64, limit: DESK_PAGE_BUDGET as u64 }));
+                    return Ok(DeskResponse::NotStored(fork_too_large(key, *order)));
                 }
                 list.forks.pop();
                 list.next_cursor = before;
@@ -1061,9 +1061,11 @@ mod tests {
         assert_eq!(ids(&first), [fork(3)]);
         assert_eq!(first.next_cursor.map(ForkCursor::order), Some(2));
         let response = mind.desk_forks(&list(&LIST), None, first.next_cursor, limit(50));
-        let DeskResponse::Refused(MindRefusal::DeskForkTooLarge { fork: named, order, bytes, limit: bound }) = response else { panic!("{response:?}") };
-        assert_eq!((named, order, bound), (fork(2), 1, DESK_PAGE_BUDGET as u64));
-        assert!(bytes > bound);
+        let DeskResponse::NotStored(refusal) = response else { panic!("{response:?}") };
+        assert_eq!(refusal.code, ForkRefusalCode::ForkTooLarge);
+        let order = 1;
+        assert_eq!(refusal.error, fork_too_large(&fork(2), order).error);
+        assert!(refusal.error.contains(&fork(2)) && refusal.error.contains("order 1"), "{refusal:?}");
         // The order the refusal names is the order of the fork it names; its cursor goes on past it.
         assert_eq!(entry(mind.desk_fork(&list(&LIST), &fork(2))).0.card.order, order);
         let (rest, _) = page(mind.desk_forks(&list(&LIST), None, ForkCursor::new(order), limit(50)));

crates/huginn-mind/src/refusal.rs

@@ -109,11 +109,6 @@ pub enum MindRefusal {
     /// truncated or paginated on its behalf. The daemon owns it and its
     /// deferral bound decides it; no rule of the mind is involved.
     ResponseTooLarge { bytes: u64, limit: u64 },
-    /// A desk page whose first entry alone is past the page budget: `fork` is that fork, `order` its
-    /// rank (a caller goes on past it with the cursor `c<order>`, which returns the forks below it),
-    /// `bytes` what the page holding it would encode to (without the envelope) and `limit` the page
-    /// budget it exceeds. Raised by the mind's page fill, which is why it is not `ResponseTooLarge`.
-    DeskForkTooLarge { fork: String, order: u64, bytes: u64, limit: u64 },
     /// A selection the substrate or the organ's value door refused: `field`
     /// is the selection field the client sent, `value` the offending part.
     SelectionInvalid { field: String, value: Option<String>, message: String },

schemas/cultnet/huginn.desk_response.v1.schema.json

@@ -471,12 +471,13 @@
       ]
     },
     "ForkRefusalCode": {
-      "description": "The three requests Huginn refuses without storing, by code.",
+      "description": "The refusals Huginn sends without storing, by code. `ForkTooLarge` is the list's alone.",
       "type": "string",
       "enum": [
         "fork_unknown",
         "body_invalid",
-        "request_id_conflict"
+        "request_id_conflict",
+        "fork_too_large"
       ]
     },
     "ListSchema": {
@@ -1562,45 +1563,6 @@
             "ResponseTooLarge"
           ]
         },
-        {
-          "description": "A desk page whose first entry alone is past the page budget: `fork` is that fork, `order` its\nrank (a caller goes on past it with the cursor `c<order>`, which returns the forks below it),\n`bytes` what the page holding it would encode to (without the envelope) and `limit` the page\nbudget it exceeds. Raised by the mind's page fill, which is why it is not `ResponseTooLarge`.",
-          "type": "object",
-          "properties": {
-            "DeskForkTooLarge": {
-              "type": "object",
-              "properties": {
-                "bytes": {
-                  "type": "integer",
-                  "format": "uint64",
-                  "minimum": 0
-                },
-                "fork": {
-                  "type": "string"
-                },
-                "limit": {
-                  "type": "integer",
-                  "format": "uint64",
-                  "minimum": 0
-                },
-                "order": {
-                  "type": "integer",
-                  "format": "uint64",
-                  "minimum": 0
-                }
-              },
-              "required": [
-                "fork",
-                "order",
-                "bytes",
-                "limit"
-              ]
-            }
-          },
-          "additionalProperties": false,
-          "required": [
-            "DeskForkTooLarge"
-          ]
-        },
         {
           "description": "A selection the substrate or the organ's value door refused: `field`\nis the selection field the client sent, `value` the offending part.",
           "type": "object",

schemas/cultnet/huginn.mind_response.v1.schema.json

@@ -2059,45 +2059,6 @@
             "ResponseTooLarge"
           ]
         },
-        {
-          "description": "A desk page whose first entry alone is past the page budget: `fork` is that fork, `order` its\nrank (a caller goes on past it with the cursor `c<order>`, which returns the forks below it),\n`bytes` what the page holding it would encode to (without the envelope) and `limit` the page\nbudget it exceeds. Raised by the mind's page fill, which is why it is not `ResponseTooLarge`.",
-          "type": "object",
-          "properties": {
-            "DeskForkTooLarge": {
-              "type": "object",
-              "properties": {
-                "bytes": {
-                  "type": "integer",
-                  "format": "uint64",
-                  "minimum": 0
-                },
-                "fork": {
-                  "type": "string"
-                },
-                "limit": {
-                  "type": "integer",
-                  "format": "uint64",
-                  "minimum": 0
-                },
-                "order": {
-                  "type": "integer",
-                  "format": "uint64",
-                  "minimum": 0
-                }
-              },
-              "required": [
-                "fork",
-                "order",
-                "bytes",
-                "limit"
-              ]
-            }
-          },
-          "additionalProperties": false,
-          "required": [
-            "DeskForkTooLarge"
-          ]
-        },
         {
           "description": "A selection the substrate or the organ's value door refused: `field`\nis the selection field the client sent, `value` the offending part.",
           "type": "object",