diff --git a/crates/spt-daemon/src/brainproc.rs b/crates/spt-daemon/src/brainproc.rs index a327d0f4..1f28ac52 100644 --- a/crates/spt-daemon/src/brainproc.rs +++ b/crates/spt-daemon/src/brainproc.rs @@ -872,6 +872,11 @@ fn run_trial( ) -> TrialStep { let deadline = Instant::now() + window; let mut ready_seen = false; + // ONCE PER TRIAL (releases#49/#267 Q3). The gate is a pure read that says + // nothing when it says no, so a trial that waited on the drain and a trial + // that never had to are the same silence. One line on the FIRST not-drained + // read ends that; repeating it every tick would turn an answer into a log. + let mut not_drained_reported = false; loop { if stop.load(Ordering::Relaxed) { return TrialStep::Stopped; @@ -881,8 +886,39 @@ fn run_trial( if !ready_seen && env.ready_generation() == Some(generation) { ready_seen = true; } - if ready_seen && env.old_gen_drained() { - return TrialStep::Promoted; + if ready_seen { + // ONE observation, read once and used twice — the promote decision and + // the diagnostic line are the SAME read. A second `old_gen_drained()` + // call to describe what the first one decided could disagree with it, + // and a diagnostic that can contradict the branch it reports on is + // worse than no diagnostic. + let drained = env.old_gen_drained(); + if drained { + return TrialStep::Promoted; + } + // THE NARROWER FACT, DELIBERATELY. The gate answers one bool over every + // hosted session; naming WHICH session is wedged would need a second, + // separately-taken scan, so this reports what this observation actually + // established — that the gate was EVALUATED and read not-drained — and + // nothing about which conn held it. It is scoped to THIS trial: earlier + // trials evaluated the same gate and this line says nothing about them, so + // it must not be read as the first evaluation the daemon ever made. It is + // also not evidence that the + // heartbeat's drain-driving `brain.sessions()` call completed: this says + // the gate was consulted, never that the reap ran. + // [impl->REQ-REFRESH-WAIT-ATTRIBUTION] + if !not_drained_reported { + not_drained_reported = true; + if crate::brain::diag_trace_enabled() { + spt_proto::emit_line_err!( + "BRAIN_PROMOTE_GATE_NOT_DRAINED: generation {generation} — ready latched \ + and the DRAINED gate read NOT drained; this is the first such \ + observation in THIS trial, not the first evaluation of the gate; the trial \ + keeps polling to its window \ + [REQ-REFRESH-WAIT-ATTRIBUTION]" + ); + } + } } // Only watch for a pre-ready exit while the candidate has NOT reached ready; // once ready-seen we are waiting on the OLD gen to drain, and a ready-then- diff --git a/crates/spt-daemon/src/broker.rs b/crates/spt-daemon/src/broker.rs index 4c58d297..820b7f85 100644 --- a/crates/spt-daemon/src/broker.rs +++ b/crates/spt-daemon/src/broker.rs @@ -2718,14 +2718,57 @@ impl OutputLog { } } + /// The drain timed out the controller's `send_timeout` and the epoch guard + /// still matched ([`mark_controller_gone`], W1). The writer was BLOCKED past + /// the deadline -- it is NOT established to have exited or dropped its `tx`, + /// so this names the deadline the drain observed and not the writer's fate. + /// This is the edge releases#49/#267 Q1 asks about: a cleanly-closed + /// OLD-generation conn arrives here too, and from this site the two are + /// indistinguishable. + // [impl->REQ-REFRESH-WAIT-ATTRIBUTION] + const CLOSE_SEND_DEADLINE: &'static str = "send_deadline"; + /// The wedge evict released the slot ([`stall_evict_controller`], inline stamp + /// mode). `BRAIN_SUBSCRIBER_STALL_EVICT` already names the wedge; this names + /// the slot transition that followed it. + // [impl->REQ-REFRESH-WAIT-ATTRIBUTION] + const CLOSE_STALL_EVICT: &'static str = "stall_evict"; + /// A subscriber detached and it happened to be the controller + /// ([`detach_if`]) — an ordinary departure, not a wedge and not a kill. + // [impl->REQ-REFRESH-WAIT-ATTRIBUTION] + const CLOSE_DETACH: &'static str = "detach"; + /// Detach the controller and UNLATCH the perch — the ONE unlatch path shared /// by W1's deadline-evict ([`mark_controller_gone`]), [`detach_if`], and (per /// the W1↔W5 synergy) W5's reconcile self-heal. Dropping the sink ends its /// writer thread; clearing the slot re-stamps `driven_by` to `None` (so an /// `ONLINE+CONTROLLED` latch cannot outlive a gone controller). - fn clear_controller(&mut self) { + /// + /// `cause` is the CALLER'S OWN knowledge of which edge closed the slot, and it + /// is a parameter for the reason releases#49/#267 needed this line at all: at + /// this site a writer that exited, an evicted wedge and a deliberate detach are + /// indistinguishable, so a cause DERIVED here would be a guess wearing a label. + /// Each caller names what it knows; nothing infers. + // [impl->REQ-REFRESH-WAIT-ATTRIBUTION] + fn clear_controller(&mut self, cause: &'static str) { let prev_gen = self.controller.as_ref().map(|c| c.attach_gen); + // Captured BEFORE the take: after it there is no sink left to ask. + let conn = self.controller.as_ref().map(|c| c.send.id()); if self.controller.take().is_some() { + // ONLY ON THE REAL TRANSITION (releases#49/#267 Q1). The epoch-guarded + // no-op above returns without taking anything, and a line there would + // report a closure that did not happen — which is exactly the confusion + // the question exists to remove: today a clean close and a slot that was + // already empty leave identical stderr. + // [impl->REQ-REFRESH-WAIT-ATTRIBUTION] + if crate::brain::diag_trace_enabled() { + spt_proto::emit_line_err!( + "CONTROLLER_SLOT_CLOSED:{} conn={} cause={cause} {} — the controller slot \ + was occupied and is now empty [REQ-REFRESH-WAIT-ATTRIBUTION]", + self.session_id, + conn.unwrap_or(0), + crate::conn::log_stamp() + ); + } self.stamp_driven_by(); // Detach is a drop edge: the posture goes offline (derived from the // stamp just written) and the empowerments go with it. @@ -2784,7 +2827,7 @@ impl OutputLog { /// attached during the stall is never unseated. fn mark_controller_gone(&mut self, epoch: u64) { if self.controller_epoch.load(Ordering::Acquire) == epoch { - self.clear_controller(); + self.clear_controller(Self::CLOSE_SEND_DEADLINE); } } @@ -2850,7 +2893,7 @@ impl OutputLog { match stamp { // Inline: drop the sink AND re-stamp the perch (become_controller's // sibling — the take/reattach path is already I/O-under-log-lock). - StampMode::Inline => self.clear_controller(), + StampMode::Inline => self.clear_controller(Self::CLOSE_STALL_EVICT), // Deferred: in-memory drop ONLY (the reap closure forbids I/O under the // shared sessions lock); converge_perch_stamps writes the release off-lock. StampMode::Deferred => self.controller = None, @@ -3522,7 +3565,7 @@ impl OutputLog { .as_ref() .is_some_and(|c| Arc::ptr_eq(&c.send, sub)); if is_ctrl { - self.clear_controller(); // drops the sink (ends its writer) + unlatches + self.clear_controller(Self::CLOSE_DETACH); // drops the sink (ends its writer) + unlatches } let before = self.viewers.len(); self.viewers.retain(|_, sink| !Arc::ptr_eq(&sink.send, sub)); @@ -3818,6 +3861,8 @@ enum CtrlMsg { /// session output and must not perturb the resume cursor. // [impl->REQ-SEAL-CEREMONY-RC-CLIENT] Ceremony(Envelope), + /// A receipt-bound file request, never output and never a cursor advance. + InputPaths(Envelope), } /// Per-sink queue handles (+ each sink's conn for the bounded degrade) the exit @@ -4022,6 +4067,7 @@ fn controller_writer( // A ceremony frame (WAX-SEAL W2): written like any frame, no // cursor advance (not session output). CtrlMsg::Ceremony(frame) => (frame, None, None), + CtrlMsg::InputPaths(frame) => (frame, None, None), }; // NO epoch gate on the live path (P1c): new output only ever flows to the // CURRENT controller's channel (the drain clones `self.controller.tx`), so @@ -4119,6 +4165,8 @@ struct HostedSession { /// so a paste burst that fills the harness's input buffer parks only the /// writer thread, never the broker dispatch thread. input: Arc, + /// Live-reference dedup and reply custody survive a brain restart. + input_reports: Arc>, /// The output drain pump (kept so teardown can stop it). #[allow(dead_code)] // held for ownership/teardown; not read directly drain: Drain, @@ -4162,9 +4210,8 @@ struct HostedSession { process_started_at: Option, } -/// One record queued to a session's dedicated PTY input-writer thread — the raw -/// bytes to write (v0.13.0 P0, REQ-HAZARD-PTY-INPUT-WRITER-WEDGE). -type InputRecord = Vec; +/// Source-tagged records retain delivery custody through the single physical writer. +use crate::deliverybytes::{DeliveryAttempt, DeliveryBytes, DeliveryPart, InputRecord}; /// The per-session PTY input FIFO depth (records). Sized for a generous paste — /// a wedged harness fills this and then DROPS, never blocking the dispatch @@ -4200,6 +4247,8 @@ struct InputWriter { /// The bounded input FIFO to the writer thread. `try_send` only — a full /// queue (a genuinely wedged harness) DROPS, it never blocks. tx: SyncSender, + /// Successful delivery evidence survives translation respawn for this session. + deliveries: Arc, /// True while the FIFO is saturated and input is being dropped; cleared on /// the next accepted enqueue (heal-on-resume). Mirrored to the perch's /// `input_backpressure` so the operator sees the drop. @@ -4217,23 +4266,39 @@ impl InputWriter { /// whose PTY write handle this thread now exclusively owns. fn spawn(session: Arc, endpoint: String) -> Arc { let (tx, rx) = sync_channel::(input_queue_depth()); - let writer = thread::spawn(move || input_writer(session, rx)); + let deliveries = Arc::new(DeliveryBytes::default()); + let writer_deliveries = Arc::clone(&deliveries); + let writer = thread::spawn(move || input_writer(session, rx, writer_deliveries)); Arc::new(InputWriter { tx, + deliveries, backpressure: AtomicBool::new(false), endpoint, _writer: writer, }) } - /// Enqueue `bytes` for the PTY write thread — NON-BLOCKING. Returns `true` - /// when accepted (ordered, will land), `false` when the FIFO was full and the + /// Enqueue controller bytes for the PTY write thread — NON-BLOCKING. Returns + /// `true` when accepted (ordered, not yet physically written), `false` when the /// record was DROPPED. A full queue stamps `INPUT_BACKPRESSURE` (once, on the /// rising edge); the next accepted enqueue clears it (heal). The dispatch /// thread never blocks here, however stuck the harness is. // [impl->REQ-HAZARD-PTY-INPUT-WRITER-WEDGE] - fn enqueue(&self, bytes: InputRecord) -> bool { - match self.tx.try_send(bytes) { + fn enqueue(&self, bytes: Vec) -> bool { + self.enqueue_record(InputRecord::controller(bytes)) + } + + // [impl->REQ-INPUT-PROVENANCE-ACCEPTANCE-REPORT] + fn delivery_matches(&self, payload: &[u8]) -> bool { + self.deliveries.matches(payload) + } + + fn delivery_unavailable_reason(&self) -> Option<&'static str> { + self.deliveries.unavailable_reason() + } + + fn enqueue_record(&self, record: InputRecord) -> bool { + match self.tx.try_send(record) { Ok(()) => { // Heal: input is flowing again — clear a prior backpressure stamp. if self.backpressure.swap(false, Ordering::AcqRel) { @@ -4241,7 +4306,8 @@ impl InputWriter { } true } - Err(TrySendError::Full(_)) => { + Err(TrySendError::Full(record)) => { + record.dropped(); // DROP + surface, but stamp only on the rising edge. if !self.backpressure.swap(true, Ordering::AcqRel) { self.stamp_backpressure(true); @@ -4249,7 +4315,10 @@ impl InputWriter { false } // The writer thread is gone (session teardown) — nothing to do. - Err(TrySendError::Disconnected(_)) => false, + Err(TrySendError::Disconnected(record)) => { + record.dropped(); + false + } } } @@ -4268,12 +4337,15 @@ impl InputWriter { /// A session's input-writer thread (REQ-HAZARD-PTY-INPUT-WRITER-WEDGE): drain /// records from `rx` and apply each via the BLOCKING /// [`SessionSurface::write_input`] — the ONE place that touches the PTY writer. A -/// wedged harness parks this thread (and only this thread); a write error is -/// swallowed (the same best-effort the inline callers had). The thread exits -/// when the last `tx` clone drops at session teardown. -fn input_writer(session: Arc, rx: Receiver) { - while let Ok(bytes) = rx.recv() { - let _ = session.write_input(&bytes); +/// wedged harness parks this thread (and only this thread). A failed write cannot +/// certify delivery bytes; the thread exits after all senders drop at teardown. +fn input_writer( + session: Arc, + rx: Receiver, + deliveries: Arc, +) { + while let Ok(record) = rx.recv() { + record.write_to(session.as_ref(), &deliveries); } } @@ -4766,7 +4838,7 @@ fn settle_before_inject(input: &InputWriter, log: &Mutex, endpoint: & // child's input then echoes back as output and advances the ring one // hop later, well within the poll cadence. A re-render works the same // as before; a non-echoing ConPTY stays unobservable (the latch). - input.enqueue(INJECT_SETTLE_PROBE.to_vec()); + input.enqueue_record(InputRecord::probe(INJECT_SETTLE_PROBE.to_vec())); thread::sleep(INJECT_SETTLE_POLL); if recover_log(log).high_water() > baseline { return true; // output produced since baseline → the input reader is live @@ -4840,9 +4912,15 @@ fn echo_verify_after(log: &Mutex, seq_start: u64, sent_text: &[u8]) - /// FIFO, single writer). The reader is already confirmed live by the settle-gate before /// the first chunk lands. // [impl->REQ-INJECT-MULTILINE-INTEGRITY] -fn enqueue_text_chunked(input: &InputWriter, bytes: Vec) { +fn enqueue_text_chunked(input: &InputWriter, attempt: &Arc, bytes: Vec) { + let mut remaining = bytes.len(); chunk_text(&bytes, inject_text_chunk(), |part| { - input.enqueue(part.to_vec()); + remaining -= part.len(); + input.enqueue_record(InputRecord::delivery( + part.to_vec(), + attempt, + DeliveryPart::Text { end: remaining == 0 }, + )); }); } @@ -4897,6 +4975,7 @@ fn drive_one_sequence( let mut committed = false; let mut disconnected = false; let mut sent_text: Vec = Vec::new(); + let delivery = Arc::new(DeliveryAttempt::default()); loop { let now = Instant::now(); let remaining = deadline.saturating_duration_since(now); @@ -4906,14 +4985,18 @@ fn drive_one_sequence( match cmd_rx.recv_timeout(remaining) { Ok(KeyCmd::Key { key }) => { if let Some(bytes) = key_to_bytes(&key) { - input.enqueue(bytes); + input.enqueue_record(InputRecord::delivery( + bytes, + &delivery, + DeliveryPart::Key, + )); } } Ok(KeyCmd::Text { text }) => { // Accumulate the payload for echo-verify, then type it paced-chunked. let bytes = text.into_bytes(); sent_text.extend_from_slice(&bytes); - enqueue_text_chunked(input, bytes); + enqueue_text_chunked(input, &delivery, bytes); } Ok(KeyCmd::Delay { delay_ms }) => { let want = Duration::from_millis(delay_ms); @@ -4931,6 +5014,8 @@ fn drive_one_sequence( } } } + // Logical Commit (or timeout/death) does not certify queued bytes. Each + // record retains this attempt until the physical writer confirms it. // Flush buffered controller input AFTER the injected bytes + release the floor — // in EVERY exit path (the ANTI-STALL guarantee). flush_inject_floor(floor, input); @@ -6097,6 +6182,7 @@ impl Broker { } } KIND_NET_STATUS => self.dispatch_net_status(&send), + crate::msg::KIND_NET_REACTOR_DIAG => self.dispatch_net_reactor_diag(&send), // [impl->REQ-UPDATE-RUNNING-IMAGE-SURFACE] KIND_BROKER_IMAGE => self.dispatch_broker_image(&send), // [impl->REQ-UPDATE-RUNNING-IMAGE-SURFACE] @@ -6293,6 +6379,19 @@ impl Broker { ); send_frame(&send, &frame); } + crate::msg::KIND_USER_INPUT_REPORT => { + let result = self.dispatch_user_input_report(env); + let reply = crate::msg::UserInputReported { declined: result.err() }; + send_frame(&send, &Envelope::new(crate::msg::KIND_USER_INPUT_REPORTED, + serde_json::to_value(reply).expect("UserInputReported serializes"))); + } + crate::msg::KIND_USER_INPUT_PATHS_REPLY => { + if let Err(reason) = self.dispatch_input_paths_reply(env, &send) { + spt_proto::emit_line_err!( + "HELPER_SERVE_FOR: outcome=declined reason={reason}" + ); + } + } KIND_ENDPOINT_INPUT => { if let Err(msg) = self.dispatch_endpoint_input(env, &send) { send_error(&send, &msg); @@ -8601,6 +8700,7 @@ impl Broker { HostedSession { session, input, + input_reports: Arc::new(Mutex::new(crate::inputreceipt::ReceiptBook::default())), drain, log, endpoint: req.endpoint.clone(), @@ -8913,6 +9013,107 @@ impl Broker { } } + /// Bind to the live controller once, then retain that exact connection. + /// Neither the timeout worker nor the reply handler samples a later seat. + // [impl->REQ-INPUT-PROVENANCE-ACCEPTANCE-REPORT] + fn dispatch_user_input_report(&self, env: Envelope) -> Result<(), String> { + let req: crate::msg::UserInputReport = serde_json::from_value(env.payload) + .map_err(|e| format!("bad USER_INPUT report: {e}"))?; + let (log, input, reports) = { + let sessions = recover(&self.sessions); + let Some(h) = sessions.values().find(|h| h.endpoint == req.endpoint) else { + return Ok(()); + }; + (Arc::clone(&h.log), Arc::clone(&h.input), Arc::clone(&h.input_reports)) + }; + // Capture the seat before any auth-store or helper work. Ordinary local, + // absent, and viewer-only sessions owe neither serving nor a diagnostic. + let (controller_conn, controller_tx, by) = { + let mut log = recover_log(&log); + log.reap_dead_controller(); + let Some(controller) = log.controller.as_ref() else { return Ok(()); }; + if controller.by.is_none() { return Ok(()); } + (controller.send.id(), controller.tx.clone(), controller.by.clone()) + }; + let local_node = crate::access::local_node_hex(); + if crate::inputreceipt::remote_origin(Some(by.as_deref()), local_node.as_deref()) + .map_err(str::to_owned)?.is_none() + { + return Ok(()); + } + let perch = resolve_perch_path(&req.endpoint, ParentHint::Infer); + let rec = spt_store::info::read_info(&perch) + .ok_or("report endpoint has no authenticated session")?; + if req.session.is_empty() || req.session != rec.session_id { + return Err("report session authentication failed".into()); + } + if req.expires_at_ms <= crate::brain::now_ms() { + return Err("input report expired before broker receipt".into()); + } + // Status first: an in-progress physical write has not published its + // digest yet. A remote report must not guess while evidence is uncertain. + if let Some(reason) = input.delivery_unavailable_reason() { + return Err(reason.into()); + } + if input.delivery_matches(req.payload.as_bytes()) { + return Err("payload matches core-written PTY delivery bytes".into()); + } + let request = recover(&reports).begin( + &req.session, &req.payload, controller_conn, crate::brain::now_ms(), + ).map_err(str::to_owned)?; + let Some(request) = request else { return Ok(()); }; + let frame = Envelope::new(crate::msg::KIND_USER_INPUT_PATHS, + serde_json::to_value(&request).expect("InputPathRequest serializes")); + if controller_tx.try_send(CtrlMsg::InputPaths(frame)).is_err() { + recover(&reports).cancel_unsent(&request.receipt_id); + return Err("receipt-bound controller queue unavailable".into()); + } + // Bounded independently of every rc pump and hook. In-flight admission + // is capped by ReceiptBook, so silent clients cannot grow unlimited waiters. + thread::spawn(move || { + thread::sleep(crate::inputreceipt::REPLY_BOUND); + if recover(&reports).timeout(&request.receipt_id) { + spt_proto::emit_line_err!( + "HELPER_SERVE_FOR: target={} receipt={} outcome=unanswered reason=controller reply deadline", + req.endpoint, request.receipt_id, + ); + } + }); + Ok(()) + } + + // [impl->REQ-INPUT-PROVENANCE-ACCEPTANCE-REPORT] + fn dispatch_input_paths_reply(&self, env: Envelope, send: &SharedSend) -> Result<(), String> { + let reply: spt_net::net::attach::InputPathReply = serde_json::from_value(env.payload) + .map_err(|e| format!("bad input paths reply: {e}"))?; + let books: Vec<_> = { + let sessions = recover(&self.sessions); + sessions.values().map(|h| (h.endpoint.clone(), Arc::clone(&h.input_reports))).collect() + }; + for (endpoint, reports) in books { + let completion = { + let mut book = recover(&reports); + if !book.contains(&reply.receipt_id) { continue; } + book.complete(&reply, send.id(), crate::brain::now_ms()).map_err(str::to_owned)? + }; + let perch = resolve_perch_path(&endpoint, ParentHint::Infer); + for line in completion.lines { + if let Err(error) = spt_store::helperline::append_at(&perch, &line) { + spt_proto::emit_line_err!( + "HELPER_SERVE_FOR: target={endpoint} outcome=failed reason=helper record: {error}" + ); + } + } + for reason in completion.declines { + spt_proto::emit_line_err!( + "HELPER_SERVE_FOR: target={endpoint} outcome=declined {reason}" + ); + } + return Ok(()); + } + Err("reply has no live broker-session receipt".into()) + } + /// Deliver an inbound message to an spt-hosted endpoint by ENDPOINT ID: /// resolve the endpoint to its hosted session and deliver the bytes through its /// translation binary (REQ-SEND-SPT-HOSTED / REQ-MSG-IDLE-TRANSLATION-BINARY). @@ -9085,6 +9286,23 @@ impl Broker { } } + // DIAGNOSTIC releases#302: sampling must not queue behind the suspect runtime, + // sessions, connection table, or output logs. No diagnosis is inferred here. + // [impl->REQ-RC-HITCH-DISCRIMINATOR] + fn dispatch_net_reactor_diag(&self, send: &SharedSend) { + let host = self.net.get(); + let reply = serde_json::json!({ + "diagnostic": "rc-hitch", + "broker_pid": std::process::id(), + "broker_version": env!("CARGO_PKG_VERSION"), + "sampled_at_ms": crate::brain::now_ms(), + "net_enabled": host.is_some(), + "net_canary_age_ms": host.map(|h| h.net_canary_age_ms()), + "active_dial_tasks": host.map(|h| h.active_dial_tasks()), + }); + send_frame(send, &Envelope::new(crate::msg::KIND_NET_REACTOR_DIAG_REPLY, reply)); + } + /// Report the broker-owned net endpoint's status (D4a). Answered even when /// the broker has no network host (`enabled: false`) so a brain can probe /// capability without treating absence as an error. @@ -10262,6 +10480,88 @@ mod tests { }); } + // [unit->REQ-INPUT-PROVENANCE-ACCEPTANCE-REPORT] + #[test] + fn input_receipt_local_is_silent_then_remote_stays_bound_across_take() { + use spt_net::net::attach::{InputPathOutcome, InputPathReply, InputPathRequest, InputPathResult}; + crate::test_home::with_home(|_| { + struct Child(Arc); + impl Drop for Child { + fn drop(&mut self) { let _ = self.0.kill(); } + } + spt_store::nodeid::load_or_create().unwrap(); + let broker = admit_test_broker(); + let endpoint = "receipt-seat"; + let perch = resolve_perch_path(endpoint, ParentHint::Infer); + std::fs::create_dir_all(&perch).unwrap(); + spt_store::info::write_info(&perch, &spt_store::info::InfoJson::new( + endpoint, "t", std::process::id(), "session", "live_agent", + )).unwrap(); + #[cfg(windows)] + let (program, args) = ("findstr", vec![".".to_string()]); + #[cfg(unix)] + let (program, args) = ("cat", Vec::new()); + let (original, mut wire, _reader) = controller_socket_pair(); + let sid = broker.dispatch_spawn_policy(SpawnReq { + program: program.into(), args, rows: 24, cols: 80, + endpoint: endpoint.into(), cwd: None, env: Default::default(), + translation_binary: None, adapter: String::new(), install_dir: None, + }, &original, true).unwrap().unwrap(); + let (log, _child) = { + let sessions = recover(&broker.sessions); + let hosted = sessions.get(&sid).unwrap(); + (Arc::clone(&hosted.log), Child(Arc::clone(&hosted.session))) + }; + let report = |session| Envelope::new(crate::msg::KIND_USER_INPUT_REPORT, + serde_json::json!({ + "endpoint": endpoint, "session": session, "payload": "read /input", + "expires_at_ms": crate::brain::now_ms() + crate::msg::USER_INPUT_REPORT_BOUND.as_millis() as u64, + })); + // A local seat is an ordinary silent no-op, even without session + // proof. No helper observation is created. + broker.dispatch_user_input_report(report("wrong-session")).unwrap(); + broker.dispatch_user_input_report(report("session")).unwrap(); + let mut unhosted = report("session"); + unhosted.payload["endpoint"] = serde_json::json!("not-hosted"); + broker.dispatch_user_input_report(unhosted).unwrap(); + assert!(spt_store::helperline::read_at(&perch).is_empty()); + recover_log(&log).become_controller(Arc::clone(&original), Some("11".repeat(32)), 0, 1); + assert!(broker.dispatch_user_input_report(report("wrong-session")).is_err()); + let mut expired = report("session"); + expired.payload["expires_at_ms"] = serde_json::json!(0); + assert!(broker.dispatch_user_input_report(expired).unwrap_err().contains("expired")); + broker.dispatch_user_input_report(report("session")).unwrap(); + let request: InputPathRequest = loop { + let frame = read_frame(&mut wire).unwrap(); + if frame.kind == crate::msg::KIND_USER_INPUT_PATHS { + break serde_json::from_value(frame.payload).unwrap(); + } + }; + let (replacement, _replacement_wire, _replacement_reader) = controller_socket_pair(); + recover_log(&log).become_controller(Arc::clone(&replacement), Some("22".repeat(32)), 0, 2); + let reply = InputPathReply { + receipt_id: request.receipt_id, + results: vec![InputPathResult { + path: "/input".into(), + outcome: InputPathOutcome::Registered { + url: "http://localhost:5474/owner/f/input".into(), + expires_at_ms: request.received_at_ms + spt_store::serving::HELPER_ENTRY_TTL_MS, + }, + }], + }; + let frame = || Envelope::new(crate::msg::KIND_USER_INPUT_PATHS_REPLY, + serde_json::to_value(&reply).unwrap()); + assert!(broker.dispatch_input_paths_reply(frame(), &replacement).is_err()); + assert!(spt_store::helperline::read_at(&perch).is_empty()); + broker.dispatch_input_paths_reply(frame(), &original).unwrap(); + broker.dispatch_user_input_report(report("session")).unwrap(); + assert!(broker.dispatch_input_paths_reply(frame(), &original).is_err()); + let notices = spt_store::helperline::read_at(&perch); + assert_eq!(notices.len(), 1); + assert_eq!(notices[0].url, "http://localhost:5474/owner/f/input"); + }); + } + // [unit->REQ-BROKER-ZOMBIE-IDENTITY] #[test] fn zombie_reap_requires_positive_identity_but_dead_root_needs_no_kill() { @@ -11019,18 +11319,88 @@ mod tests { let (tx, rx) = sync_channel::(depth); let w = InputWriter { tx, + deliveries: Arc::new(DeliveryBytes::default()), backpressure: AtomicBool::new(false), endpoint: String::new(), _writer: thread::spawn(|| {}), }; (w, rx) } + struct InputCustodySurface; + + impl SessionSurface for InputCustodySurface { + fn write_input(&self, _: &[u8]) -> Result<(), spt_term::surface::SurfaceError> { + Ok(()) + } + + fn resize(&self, _: spt_term::surface::SurfaceSize) -> Result<(), spt_term::surface::SurfaceError> { + Ok(()) + } + } + + // [unit->REQ-INPUT-PROVENANCE-ACCEPTANCE-REPORT] + #[test] + fn input_delivery_queue_drop_never_certifies_missing_text() { + let (input, rx) = test_input_writer(1); + let attempt = Arc::new(DeliveryAttempt::default()); + assert!(input.enqueue_record(InputRecord::delivery( + b"first".to_vec(), &attempt, DeliveryPart::Text { end: false }, + ))); + assert!(!input.enqueue_record(InputRecord::delivery( + b"missing".to_vec(), &attempt, DeliveryPart::Text { end: false }, + ))); + rx.recv().unwrap().write_to(&InputCustodySurface, &input.deliveries); + assert!(input.enqueue_record(InputRecord::delivery( + b"last".to_vec(), &attempt, DeliveryPart::Text { end: true }, + ))); + rx.recv().unwrap().write_to(&InputCustodySurface, &input.deliveries); + assert!(!input.delivery_matches(b"firstmissinglast")); + assert!(!input.delivery_matches(b"firstlast")); + + let next_attempt = Arc::new(DeliveryAttempt::default()); + assert!(input.enqueue_record(InputRecord::delivery( + b"next".to_vec(), &next_attempt, DeliveryPart::Text { end: true }, + ))); + rx.recv().unwrap().write_to(&InputCustodySurface, &input.deliveries); + assert!(input.delivery_matches(b"next")); + } + + // [unit->REQ-INPUT-PROVENANCE-ACCEPTANCE-REPORT] + #[test] + fn chunked_delivery_is_visible_before_enter_and_floor_flush_stays_human() { + let (input, rx) = test_input_writer(16); + let attempt = Arc::new(DeliveryAttempt::default()); + let text = format!("{}\nC:/exact", "a".repeat(inject_text_chunk() + 1)); + enqueue_text_chunked(&input, &attempt, text.as_bytes().to_vec()); + assert!(!input.delivery_matches(text.as_bytes())); + while let Ok(record) = rx.try_recv() { + record.write_to(&InputCustodySurface, &input.deliveries); + } + assert!(input.delivery_matches(text.as_bytes())); + assert!(input.enqueue_record(InputRecord::delivery( + b"\r".to_vec(), &attempt, DeliveryPart::Key, + ))); + + let floor = Mutex::new(InjectFloor::default()); + lock_floor(&floor).open(); + assert!(lock_floor(&floor).buffer_if_held(b"human typeahead C:/private")); + flush_inject_floor(&floor, &input); + assert!(input.enqueue(b"direct controller C:/other".to_vec())); + while let Ok(record) = rx.try_recv() { + record.write_to(&InputCustodySurface, &input.deliveries); + } + assert!(input.delivery_matches(text.as_bytes())); + assert!(!input.delivery_matches(b"human typeahead C:/private")); + assert!(!input.delivery_matches(b"direct controller C:/other")); + assert!(!input.delivery_matches(b"\r")); + } + /// Drain ORDER: records enqueued in sequence reach the sole writer in strict /// FIFO order. We drive the REAL `enqueue` into the REAL channel, then drain it /// through a single consumer that applies each record to a stub `SessionSurface` /// recording the bytes it receives — EXACTLY mirroring the production - /// `input_writer` loop (`while let Ok(b) = rx.recv() { surface.write_input(&b) }`). + /// `input_writer` loop via the same source-tagged record's `write_to` seam. /// One FIFO + one writer ⇒ exact input order. Non-vacuous: a reorder/LIFO/dedup /// mutation of the single-writer contract makes the recorded sequence diverge. // [unit->REQ-HAZARD-PTY-INPUT-WRITER-WEDGE] @@ -11074,8 +11444,9 @@ mod tests { // Close the FIFO so the drain loop terminates, then drain through the SOLE // writer exactly as `input_writer` does. drop(w); - while let Ok(bytes) = rx.recv() { - surface.write_input(&bytes).unwrap(); + let deliveries = DeliveryBytes::default(); + while let Ok(record) = rx.recv() { + record.write_to(&surface, &deliveries); } assert_eq!( @@ -11423,7 +11794,7 @@ mod tests { "the watermark advance moves delivered_through past the suppressed range" ); - log.clear_controller(); + log.clear_controller("test"); drop(log); // Nothing further leaks to the viewer after its sync frame. assert!( @@ -11456,7 +11827,7 @@ mod tests { log.commit_resize(4, 10); let _ = read_frame(&mut ctrl_client).expect("sync frame"); assert_eq!(wait_delivered(&log.delivered_through, 3), 3); - log.clear_controller(); + log.clear_controller("test"); // Resume from the cursor-of-record, exactly as a cold-starting brain does // (KIND_SESSIONS resume_seq = delivered_through). @@ -11471,7 +11842,7 @@ mod tests { "the resumed controller starts at the live frame — a suppressed seq \ (1 or 2) here is the raw mixed-geometry replay the barrier forbids" ); - log.clear_controller(); + log.clear_controller("test"); drop(log); } @@ -11486,7 +11857,7 @@ mod tests { log.become_controller(Arc::clone(&send), None, 0, 0); let _ = read_frame(&mut client).expect("initial repaint"); wait_delivered(&log.delivered_through, 1); - log.clear_controller(); // detached BEFORE the resize, cursor rests at 1 + log.clear_controller("test"); // detached BEFORE the resize, cursor rests at 1 log.begin_resize(4, 10).expect("barrier closes"); assert!(log.append(b"\x1b[2;1Hheld-one").is_none()); // seq 1, suppressed log.mark_resize_issued(); @@ -11527,7 +11898,7 @@ mod tests { let f = read_frame(&mut client2).expect("the live frame after the repaint"); let ev: crate::msg::OutputEvent = serde_json::from_value(f.payload).unwrap(); assert_eq!(ev.seq, 3, "live frames stream raw + in-order after the repaint"); - log.clear_controller(); + log.clear_controller("test"); drop(log); } @@ -11550,7 +11921,7 @@ mod tests { b"after-commit-raw", "an at/above-floor resume is the raw ring slice, byte-for-byte" ); - log.clear_controller(); + log.clear_controller("test"); drop(log); } @@ -11796,7 +12167,7 @@ mod tests { assert_eq!(ev.seq, 1); assert_eq!(wait_delivered(&log.delivered_through, 2), 2); - log.clear_controller(); + log.clear_controller("test"); drop(log); } @@ -12158,7 +12529,9 @@ mod tests { // Drain the FIFO: the bytes were enqueued EXACTLY ONCE despite two apply_once // calls (a re-driven keystroke must not double-type into the paste). drop(w); - let drained: Vec> = std::iter::from_fn(|| rx.recv().ok()).collect(); + let drained: Vec> = std::iter::from_fn(|| rx.recv().ok()) + .map(InputRecord::into_bytes) + .collect(); assert_eq!( drained, vec![bytes], @@ -12313,7 +12686,7 @@ mod tests { "a superseded writer-A must emit NOTHING; got a leaked frame {leaked:?}" ); - log.clear_controller(); + log.clear_controller("test"); drop(send); drop(log); } @@ -12390,7 +12763,7 @@ mod tests { assert!(matches!(out, SubscribeOutcome::BusyControlled { .. }), "got {out:?}"); assert_eq!(decision, "busy"); - log.clear_controller(); + log.clear_controller("test"); } /// The COLLISION the `from_seq` key element exists for: core itself @@ -12438,7 +12811,7 @@ mod tests { ); assert_eq!(log.controller.as_ref().unwrap().attach_gen, 700, "the generation is preserved (fix 6)"); - log.clear_controller(); + log.clear_controller("test"); } /// Ownership/generation validation (ADR-0038 Amendment fix 6): the @@ -12502,7 +12875,7 @@ mod tests { 0, None); assert!(matches!(out, SubscribeOutcome::Controller), "legacy gen 0 re-takes, got {out:?}"); - log.clear_controller(); + log.clear_controller("test"); } /// The gap-resume re-subscribe (gate round 1 must-fix 1): a serve worker's @@ -12665,7 +13038,7 @@ mod tests { "the initial batch is ONE clean repaint, not the raw ring replayed chunk-by-chunk" ); - log.clear_controller(); + log.clear_controller("test"); drop(send); drop(log); } @@ -12732,7 +13105,7 @@ mod tests { assert_eq!(auth[1], "", "overflow row stays blank"); assert_eq!(auth[2], "Found:世界"); - log.clear_controller(); + log.clear_controller("test"); drop(send); drop(log); } @@ -12817,7 +13190,7 @@ mod tests { "an own-node controller latches driven_by to its own hex (CONTEXT:386)" ); assert!(after.controlled, "controlled stays true (any-controller truth)"); - log.clear_controller(); + log.clear_controller("test"); // A remote hex latches identically. let (send2, _client2, _recv2) = controller_socket_pair(); @@ -12829,7 +13202,7 @@ mod tests { "a remote controller latches driven_by verbatim" ); assert!(after2.controlled); - log.clear_controller(); + log.clear_controller("test"); }); } @@ -15360,7 +15733,7 @@ mod tests { // A DETACH drops it. empower("bignet"); assert!(holds(), "re-empowered for the detach leg"); - log.clear_controller(); + log.clear_controller("test"); assert!(!holds(), "detach revokes every empowerment"); // And the posture that derives from the now-empty seat reads @@ -15587,7 +15960,7 @@ mod tests { log.stamp_controller_seal_ceremony(conn, true); assert!(log.ceremony_surface().unwrap().3, "the declaring conn stamps"); - log.clear_controller(); + log.clear_controller("test"); assert!( log.ceremony_surface().is_none(), "detach clears the surface with the seat" @@ -15708,7 +16081,7 @@ mod tests { let sessions = recover(&broker.sessions); Arc::clone(&sessions.get(&sid).unwrap().log) }; - recover_log(&log).clear_controller(); + recover_log(&log).clear_controller("test"); broker .dispatch_seal_ceremony(seal_ceremony_env("ling", "sealnet", b"content"), &req_send) .unwrap(); diff --git a/crates/spt-daemon/src/lib.rs b/crates/spt-daemon/src/lib.rs index 359cc206..8b17f864 100644 --- a/crates/spt-daemon/src/lib.rs +++ b/crates/spt-daemon/src/lib.rs @@ -121,6 +121,7 @@ pub mod crc_swap; pub mod daemon; pub mod deadline; pub mod deelevate; +mod deliverybytes; pub mod digest; pub mod digesthub; pub mod digestlink; @@ -132,6 +133,7 @@ pub mod effect; pub mod endpoint; pub mod failedaddr; pub mod firewall; +pub mod bootstrap_firewall; pub mod answerop; pub mod forkop; pub mod redeemop; @@ -139,6 +141,7 @@ pub mod frame; pub mod grants; pub mod harnesshost; pub mod inject; +mod inputreceipt; pub mod iobus; pub mod knocknotif; pub mod lifecycle;