diff --git a/CHANGELOG.md b/CHANGELOG.md index f39f7e40..97ffa493 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,6 +10,17 @@ breaks something, or changes the observable behavior of existing surfaces broadly; **patch** for fixes, and for additive opt-in capability — a new key, flag, or page that no existing user can encounter without opting into it. +## [Unreleased] + +### Fixed + + +- Quoting a filepath while controlling an agent through remote `spt rc` can now + offer that agent a fetch command for the live file on your machine. The reference + lasts up to 24 hours and is addressed only to that agent. Repeated reports do not + extend it or repeat the notice. Missing files stay silent; unavailable provenance + and other refusals are named. No harness-adapter changes are required. + ## [0.69.0] - 2026-09-11 ### Fixed diff --git a/Cargo.lock b/Cargo.lock index c8fa4e8c..2cff96ff 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4158,6 +4158,7 @@ dependencies = [ "strum 0.28.0", "tempfile", "tokio", + "uuid", ] [[package]] diff --git a/crates/spt-daemon/Cargo.toml b/crates/spt-daemon/Cargo.toml index 5748d6d4..615159e4 100644 --- a/crates/spt-daemon/Cargo.toml +++ b/crates/spt-daemon/Cargo.toml @@ -67,6 +67,7 @@ hyper-util = { version = "0.1", features = ["tokio"] } http-body-util = "0.1" serde = { version = "1", features = ["derive"] } serde_json = { version = "1", features = ["preserve_order"] } +uuid = { version = "1", features = ["v4"] } # PTY output is arbitrary bytes; the forward-compat JSON envelope carries it as # a base64 string in the payload rather than forking the codec for binary. base64 = "0.22" diff --git a/crates/spt-daemon/src/attach.rs b/crates/spt-daemon/src/attach.rs index f77f45f4..3da33ffe 100644 --- a/crates/spt-daemon/src/attach.rs +++ b/crates/spt-daemon/src/attach.rs @@ -687,6 +687,11 @@ pub fn serve_attach( facts, )?; } + AttachRecord::InputPathsReply { reply } + if attached && role == Some(ServeRole::Controller) => + { + brain.complete_input_paths(&reply)?; + } // Input/Resize before Request, a viewer's input/resize, // or target-direction records echoed back: noise — ignore. _ => {} @@ -734,6 +739,16 @@ pub fn serve_attach( } } } + BrokerEvent::Other(env) + if attached + && role == Some(ServeRole::Controller) + && env.kind == crate::msg::KIND_USER_INPUT_PATHS => + { + let request = serde_json::from_value(env.payload) + .map_err(|error| io::Error::new(io::ErrorKind::InvalidData, error))?; + let line = ndjson::encode_line(&AttachRecord::InputPaths { request }); + wire.net_stream_send(stream_id, &line, None, false)?; + } // Seal-ceremony overlay open / per-attempt verdict (WAX-SEAL W2): // broker-authored records forwarded onto the attach wire for the // controller's rc to render. Additive record kinds — pushed only diff --git a/crates/spt-daemon/src/brain.rs b/crates/spt-daemon/src/brain.rs index afb9d92c..54a2b022 100644 --- a/crates/spt-daemon/src/brain.rs +++ b/crates/spt-daemon/src/brain.rs @@ -2293,6 +2293,38 @@ impl Brain { } } + /// Bind the existing hook report before any async helper or roster work. + // [impl->REQ-INPUT-PROVENANCE-ACCEPTANCE-REPORT] + pub fn report_user_input( + &mut self, + report: crate::msg::UserInputReport, + ) -> io::Result { + self.send( + crate::msg::KIND_USER_INPUT_REPORT, + serde_json::to_value(report).expect("UserInputReport serializes"), + )?; + let deadline = self.call_deadline(); + loop { + match self.read_event_until(deadline)? { + BrokerEvent::Other(env) if env.kind == crate::msg::KIND_USER_INPUT_REPORTED => { + return serde_json::from_value(env.payload).map_err(io::Error::other); + } + BrokerEvent::Error { message } => return Err(io::Error::other(message)), + _ => {} + } + } + } + + /// The answer returns on the same broker connection that owns the rc seat. + // [impl->REQ-INPUT-PROVENANCE-ACCEPTANCE-REPORT] + pub fn complete_input_paths( + &mut self, + reply: &spt_net::net::attach::InputPathReply, + ) -> io::Result<()> { + self.send(crate::msg::KIND_USER_INPUT_PATHS_REPLY, + serde_json::to_value(reply).map_err(io::Error::other)?) + } + /// List the broker's hosted sessions (id + owning endpoint) — the inbound /// dispatcher's session→endpoint resolution (D9-1). pub fn sessions(&mut self) -> io::Result { diff --git a/crates/spt-daemon/src/broker.rs b/crates/spt-daemon/src/broker.rs index 4c58d297..94d580c3 100644 --- a/crates/spt-daemon/src/broker.rs +++ b/crates/spt-daemon/src/broker.rs @@ -3818,6 +3818,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 +4024,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 +4122,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 +4167,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 +4204,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 +4223,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 +4263,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 +4272,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 +4294,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 +4795,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 +4869,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 +4932,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 +4942,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 +4971,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); @@ -6293,6 +6335,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 +8656,7 @@ impl Broker { HostedSession { session, input, + input_reports: Arc::new(Mutex::new(crate::inputreceipt::ReceiptBook::default())), drain, log, endpoint: req.endpoint.clone(), @@ -8913,6 +8969,102 @@ 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 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()); + } + let (log, input, reports) = { + let sessions = recover(&self.sessions); + let h = sessions.values().find(|h| h.endpoint == req.endpoint) + .ok_or("no broker-held controller seat")?; + (Arc::clone(&h.log), Arc::clone(&h.input), Arc::clone(&h.input_reports)) + }; + // The source classifier is consulted before attributing any payload. + 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 local_node = crate::access::local_node_hex(); + let (request, controller_tx) = { + let mut log = recover_log(&log); + if req.expires_at_ms <= crate::brain::now_ms() { + return Err("input report expired before broker receipt".into()); + } + log.reap_dead_controller(); + crate::inputreceipt::remote_origin( + log.controller.as_ref().map(|c| c.by.as_deref()), + !log.viewers.is_empty(), + local_node.as_deref(), + ).map_err(str::to_owned)?; + let controller = log.controller.as_ref().expect("remote origin requires a controller"); + let request = recover(&reports).begin( + &req.session, &req.payload, controller.send.id(), crate::brain::now_ms(), + ).map_err(str::to_owned)?; + (request, controller.tx.clone()) + }; + 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). @@ -10262,6 +10414,82 @@ mod tests { }); } + // [unit->REQ-INPUT-PROVENANCE-ACCEPTANCE-REPORT] + #[test] + fn input_receipt_keeps_its_controller_across_take_and_publishes_once() { + 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, + })); + assert!(broker.dispatch_user_input_report(report("wrong-session")).is_err()); + assert!(broker.dispatch_user_input_report(report("session")) + .unwrap_err().contains("local controller")); + recover_log(&log).become_controller(Arc::clone(&original), Some("11".repeat(32)), 0, 1); + 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 +11247,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 +11372,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!( @@ -12158,7 +12457,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], diff --git a/crates/spt-daemon/src/deliverybytes.rs b/crates/spt-daemon/src/deliverybytes.rs new file mode 100644 index 00000000..8ee0f959 --- /dev/null +++ b/crates/spt-daemon/src/deliverybytes.rs @@ -0,0 +1,385 @@ +//! Session-lifetime evidence from successful physical delivery writes, not enqueue intent. + +use sha2::{Digest, Sha256}; +use spt_term::surface::SessionSurface; +use std::collections::HashSet; +use std::sync::{Arc, Mutex}; + +// Digests, not retained payloads. Never evict evidence from a live hosted session: +// exhaustion disables origin grants with a named reason instead. +const MAX_DELIVERY_DIGESTS: usize = 65_536; + +#[derive(Default)] +pub(crate) struct DeliveryBytes { + completed: Mutex, +} + +#[derive(Default)] +struct Completed { + digests: HashSet<[u8; 32]>, + exhausted: bool, + writing: bool, + unproven: bool, +} + +impl DeliveryBytes { + pub(crate) fn matches(&self, payload: &[u8]) -> bool { + let digest: [u8; 32] = Sha256::digest(payload).into(); + self.completed + .lock() + .unwrap_or_else(|p| p.into_inner()) + .digests + .contains(&digest) + } + + pub(crate) fn unavailable_reason(&self) -> Option<&'static str> { + let completed = self.completed.lock().unwrap_or_else(|p| p.into_inner()); + if completed.unproven { + Some("delivery-write-unproven") + } else if completed.writing { + Some("delivery-write-in-progress") + } else if completed.exhausted { + Some("delivery-evidence-capacity") + } else { + None + } + } + + fn begin_write(&self) { + self.completed + .lock() + .unwrap_or_else(|p| p.into_inner()) + .writing = true; + } + + fn finish_write(&self, success: bool) { + let mut completed = self.completed.lock().unwrap_or_else(|p| p.into_inner()); + completed.unproven |= !success; + completed.writing = false; + } + + fn remember(&self, digest: [u8; 32]) { + self.remember_with_limit(digest, MAX_DELIVERY_DIGESTS); + } + + fn remember_with_limit(&self, digest: [u8; 32], limit: usize) { + let mut completed = self.completed.lock().unwrap_or_else(|p| p.into_inner()); + if completed.digests.contains(&digest) { + return; + } + if completed.digests.len() >= limit { + completed.exhausted = true; + } else { + completed.digests.insert(digest); + } + } +} + +/// One translation attempt; queued records keep its constant-size streaming state +/// alive until the lone writer has processed them, independently of logical Commit. +#[derive(Default)] +pub(crate) struct DeliveryAttempt { + text: Mutex, +} + +#[derive(Default)] +struct AttemptText { + hash: Sha256, + has_text: bool, + failed: bool, +} + +#[derive(Clone, Copy)] +pub(crate) enum DeliveryPart { + Text { end: bool }, + Key, +} + +impl DeliveryAttempt { + fn failed(&self) { + self.text.lock().unwrap_or_else(|p| p.into_inner()).failed = true; + } + + fn written(&self, bytes: &[u8], part: DeliveryPart, ledger: &DeliveryBytes) { + let mut text = self.text.lock().unwrap_or_else(|p| p.into_inner()); + if text.failed { + return; + } + if let DeliveryPart::Text { end } = part { + text.hash.update(bytes); + text.has_text |= !bytes.is_empty(); + if end && text.has_text { + // Publish each complete Text command's cumulative text, including + // exact embedded newlines. Keys are source-tagged controls, never + // guessed framing stripped from the reported payload. + ledger.remember(text.hash.clone().finalize().into()); + } + } + } +} + +pub(crate) struct InputRecord { + bytes: Vec, + origin: InputOrigin, +} + +enum InputOrigin { + Controller, + Probe, + Delivery { + attempt: Arc, + part: DeliveryPart, + }, +} + +impl InputRecord { + #[cfg(test)] + pub(crate) fn into_bytes(self) -> Vec { + self.bytes + } + + pub(crate) fn controller(bytes: Vec) -> Self { + Self { + bytes, + origin: InputOrigin::Controller, + } + } + + pub(crate) fn probe(bytes: Vec) -> Self { + Self { + bytes, + origin: InputOrigin::Probe, + } + } + + pub(crate) fn delivery( + bytes: Vec, + attempt: &Arc, + part: DeliveryPart, + ) -> Self { + Self { + bytes, + origin: InputOrigin::Delivery { + attempt: Arc::clone(attempt), + part, + }, + } + } + + pub(crate) fn dropped(&self) { + if let InputOrigin::Delivery { attempt, .. } = &self.origin { + attempt.failed(); + } + } + + // [impl->REQ-INPUT-PROVENANCE-ACCEPTANCE-REPORT] + pub(crate) fn write_to(&self, surface: &impl SessionSurface, ledger: &DeliveryBytes) { + if let InputOrigin::Delivery { attempt, part } = &self.origin { + // Mark uncertainty BEFORE input can cause a report. No custody lock + // spans the blocking PTY call; receipt declines immediately instead. + ledger.begin_write(); + let success = surface.write_input(&self.bytes).is_ok(); + if success { + attempt.written(&self.bytes, *part, ledger); + } else { + // Err can follow a partial write; this API exposes no byte count. + // Keep session evidence unavailable, rather than grant an origin + // merely because the partly delivered payload lacks a digest. + attempt.failed(); + } + // Successful candidate publication precedes clearing the active flag. + // Receipt checks unavailable_reason BEFORE matches so it cannot miss + // an unpublished digest and then observe an already-cleared flag. + ledger.finish_write(success); + } else { + let _ = surface.write_input(&self.bytes); + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use spt_term::surface::{SurfaceError, SurfaceSize}; + + struct Sink; + impl SessionSurface for Sink { + fn write_input(&self, _: &[u8]) -> Result<(), SurfaceError> { + Ok(()) + } + fn resize(&self, _: SurfaceSize) -> Result<(), SurfaceError> { + Ok(()) + } + } + + // [unit->REQ-INPUT-PROVENANCE-ACCEPTANCE-REPORT] + #[test] + fn split_multiline_delivery_matches_before_commit_without_normalizing() { + let ledger = DeliveryBytes::default(); + let attempt = Arc::new(DeliveryAttempt::default()); + let first = InputRecord::delivery( + b"\n C:/a".to_vec(), + &attempt, + DeliveryPart::Text { end: false }, + ); + let last = InputRecord::delivery( + b"\n ".to_vec(), + &attempt, + DeliveryPart::Text { end: true }, + ); + let payload = b"\n C:/a\n "; + assert!(!ledger.matches(payload)); + first.write_to(&Sink, &ledger); + assert!(!ledger.matches(payload)); + last.write_to(&Sink, &ledger); + assert!(ledger.matches(payload)); + InputRecord::delivery(b"\r".to_vec(), &attempt, DeliveryPart::Key).write_to(&Sink, &ledger); + assert!(ledger.matches(payload)); + assert!(!ledger.matches(b"\n C:/b\n ")); + assert!(!ledger.matches(b"\n C:/a\n")); + assert!(!ledger.matches(b"\r\n C:/a\r\n ")); + assert!(!ledger.matches(b"\r")); + } + + // [unit->REQ-INPUT-PROVENANCE-ACCEPTANCE-REPORT] + #[test] + fn controller_and_probe_bytes_never_become_delivery_evidence() { + let ledger = DeliveryBytes::default(); + InputRecord::controller(b"human C:/private".to_vec()).write_to(&Sink, &ledger); + InputRecord::probe(b"\x1b[6n".to_vec()).write_to(&Sink, &ledger); + assert!(!ledger.matches(b"human C:/private")); + assert!(!ledger.matches(b"\x1b[6n")); + assert!(!ledger.matches(b"unknown human payload")); + } + + // [unit->REQ-INPUT-PROVENANCE-ACCEPTANCE-REPORT] + #[test] + fn report_during_physical_write_declines_until_success_is_published() { + struct ReportingSurface<'a>(&'a DeliveryBytes); + impl SessionSurface for ReportingSurface<'_> { + fn write_input(&self, bytes: &[u8]) -> Result<(), SurfaceError> { + assert_eq!( + self.0.unavailable_reason(), + Some("delivery-write-in-progress") + ); + assert!(!self.0.matches(bytes)); + Ok(()) + } + fn resize(&self, _: SurfaceSize) -> Result<(), SurfaceError> { + Ok(()) + } + } + + let ledger = DeliveryBytes::default(); + let attempt = Arc::new(DeliveryAttempt::default()); + InputRecord::delivery( + b"submission C:/exact".to_vec(), + &attempt, + DeliveryPart::Text { end: true }, + ) + .write_to(&ReportingSurface(&ledger), &ledger); + assert_eq!(ledger.unavailable_reason(), None); + assert!(ledger.matches(b"submission C:/exact")); + } + + // [unit->REQ-INPUT-PROVENANCE-ACCEPTANCE-REPORT] + #[test] + fn failed_physical_write_cannot_certify_omitted_or_partial_bytes() { + struct PartialFailure(Mutex>); + impl SessionSurface for PartialFailure { + fn write_input(&self, bytes: &[u8]) -> Result<(), SurfaceError> { + self.0 + .lock() + .unwrap_or_else(|p| p.into_inner()) + .extend_from_slice(&bytes[..2]); + Err(SurfaceError::Io(std::io::Error::from( + std::io::ErrorKind::BrokenPipe, + ))) + } + fn resize(&self, _: SurfaceSize) -> Result<(), SurfaceError> { + Ok(()) + } + } + + let ledger = DeliveryBytes::default(); + let attempt = Arc::new(DeliveryAttempt::default()); + InputRecord::delivery( + b"before".to_vec(), + &attempt, + DeliveryPart::Text { end: false }, + ) + .write_to(&Sink, &ledger); + let partial = PartialFailure(Mutex::new(Vec::new())); + InputRecord::delivery( + b"middle".to_vec(), + &attempt, + DeliveryPart::Text { end: false }, + ) + .write_to(&partial, &ledger); + InputRecord::delivery( + b"after".to_vec(), + &attempt, + DeliveryPart::Text { end: true }, + ) + .write_to(&Sink, &ledger); + assert_eq!(*partial.0.lock().unwrap_or_else(|p| p.into_inner()), b"mi"); + assert!(!ledger.matches(b"beforemiddleafter")); + assert!(!ledger.matches(b"beforeafter")); + assert!(!ledger.matches(b"beforemiafter")); + assert_eq!(ledger.unavailable_reason(), Some("delivery-write-unproven")); + + let recovered_attempt = Arc::new(DeliveryAttempt::default()); + InputRecord::delivery( + b"later success".to_vec(), + &recovered_attempt, + DeliveryPart::Text { end: true }, + ) + .write_to(&Sink, &ledger); + assert!(ledger.matches(b"later success")); + assert_eq!(ledger.unavailable_reason(), Some("delivery-write-unproven")); + } + + // [unit->REQ-INPUT-PROVENANCE-ACCEPTANCE-REPORT] + #[test] + fn successful_text_commands_aggregate_without_key_bytes() { + let ledger = DeliveryBytes::default(); + let attempt = Arc::new(DeliveryAttempt::default()); + InputRecord::delivery( + b"one\n".to_vec(), + &attempt, + DeliveryPart::Text { end: true }, + ) + .write_to(&Sink, &ledger); + InputRecord::delivery(b"\x1b[A".to_vec(), &attempt, DeliveryPart::Key) + .write_to(&Sink, &ledger); + InputRecord::delivery(b"two".to_vec(), &attempt, DeliveryPart::Text { end: true }) + .write_to(&Sink, &ledger); + assert!(ledger.matches(b"one\ntwo")); + assert!(!ledger.matches(b"one\n\x1b[Atwo")); + let next_attempt = Arc::new(DeliveryAttempt::default()); + InputRecord::delivery( + b"three".to_vec(), + &next_attempt, + DeliveryPart::Text { end: true }, + ) + .write_to(&Sink, &ledger); + assert!(ledger.matches(b"three")); + assert!(!ledger.matches(b"one\ntwothree")); + } + + // [unit->REQ-INPUT-PROVENANCE-ACCEPTANCE-REPORT] + #[test] + fn exhausted_evidence_keeps_old_matches_and_declines_future_origin_grants() { + let ledger = DeliveryBytes::default(); + ledger.remember_with_limit(Sha256::digest(b"first").into(), 1); + ledger.remember_with_limit(Sha256::digest(b"first").into(), 1); + assert_eq!(ledger.unavailable_reason(), None); + ledger.remember_with_limit(Sha256::digest(b"second").into(), 1); + assert!(ledger.matches(b"first")); + assert!(!ledger.matches(b"second")); + assert_eq!( + ledger.unavailable_reason(), + Some("delivery-evidence-capacity") + ); + } +} diff --git a/crates/spt-daemon/src/inputreceipt.rs b/crates/spt-daemon/src/inputreceipt.rs new file mode 100644 index 00000000..926292c7 --- /dev/null +++ b/crates/spt-daemon/src/inputreceipt.rs @@ -0,0 +1,368 @@ +//! Broker-session custody for receipt-bound input references. No payload is retained. + +use sha2::{Digest, Sha256}; +use spt_net::net::attach::{InputPathOutcome, InputPathReply, InputPathRequest}; +use spt_store::helperline::{quoted_path_candidates, HelperLine, MAX_QUOTED_PATHS}; +use spt_store::serving::HELPER_ENTRY_TTL_MS; +use std::collections::{HashMap, HashSet}; +use std::time::{Duration, Instant}; + +pub(crate) const REPLY_BOUND: Duration = Duration::from_secs(10); +const LIVE_RECEIPTS: usize = 4096; +const PENDING_RECEIPTS: usize = 64; + +/// The seat read is the broker's live slot, never the perch's recorded mirror. +// [impl->REQ-INPUT-PROVENANCE-ACCEPTANCE-REPORT] +pub(crate) fn remote_origin<'a>( + controller: Option>, + viewers: bool, + local_node: Option<&str>, +) -> Result<&'a str, &'static str> { + match controller { + None if viewers => Err("viewer-only: no controller seat"), + None => Err("no controller seat"), + Some(None) => Err("local controller seat"), + Some(Some(by)) if Some(by) == local_node => Err("local controller seat"), + Some(Some(_)) if local_node.is_none() => Err("local node identity unavailable"), + Some(Some(by)) => Ok(by), + } +} + +struct Receipt { + request: InputPathRequest, + controller_conn: u64, + expires_at_ms: u64, + answered: bool, + received: Instant, +} + +/// Session-owned and therefore survives a brain swap, but not PTY teardown. +/// Unanswered requests retain their dedup fence for the possible exposure's full +/// lifetime: a lost reply is not proof that the owner registered nothing. +#[derive(Default)] +pub(crate) struct ReceiptBook { + receipts: HashMap<(String, [u8; 32]), Receipt>, + // Includes answered requests until their timer exits: fast missing/refused + // replies must not allow an unbounded number of sleeping deadline threads. + deadlines: HashSet, +} + +pub(crate) struct Completion { + pub lines: Vec, + pub declines: Vec, +} + +impl ReceiptBook { + // [impl->REQ-INPUT-PROVENANCE-ACCEPTANCE-REPORT] + pub fn begin( + &mut self, + session: &str, + payload: &str, + controller_conn: u64, + now: u64, + ) -> Result, &'static str> { + self.receipts.retain(|_, r| r.expires_at_ms > now); + let key = ( + session.to_owned(), + Sha256::digest(payload.as_bytes()).into(), + ); + if self.receipts.contains_key(&key) { + return Ok(None); + } + let paths = quoted_path_candidates(payload, MAX_QUOTED_PATHS); + if paths.is_empty() { + return Err("payload quoted no absolute path"); + } + if self.receipts.len() >= LIVE_RECEIPTS || self.deadlines.len() >= PENDING_RECEIPTS { + return Err("input receipt capacity reached"); + } + let request = InputPathRequest { + receipt_id: uuid::Uuid::new_v4().to_string(), + session: session.to_owned(), + paths, + received_at_ms: now, + }; + self.deadlines.insert(request.receipt_id.clone()); + self.receipts.insert( + key, + Receipt { + request: request.clone(), + controller_conn, + received: Instant::now(), + expires_at_ms: now.saturating_add(HELPER_ENTRY_TTL_MS), + answered: false, + }, + ); + Ok(Some(request)) + } + + /// A queue refusal proves nothing was handed to the controller. + pub fn cancel_unsent(&mut self, receipt_id: &str) { + self.deadlines.remove(receipt_id); + self.receipts + .retain(|_, r| r.request.receipt_id != receipt_id); + } + + /// Exactly one deadline diagnostic; a late reply cannot revive this request. + pub fn timeout(&mut self, receipt_id: &str) -> bool { + self.deadlines.remove(receipt_id); + let Some(r) = self + .receipts + .values_mut() + .find(|r| r.request.receipt_id == receipt_id) + else { + return false; + }; + if r.answered { + return false; + } + r.answered = true; + true + } + + pub fn contains(&self, receipt_id: &str) -> bool { + self.receipts + .values() + .any(|r| r.request.receipt_id == receipt_id) + } + + // [impl->REQ-INPUT-PROVENANCE-ACCEPTANCE-REPORT] + pub fn complete( + &mut self, + reply: &InputPathReply, + controller_conn: u64, + now: u64, + ) -> Result { + let Some(r) = self + .receipts + .values_mut() + .find(|r| r.request.receipt_id == reply.receipt_id) + else { + return Err("unknown input receipt"); + }; + if r.controller_conn != controller_conn { + return Err("reply is not from the receipt-bound controller connection"); + } + if r.answered || r.expires_at_ms <= now || r.received.elapsed() >= REPLY_BOUND { + return Err("input receipt already answered or past its reply bound"); + } + if reply.results.len() != r.request.paths.len() + || r.request + .paths + .iter() + .any(|p| reply.results.iter().filter(|v| &v.path == p).count() != 1) + { + return Err("reply paths differ from the receipt"); + } + for result in &reply.results { + if let InputPathOutcome::Registered { expires_at_ms, .. } = &result.outcome { + if *expires_at_ms > r.expires_at_ms || *expires_at_ms <= now { + return Err("reply reference lifetime is outside the receipt bound"); + } + } + } + let mut out = Completion { + lines: Vec::new(), + declines: Vec::new(), + }; + let mut live_until = now; + for result in &reply.results { + match &result.outcome { + InputPathOutcome::Registered { url, expires_at_ms } => { + live_until = live_until.max(*expires_at_ms); + out.lines.push(HelperLine { + msg_id: r.request.receipt_id.clone(), + path: result.path.clone(), + url: url.clone(), + at_ms: now, + }); + } + InputPathOutcome::Missing => {} + InputPathOutcome::Unanswered { reason } => { + live_until = r.expires_at_ms; + out.declines.push(format!( + "path={} outcome=unanswered reason={reason}", + result.path + )); + } + InputPathOutcome::Declined { reason } => { + out.declines + .push(format!("path={} reason={reason}", result.path)); + } + } + } + r.answered = true; + r.expires_at_ms = live_until; + Ok(out) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use spt_net::net::attach::InputPathResult; + + // [unit->REQ-INPUT-PROVENANCE-ACCEPTANCE-REPORT] + #[test] + fn receipt_origin_requires_a_remote_controller_not_a_viewer_or_local_seat() { + assert_eq!( + remote_origin(Some(Some("remote")), true, Some("local")), + Ok("remote") + ); + assert_eq!( + remote_origin(Some(None), true, Some("local")), + Err("local controller seat") + ); + assert_eq!( + remote_origin(Some(Some("local")), false, Some("local")), + Err("local controller seat") + ); + assert_eq!( + remote_origin(None, true, Some("local")), + Err("viewer-only: no controller seat") + ); + assert_eq!( + remote_origin(None, false, Some("local")), + Err("no controller seat") + ); + } + + fn registered(request: &InputPathRequest, expires_at_ms: u64) -> InputPathReply { + InputPathReply { + receipt_id: request.receipt_id.clone(), + results: vec![InputPathResult { + path: request.paths[0].clone(), + outcome: InputPathOutcome::Registered { + url: "http://owner:5474/owner/input.txt".into(), + expires_at_ms, + }, + }], + } + } + + // [unit->REQ-INPUT-PROVENANCE-ACCEPTANCE-REPORT] + #[test] + fn repeated_payload_has_one_notice_and_never_extends_a_live_reference() { + let mut book = ReceiptBook::default(); + let payload = "read /input.txt"; + let first = book.begin("s1", payload, 7, 100).unwrap().unwrap(); + assert!(book.begin("s1", payload, 9, 101).unwrap().is_none()); + let reply = registered(&first, 500); + assert_eq!( + book.complete(&reply, 7, 110).unwrap().lines[0].url, + "http://owner:5474/owner/input.txt" + ); + assert!(book.complete(&reply, 7, 111).is_err()); + assert!(book.begin("s1", payload, 9, 499).unwrap().is_none()); + assert!(book.begin("s2", payload, 9, 499).unwrap().is_some()); + let next = book.begin("s1", payload, 9, 500).unwrap().unwrap(); + assert_ne!(next.receipt_id, first.receipt_id); + } + + // [unit->REQ-INPUT-PROVENANCE-ACCEPTANCE-REPORT] + #[test] + fn missing_file_is_silent_but_a_refusal_is_named() { + let mut book = ReceiptBook::default(); + let request = book + .begin("s", "read /missing and /denied", 1, 10) + .unwrap() + .unwrap(); + let reply = InputPathReply { + receipt_id: request.receipt_id, + results: vec![ + InputPathResult { + path: "/missing".into(), + outcome: InputPathOutcome::Missing, + }, + InputPathResult { + path: "/denied".into(), + outcome: InputPathOutcome::Declined { + reason: "access denied".into(), + }, + }, + ], + }; + let result = book.complete(&reply, 1, 20).unwrap(); + assert!(result.lines.is_empty()); + assert_eq!(result.declines, ["path=/denied reason=access denied"]); + } + + // [unit->REQ-INPUT-PROVENANCE-ACCEPTANCE-REPORT] + #[test] + fn replies_cannot_change_controller_paths_or_lifetime() { + let mut book = ReceiptBook::default(); + let request = book.begin("s", "read /input", 7, 100).unwrap().unwrap(); + let mut reply = registered(&request, 200); + assert!(book.complete(&reply, 8, 110).is_err()); + reply.results[0].path = "/different".into(); + assert!(book.complete(&reply, 7, 110).is_err()); + reply = registered(&request, 101 + HELPER_ENTRY_TTL_MS); + assert!(book.complete(&reply, 7, 110).is_err()); + reply = registered(&request, 200); + assert_eq!(book.complete(&reply, 7, 110).unwrap().lines.len(), 1); + } + + // [unit->REQ-INPUT-PROVENANCE-ACCEPTANCE-REPORT] + #[test] + fn silent_controller_times_out_once_without_regranting_an_uncertain_exposure() { + let mut book = ReceiptBook::default(); + let request = book.begin("s", "read /input", 7, 100).unwrap().unwrap(); + assert!(book.timeout(&request.receipt_id)); + assert!(!book.timeout(&request.receipt_id)); + assert!(book.complete(®istered(&request, 300), 7, 200).is_err()); + assert!(book.begin("s", "read /input", 9, 200).unwrap().is_none()); + assert!(book + .begin("s", "read /input", 9, 100 + HELPER_ENTRY_TTL_MS) + .unwrap() + .is_some()); + } + + // [unit->REQ-INPUT-PROVENANCE-ACCEPTANCE-REPORT] + #[test] + fn uncertain_owner_result_keeps_the_fence_across_controller_changes() { + let mut book = ReceiptBook::default(); + let request = book.begin("s", "read /input", 7, 100).unwrap().unwrap(); + let reply = InputPathReply { + receipt_id: request.receipt_id, + results: vec![InputPathResult { + path: "/input".into(), + outcome: InputPathOutcome::Unanswered { + reason: "owner transport lost after request".into(), + }, + }], + }; + let result = book.complete(&reply, 7, 110).unwrap(); + assert!(result.lines.is_empty()); + assert_eq!( + result.declines, + ["path=/input outcome=unanswered reason=owner transport lost after request",] + ); + assert!(book.begin("s", "read /input", 9, 200).unwrap().is_none()); + } + + // [unit->REQ-INPUT-PROVENANCE-ACCEPTANCE-REPORT] + #[test] + fn fast_missing_replies_cannot_accumulate_unbounded_deadline_work() { + let mut book = ReceiptBook::default(); + let mut first = None; + for _ in 0..PENDING_RECEIPTS { + let request = book.begin("s", "read /input", 7, 100).unwrap().unwrap(); + first.get_or_insert_with(|| request.receipt_id.clone()); + book.complete( + &InputPathReply { + receipt_id: request.receipt_id, + results: vec![InputPathResult { + path: "/input".into(), + outcome: InputPathOutcome::Missing, + }], + }, + 7, + 100, + ) + .unwrap(); + } + assert!(book.begin("s", "read /input", 7, 100).is_err()); + assert!(!book.timeout(first.as_deref().unwrap())); + assert!(book.begin("s", "read /input", 7, 100).unwrap().is_some()); + } +} diff --git a/crates/spt-daemon/src/lib.rs b/crates/spt-daemon/src/lib.rs index 359cc206..2b581b7b 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; @@ -139,6 +140,7 @@ pub mod frame; pub mod grants; pub mod harnesshost; pub mod inject; +mod inputreceipt; pub mod iobus; pub mod knocknotif; pub mod lifecycle; diff --git a/crates/spt-daemon/src/msg.rs b/crates/spt-daemon/src/msg.rs index 2576c0cd..bc93f60f 100644 --- a/crates/spt-daemon/src/msg.rs +++ b/crates/spt-daemon/src/msg.rs @@ -450,6 +450,30 @@ pub const KIND_ENDPOINT_INPUT: &str = "endpoint-input"; /// (no matching broker session) tells the caller to fall back to the spool. pub const KIND_ENDPOINT_INJECTED: &str = "endpoint-injected"; +/// Authenticated USER_INPUT receipt, bound to the current broker controller. +// [impl->REQ-INPUT-PROVENANCE-ACCEPTANCE-REPORT] +pub const KIND_USER_INPUT_REPORT: &str = "user-input-report"; +pub const KIND_USER_INPUT_REPORTED: &str = "user-input-reported"; +pub const KIND_USER_INPUT_PATHS: &str = "user-input-paths"; +pub const KIND_USER_INPUT_PATHS_REPLY: &str = "user-input-paths-reply"; +pub const USER_INPUT_REPORT_BOUND: std::time::Duration = std::time::Duration::from_millis(500); + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct UserInputReport { + pub endpoint: String, + pub session: String, + pub payload: String, + /// Same-node deadline set before connection/handshake. A delayed worker + /// cannot attribute the report to a seat acquired after the caller gave up. + pub expires_at_ms: u64, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct UserInputReported { + /// Empty on admitted or duplicate reports. Non-grants name their condition. + pub declined: Option, +} + /// Client→broker: trigger a **planned brain-process restart** (ADR-0018 D3-3, /// Q3). `spt update apply` sends this after swapping the binary on disk; the /// broker signals its brain supervisor to cycle the brain child with diff --git a/crates/spt-daemon/src/servehost.rs b/crates/spt-daemon/src/servehost.rs index e995a306..91b1c228 100644 --- a/crates/spt-daemon/src/servehost.rs +++ b/crates/spt-daemon/src/servehost.rs @@ -57,6 +57,14 @@ pub enum ServeRequest { ttl_ms: u64, audience: String, }, + /// Only the locally held remote-controller viewport requests this operation. + /// The audience is its established target, never a field supplied by the peer. + AddInputReference { + path: PathBuf, + receipt_id: String, + audience: String, + received_at_ms: u64, + }, Remove { name_or_id: String, }, @@ -86,6 +94,8 @@ pub enum ServeRequest { #[serde(tag = "outcome", rename_all = "snake_case")] pub enum ServeResult { Entry { entry: ServedEntry }, + /// An input reference resolved to no source. No diagnostic is attached. + Missing, /// An attachment, with the SNAPSHOT's size — the sender needs it for the /// envelope, and the daemon is the only party that ever saw the copy. // [impl->REQ-WEB-ATTACHMENT-PULL] @@ -149,7 +159,13 @@ pub(crate) fn apply_at(home: &Path, request: ServeRequest, now_ms: u64) -> io::R .lock() .map_err(|_| io::Error::other("SERVE_REGISTRY_POISONED: writer state is unproven"))?; let path = spt_store::perch::serving_registry_file_in(home); - let mut registry = ServingRegistry::load_at(&path)?; + let mut registry = ServingRegistry::load_at(&path).map_err(|error| { + if matches!(&request, ServeRequest::AddInputReference { .. }) { + io::Error::new(error.kind(), format!("INPUT_PATH_REGISTRY: {error}")) + } else { + error + } + })?; match request { ServeRequest::Add { path: source, as_name, origin } => { let entry = registry.add_reference( @@ -214,6 +230,28 @@ pub(crate) fn apply_at(home: &Path, request: ServeRequest, now_ms: u64) -> io::R registry.save_at(&path)?; Ok(ServeResult::Entry { entry: scoped }) } + ServeRequest::AddInputReference { path: source, receipt_id, audience, received_at_ms } => { + // Resolve the submitter's home HERE, never on the receiving node. + let source = if let Some(relative) = source.to_str().and_then(|p| p.strip_prefix("~/")) { + std::env::home_dir() + .ok_or_else(|| io::Error::other("INPUT_PATH_HOME_UNAVAILABLE"))? + .join(relative) + } else { + source + }; + let previous_id = registry.entries().find(|entry| entry.path == source) + .map(|entry| entry.id.clone()); + match registry.add_input_reference(&source, &receipt_id, &audience, received_at_ms, now_ms)? { + None => Ok(ServeResult::Missing), + Some(entry) => { + if previous_id.as_deref() != Some(entry.id.as_str()) { + registry.save_at(&path) + .map_err(|error| io::Error::new(error.kind(), format!("INPUT_PATH_REGISTRY: {error}")))?; + } + Ok(ServeResult::Entry { entry }) + } + } + } ServeRequest::Remove { name_or_id } => { let entry = registry.remove(&name_or_id).ok_or_else(|| { io::Error::new(io::ErrorKind::NotFound, format!("SERVE_NOT_FOUND: {name_or_id}")) @@ -410,6 +448,135 @@ mod tests { ServeRequest::Add { path: path.to_owned(), as_name: None, origin: Some("alice".into()) } } + fn input_reference(path: &Path, receipt: &str, audience: &str, received_at_ms: u64) -> ServeRequest { + ServeRequest::AddInputReference { + path: path.to_owned(), + receipt_id: receipt.into(), + audience: audience.into(), + received_at_ms, + } + } + + #[test] + fn input_reference_missing_is_a_silent_skip_without_registry_publication() { + let tmp = tempfile::tempdir().unwrap(); + let home = tmp.path().join("home"); + let result = apply_at(&home, input_reference(&tmp.path().join("missing"), "r1", "agent", 1), 1) + .unwrap(); + assert!(matches!(result, ServeResult::Missing)); + assert!(!spt_store::perch::serving_registry_file_in(&home).exists()); + } + + #[test] + fn input_reference_replay_preserves_the_live_entry_and_deadline() { + let tmp = tempfile::tempdir().unwrap(); + let home = tmp.path().join("home"); + let source = tmp.path().join("report.txt"); + std::fs::write(&source, b"before").unwrap(); + let ServeResult::Entry { entry } = apply_at(&home, input_reference(&source, "r1", "agent", 10), 20) + .unwrap() else { panic!("expected reference") }; + let ServeResult::Entry { entry: replay } = + apply_at(&home, input_reference(&source, "r1", "agent", 50), 50).unwrap() + else { panic!("expected replay") }; + assert_eq!(replay, entry); + // A second submission can reuse the same narrow exposure, not renew it. + let ServeResult::Entry { entry: second } = + apply_at(&home, input_reference(&source, "r2", "agent", 60), 60).unwrap() + else { panic!("expected shared live reference") }; + assert_eq!(second, entry); + assert_eq!(list(&home), vec![entry]); + } + + #[test] + fn input_reference_cannot_rescope_another_audience_or_unrelated_entry() { + let tmp = tempfile::tempdir().unwrap(); + let home = tmp.path().join("home"); + let source = tmp.path().join("report.txt"); + let ordinary = tmp.path().join("ordinary.txt"); + std::fs::write(&source, b"scoped").unwrap(); + std::fs::write(&ordinary, b"ordinary").unwrap(); + apply_at(&home, input_reference(&source, "r1", "agent", 10), 10).unwrap(); + apply_at(&home, add(&ordinary), 10).unwrap(); + let before = list(&home); + for request in [ + input_reference(&source, "r1", "other", 20), + input_reference(&ordinary, "r2", "agent", 20), + ] { + let error = apply_at(&home, request, 20).unwrap_err(); + assert_eq!(error.kind(), io::ErrorKind::PermissionDenied); + assert!(error.to_string().contains("INPUT_PATH_SCOPE_CONFLICT")); + } + assert_eq!(list(&home), before); + } + + #[test] + fn input_reference_expiry_cannot_be_renewed_by_replay_before_or_after_reaping() { + let tmp = tempfile::tempdir().unwrap(); + let home = tmp.path().join("home"); + let source = tmp.path().join("report.txt"); + std::fs::write(&source, b"live").unwrap(); + let request = input_reference(&source, "r1", "agent", 10); + let ServeResult::Entry { entry } = apply_at(&home, request.clone(), 10).unwrap() + else { panic!("expected reference") }; + let expires = 10 + spt_store::serving::HELPER_ENTRY_TTL_MS; + assert!(apply_at(&home, request.clone(), expires).unwrap_err().to_string() + .contains("INPUT_PATH_EXPIRED")); + apply_at(&home, ServeRequest::Remove { name_or_id: entry.id }, expires).unwrap(); + assert!(apply_at(&home, request, expires).unwrap_err().to_string() + .contains("INPUT_PATH_EXPIRED")); + let ServeResult::Entry { entry: renewed } = + apply_at(&home, input_reference(&source, "r2", "agent", expires), expires).unwrap() + else { panic!("a new receipt may register after expiry") }; + assert_eq!(renewed.registered_at_ms, expires); + let next_expiry = expires + spt_store::serving::HELPER_ENTRY_TTL_MS; + let ServeResult::Entry { entry: replaced } = + apply_at(&home, input_reference(&source, "r3", "agent", next_expiry), next_expiry).unwrap() + else { panic!("a fresh receipt must not wait for the periodic reaper") }; + assert_ne!(replaced.id, renewed.id); + assert_eq!(list(&home), vec![replaced]); + } + + #[test] + fn input_reference_clock_skew_never_extends_either_nodes_lifetime() { + let tmp = tempfile::tempdir().unwrap(); + let home = tmp.path().join("home"); + let source = tmp.path().join("report.txt"); + std::fs::write(&source, b"live").unwrap(); + let ServeResult::Entry { entry } = + apply_at(&home, input_reference(&source, "r1", "agent", 500), 100).unwrap() + else { panic!("a future remote clock does not refuse a live source") }; + assert_eq!(entry.registered_at_ms, 100); + let ServeResult::Entry { entry: replay } = + apply_at(&home, input_reference(&source, "r1", "agent", 500), 600).unwrap() + else { panic!("expected unchanged replay") }; + assert_eq!(replay, entry); + } + + #[test] + fn input_reference_http_reads_observe_edits_and_deletion() { + use http_body_util::BodyExt; + let tmp = tempfile::tempdir().unwrap(); + let home = tmp.path().join("home"); + let source = tmp.path().join("report.txt"); + std::fs::write(&source, b"before").unwrap(); + let now = crate::brain::now_ms(); + let ServeResult::Entry { entry } = + apply_at(&home, input_reference(&source, "r1", "agent", now), now).unwrap() + else { panic!("expected reference") }; + let uri = format!("/local/f/{}", entry.served_name); + let get = || crate::webserve::handle_path(&home, &home.join("docs"), "LOCAL", 5474, &uri, None, false); + let runtime = tokio::runtime::Builder::new_current_thread().build().unwrap(); + assert_eq!(runtime.block_on(get().into_body().collect()).unwrap().to_bytes().as_ref(), b"before"); + std::fs::write(&source, b"edited").unwrap(); + assert_eq!(runtime.block_on(get().into_body().collect()).unwrap().to_bytes().as_ref(), b"edited"); + std::fs::remove_file(&source).unwrap(); + assert_eq!(get().status(), hyper::StatusCode::NOT_FOUND); + assert!(matches!( + apply_at(&home, input_reference(&source, "r1", "agent", now), now).unwrap(), + ServeResult::Missing, + )); + } + /// LanStatus is a QUESTION: it writes nothing, and it does not re-run the gate. /// /// Both halves are load-bearing, and both are asserted against a home the @@ -685,4 +852,33 @@ mod tests { assert_eq!(entry_url("node", 5474, &facet_only), "http://localhost:5474/node/a/reports/"); assert_eq!(alias_url("node", 5474, &facet_only), None); } + + // [unit->REQ-INPUT-PROVENANCE-ACCEPTANCE-REPORT] + #[test] + fn input_reference_tilde_resolves_on_the_owners_machine() { + let _env = crate::test_home::env_lock(); + let tmp = tempfile::tempdir().unwrap(); + let user_home = tmp.path().join("human"); + std::fs::create_dir_all(&user_home).unwrap(); + let source = user_home.join("report.txt"); + std::fs::write(&source, "owner's live file").unwrap(); + let key = if cfg!(windows) { "USERPROFILE" } else { "HOME" }; + let previous = std::env::var_os(key); + std::env::set_var(key, &user_home); + let result = std::panic::catch_unwind(|| apply_at( + &tmp.path().join("spt"), + input_reference(Path::new("~/report.txt"), "r1", "receiver", 100), + 100, + )); + match previous { + Some(value) => std::env::set_var(key, value), + None => std::env::remove_var(key), + } + let ServeResult::Entry { entry } = result.unwrap().unwrap() else { + panic!("an existing owner-home file must register"); + }; + assert_eq!(std::fs::read_to_string(&entry.path).unwrap(), "owner's live file"); + assert_eq!(entry.audience.as_deref(), Some("receiver")); + assert_eq!(entry.ttl_ms, Some(spt_store::serving::HELPER_ENTRY_TTL_MS)); + } } diff --git a/crates/spt-net/src/net/attach.rs b/crates/spt-net/src/net/attach.rs index 9f75464a..a09c02cc 100644 --- a/crates/spt-net/src/net/attach.rs +++ b/crates/spt-net/src/net/attach.rs @@ -29,6 +29,39 @@ use serde::{Deserialize, Serialize}; use crate::net::ndjson::{self, NdjsonDecoder}; +/// Core's receipt-bound request on the already authenticated controller stream. +/// Audience is deliberately absent: the owner derives it from its rc target. +// [impl->REQ-INPUT-PROVENANCE-ACCEPTANCE-REPORT] +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct InputPathRequest { + pub receipt_id: String, + pub session: String, + pub paths: Vec, + pub received_at_ms: u64, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct InputPathResult { + pub path: String, + pub outcome: InputPathOutcome, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(tag = "outcome", rename_all = "snake_case")] +pub enum InputPathOutcome { + Registered { url: String, expires_at_ms: u64 }, + Missing, + Declined { reason: String }, + /// The owner may have registered, but could not prove the final result. + Unanswered { reason: String }, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct InputPathReply { + pub receipt_id: String, + pub results: Vec, +} + /// What an operator's attach is *for* (REQ-RCVIEW-1 / REQ-KICK-1). Three-valued, /// not a binary role — a plain `Control` request to an already-controlled endpoint /// is REFUSED (never an auto-viewer, never a silent displace); explicit `Take` is @@ -139,6 +172,10 @@ pub enum AttachRecord { /// exactly-once input path (a replayed record cannot double-type). Honored /// only from the CONTROLLER's stream (a viewer sends none). Input { data_b64: String, op_id: u64 }, + /// Target → owner controller: resolve paths from one USER_INPUT receipt. + InputPaths { request: InputPathRequest }, + /// Owner controller → target, only on the originating attach stream. + InputPathsReply { reply: InputPathReply }, /// Operator(controller) → target: resize the PTY to the controller's viewport. /// Controller-EXCLUSIVE — the broker rejects a `Resize` arriving on a viewer's /// stream (REQ-RCVIEW-1, resize is controller-exclusive for ConPTY repaint cost). @@ -277,6 +314,34 @@ mod tests { #[test] fn records_round_trip_and_unknown_kind_is_skipped() { let records = vec![ + AttachRecord::InputPaths { + request: InputPathRequest { + receipt_id: "receipt".into(), + session: "session".into(), + paths: vec!["/file".into()], + received_at_ms: 10, + }, + }, + AttachRecord::InputPathsReply { + reply: InputPathReply { + receipt_id: "receipt".into(), + results: vec![ + InputPathResult { + path: "/file".into(), + outcome: InputPathOutcome::Registered { url: "http://localhost/node/f/file".into(), expires_at_ms: 20 }, + }, + InputPathResult { path: "/missing".into(), outcome: InputPathOutcome::Missing }, + InputPathResult { + path: "/denied".into(), + outcome: InputPathOutcome::Declined { reason: "INPUT_PATH_VIEWER".into() }, + }, + InputPathResult { + path: "/uncertain".into(), + outcome: InputPathOutcome::Unanswered { reason: "owner RPC timed out".into() }, + }, + ], + }, + }, AttachRecord::Request { session_id: 7, from_seq: 0, diff --git a/crates/spt-store/src/helperline.rs b/crates/spt-store/src/helperline.rs index eafae470..26db2962 100644 --- a/crates/spt-store/src/helperline.rs +++ b/crates/spt-store/src/helperline.rs @@ -40,8 +40,8 @@ const KEEP_LINES: usize = 256; /// One registration a peer performed on this endpoint's behalf. #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] pub struct HelperLine { - /// The message whose text quoted the path — the delta key's first half, and - /// the entry's `origin` on the owning node. + /// The message short-ID or USER_INPUT receipt ID — the delta key's first + /// half. Input references use `user-input:` as their origin. pub msg_id: String, /// The path as the USER wrote it, which is what a human recognizes; the /// URL below is what a machine uses. @@ -132,11 +132,9 @@ pub const MAX_QUOTED_PATHS: usize = 5; /// The filepath candidates a message's text quotes, capped and deduplicated. /// -/// PURE, and deliberately so: it decides nothing about whether a path exists, -/// because the two callers stand on different machines. The receiving daemon -/// uses it to decide what to ASK the sender's node about (it cannot stat the -/// sender's disk), and the local arm of the signal adds its own existence check -/// on the node that actually holds the file. +/// PURE, and deliberately so: it decides nothing about whether a path exists. +/// The receiving daemon asks the sender or receipt-bound controller's node; +/// only that owner resolves its home directory and checks its own disk. /// /// ABSOLUTE OR `~`-ROOTED ONLY. A relative path has no anchor, so across nodes /// it would silently name a different file — or nothing. diff --git a/crates/spt-store/src/serving.rs b/crates/spt-store/src/serving.rs index 3fd09b21..52317a20 100644 --- a/crates/spt-store/src/serving.rs +++ b/crates/spt-store/src/serving.rs @@ -242,6 +242,65 @@ impl ServingRegistry { Ok(entry) } + /// Receipt-scoped live reference. A duplicate never changes exposure or its + /// deadline; an unrelated registration is not authority to rescope it. + /// `None` means the source is absent, not a diagnostic. + // [impl->REQ-INPUT-PROVENANCE-ACCEPTANCE-REPORT] + pub fn add_input_reference( + &mut self, + source: &Path, + receipt_id: &str, + audience: &str, + received_at_ms: u64, + now_ms: u64, + ) -> io::Result> { + if !valid_reference_path(source) || receipt_id.is_empty() || audience.trim().is_empty() { + return Err(io::Error::new(io::ErrorKind::InvalidInput, "INPUT_PATH_INVALID")); + } + match fs::metadata(source) { + Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok(None), + Err(error) => return Err(io::Error::new(error.kind(), format!("INPUT_PATH_METADATA: {error}"))), + Ok(metadata) if !metadata.is_file() && !metadata.is_dir() => { + return Err(io::Error::new(io::ErrorKind::InvalidInput, "INPUT_PATH_INVALID")); + } + Ok(_) => {} + } + let origin = format!("user-input:{receipt_id}"); + if let Some(entry) = self.entries.iter().find(|entry| entry.path == source) { + if !entry.origin.as_deref().is_some_and(|value| value.starts_with("user-input:")) + || entry.audience.as_deref() != Some(audience) + || entry.ttl_ms != Some(HELPER_ENTRY_TTL_MS) + || entry.adapter.is_some() + || entry.kind == ServedKind::Attachment + { + return Err(io::Error::new(io::ErrorKind::PermissionDenied, "INPUT_PATH_SCOPE_CONFLICT")); + } + if entry.registered_at_ms.saturating_add(HELPER_ENTRY_TTL_MS) > now_ms { + return Ok(Some(entry.clone())); + } + // A later submission may replace its same-audience expired input + // reference. Replaying the original receipt cannot renew it. + if entry.origin.as_deref() == Some(origin.as_str()) { + return Err(io::Error::new(io::ErrorKind::PermissionDenied, "INPUT_PATH_EXPIRED")); + } + let id = entry.id.clone(); + self.remove(&id); + } + let registered_at_ms = received_at_ms.min(now_ms); + if registered_at_ms.saturating_add(HELPER_ENTRY_TTL_MS) <= now_ms + { + return Err(io::Error::new(io::ErrorKind::PermissionDenied, "INPUT_PATH_EXPIRED")); + } + let entry = match self.add_reference(source, None, Some(&origin), registered_at_ms) { + Ok(entry) => entry, + Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok(None), + Err(error) => return Err(io::Error::new(error.kind(), format!("INPUT_PATH_REGISTRATION: {error}"))), + }; + let scoped = self.scope_entry(&entry.id, Some(HELPER_ENTRY_TTL_MS), Some(audience), None) + .ok_or_else(|| io::Error::other("INPUT_PATH_REGISTRATION_LOST"))?; + Ok(Some(scoped)) + } + /// Activate the core-owned adapter web directory. `root` is the directory /// returned by `perch::adapter_web_dir_in`, created by core before this call. /// A manifest alias is one assignment, not another entry or directory. diff --git a/crates/spt/src/api/delivery.rs b/crates/spt/src/api/delivery.rs index ec3f547b..71d90f75 100644 --- a/crates/spt/src/api/delivery.rs +++ b/crates/spt/src/api/delivery.rs @@ -112,7 +112,7 @@ pub fn validate_mid(state: &str, mid: bool, has_payload: bool) -> Result<(), i32 /// on it. A future edit that wanted edge-conditional emission would have to /// change this signature, which is a visible act rather than a silent drift. // [impl->REQ-IO-INGEST-STATE-PAYLOAD] -fn state_io_kind(state: &str, has_payload: bool, mid: bool) -> Option<&'static str> { +pub(super) fn state_io_kind(state: &str, has_payload: bool, mid: bool) -> Option<&'static str> { if !has_payload { // No payload, no event — the back-compat arm. return None; diff --git a/crates/spt/src/api/mod.rs b/crates/spt/src/api/mod.rs index 145a99ea..c0d05493 100644 --- a/crates/spt/src/api/mod.rs +++ b/crates/spt/src/api/mod.rs @@ -654,6 +654,9 @@ pub fn run(args: ApiArgs, json: bool) -> i32 { spec_manifest, spec_file, } => { + if !user_input.is_empty() { + nowsignal::report_user_input(&id, &session, &user_input); + } let spec = nowsignal::resolve_spec(&ctx, spec_manifest, spec_file.as_deref()); nowsignal::cmd_now_signal(&ctx, &id, &session, &user_input, &agent_output, &spec) } @@ -771,6 +774,17 @@ pub fn run(args: ApiArgs, json: bool) -> i32 { .as_ref() .is_some_and(|m| m.shortform_enabled()); gated(&id, &auth, |id| { + // [impl->REQ-INPUT-PROVENANCE-ACCEPTANCE-REPORT] + if delivery::state_io_kind(&state, payload.is_some(), mid) + == Some(spt_proto::ioevent::IO_KIND_USER_INPUT) + { + // Carry the caller's proof, never borrow a session from disk + // merely because a capability token authorized the mutation. + let proof = auth.proof(); + nowsignal::report_user_input( + id, proof.session_id.as_deref().unwrap_or_default(), payload.as_deref().unwrap_or_default(), + ); + } delivery::cmd_state(id, &state, no_gate, payload.as_deref(), shortform, mid) }) } diff --git a/crates/spt/src/api/nowsignal.rs b/crates/spt/src/api/nowsignal.rs index 8852d6f8..f3bb2b4a 100644 --- a/crates/spt/src/api/nowsignal.rs +++ b/crates/spt/src/api/nowsignal.rs @@ -696,8 +696,8 @@ pub fn gather_last_msgs(input: &PollInput, now: u64, seen: &mut SeenSet) -> Vec< /// (b) a user's message QUOTES A FILEPATH that exists — the path is registered /// for this endpoint alone and the same fetch line is handed over. /// -/// The SAME-NODE case of (b) registers NOTHING and says so: serving a file to a -/// process that can already open it buys an audit entry and no access. +/// USER_INPUT is bound at receipt, not here. This gatherer never looks for the +/// submitter's files on the receiving node or samples a later controller. /// /// Delta discipline is per (message, path): a re-poll in the same session emits /// nothing, which is what keeps a standing signal from becoming a nag. @@ -742,16 +742,6 @@ pub fn gather_file_access_helper(input: &PollInput, seen: &mut SeenSet) -> Vec VecREQ-NOW-SIGNAL-FILE-ACCESS-HELPER] -pub fn quoted_paths(text: &str) -> Vec { - let mut out: Vec = Vec::new(); - for token in spt_store::helperline::quoted_path_candidates(text, MAX_QUOTED_PATHS) { - let candidate = if let Some(rest) = token.strip_prefix("~/") { - match std::env::var_os("HOME").or_else(|| std::env::var_os("USERPROFILE")) { - Some(home) => std::path::PathBuf::from(home).join(rest), - None => continue, - } - } else { - std::path::PathBuf::from(&token) - }; - // A quoted path that is not there emits NOTHING rather than a dead link. - if !candidate.exists() || out.contains(&candidate) { - continue; - } - out.push(candidate); - } - out -} /// EDGE_TRANSITIONS — endpoint/node on-offline edges and subnet joins. /// @@ -1120,6 +1079,45 @@ pub fn resolve_spec(ctx: &Ctx, spec_manifest: bool, spec_file: Option<&Path>) -> NowSpec::default() } +/// Submit the hook's exact payload to the live broker before any slow gathering. +/// No daemon start, no owner round trip, and no input-processing failure on a miss. +// [impl->REQ-INPUT-PROVENANCE-ACCEPTANCE-REPORT] +pub(super) fn report_user_input(id: &str, session: &str, payload: &str) { + use spt_daemon::brain::{Brain, PumpTrace}; + use spt_daemon::msg::{UserInputReport, USER_INPUT_REPORT_BOUND}; + let report = UserInputReport { + endpoint: id.to_owned(), + session: session.to_owned(), + payload: payload.to_owned(), + expires_at_ms: now_ms().saturating_add(USER_INPUT_REPORT_BOUND.as_millis() as u64), + }; + let (tx, rx) = std::sync::mpsc::sync_channel(1); + // This CLI's one receipt worker also bounds connect, hello, and writes: + // pump read deadlines alone do not bound a stalled initial handshake. + let worker = std::thread::Builder::new().name("input-receipt".into()).spawn(move || { + let result = Brain::cold_start_pump( + &spt_daemon::endpoint::broker_socket_name(), + now_ms(), + USER_INPUT_REPORT_BOUND, + PumpTrace::Silent, + ).and_then(|mut brain| brain.report_user_input(report)); + let _ = tx.send(result); + }); + let result = worker.and_then(|_| { + rx.recv_timeout(USER_INPUT_REPORT_BOUND) + .map_err(|e| std::io::Error::new(std::io::ErrorKind::TimedOut, e))? + }); + let declined = match result { + Ok(reply) => reply.declined, + Err(error) => Some(format!("receipt broker unanswered: {error}")), + }; + if let Some(reason) = declined { + spt_proto::emit_line_err!( + "HELPER_SERVE_FOR: target={id} outcome=declined reason={reason}" + ); + } +} + /// Run one poll and print what it found. /// /// **Best-effort end to end, and structurally unable to fail its caller**: this @@ -1536,129 +1534,21 @@ mod tests { clear_session(&session); } - /// FILE_ACCESS_HELPER (b) guard 1 of 4 — the path is HONORED when it exists - /// and is anchored, punctuation around it trimmed to a fixed point. - /// - /// These four guards were one cell. Split because a bundled cell reports - /// only its FIRST failing assertion: break the dedup guard and the cap guard - /// together and the run names one of them, so the second is repaired blind - /// or not at all. One guard per cell means the count of reds is the count of - /// broken guards. - // [unit->REQ-NOW-SIGNAL-FILE-ACCESS-HELPER] - #[test] - fn a_quoted_path_that_exists_and_is_anchored_is_honored() { - let dir = tempfile::tempdir().unwrap(); - let real = dir.path().join("notes.md"); - std::fs::write(&real, b"x").unwrap(); - let real_text = real.display().to_string(); - - assert_eq!( - quoted_paths(&format!("look at {real_text} when you can")), - vec![real.clone()], - "an existing absolute path is honored" - ); - assert_eq!( - quoted_paths(&format!("look at \"{real_text}\".")), - vec![real], - "sentence punctuation around the path is trimmed" - ); - } - - /// Guard 2 of 4 — a path that is NOT THERE emits nothing. - /// - /// The alternative is minting a link that 404s on the far side, which costs - /// the receiver a round trip to learn what this node already knew. - // [unit->REQ-NOW-SIGNAL-FILE-ACCESS-HELPER] - #[test] - fn a_quoted_path_that_does_not_exist_is_not_honored() { - let dir = tempfile::tempdir().unwrap(); - assert!( - quoted_paths(&format!("{}", dir.path().join("missing.md").display())).is_empty(), - "a path that is not there emits nothing rather than a dead link" - ); - } - - /// Guard 3 of 4 — a RELATIVE path is not honored. - /// - /// It has no anchor on the other node: `notes.md` names whatever the - /// receiver's working directory happens to hold, which is either nothing or, - /// worse, a different file with the same name. - // [unit->REQ-NOW-SIGNAL-FILE-ACCESS-HELPER] - #[test] - fn a_relative_quoted_path_is_not_honored() { - assert!( - quoted_paths("look at notes.md and ./src/main.rs").is_empty(), - "a relative path has no anchor on the other node" - ); - } - /// Guard 4 of 4 — the same path twice is one datum, and at most five are - /// honored per message. - // [unit->REQ-NOW-SIGNAL-FILE-ACCESS-HELPER] + // [unit->REQ-INPUT-PROVENANCE-ACCEPTANCE-REPORT] #[test] - fn quoted_paths_are_deduplicated_and_capped() { - let dir = tempfile::tempdir().unwrap(); - let real = dir.path().join("notes.md"); - std::fs::write(&real, b"x").unwrap(); - let real_text = real.display().to_string(); - assert_eq!( - quoted_paths(&format!("{real_text} {real_text}")), - vec![real], - "the same path twice in one message is one datum" - ); - - let mut many = String::new(); - for index in 0..7 { - let path = dir.path().join(format!("f{index}.md")); - std::fs::write(&path, b"x").unwrap(); - many.push_str(&format!("{} ", path.display())); - } - // The LITERAL, not `MAX_QUOTED_PATHS`: comparing the product to its own - // constant means changing the cap from 5 to 3 keeps this cell green, and - // the cap is the claim. (`helperline.rs` already pins its 5 this way.) - assert_eq!(quoted_paths(&many).len(), 5, "at most five per message"); - } - - /// FILE_ACCESS_HELPER (b), SAME NODE: nothing is registered and the signal - /// says the file is readable where it already is. Serving a file to a - /// process that can open it buys an audit entry and no access. - // [unit->REQ-NOW-SIGNAL-FILE-ACCESS-HELPER] - #[test] - fn a_locally_quoted_path_registers_nothing_and_says_so() { + fn a_report_never_resolves_its_path_on_the_receiving_node() { let _home = crate::testutil::isolated_home(); - let session = session_id("fah-local"); - clear_session(&session); - let id = "fah-local-agent"; + let session = session_id("fah-receipt"); let dir = tempfile::tempdir().unwrap(); - let quoted = dir.path().join("plan.md"); - std::fs::write("ed, b"x").unwrap(); - let words = format!("have a look at {}", quoted.display()); - let input = - PollInput { id, session: &session, user_input: &words, agent_output: "" }; - - let mut seen = SeenSet::load(&session, Category::FileAccessHelper); - let lines = gather_file_access_helper(&input, &mut seen); - seen.flush(); - assert_eq!(lines.len(), 1, "{lines:?}"); - assert!(lines[0].contains("readable directly"), "{lines:?}"); - assert!(!lines[0].starts_with("spt fetch"), "nothing to fetch on one machine"); - - let registry = spt_store::serving::ServingRegistry::load_at( - &spt_store::perch::serving_registry_file(), - ) - .unwrap(); - assert_eq!( - registry.entries().count(), - 0, - "the same-node case exposes nothing at all" - ); - + let local = dir.path().join("exists-only-here"); + std::fs::write(&local, b"not the submitter's file").unwrap(); + let words = local.display().to_string(); + let input = PollInput { + id: "fah-receipt", session: &session, user_input: &words, agent_output: "", + }; let mut seen = SeenSet::load(&session, Category::FileAccessHelper); - assert!( - gather_file_access_helper(&input, &mut seen).is_empty(), - "once per (message, path)" - ); - seen.flush(); + assert!(gather_file_access_helper(&input, &mut seen).is_empty()); clear_session(&session); } diff --git a/crates/spt/src/rc.rs b/crates/spt/src/rc.rs index 20d7227f..2dafd677 100644 --- a/crates/spt/src/rc.rs +++ b/crates/spt/src/rc.rs @@ -32,7 +32,10 @@ use spt_daemon::attach::{ use spt_daemon::effect::{Minter, MintedOp}; use spt_daemon::brain::{now_ms, Brain, BrokerEvent}; use spt_daemon::msg::{decode_bytes, SEAL_CEREMONY_ADMITTED, SEAL_CEREMONY_REFUSED}; -use spt_net::net::attach::{AttachDecoder, AttachIntent, AttachRecord}; +use spt_net::net::attach::{ + AttachDecoder, AttachIntent, AttachRecord, InputPathOutcome, InputPathReply, + InputPathRequest, InputPathResult, +}; /// The detach prefix: ctrl-b (0x02), matching the legacy capsule's prefix so the /// muscle memory carries over. `ctrl-b d` detaches; `ctrl-b ctrl-b` sends one @@ -2602,6 +2605,8 @@ fn attach_viewport( &mut decoder, out, !view, // controller drives resize + endpoint_id, + est.remote_node.as_deref(), mouse_mode, status.as_deref_mut(), ); @@ -2872,6 +2877,116 @@ fn classify_read_err(kind: std::io::ErrorKind) -> ReadDisposition { } } +const INPUT_PATH_QUEUE: usize = 8; + +// [impl->REQ-INPUT-PROVENANCE-ACCEPTANCE-REPORT] +fn input_path_audience( + controller: bool, + remote_node: Option<&str>, + endpoint: &str, +) -> Result { + if !controller { + return Err("INPUT_PATH_VIEWER"); + } + // Established::remote_node is set only by the non-local owner-dial arm. + // Both plain local targets and qualified own-node targets use None. + if remote_node.is_none() { + return Err("INPUT_PATH_LOCAL_CONTROLLER"); + } + Ok(canonical_wire_id(endpoint)) +} + +fn decline_input_paths(request: InputPathRequest, reason: &str) -> InputPathReply { + InputPathReply { + receipt_id: request.receipt_id, + results: request.paths.into_iter() + .take(spt_store::helperline::MAX_QUOTED_PATHS) + .map(|path| InputPathResult { + path, + outcome: InputPathOutcome::Declined { reason: reason.to_owned() }, + }) + .collect(), + } +} + +struct InputPathWork { + request: InputPathRequest, + audience: String, + replies: mpsc::SyncSender, + deadline: Instant, +} + +/// One bounded worker for this rc process, not one thread per path or reconnect. +/// Each viewport owns its reply receiver; dropping it severs the return authority +/// permanently, so an old result can never enter a replacement stream. +// [impl->REQ-INPUT-PROVENANCE-ACCEPTANCE-REPORT] +fn input_path_worker() -> &'static mpsc::SyncSender { + static WORKER: std::sync::LazyLock> = std::sync::LazyLock::new(|| { + let (tx, rx) = mpsc::sync_channel::(INPUT_PATH_QUEUE); + std::thread::spawn(move || { + while let Ok(work) = rx.recv() { + let (reply, timed_out) = register_input_paths(work.request, &work.audience, work.deadline); + let _ = work.replies.try_send(reply); + if timed_out { + // The existing RPC bounds waiting with a helper thread. Do + // not start further RPCs after one was abandoned: at most ONE + // blocked transport thread can survive in this process. + for queued in rx.try_iter() { + let _ = queued.replies.try_send(decline_input_paths( + queued.request, "INPUT_PATH_OWNER_UNANSWERED", + )); + } + break; + } + } + }); + tx + }); + &WORKER +} + +// [impl->REQ-INPUT-PROVENANCE-ACCEPTANCE-REPORT] +fn register_input_paths(request: InputPathRequest, audience: &str, deadline: Instant) -> (InputPathReply, bool) { + use spt_daemon::servehost::{call_with_timeout, ServeRequest, ServeResult}; + let mut timed_out = false; + let mut results = Vec::with_capacity(request.paths.len()); + let socket = spt_daemon::endpoint::seed_socket_name(); + for path in request.paths { + let remaining = deadline.saturating_duration_since(Instant::now()); + let outcome = if timed_out || remaining.is_zero() { + InputPathOutcome::Declined { reason: "INPUT_PATH_OWNER_UNANSWERED".into() } + } else { + match call_with_timeout(&socket, ServeRequest::AddInputReference { + path: std::path::PathBuf::from(&path), + receipt_id: request.receipt_id.clone(), + audience: audience.to_owned(), + received_at_ms: request.received_at_ms, + }, remaining.min(Duration::from_secs(1))) { + Ok(ServeResult::Missing) => InputPathOutcome::Missing, + Ok(ServeResult::Entry { entry }) => match crate::serveverb::node_and_port() { + Ok((node, port)) => InputPathOutcome::Registered { + url: spt_store::serving::entry_url(&node, port, &entry), + expires_at_ms: entry.registered_at_ms.saturating_add( + spt_store::serving::HELPER_ENTRY_TTL_MS, + ), + }, + Err(reason) => InputPathOutcome::Unanswered { reason }, + }, + Ok(ServeResult::Refused { error }) => InputPathOutcome::Declined { reason: error }, + Ok(_) => InputPathOutcome::Declined { reason: "INPUT_PATH_UNEXPECTED_REPLY".into() }, + Err(error) => { + timed_out = error.kind() == std::io::ErrorKind::TimedOut; + InputPathOutcome::Unanswered { + reason: format!("INPUT_PATH_OWNER_UNANSWERED: {error}"), + } + } + } + }; + results.push(InputPathResult { path, outcome }); + } + (InputPathReply { receipt_id: request.receipt_id, results }, timed_out) +} + /// The full-duplex pump: drain stdin → `send_attach_input`; render inbound /// `AttachRecord::Output` → stdout; end on `Exit`, EOF, or detach. #[allow(clippy::too_many_arguments)] @@ -2883,9 +2998,13 @@ fn pump( decoder: &mut AttachDecoder, stdout: &mut impl std::io::Write, controller: bool, + endpoint_id: &str, + remote_node: Option<&str>, mouse_mode: &MouseMode, mut status: Option<&mut StatusRow>, ) -> Result { + let input_audience = input_path_audience(controller, remote_node, endpoint_id); + let (input_replies_tx, input_replies_rx) = mpsc::sync_channel(INPUT_PATH_QUEUE); // Track the harness's mouse-reporting mode from its output DECSET sequences so // the stdin reader knows whether to forward scroll (REQ-RC-MOUSE-FORWARD). The // scanner is pump-local (single producer) and survives a sequence split across @@ -2941,6 +3060,11 @@ fn pump( // this before the serve would ever finish the stream. (#4, REQ-RC-CROSS-NODE-ATTACH) let mut rendered_any = false; loop { + while let Ok(reply) = input_replies_rx.try_recv() { + let line = AttachRecord::InputPathsReply { reply }.encode_line(); + brain.net_stream_send(stream_id, &line, None, false) + .map_err(|error| format!("send input paths reply: {error}"))?; + } // ── controller window-change → resize ────────────────────────────── if controller { if let Ok((cols, rows)) = crossterm::terminal::size() { @@ -3117,6 +3241,37 @@ fn pump( BrokerEvent::NetStreamData { stream_id: sid, bytes, .. } if sid == stream_id => { for rec in decoder.push(&bytes) { match rec { + AttachRecord::InputPaths { request } => { + let refusal = if request.paths.len() > spt_store::helperline::MAX_QUOTED_PATHS { + Some("INPUT_PATH_LIMIT") + } else { + input_audience.as_ref().err().copied() + }; + let declined = if let Some(reason) = refusal { + Some(decline_input_paths(request, reason)) + } else { + let work = InputPathWork { + request, + audience: input_audience.as_ref().expect("authority checked").clone(), + replies: input_replies_tx.clone(), + deadline: Instant::now() + Duration::from_secs(5), + }; + match input_path_worker().try_send(work) { + Ok(()) => None, + Err(mpsc::TrySendError::Full(work)) => Some(decline_input_paths( + work.request, "INPUT_PATH_OWNER_BUSY", + )), + Err(mpsc::TrySendError::Disconnected(work)) => Some(decline_input_paths( + work.request, "INPUT_PATH_OWNER_UNANSWERED", + )), + } + }; + if let Some(reply) = declined { + brain.net_stream_send( + stream_id, &AttachRecord::InputPathsReply { reply }.encode_line(), None, false, + ).map_err(|error| format!("send input paths decline: {error}"))?; + } + } AttachRecord::Output { seq, data_b64 } => { if seq < cursor { continue; // already rendered — re-serve dedup @@ -3346,6 +3501,31 @@ fn pump( mod tests { use super::*; + #[test] + fn input_paths_require_remote_control_and_use_the_local_target() { + assert_eq!(input_path_audience(false, Some("peer"), "agent"), Err("INPUT_PATH_VIEWER")); + assert_eq!(input_path_audience(true, None, "agent"), Err("INPUT_PATH_LOCAL_CONTROLLER")); + assert_eq!( + input_path_audience(true, Some("peer"), "subnet:agent@peer"), + Ok("agent".to_owned()), + ); + } + + #[test] + fn declined_input_paths_remain_receipt_bound_and_bounded() { + let reply = decline_input_paths(InputPathRequest { + receipt_id: "receipt".into(), + session: "session".into(), + paths: (0..20).map(|n| format!("/file-{n}")).collect(), + received_at_ms: 1, + }, "INPUT_PATH_VIEWER"); + assert_eq!(reply.receipt_id, "receipt"); + assert_eq!(reply.results.len(), 5); + assert!(reply.results.iter().all(|result| matches!( + &result.outcome, InputPathOutcome::Declined { reason } if reason == "INPUT_PATH_VIEWER" + ))); + } + // [unit->REQ-SEAL-CEREMONY-FIDO2] [unit->REQ-SEAL-FIDO2-RC-CLIENT] the // client-side offer decision: the offer STANDS only when both offer // fields shipped AND the enrolled node is THIS node — a node mismatch diff --git a/crates/spt/src/serveverb.rs b/crates/spt/src/serveverb.rs index 46265597..bc68469b 100644 --- a/crates/spt/src/serveverb.rs +++ b/crates/spt/src/serveverb.rs @@ -175,6 +175,7 @@ pub(crate) fn run(command: ServeCmd, json: bool) -> i32 { } }; match result { + ServeResult::Missing => 0, ServeResult::Refused { error } => { eprintln!("{error}"); 1 diff --git a/docs-site/src/changelog.md b/docs-site/src/changelog.md index 854d67da..2b8e2ecb 100644 --- a/docs-site/src/changelog.md +++ b/docs-site/src/changelog.md @@ -12,6 +12,16 @@ breaks something, or changes the observable behavior of existing surfaces broadly; **patch** for fixes, and for additive opt-in capability — a new key, flag, or page that no existing user can encounter without opting into it. +## [Unreleased] + +### Fixed + +- Quoting a filepath while controlling an agent through remote `spt rc` can now + offer that agent a fetch command for the live file on your machine. The reference + lasts up to 24 hours and is addressed only to that agent. Repeated reports do not + extend it or repeat the notice. Missing files stay silent; unavailable provenance + and other refusals are named. No harness-adapter changes are required. + ## [0.69.0] - 2026-09-11 ### Fixed diff --git a/docs-site/src/serving/attachments.md b/docs-site/src/serving/attachments.md index f241dfaa..eb9063d4 100644 --- a/docs-site/src/serving/attachments.md +++ b/docs-site/src/serving/attachments.md @@ -107,19 +107,32 @@ spt fetch http://localhost:5474/kitsubito/f/report.md ``` + The same category answers the other way a file arrives: **a user quoting a -filepath**. When the user and the agent are on one node, nothing is registered -and the signal says the file is readable where it already is — serving a file to -a process that can already open it buys an audit entry and no access. - -Guards, each of which produces silence rather than a bad line: - -- The path must **exist** at signal time; a quoted path that is not there emits - nothing rather than a dead link. -- **Absolute or `~`-rooted only.** A relative path has no anchor and would name - a different file on another node. -- At most **five** per message. -- Once per (message, path): a re-poll in the same session emits nothing. +filepath while controlling the receiving agent through remote `spt rc`**. +Core binds the existing session-authenticated input report to the broker's live +remote controller when the report arrives. A later controller change cannot +reattribute it. Registration happens on that controller's machine, not by looking +for a similarly named file on the receiving machine. + +The file is a **live reference**, not an attachment snapshot: edits are visible +and deletion returns not found. The reference lasts up to **24 hours**, addressed +only to the receiving endpoint. The fetch command appears on a subsequent signal +poll once the owner answers; input processing never waits for that answer. + +- **Absolute or `~`-rooted only.** The owner resolves `~` against its own home. + A relative path has no anchor on another node. At most **five** paths per report; + directories receive the same lifetime and audience. +- A **missing file is silent**. Other non-grants name their reason, including + local/viewer/no controller seat, an invalid session, uncertain delivery-byte + evidence, or an owner that did not answer. +- A repeated payload in the same harness session does not register again, repeat + the helper notice, or extend an existing reference's deadline. +- Core excludes bytes it knows it physically wrote for peer delivery, rather than + guessing from message syntax. This is **receipt-seat attribution**, not proof + of who typed every byte: external automation and native harness re-submission + cannot be distinguished beyond the current controller seat. No adapter token + or integration change is required. Every entry the helper registers is enumerable in `spt serve list` with its origin, so an automatic exposure is exactly as visible as a deliberate one. diff --git a/docs-site/src/serving/overview.md b/docs-site/src/serving/overview.md index 9a2c250c..0a6c76ff 100644 --- a/docs-site/src/serving/overview.md +++ b/docs-site/src/serving/overview.md @@ -181,7 +181,7 @@ narrowing that did not render would make that answer a half-truth. |---|---| | `ttl` | A lifetime from registration. Absent means the entry lives until removed. Attachments carry 30 days by default; entries the `FILE_ACCESS_HELPER` registers carry 24 hours. | | `audience` | The one endpoint allowed to fetch it. Absent means anyone the `WEB` surface admits. | -| `origin` | The endpoint that registered it — for a helper-registered entry, the message short-ID it came from. | +| `origin` | Who or what registered it — for a message helper, the message short-ID; for an input-report helper, `user-input:`. | An **audience** is a cross-node narrowing. The owner serves the entry only to the named endpoint and answers everyone else `403` naming the surface — diff --git a/docs/INPUT-PROVENANCE-CONTRACT.md b/docs/INPUT-PROVENANCE-CONTRACT.md index 116d37fd..b7bde5c7 100644 --- a/docs/INPUT-PROVENANCE-CONTRACT.md +++ b/docs/INPUT-PROVENANCE-CONTRACT.md @@ -1,9 +1,9 @@ -# Submission-bound input provenance — the public harness contract +# Receipt-bound input provenance — core-only contract -**Status: CONTRACT DRAFT for adapter-owner assessment. Nothing here is implemented.** No `REQ-*` is -minted or activated by this document, no core behaviour changes on its account, and it grants nothing. -Authorized by doyle 2026-09-12 for #300 direction C (confirmed by the operator, relayed -operator → lia → doyle); the direction record is #300 comment 5645159945. +**Current contract: §8, core-only.** Sections 0–7 retain the superseded +adapter-design discussion and measurements; they impose no adapter obligations. +The implementation uses the existing USER_INPUT reports and controller rc stream. +The two-node field acceptance remains separate from the doc/impl/unit lane. ## The operator decision this contract is bound to @@ -398,6 +398,8 @@ Scope carried with the answer: no experiment, no installed-runtime measurement, ## 8. RULING (doyle, 2026-09-13; RE-CUT the same day on the operator's direction) — core-only; no token, nothing asked of adapters + + **Operator, 2026-09-13, direct:** "#300 needs to be developed entirely independently of harness adapters. spt-core does not concern itself with harness adapter integration. omp-spt and/or claude-spt should not have anything to do with #300. they already submit USER_INPUT kind event hooks, and that's all spt-core @@ -432,5 +434,45 @@ exact payload, at whatever moment the harness reports it. 6. **§4.1 token issuance and §4.5 grace period: STRUCK.** §7's adapter answers stay on record as measurements of the ceiling in (5), not as obligations. -**Build:** `REQ-INPUT-PROVENANCE-ACCEPTANCE-REPORT` (traceable-reqs.toml, minted unactivated). Product -lane: todlando. Adapter side: NONE. +**Build:** `REQ-INPUT-PROVENANCE-ACCEPTANCE-REPORT` activates doc/impl/unit in +the product lane. Adapter side: NONE. The integration stage awaits the separately +admitted two-node field leg. + +### Implemented boundaries + +- Both `api state` USER_INPUT and `api now-signal --user-input` report before + event fanout, shortform processing, or roster gathering. The broker validates + the harness session, checks its physical delivery-byte evidence, and snapshots + the live remote controller connection. A later Take never changes that origin. + `api state` carries the caller's resolved session proof. A capability token + can still authorize the state event, but cannot borrow a session ID from disk + to mint input provenance. +- File requests and replies ride that same authenticated rc stream. The owner + derives the audience from its own rc target, never from request-supplied + audience text. The existing MSG_OUT-backed WEB ServeFor guard is unchanged. +- The receipt helper waits at most 500 ms for acknowledgement, including a stalled + connect/hello/write, not for serving. A same-node report deadline prevents a + worker that outlives that wait from binding to a later seat. This is not a bound + on the entire hook (its other work has separate costs). The broker allows + 10 seconds for the controller reply, then names the unanswered outcome. Owner + work runs off the rc display/input pump. An unanswered request or uncertain + owner RPC result retains its dedup fence for the possible reference lifetime: + a lost reply is not proof that the owner exposed nothing. +- Dedup is per harness session and exact payload hash in the broker-held PTY + session; it survives brain replacement. Live records are never evicted to make + room: 4096 retained receipts or 64 unretired reply deadlines refuse new grants + by name. Answered requests still count until their deadline worker exits. + The owner reuses a live same-audience input reference without extending it or + changing unrelated exposure. Its timestamp is the earlier of receipt time and + owner time, so clock skew cannot lengthen either side's 24-hour bound. +- Physical delivery writes retain exact text hashes, not payload shapes. The + source-tagged controller/typeahead and choreography keys are not delivery text. + In-progress or partially failed physical delivery writes make provenance + unavailable by name rather than guessing; no receipt blocks on a PTY writer. + Session-lifetime evidence is never silently evicted (65536 distinct hashes is + the named capacity limit). An identical later human payload is indistinguishable + from these known bytes and is excluded too; external automation and native + harness re-submission remain subject to the receipt-seat ceiling above. +- Missing files produce neither helper notices nor declined diagnostics. + Wrong session, local/viewer/no seat, unknown delivery evidence, queue pressure, + owner refusal, reply timeout, and persistence failure each name their condition. diff --git a/docs/adr/0058-attachments-are-pull-model.md b/docs/adr/0058-attachments-are-pull-model.md index 5e9c72ca..fc1b1fe6 100644 --- a/docs/adr/0058-attachments-are-pull-model.md +++ b/docs/adr/0058-attachments-are-pull-model.md @@ -78,3 +78,26 @@ the user's live file, edits visible, 404 once deleted — not a snapshot. The us this", not "keep this as it was", and a live reference costs no copy. It is therefore a *file* (or *dir*) entry carrying a TTL and an audience, not an *attachment* entry; the TTL/audience fields are per-entry and kind-independent. + +## Amendment 2 — receipt-seat authority for USER_INPUT (operator re-cut, 2026-09-13; releases#300) + + +The core-only ruling in [INPUT-PROVENANCE-CONTRACT §8](../INPUT-PROVENANCE-CONTRACT.md#8-ruling-doyle-2026-09-13-re-cut-the-same-day-on-the-operators-direction--core-only-no-token-nothing-asked-of-adapters) +supersedes Amendment 1's local-readable notice and its proposed new WEB request +authority for USER_INPUT. Existing MSG_OUT-backed ServeFor authorization remains +unchanged. + +At receipt of an existing session-authenticated input report, the broker snapshots +its live remote controller connection. Requests and replies ride that same rc +stream; the owner derives the audience from its locally established target. Local, +viewer-only, and absent seats decline by name, without inspecting receiver-side +paths. The reference origin is `user-input:`. The live-reference kind, +one-endpoint audience, bounded 24-hour lifetime, and receiver-side helper record +remain the existing mechanisms. + +Identical payload/session reports cannot renew a live reference or publish another +notice. Lost or uncertain replies retain a dedup fence: absence of an answer is not +proof that registration did not happen. Missing files alone are silent; other +non-grants are named. The hook receipt and owner-reply waits are bounded separately. +This is receipt-seat attribution, not reconstructed keystroke provenance; adapters +receive no new integration obligation. diff --git a/traceable-reqs.toml b/traceable-reqs.toml index 46300fda..9d01b5d1 100644 --- a/traceable-reqs.toml +++ b/traceable-reqs.toml @@ -7638,4 +7638,4 @@ required_stages = [] # NOT ACTIVATED -- doyle 2026-09-10, registry-first per th [[requirements]] id = "REQ-INPUT-PROVENANCE-ACCEPTANCE-REPORT" title = "A REMOTE SUBMITTER'S QUOTED ABSOLUTE PATHS ARE SERVED FROM THE SUBMITTER'S MACHINE, BOUND BY CORE ALONE AT RECEIPT OF THE USER_INPUT REPORT ADAPTERS ALREADY SEND -- NO TOKEN, NO DECLARATION, NOTHING ASKED OF ANY ADAPTER (releases#300; operator-ruled 2026-09-13 that #300 is developed entirely independently of harness adapters, USER_INPUT being all core asks of them; doyle re-cut docs/INPUT-PROVENANCE-CONTRACT.md section 8 the same day). Operator experience (relayed, comment 5645159945): the origin is the SENDER'S computer; a path that names no file there is SILENTLY SKIPPED, no prompt, no fallback origin. MECHANISM, core-only: (1) at RECEIPT of every USER_INPUT report core binds origin to the session's current authenticated REMOTE controller seat, the broker-owned driven_by; local seat, viewer, or none binds nothing and the declined line names it. (2) OWN-INJECTION EXCLUSION: a report whose payload is a delivery core itself wrote into that session's PTY binds no origin, recognised by the bytes core wrote and never by payload shape; resumed-session and external automation remain uncharacterised and bind the seat per the submitter rule. (3) core-side dedup: a repeat of the same payload in the same session while its reference is live re-registers nothing, emits no second helper notice, extends no TTL. (4) serving unchanged from the live-reference contract: submitter's machine, 24h TTL, audience the receiving endpoint, helper notice once per submission and path; every non-grant named EXCEPT the missing file, silent by decision. STATED CEILING, not an adapter promise: the report lags acceptance by the harness's own path (0.5-1.0 s measured on claude-spt), and a cross-operator Take inside that lag attributes to the new holder, whose own file is then served to the agent that holder controls. Gate: doc -- section 8 of the contract; impl -- the receipt-time bind, the own-injection exclusion, the dedup, wired to the existing live-reference serve; unit -- remote seat binds that seat, viewer/local/none binds nothing with the declined line naming it, a core-written delivery payload binds nothing, a repeated payload is idempotent with no TTL extension, a missing file on the bound origin emits no line; int -- a real two-node rc session where a remote controller's quoted path is served to the agent and a locally typed one is not." -required_stages = [] # MINTED releases#300 (doyle 2026-09-13), registry-first per the activate-don't-pre-fail rule; RE-CUT the same day to core-only on the operator's direction (no adapter declaration, no submission id). Activate doc/impl/unit at todlando's lane start; int at the field leg. +required_stages = ["doc", "impl", "unit"] # releases#300 core-only lane; int activates at the separately admitted field leg.