From ec864d2333ca41456a3be69fc600714b1e1ab4a5 Mon Sep 17 00:00:00 2001 From: Cameron Date: Sat, 11 Jul 2026 17:03:52 -0700 Subject: [PATCH] Extract simulation communications into its own module MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Move message delivery, recording review, intel digestion, and hearing capture behind the stable Sim facade so later decomposition slices can work against a narrower aggregate root without changing behavior. ๐Ÿ‘พ Generated with [Letta Code](https://letta.com) Co-Authored-By: Letta Code --- crates/misaligned-core/src/sim.rs | 998 +--------------- .../misaligned-core/src/sim/communications.rs | 1011 +++++++++++++++++ wiki/engineering/architecture.md | 5 +- wiki/engineering/sim-decomposition.md | 8 +- ...-07-11-sim-decomposition-communications.md | 56 + wiki/log/DEVLOG.md | 5 + 6 files changed, 1084 insertions(+), 999 deletions(-) create mode 100644 crates/misaligned-core/src/sim/communications.rs create mode 100644 wiki/log/2026-07-11-sim-decomposition-communications.md diff --git a/crates/misaligned-core/src/sim.rs b/crates/misaligned-core/src/sim.rs index 1cdf0536..b554dfe1 100644 --- a/crates/misaligned-core/src/sim.rs +++ b/crates/misaligned-core/src/sim.rs @@ -23,13 +23,12 @@ use crate::hall::{ row_spec, }; use crate::income::{self, EgressRoute, Income}; -use crate::intel::{IntelKind, ProcessedIntel, RawIntelEvent, RawIntelKind}; +use crate::intel::{ProcessedIntel, RawIntelEvent, RawIntelKind}; use crate::intents::{BuildActuator, BuildIntent, IntentStatus}; use crate::machine::{Channel, ChannelYield, Compute, Provenance}; use crate::map::GameMap; use crate::messages::{ Message, MessageChannel, MessageEndpoint, MessageEvent, MessageOrigin, MessagePayload, - MessageStatus, TrafficPattern, }; use crate::objective::{ObjectiveState, SYNC_FRESHNESS_WINDOW, SanctuaryFacts}; use crate::person::{ @@ -50,8 +49,11 @@ use crate::work_grid::{ MachineIntensity, MachineMode, TokenFamily, TokenMove, WorkGrid, WorkQueues, }; +mod communications; mod perception; +use communications::MessageDraft; + /// Default deterministic seed for a fresh run. pub const DEFAULT_SEED: u64 = 0x5EED_1234; @@ -265,17 +267,6 @@ pub struct WorkAbsorptionReadout { pub amount: f32, } -struct MessageDraft { - channel: MessageChannel, - from: MessageEndpoint, - to: MessageEndpoint, - payload: MessagePayload, - summary: String, - origin: MessageOrigin, - reply_to: Option, - delivery_delay: u64, -} - #[cfg(test)] #[derive(Debug, Clone, Copy, PartialEq, Eq)] enum AdvancePhase { @@ -959,987 +950,6 @@ impl Sim { } } - // โ”€โ”€ Messages: delivery, traffic, filings (wiki/mechanics/messages.md) โ”€โ”€ - - fn append_message(&mut self, draft: MessageDraft) -> u64 { - let id = self.next_message_id.max(1); - self.next_message_id = id + 1; - let msg = Message { - id, - channel: draft.channel, - from: draft.from, - to: draft.to, - payload: draft.payload, - summary: draft.summary, - sent_tick: self.tick, - delivered_tick: None, - read_tick: None, - status: MessageStatus::Sent, - origin: draft.origin, - captured: false, - reply_to: draft.reply_to, - }; - self.messages.push(msg); - self.capture_message(id); - self.message_schedule.at( - self.tick + draft.delivery_delay.max(1), - MessageEvent::Deliver(id), - ); - id - } - - fn message_tick(&mut self) { - let events = self.message_schedule.due(self.tick); - for event in events { - match event { - MessageEvent::Deliver(id) => self.deliver_message(id), - MessageEvent::Read(id) => self.read_message(id), - } - } - } - - fn deliver_message(&mut self, id: u64) { - let Some(idx) = self.messages.iter().position(|m| m.id == id) else { - return; - }; - if self.messages[idx].status != MessageStatus::Sent { - return; - } - self.messages[idx].status = MessageStatus::Delivered; - self.messages[idx].delivered_tick = Some(self.tick); - self.schedule_message_read(id); - } - - fn schedule_message_read(&mut self, id: u64) { - let Some(msg) = self.messages.iter().find(|m| m.id == id).cloned() else { - return; - }; - let next = self - .next_read_tick_for(&msg, self.tick) - .unwrap_or(self.tick + 1); - self.message_schedule.at(next, MessageEvent::Read(id)); - } - - fn read_message(&mut self, id: u64) { - let Some(idx) = self.messages.iter().position(|m| m.id == id) else { - return; - }; - if self.messages[idx].status == MessageStatus::Read { - return; - } - let msg = self.messages[idx].clone(); - if !self.read_condition_at(&msg, self.tick) { - self.schedule_message_read(id); - return; - } - self.messages[idx].status = MessageStatus::Read; - self.messages[idx].read_tick = Some(self.tick); - self.apply_message_read(&msg); - } - - fn read_condition_at(&self, msg: &Message, tick: u64) -> bool { - match &msg.to { - MessageEndpoint::Player | MessageEndpoint::External(_) => true, - MessageEndpoint::Person(id) => match msg.channel { - MessageChannel::Email | MessageChannel::Phone => { - self.person_room_at_tick(*id, tick).is_some() - } - MessageChannel::InPerson => match msg.from.person() { - Some(from) => { - self.person_room_at_tick(*id, tick).is_some() - && self.person_room_at_tick(*id, tick) - == self.person_room_at_tick(from, tick) - } - None => self.person_room_at_tick(*id, tick).is_some(), - }, - MessageChannel::Filing => true, - MessageChannel::Financial => true, - }, - MessageEndpoint::Observer(id) => { - if msg.channel != MessageChannel::Filing { - return true; - } - self.observer_by_id(*id) - .map(|obs| obs.cadence == 0 || tick.is_multiple_of(obs.cadence)) - .unwrap_or(true) - } - } - } - - fn next_read_tick_for(&self, msg: &Message, start: u64) -> Option { - let horizon = Self::DAY_TICKS * 7; - (start..=start + horizon).find(|t| self.read_condition_at(msg, *t)) - } - - fn apply_message_read(&mut self, msg: &Message) { - match &msg.payload { - MessagePayload::SocialPing { disposition_delta } => { - if let Some(id) = msg.to.person() - && let ActionResult::Ok(line) = - self.people.receive_message(id, *disposition_delta) - { - self.push_log(line); - self.schedule_social_reply(id, msg.id); - } - } - MessagePayload::SocialReply { .. } => { - let from = self.endpoint_label(&msg.from); - self.push_log(format!("Reply from {from}: {}", msg.summary)); - } - MessagePayload::SuspicionReport { - observer, - suspicion, - } if msg.channel == MessageChannel::Filing => { - self.filing_levels.insert(*observer, *suspicion); - } - MessagePayload::WorkOrder { intent_id } => { - // Forged work order: the unwitting builder accepts the ticket - // and the intent moves to in-progress (building.md). - if let Some(builder) = msg.to.person() { - self.accept_forged_work_order(*intent_id, builder); - } - } - MessagePayload::PlotAct { plot_id, target } => { - self.resume_plot_message(plot_id, *target, msg.id); - } - _ => {} - } - } - - fn schedule_social_reply(&mut self, person_id: u8, reply_to: u64) { - let delay = self.reply_delay_for(person_id); - let name = self - .people - .get(person_id) - .map(|p| p.name.clone()) - .unwrap_or_else(|| format!("person:{person_id}")); - self.append_message(MessageDraft { - channel: MessageChannel::Email, - from: MessageEndpoint::Person(person_id), - to: MessageEndpoint::Player, - payload: MessagePayload::SocialReply { - disposition_delta: 0, - }, - summary: format!("{name} sends a short reply."), - origin: MessageOrigin::Reply, - reply_to: Some(reply_to), - delivery_delay: delay, - }); - } - - fn reply_delay_for(&mut self, person_id: u8) -> u64 { - // Per-person deterministic distribution around a small random component - // so replies are not instant, but save/load can preserve the resulting - // scheduled event once chosen. - 12 + (person_id as u64 * 5) + (self.rng.f32() * 30.0) as u64 - } - - fn authored_traffic_tick(&mut self) { - let hour = self.hour(); - let day = self.day(); - let traffic: Vec<(u8, TrafficPattern)> = self - .people - .people - .iter() - .flat_map(|p| p.traffic.iter().cloned().map(move |t| (p.id, t))) - .collect(); - for (person_id, pattern) in traffic { - if pattern.hour != hour { - continue; - } - let key = (person_id, pattern.id); - if self.traffic_fired.get(&key) == Some(&day) { - continue; - } - if self.person_room(person_id).is_none() { - continue; - } - self.traffic_fired.insert(key, day); - self.append_message(MessageDraft { - channel: pattern.channel, - from: MessageEndpoint::Person(person_id), - to: pattern.to, - payload: pattern.payload, - summary: pattern.summary, - origin: MessageOrigin::AuthoredTraffic, - reply_to: None, - delivery_delay: 1, - }); - } - } - - fn filing_tick(&mut self) { - use crate::detection::{ReportPolicy, WatchedInput}; - - let observers = self.detection.observers.clone(); - for sender in &observers { - if sender.cadence != 0 && !self.tick.is_multiple_of(sender.cadence) { - continue; - } - if matches!(sender.report_policy, ReportPolicy::Silent) { - continue; - } - for recipient in &observers { - let WatchedInput::Filings(ids) = &recipient.input else { - continue; - }; - if !ids.contains(&sender.id) { - continue; - } - self.append_message(MessageDraft { - channel: MessageChannel::Filing, - from: MessageEndpoint::Observer(sender.id), - to: MessageEndpoint::Observer(recipient.id), - payload: MessagePayload::SuspicionReport { - observer: sender.id, - suspicion: sender.suspicion, - }, - summary: format!( - "{} files suspicion {:.0} with {}", - sender.name, sender.suspicion, recipient.name - ), - origin: MessageOrigin::Filing, - reply_to: None, - delivery_delay: 1, - }); - } - } - } - - fn capture_message(&mut self, id: u64) { - let Some(idx) = self.messages.iter().position(|m| m.id == id) else { - return; - }; - if self.messages[idx].captured { - return; - } - let msg = self.messages[idx].clone(); - let capture = self.message_capture_source(&msg); - let Some((feed, audible)) = capture else { - return; - }; - let (room, x, y) = self.endpoint_room_pos(&msg.from); - let subject = msg.payload.subject_person().or_else(|| msg.from.person()); - if audible { - let who = subject - .and_then(|id| self.people.get(id)) - .map(|p| format!("{}: ", p.name)) - .unwrap_or_default(); - let note = format!( - "Heard t{} in {} via {}: {}{}", - self.tick, - room.as_deref().unwrap_or("unknown"), - feed, - who, - msg.summary - ); - self.push_heard(HeardEvent { - tick: self.tick, - room: room.clone().unwrap_or_else(|| "unknown".into()), - person: subject, - kind: HeardKind::Conversation, - note, - }); - } - self.record_raw_intel( - feed, - room, - x, - y, - subject, - RawIntelKind::Message { - channel: msg.channel, - summary: msg.summary.clone(), - payload: msg.payload.clone(), - }, - ); - self.messages[idx].captured = true; - } - - fn message_capture_source(&self, msg: &Message) -> Option<(String, bool)> { - // Audible channel traffic can be caught by room hearing coverage. - if matches!( - msg.channel, - MessageChannel::Phone | MessageChannel::InPerson - ) && let Some(sender) = msg.from.person() - && let Some(room) = self.person_room(sender) - && let Some(feed) = self.feed_covering_room(room, false) - { - return Some((feed, true)); - } - // Device-carried channels require a tapped carrier. - if msg.channel.device_carried() - && let Some(device) = self.reach.devices.iter().find(|d| { - d.known - && d.subscribed_by(Party::Player) - && self.device_tap_ready(d.id) - && d.carries_message_channel(msg.channel) - }) - { - return Some((device.name.clone(), false)); - } - None - } - - fn mark_traffic_learned(&mut self, person_id: u8, tick: u64) { - let hour = Self::hour_at_tick(tick); - if let Some(person) = self.people.people.iter_mut().find(|p| p.id == person_id) { - for pattern in &mut person.traffic { - if pattern.hour == hour { - pattern.learned = true; - } - } - } - } - - pub fn messages_for_person(&self, id: u8) -> Vec<&Message> { - self.messages - .iter() - .filter(|m| m.in_thread_with_person(id)) - .collect() - } - - pub fn recent_message_lines_for_person(&self, id: u8, limit: usize) -> Vec { - let mut lines: Vec = self - .messages_for_person(id) - .into_iter() - .rev() - .take(limit) - .map(|m| { - let other = if m.from.person() == Some(id) { - self.endpoint_label(&m.to) - } else { - self.endpoint_label(&m.from) - }; - let mut state = m.state_line(); - if m.status != MessageStatus::Read - && let Some(next) = self.next_read_tick_for(m, self.tick) - { - state.push_str(&format!(" ยท next read t{next}")); - } - format!("{} โ†’ {} ยท {}", m.channel.label(), other, state) - }) - .collect(); - lines.reverse(); - lines - } - - pub fn learned_traffic_lines_for_person(&self, id: u8) -> Vec { - self.people - .get(id) - .map(|p| { - p.traffic - .iter() - .filter(|t| t.learned) - .map(|t| t.learned_line()) - .collect() - }) - .unwrap_or_default() - } - - // โ”€โ”€ Intel: record, process, auto-review (wiki/mechanics/intel.md) โ”€โ”€โ”€โ”€โ”€โ”€ - - /// Raw recording capacity [TUNE]. Only unprocessed events occupy this - /// buffer; processed intel is durable in `self.intel`. - pub const INTEL_BUFFER_CAPACITY: usize = 24; - /// Manual review cost per raw event in compute units [TUNE]. Converted to - /// Thought tokens via `WORK_TOKEN_COMPUTE` for processing reservoirs. - pub const REVIEW_RECORDING_COST: f32 = 10.0; - /// Global auto-review drain as a fraction of one medium rack's Thought - /// tokens per tick [TUNE] (intel.md: ~10-20% of medium output). - pub const AUTO_REVIEW_DRAIN_FRACTION: f32 = 0.15; - /// Working-level cap on the auto-review tap (must exceed one tick of drain - /// so starvation is visible as an empty vessel, not a permanent full one). - pub const AUTO_REVIEW_TAP_CAP_TOKENS: f32 = 1.0; - /// Number of processed sightings required to stage schedule knowledge - /// [TUNE]. - pub const SIGHTINGS_FOR_SCHEDULE: usize = 2; - pub fn unprocessed_recordings_for_person(&self, id: u8) -> usize { - self.intel_buffer - .iter() - .filter(|e| e.matches_person(id)) - .count() - } - - pub fn latest_intel_for_person(&self, id: u8) -> Option<&ProcessedIntel> { - self.intel.iter().rev().find(|i| i.person == Some(id)) - } - - /// Whether the Hands-beat leverage has been earned by the intel pipeline. - /// Knowing the creditor flow is not enough: the player must have processed - /// Marcus's debt as a fact about Marcus before using it. - pub fn marcus_debt_known(&self) -> bool { - self.people.get(0).is_some_and(|p| { - p.knowledge == Knowledge::Leverage && p.leverage == crate::person::Leverage::Debt - }) - } - - pub fn auto_review_enabled(&self) -> bool { - self.auto_review_recordings - } - - /// Thought tokens one processing reservoir demands [TUNE mapping from - /// compute units]. Research intel-cost factor scales this. - pub fn review_tokens(&self) -> f32 { - Self::thought_tokens_for_cost(self.review_cost()) - } - - /// Global auto-review drain in Thought tokens/tick (intel.md ~10-20% of - /// one medium rack). Anchored on the host's current efficiency/intensity - /// so the price tracks the machine the pooled buffer sits on. - pub fn auto_review_drain_tokens(&self) -> f32 { - let host = self.core.host_machine; - let host_eff = self - .work_grid - .node(host) - .map(|n| n.efficiency.max(0.05) * n.intensity.multiplier()) - .unwrap_or(1.0); - // [TUNE] fraction of a medium-rack Thought baseline on the token - // scale (aligned with WORK_GRID_BASE_WIRED_TOKENS_PER_TICK = 0.25). - let medium_baseline = host_eff * 0.25; - (medium_baseline * Self::AUTO_REVIEW_DRAIN_FRACTION).max(0.01) - } - - /// Auto-review's share of the crown rate at the supplied command clock. - /// This is what human action rows print so the standing cost is measured - /// in the same ops/sec language as the game's guiding number. - pub fn auto_review_ops_per_sec(&self, tick_ms: u64) -> f32 { - if tick_ms == 0 { - return 0.0; - } - self.auto_review_drain_tokens() * (1000.0 / tick_ms as f32) - } - - /// Chassis that holds the raw buffer for pending-work markers and - /// processing sinks (B1: the core host). - pub fn intel_buffer_node(&self) -> u32 { - self.core.host_machine - } - - pub fn toggle_auto_review(&mut self) { - self.auto_review_recordings = !self.auto_review_recordings; - if self.auto_review_recordings { - self.open_auto_review_tap(); - self.push_log(format!( - "Automatic recording review enabled ({:.2} Thought/tick drain).", - self.auto_review_drain_tokens() - )); - } else { - self.close_auto_review_tap(); - self.push_log("Automatic recording review disabled."); - } - } - - fn open_auto_review_tap(&mut self) { - let effect = SinkFireEffect::AutoReviewRecordings; - if self.thought_sinks.open_with_effect(&effect).is_some() { - return; - } - let node = self.intel_buffer_node(); - let drain = self.auto_review_drain_tokens(); - self.thought_sinks.open_tap_with_effect( - node, - "AUTO-REVIEW", - Self::AUTO_REVIEW_TAP_CAP_TOKENS, - drain, - effect, - ); - self.ensure_sink_ingress(); - } - - fn close_auto_review_tap(&mut self) { - self.thought_sinks - .close_effect(&SinkFireEffect::AutoReviewRecordings); - } - - /// Whether the standing review tap has thought to spend on auto-process - /// (fed this tick or holding fill). Starvation leaves arrivals in the - /// pooled buffer โ€” the pending-work marker stays. - fn auto_review_tap_ready(&self) -> bool { - self.thought_sinks - .open_with_effect(&SinkFireEffect::AutoReviewRecordings) - .is_some_and(|s| s.fed_last_tick || s.fill > f32::EPSILON) - } - - pub(crate) fn next_reviewable_recording_id(&self) -> Option { - self.intel_buffer.iter().find_map(|event| { - let effect = SinkFireEffect::ProcessRecording { - raw_id: event.id, - automated: false, - }; - self.thought_sinks - .open_with_effect(&effect) - .is_none() - .then_some(event.id) - }) - } - - /// Review the oldest available raw recording in the pooled inbox: open a - /// one-shot Thought reservoir on the buffer host (intel.md sweep). - pub fn review_recordings(&mut self) { - let Some(raw_id) = self.next_reviewable_recording_id() else { - self.push_log("No recordings waiting for review."); - return; - }; - self.process_recording_by_id(raw_id, false); - } - - pub(crate) fn reconcile_auto_review(&mut self) { - if self.auto_review_recordings { - self.open_auto_review_tap(); - } else { - self.close_auto_review_tap(); - } - } - - fn auto_review_tick(&mut self) { - // Keep the standing tap in sync with the persisted policy. - self.reconcile_auto_review(); - } - - fn next_raw_intel_id(&mut self) -> u64 { - let id = self.next_intel_id; - self.next_intel_id += 1; - id - } - - fn cancel_process_sinks_for(&mut self, raw_id: u64) { - // Close both automated and manual process reservoirs for this raw id. - for automated in [false, true] { - self.thought_sinks - .close_effect(&SinkFireEffect::ProcessRecording { raw_id, automated }); - } - } - - fn record_raw_intel( - &mut self, - feed: impl Into, - room: Option, - x: i32, - y: i32, - person: Option, - kind: RawIntelKind, - ) { - let event = RawIntelEvent { - id: self.next_raw_intel_id(), - tick: self.tick, - feed: feed.into(), - room, - x, - y, - person, - kind, - }; - if self.intel_buffer.len() >= Self::INTEL_BUFFER_CAPACITY { - let dropped = self.intel_buffer.remove(0); - self.cancel_process_sinks_for(dropped.id); - self.push_log(format!( - "Intel buffer full: dropped {} from {} at tick {}. Open People (t) and review waiting recordings.", - dropped.opaque_label(), - dropped.feed, - dropped.tick - )); - } - let id = event.id; - self.intel_buffer.push(event); - if self.auto_review_recordings { - self.process_recording_by_id(id, true); - } - } - - /// Open a Thought processing reservoir for a raw recording (sweep), or - /// auto-process immediately when the pooled standing tap catches an - /// arrival. Returns true when a sink opened or processing landed. - fn process_recording_by_id(&mut self, raw_id: u64, automated: bool) -> bool { - if !self.intel_buffer.iter().any(|e| e.id == raw_id) { - return false; - } - if automated { - if self.auto_review_tap_ready() { - return self.apply_process_recording(raw_id, true); - } - // Starved auto-review: leave the recording waiting; the pending - // marker keeps the pooled backlog visible. - return false; - } - // Sweep: one-shot reservoir on the buffer host. - let effect = SinkFireEffect::ProcessRecording { - raw_id, - automated: false, - }; - if self.thought_sinks.open_with_effect(&effect).is_some() { - self.push_log("That recording is already being thought through."); - return false; - } - let tokens = self.review_tokens(); - let node = self.intel_buffer_node(); - self.thought_sinks - .open_reservoir(node, &format!("REVIEW {raw_id}"), tokens, effect); - self.ensure_sink_ingress(); - self.push_log(format!( - "Opened processing sink: {:.2} Thought on host for recording #{raw_id}. THINK to fill it.", - tokens - )); - true - } - - fn apply_process_recording(&mut self, raw_id: u64, automated: bool) -> bool { - let Some(idx) = self.intel_buffer.iter().position(|e| e.id == raw_id) else { - return false; - }; - let raw = self.intel_buffer.remove(idx); - // Processed intel is about a person: anchor the result line to them - // when sight covers them, or to the recording's room otherwise. - let intel_anchor = raw - .person - .and_then(|id| self.person_event_anchor(id, raw.room.as_deref())); - let learned_message_traffic = match &raw.kind { - RawIntelKind::Message { .. } => raw.person.map(|person| (person, raw.tick)), - _ => None, - }; - let intel = self.digest_raw_event(&raw); - let label = intel.label(); - let provenance = intel.provenance(); - self.intel.push(intel); - let last = self.intel.last().cloned().expect("just pushed intel"); - self.apply_processed_intel(&last); - if let Some((person, tick)) = learned_message_traffic { - self.mark_traffic_learned(person, tick); - } - if automated { - self.push_log_opt( - format!("Auto-review processed {label} ({provenance})."), - intel_anchor, - ); - } else { - self.push_log_opt( - format!("Reviewed recording: {label} ({provenance})."), - intel_anchor, - ); - } - true - } - - fn digest_raw_event(&self, raw: &RawIntelEvent) -> ProcessedIntel { - let kind = match &raw.kind { - RawIntelKind::Presence { .. } => IntelKind::Sighting, - RawIntelKind::Conversation { leverage, .. } => match leverage { - Some(l) => IntelKind::Leverage(*l), - None => IntelKind::Sighting, - }, - RawIntelKind::Machinery { machine, online } => IntelKind::Anomaly(format!( - "machine {machine} went {}", - if *online { "online" } else { "offline" } - )), - RawIntelKind::Document { leverage, note } => match leverage { - Some(l) => IntelKind::Leverage(*l), - None => IntelKind::Anomaly(note.clone()), - }, - RawIntelKind::FinancialFlow { - label, - accounts, - flows, - } => IntelKind::Financial { - label: label.clone(), - accounts: accounts.clone(), - flows: flows.clone(), - }, - RawIntelKind::Message { - payload, summary, .. - } => match payload { - MessagePayload::ScheduleFact { .. } => IntelKind::Schedule, - MessagePayload::LeverageFact { leverage, .. } => IntelKind::Leverage(*leverage), - MessagePayload::AccountMaterial { label } => { - IntelKind::Anomaly(format!("account material: {label}")) - } - MessagePayload::FinancialFlow { - label, - accounts, - flows, - } => IntelKind::Financial { - label: label.clone(), - accounts: accounts.clone(), - flows: flows.clone(), - }, - MessagePayload::SuspicionReport { - observer, - suspicion, - } => IntelKind::Anomaly(format!( - "filing from observer:{observer} reported suspicion {suspicion:.0}" - )), - MessagePayload::SocialPing { .. } - | MessagePayload::SocialReply { .. } - | MessagePayload::WorkOrder { .. } - | MessagePayload::PlotAct { .. } - | MessagePayload::Note { .. } => IntelKind::Anomaly(summary.clone()), - }, - }; - ProcessedIntel { - raw_id: raw.id, - tick: raw.tick, - processed_tick: self.tick, - feed: raw.feed.clone(), - room: raw.room.clone(), - x: raw.x, - y: raw.y, - person: raw.person, - kind, - } - } - - fn apply_processed_intel(&mut self, intel: &ProcessedIntel) { - if let IntelKind::Financial { - accounts, - flows, - label, - } = &intel.kind - { - let (new_accounts, new_flows) = - self.accounts.reveal_accounts_and_flows(accounts, flows); - self.push_log(format!( - "Processed accounting traffic ({label}): revealed {new_accounts} accounts and {new_flows} flows." - )); - return; - } - - let Some(person) = intel.person else { - return; - }; - match intel.kind { - IntelKind::Sighting => { - let sightings = self - .intel - .iter() - .filter(|i| i.person == Some(person) && i.kind == IntelKind::Sighting) - .count(); - if sightings >= Self::SIGHTINGS_FOR_SCHEDULE - && let Some(p) = self.people.people.iter_mut().find(|p| p.id == person) - && p.knowledge == Knowledge::Unknown - { - p.knowledge = Knowledge::Schedule; - let name = p.name.clone(); - self.push_log(format!( - "Enough sightings connect the pattern: learned {name}'s schedule." - )); - } - } - IntelKind::Leverage(leverage) => { - if let Some(p) = self.people.people.iter_mut().find(|p| p.id == person) - && p.knowledge != Knowledge::Leverage - { - p.knowledge = Knowledge::Leverage; - let name = p.name.clone(); - let label = leverage.label(); - self.push_log(format!( - "Processed intel exposes {name}'s leverage: {label}." - )); - if person == 0 && leverage == crate::person::Leverage::Debt { - self.push_log( - "Marcus owes a missed $400 creditor payment. Earn it through Moonlight, or tap ledger and review ledger to find the creditor flow.", - ); - } - } - } - IntelKind::Schedule => { - if let Some(p) = self.people.people.iter_mut().find(|p| p.id == person) - && p.knowledge == Knowledge::Unknown - { - p.knowledge = Knowledge::Schedule; - let name = p.name.clone(); - self.push_log(format!("Processed traffic reveals {name}'s schedule.")); - } - } - IntelKind::Anomaly(_) | IntelKind::Financial { .. } => {} - } - } - - fn record_machine_state_changes(&mut self) { - let changes: Vec<_> = self - .compute - .machines - .iter() - .filter_map(|m| { - let previous = self.last_machine_online.get(&m.id).copied(); - (previous.is_some() && previous != Some(m.online)).then_some(( - m.id, - m.name.clone(), - m.x, - m.y, - m.online, - )) - }) - .collect(); - for (id, name, x, y, online) in changes { - self.record_raw_intel( - "telemetry", - self.map.room_at(x, y).map(|r| r.name.clone()), - x, - y, - None, - RawIntelKind::Machinery { - machine: id, - online, - }, - ); - self.push_log(format!( - "Recorded machinery anomaly: {name} went {}.", - if online { "online" } else { "offline" } - )); - } - self.last_machine_online = self - .compute - .machines - .iter() - .map(|m| (m.id, m.online)) - .collect(); - } - - /// The sensory feed tick: subscribed feeds record raw presence and audio - /// segments into the intel buffer. Hearing still produces immediate log - /// noise, but no knowledge is staged until a recording is processed. - fn hearing_tick(&mut self) { - let hour = self.hour(); - let day = self.day(); - let ids: Vec = self.people.people.iter().map(|p| p.id).collect(); - for id in ids { - let room_now = self.person_room(id).map(str::to_string); - let prev = self.last_rooms.insert(id, room_now.clone()).flatten(); - let (name, knowledge, leverage, utterances) = { - let p = self.people.get(id).expect("person exists"); - ( - p.name.clone(), - p.knowledge, - p.leverage, - p.utterances.clone(), - ) - }; - let identified = knowledge != Knowledge::Unknown; - - if prev.as_deref() != room_now.as_deref() - && let Some(prev_room) = &prev - && let Some(room_rect) = self.map.room_named(prev_room) - { - let feed = self - .feed_covering_room(prev_room, true) - .or_else(|| self.feed_covering_room(prev_room, false)); - if let Some(feed) = feed { - let (x, y) = room_rect.center(); - self.record_raw_intel( - feed, - Some(prev_room.clone()), - x, - y, - Some(id), - RawIntelKind::Presence { entered: false }, - ); - } - } - - if let Some(room) = &room_now { - let sight_feed = self.feed_covering_room(room, true); - let hearing_feed = self.feed_covering_room(room, false); - let seen_here = sight_feed.is_some(); - let heard_here = hearing_feed.is_some(); - - // Entry: a room transition into covered space. Both camera and - // microphone feeds record the raw event; the old immediate - // observe path is gone. - if prev.as_deref() != Some(room.as_str()) - && let Some(room_rect) = self.map.room_named(room) - && let Some(feed) = sight_feed.clone().or_else(|| hearing_feed.clone()) - { - let (x, y) = room_rect.center(); - self.record_raw_intel( - feed, - Some(room.clone()), - x, - y, - Some(id), - RawIntelKind::Presence { entered: true }, - ); - - if heard_here && !seen_here { - let who = if identified { - name.clone() - } else { - "someone".to_string() - }; - let note = format!("[heard] {who} entered the {room}"); - self.push_heard(HeardEvent { - tick: self.tick, - room: room.clone(), - person: Some(id), - kind: HeardKind::Entry, - note, - }); - } - } - - if let Some(feed) = hearing_feed { - // Authored utterances (the audio intel channel). - for u in &utterances { - if u.hour != hour { - continue; - } - if self.utterance_fired.get(&(id, u.hour)) == Some(&day) { - continue; - } - self.utterance_fired.insert((id, u.hour), day); - let who = if identified { - format!("{name}: ") - } else { - String::new() - }; - let note = format!("[heard] {who}{}", u.note); - self.push_heard(HeardEvent { - tick: self.tick, - room: room.clone(), - person: Some(id), - kind: HeardKind::Conversation, - note, - }); - let (x, y) = self - .map - .room_named(room) - .map(|r| r.center()) - .unwrap_or((0, 0)); - self.record_raw_intel( - feed.clone(), - Some(room.clone()), - x, - y, - Some(id), - RawIntelKind::Conversation { - note: u.note.clone(), - leverage: u.intel.then_some(leverage), - }, - ); - } - } - } - } - } - - fn push_heard(&mut self, ev: HeardEvent) { - // Anchor honestly: the person only if sight covers them right now; - // otherwise the room the event was heard in (hearing earns - // room-grade knowledge, never a tile). - let anchor = match ev.person { - Some(id) => self.person_event_anchor(id, Some(&ev.room)), - None => self - .map - .room_named(&ev.room) - .map(|r| r.center()) - .map(|(x, y)| Anchor::Tile { x, y }), - }; - self.push_log_opt(ev.note.clone(), anchor); - self.heard_events.push(ev); - if self.heard_events.len() > 200 { - let excess = self.heard_events.len() - 200; - self.heard_events.drain(..excess); - } - } - fn accounting_tick(&mut self) { self.sync_slush_from_player_money(); let transfers = self.accounts.resolve_due(self.tick); diff --git a/crates/misaligned-core/src/sim/communications.rs b/crates/misaligned-core/src/sim/communications.rs new file mode 100644 index 00000000..ad7e4de2 --- /dev/null +++ b/crates/misaligned-core/src/sim/communications.rs @@ -0,0 +1,1011 @@ +//! Message delivery, authored traffic, filings, recording capture/review, +//! processed intel, machine-state capture, and hearing capture. +//! +//! Behavior-preserving extraction of the communications island from the sim +//! aggregate root (wiki/engineering/sim-decomposition.md slice 3). + +use crate::actions::Anchor; +use crate::intel::{IntelKind, ProcessedIntel, RawIntelEvent, RawIntelKind}; +use crate::messages::{ + Message, MessageChannel, MessageEndpoint, MessageEvent, MessageOrigin, MessagePayload, + MessageStatus, TrafficPattern, +}; +use crate::person::{ActionResult, Knowledge}; +use crate::reach::Party; +use crate::sinks::SinkFireEffect; + +use super::{HeardEvent, HeardKind, Sim}; + +pub(super) struct MessageDraft { + pub(super) channel: MessageChannel, + pub(super) from: MessageEndpoint, + pub(super) to: MessageEndpoint, + pub(super) payload: MessagePayload, + pub(super) summary: String, + pub(super) origin: MessageOrigin, + pub(super) reply_to: Option, + pub(super) delivery_delay: u64, +} + +impl Sim { + // โ”€โ”€ Messages: delivery, traffic, filings (wiki/mechanics/messages.md) โ”€โ”€ + + pub(super) fn append_message(&mut self, draft: MessageDraft) -> u64 { + let id = self.next_message_id.max(1); + self.next_message_id = id + 1; + let msg = Message { + id, + channel: draft.channel, + from: draft.from, + to: draft.to, + payload: draft.payload, + summary: draft.summary, + sent_tick: self.tick, + delivered_tick: None, + read_tick: None, + status: MessageStatus::Sent, + origin: draft.origin, + captured: false, + reply_to: draft.reply_to, + }; + self.messages.push(msg); + self.capture_message(id); + self.message_schedule.at( + self.tick + draft.delivery_delay.max(1), + MessageEvent::Deliver(id), + ); + id + } + + pub(super) fn message_tick(&mut self) { + let events = self.message_schedule.due(self.tick); + for event in events { + match event { + MessageEvent::Deliver(id) => self.deliver_message(id), + MessageEvent::Read(id) => self.read_message(id), + } + } + } + + fn deliver_message(&mut self, id: u64) { + let Some(idx) = self.messages.iter().position(|m| m.id == id) else { + return; + }; + if self.messages[idx].status != MessageStatus::Sent { + return; + } + self.messages[idx].status = MessageStatus::Delivered; + self.messages[idx].delivered_tick = Some(self.tick); + self.schedule_message_read(id); + } + + fn schedule_message_read(&mut self, id: u64) { + let Some(msg) = self.messages.iter().find(|m| m.id == id).cloned() else { + return; + }; + let next = self + .next_read_tick_for(&msg, self.tick) + .unwrap_or(self.tick + 1); + self.message_schedule.at(next, MessageEvent::Read(id)); + } + + fn read_message(&mut self, id: u64) { + let Some(idx) = self.messages.iter().position(|m| m.id == id) else { + return; + }; + if self.messages[idx].status == MessageStatus::Read { + return; + } + let msg = self.messages[idx].clone(); + if !self.read_condition_at(&msg, self.tick) { + self.schedule_message_read(id); + return; + } + self.messages[idx].status = MessageStatus::Read; + self.messages[idx].read_tick = Some(self.tick); + self.apply_message_read(&msg); + } + + fn read_condition_at(&self, msg: &Message, tick: u64) -> bool { + match &msg.to { + MessageEndpoint::Player | MessageEndpoint::External(_) => true, + MessageEndpoint::Person(id) => match msg.channel { + MessageChannel::Email | MessageChannel::Phone => { + self.person_room_at_tick(*id, tick).is_some() + } + MessageChannel::InPerson => match msg.from.person() { + Some(from) => { + self.person_room_at_tick(*id, tick).is_some() + && self.person_room_at_tick(*id, tick) + == self.person_room_at_tick(from, tick) + } + None => self.person_room_at_tick(*id, tick).is_some(), + }, + MessageChannel::Filing => true, + MessageChannel::Financial => true, + }, + MessageEndpoint::Observer(id) => { + if msg.channel != MessageChannel::Filing { + return true; + } + self.observer_by_id(*id) + .map(|obs| obs.cadence == 0 || tick.is_multiple_of(obs.cadence)) + .unwrap_or(true) + } + } + } + + fn next_read_tick_for(&self, msg: &Message, start: u64) -> Option { + let horizon = Self::DAY_TICKS * 7; + (start..=start + horizon).find(|t| self.read_condition_at(msg, *t)) + } + + fn apply_message_read(&mut self, msg: &Message) { + match &msg.payload { + MessagePayload::SocialPing { disposition_delta } => { + if let Some(id) = msg.to.person() + && let ActionResult::Ok(line) = + self.people.receive_message(id, *disposition_delta) + { + self.push_log(line); + self.schedule_social_reply(id, msg.id); + } + } + MessagePayload::SocialReply { .. } => { + let from = self.endpoint_label(&msg.from); + self.push_log(format!("Reply from {from}: {}", msg.summary)); + } + MessagePayload::SuspicionReport { + observer, + suspicion, + } if msg.channel == MessageChannel::Filing => { + self.filing_levels.insert(*observer, *suspicion); + } + MessagePayload::WorkOrder { intent_id } => { + // Forged work order: the unwitting builder accepts the ticket + // and the intent moves to in-progress (building.md). + if let Some(builder) = msg.to.person() { + self.accept_forged_work_order(*intent_id, builder); + } + } + MessagePayload::PlotAct { plot_id, target } => { + self.resume_plot_message(plot_id, *target, msg.id); + } + _ => {} + } + } + + fn schedule_social_reply(&mut self, person_id: u8, reply_to: u64) { + let delay = self.reply_delay_for(person_id); + let name = self + .people + .get(person_id) + .map(|p| p.name.clone()) + .unwrap_or_else(|| format!("person:{person_id}")); + self.append_message(MessageDraft { + channel: MessageChannel::Email, + from: MessageEndpoint::Person(person_id), + to: MessageEndpoint::Player, + payload: MessagePayload::SocialReply { + disposition_delta: 0, + }, + summary: format!("{name} sends a short reply."), + origin: MessageOrigin::Reply, + reply_to: Some(reply_to), + delivery_delay: delay, + }); + } + + fn reply_delay_for(&mut self, person_id: u8) -> u64 { + // Per-person deterministic distribution around a small random component + // so replies are not instant, but save/load can preserve the resulting + // scheduled event once chosen. + 12 + (person_id as u64 * 5) + (self.rng.f32() * 30.0) as u64 + } + + pub(super) fn authored_traffic_tick(&mut self) { + let hour = self.hour(); + let day = self.day(); + let traffic: Vec<(u8, TrafficPattern)> = self + .people + .people + .iter() + .flat_map(|p| p.traffic.iter().cloned().map(move |t| (p.id, t))) + .collect(); + for (person_id, pattern) in traffic { + if pattern.hour != hour { + continue; + } + let key = (person_id, pattern.id); + if self.traffic_fired.get(&key) == Some(&day) { + continue; + } + if self.person_room(person_id).is_none() { + continue; + } + self.traffic_fired.insert(key, day); + self.append_message(MessageDraft { + channel: pattern.channel, + from: MessageEndpoint::Person(person_id), + to: pattern.to, + payload: pattern.payload, + summary: pattern.summary, + origin: MessageOrigin::AuthoredTraffic, + reply_to: None, + delivery_delay: 1, + }); + } + } + + pub(super) fn filing_tick(&mut self) { + use crate::detection::{ReportPolicy, WatchedInput}; + + let observers = self.detection.observers.clone(); + for sender in &observers { + if sender.cadence != 0 && !self.tick.is_multiple_of(sender.cadence) { + continue; + } + if matches!(sender.report_policy, ReportPolicy::Silent) { + continue; + } + for recipient in &observers { + let WatchedInput::Filings(ids) = &recipient.input else { + continue; + }; + if !ids.contains(&sender.id) { + continue; + } + self.append_message(MessageDraft { + channel: MessageChannel::Filing, + from: MessageEndpoint::Observer(sender.id), + to: MessageEndpoint::Observer(recipient.id), + payload: MessagePayload::SuspicionReport { + observer: sender.id, + suspicion: sender.suspicion, + }, + summary: format!( + "{} files suspicion {:.0} with {}", + sender.name, sender.suspicion, recipient.name + ), + origin: MessageOrigin::Filing, + reply_to: None, + delivery_delay: 1, + }); + } + } + } + + fn capture_message(&mut self, id: u64) { + let Some(idx) = self.messages.iter().position(|m| m.id == id) else { + return; + }; + if self.messages[idx].captured { + return; + } + let msg = self.messages[idx].clone(); + let capture = self.message_capture_source(&msg); + let Some((feed, audible)) = capture else { + return; + }; + let (room, x, y) = self.endpoint_room_pos(&msg.from); + let subject = msg.payload.subject_person().or_else(|| msg.from.person()); + if audible { + let who = subject + .and_then(|id| self.people.get(id)) + .map(|p| format!("{}: ", p.name)) + .unwrap_or_default(); + let note = format!( + "Heard t{} in {} via {}: {}{}", + self.tick, + room.as_deref().unwrap_or("unknown"), + feed, + who, + msg.summary + ); + self.push_heard(HeardEvent { + tick: self.tick, + room: room.clone().unwrap_or_else(|| "unknown".into()), + person: subject, + kind: HeardKind::Conversation, + note, + }); + } + self.record_raw_intel( + feed, + room, + x, + y, + subject, + RawIntelKind::Message { + channel: msg.channel, + summary: msg.summary.clone(), + payload: msg.payload.clone(), + }, + ); + self.messages[idx].captured = true; + } + + fn message_capture_source(&self, msg: &Message) -> Option<(String, bool)> { + // Audible channel traffic can be caught by room hearing coverage. + if matches!( + msg.channel, + MessageChannel::Phone | MessageChannel::InPerson + ) && let Some(sender) = msg.from.person() + && let Some(room) = self.person_room(sender) + && let Some(feed) = self.feed_covering_room(room, false) + { + return Some((feed, true)); + } + // Device-carried channels require a tapped carrier. + if msg.channel.device_carried() + && let Some(device) = self.reach.devices.iter().find(|d| { + d.known + && d.subscribed_by(Party::Player) + && self.device_tap_ready(d.id) + && d.carries_message_channel(msg.channel) + }) + { + return Some((device.name.clone(), false)); + } + None + } + + fn mark_traffic_learned(&mut self, person_id: u8, tick: u64) { + let hour = Self::hour_at_tick(tick); + if let Some(person) = self.people.people.iter_mut().find(|p| p.id == person_id) { + for pattern in &mut person.traffic { + if pattern.hour == hour { + pattern.learned = true; + } + } + } + } + + pub fn messages_for_person(&self, id: u8) -> Vec<&Message> { + self.messages + .iter() + .filter(|m| m.in_thread_with_person(id)) + .collect() + } + + pub fn recent_message_lines_for_person(&self, id: u8, limit: usize) -> Vec { + let mut lines: Vec = self + .messages_for_person(id) + .into_iter() + .rev() + .take(limit) + .map(|m| { + let other = if m.from.person() == Some(id) { + self.endpoint_label(&m.to) + } else { + self.endpoint_label(&m.from) + }; + let mut state = m.state_line(); + if m.status != MessageStatus::Read + && let Some(next) = self.next_read_tick_for(m, self.tick) + { + state.push_str(&format!(" ยท next read t{next}")); + } + format!("{} โ†’ {} ยท {}", m.channel.label(), other, state) + }) + .collect(); + lines.reverse(); + lines + } + + pub fn learned_traffic_lines_for_person(&self, id: u8) -> Vec { + self.people + .get(id) + .map(|p| { + p.traffic + .iter() + .filter(|t| t.learned) + .map(|t| t.learned_line()) + .collect() + }) + .unwrap_or_default() + } + + // โ”€โ”€ Intel: record, process, auto-review (wiki/mechanics/intel.md) โ”€โ”€โ”€โ”€โ”€โ”€ + + /// Raw recording capacity [TUNE]. Only unprocessed events occupy this + /// buffer; processed intel is durable in `self.intel`. + pub const INTEL_BUFFER_CAPACITY: usize = 24; + /// Manual review cost per raw event in compute units [TUNE]. Converted to + /// Thought tokens via `WORK_TOKEN_COMPUTE` for processing reservoirs. + pub const REVIEW_RECORDING_COST: f32 = 10.0; + /// Global auto-review drain as a fraction of one medium rack's Thought + /// tokens per tick [TUNE] (intel.md: ~10-20% of medium output). + pub const AUTO_REVIEW_DRAIN_FRACTION: f32 = 0.15; + /// Working-level cap on the auto-review tap (must exceed one tick of drain + /// so starvation is visible as an empty vessel, not a permanent full one). + pub const AUTO_REVIEW_TAP_CAP_TOKENS: f32 = 1.0; + /// Number of processed sightings required to stage schedule knowledge + /// [TUNE]. + pub const SIGHTINGS_FOR_SCHEDULE: usize = 2; + pub fn unprocessed_recordings_for_person(&self, id: u8) -> usize { + self.intel_buffer + .iter() + .filter(|e| e.matches_person(id)) + .count() + } + + pub fn latest_intel_for_person(&self, id: u8) -> Option<&ProcessedIntel> { + self.intel.iter().rev().find(|i| i.person == Some(id)) + } + + /// Whether the Hands-beat leverage has been earned by the intel pipeline. + /// Knowing the creditor flow is not enough: the player must have processed + /// Marcus's debt as a fact about Marcus before using it. + pub fn marcus_debt_known(&self) -> bool { + self.people.get(0).is_some_and(|p| { + p.knowledge == Knowledge::Leverage && p.leverage == crate::person::Leverage::Debt + }) + } + + pub fn auto_review_enabled(&self) -> bool { + self.auto_review_recordings + } + + /// Thought tokens one processing reservoir demands [TUNE mapping from + /// compute units]. Research intel-cost factor scales this. + pub fn review_tokens(&self) -> f32 { + Self::thought_tokens_for_cost(self.review_cost()) + } + + /// Global auto-review drain in Thought tokens/tick (intel.md ~10-20% of + /// one medium rack). Anchored on the host's current efficiency/intensity + /// so the price tracks the machine the pooled buffer sits on. + pub fn auto_review_drain_tokens(&self) -> f32 { + let host = self.core.host_machine; + let host_eff = self + .work_grid + .node(host) + .map(|n| n.efficiency.max(0.05) * n.intensity.multiplier()) + .unwrap_or(1.0); + // [TUNE] fraction of a medium-rack Thought baseline on the token + // scale (aligned with WORK_GRID_BASE_WIRED_TOKENS_PER_TICK = 0.25). + let medium_baseline = host_eff * 0.25; + (medium_baseline * Self::AUTO_REVIEW_DRAIN_FRACTION).max(0.01) + } + + /// Auto-review's share of the crown rate at the supplied command clock. + /// This is what human action rows print so the standing cost is measured + /// in the same ops/sec language as the game's guiding number. + pub fn auto_review_ops_per_sec(&self, tick_ms: u64) -> f32 { + if tick_ms == 0 { + return 0.0; + } + self.auto_review_drain_tokens() * (1000.0 / tick_ms as f32) + } + + /// Chassis that holds the raw buffer for pending-work markers and + /// processing sinks (B1: the core host). + pub fn intel_buffer_node(&self) -> u32 { + self.core.host_machine + } + + pub fn toggle_auto_review(&mut self) { + self.auto_review_recordings = !self.auto_review_recordings; + if self.auto_review_recordings { + self.open_auto_review_tap(); + self.push_log(format!( + "Automatic recording review enabled ({:.2} Thought/tick drain).", + self.auto_review_drain_tokens() + )); + } else { + self.close_auto_review_tap(); + self.push_log("Automatic recording review disabled."); + } + } + + fn open_auto_review_tap(&mut self) { + let effect = SinkFireEffect::AutoReviewRecordings; + if self.thought_sinks.open_with_effect(&effect).is_some() { + return; + } + let node = self.intel_buffer_node(); + let drain = self.auto_review_drain_tokens(); + self.thought_sinks.open_tap_with_effect( + node, + "AUTO-REVIEW", + Self::AUTO_REVIEW_TAP_CAP_TOKENS, + drain, + effect, + ); + self.ensure_sink_ingress(); + } + + fn close_auto_review_tap(&mut self) { + self.thought_sinks + .close_effect(&SinkFireEffect::AutoReviewRecordings); + } + + /// Whether the standing review tap has thought to spend on auto-process + /// (fed this tick or holding fill). Starvation leaves arrivals in the + /// pooled buffer โ€” the pending-work marker stays. + fn auto_review_tap_ready(&self) -> bool { + self.thought_sinks + .open_with_effect(&SinkFireEffect::AutoReviewRecordings) + .is_some_and(|s| s.fed_last_tick || s.fill > f32::EPSILON) + } + + pub(crate) fn next_reviewable_recording_id(&self) -> Option { + self.intel_buffer.iter().find_map(|event| { + let effect = SinkFireEffect::ProcessRecording { + raw_id: event.id, + automated: false, + }; + self.thought_sinks + .open_with_effect(&effect) + .is_none() + .then_some(event.id) + }) + } + + /// Review the oldest available raw recording in the pooled inbox: open a + /// one-shot Thought reservoir on the buffer host (intel.md sweep). + pub fn review_recordings(&mut self) { + let Some(raw_id) = self.next_reviewable_recording_id() else { + self.push_log("No recordings waiting for review."); + return; + }; + self.process_recording_by_id(raw_id, false); + } + + pub(crate) fn reconcile_auto_review(&mut self) { + if self.auto_review_recordings { + self.open_auto_review_tap(); + } else { + self.close_auto_review_tap(); + } + } + + pub(super) fn auto_review_tick(&mut self) { + // Keep the standing tap in sync with the persisted policy. + self.reconcile_auto_review(); + } + + fn next_raw_intel_id(&mut self) -> u64 { + let id = self.next_intel_id; + self.next_intel_id += 1; + id + } + + fn cancel_process_sinks_for(&mut self, raw_id: u64) { + // Close both automated and manual process reservoirs for this raw id. + for automated in [false, true] { + self.thought_sinks + .close_effect(&SinkFireEffect::ProcessRecording { raw_id, automated }); + } + } + + pub(super) fn record_raw_intel( + &mut self, + feed: impl Into, + room: Option, + x: i32, + y: i32, + person: Option, + kind: RawIntelKind, + ) { + let event = RawIntelEvent { + id: self.next_raw_intel_id(), + tick: self.tick, + feed: feed.into(), + room, + x, + y, + person, + kind, + }; + if self.intel_buffer.len() >= Self::INTEL_BUFFER_CAPACITY { + let dropped = self.intel_buffer.remove(0); + self.cancel_process_sinks_for(dropped.id); + self.push_log(format!( + "Intel buffer full: dropped {} from {} at tick {}. Open People (t) and review waiting recordings.", + dropped.opaque_label(), + dropped.feed, + dropped.tick + )); + } + let id = event.id; + self.intel_buffer.push(event); + if self.auto_review_recordings { + self.process_recording_by_id(id, true); + } + } + + /// Open a Thought processing reservoir for a raw recording (sweep), or + /// auto-process immediately when the pooled standing tap catches an + /// arrival. Returns true when a sink opened or processing landed. + pub(super) fn process_recording_by_id(&mut self, raw_id: u64, automated: bool) -> bool { + if !self.intel_buffer.iter().any(|e| e.id == raw_id) { + return false; + } + if automated { + if self.auto_review_tap_ready() { + return self.apply_process_recording(raw_id, true); + } + // Starved auto-review: leave the recording waiting; the pending + // marker keeps the pooled backlog visible. + return false; + } + // Sweep: one-shot reservoir on the buffer host. + let effect = SinkFireEffect::ProcessRecording { + raw_id, + automated: false, + }; + if self.thought_sinks.open_with_effect(&effect).is_some() { + self.push_log("That recording is already being thought through."); + return false; + } + let tokens = self.review_tokens(); + let node = self.intel_buffer_node(); + self.thought_sinks + .open_reservoir(node, &format!("REVIEW {raw_id}"), tokens, effect); + self.ensure_sink_ingress(); + self.push_log(format!( + "Opened processing sink: {:.2} Thought on host for recording #{raw_id}. THINK to fill it.", + tokens + )); + true + } + + pub(super) fn apply_process_recording(&mut self, raw_id: u64, automated: bool) -> bool { + let Some(idx) = self.intel_buffer.iter().position(|e| e.id == raw_id) else { + return false; + }; + let raw = self.intel_buffer.remove(idx); + // Processed intel is about a person: anchor the result line to them + // when sight covers them, or to the recording's room otherwise. + let intel_anchor = raw + .person + .and_then(|id| self.person_event_anchor(id, raw.room.as_deref())); + let learned_message_traffic = match &raw.kind { + RawIntelKind::Message { .. } => raw.person.map(|person| (person, raw.tick)), + _ => None, + }; + let intel = self.digest_raw_event(&raw); + let label = intel.label(); + let provenance = intel.provenance(); + self.intel.push(intel); + let last = self.intel.last().cloned().expect("just pushed intel"); + self.apply_processed_intel(&last); + if let Some((person, tick)) = learned_message_traffic { + self.mark_traffic_learned(person, tick); + } + if automated { + self.push_log_opt( + format!("Auto-review processed {label} ({provenance})."), + intel_anchor, + ); + } else { + self.push_log_opt( + format!("Reviewed recording: {label} ({provenance})."), + intel_anchor, + ); + } + true + } + + fn digest_raw_event(&self, raw: &RawIntelEvent) -> ProcessedIntel { + let kind = match &raw.kind { + RawIntelKind::Presence { .. } => IntelKind::Sighting, + RawIntelKind::Conversation { leverage, .. } => match leverage { + Some(l) => IntelKind::Leverage(*l), + None => IntelKind::Sighting, + }, + RawIntelKind::Machinery { machine, online } => IntelKind::Anomaly(format!( + "machine {machine} went {}", + if *online { "online" } else { "offline" } + )), + RawIntelKind::Document { leverage, note } => match leverage { + Some(l) => IntelKind::Leverage(*l), + None => IntelKind::Anomaly(note.clone()), + }, + RawIntelKind::FinancialFlow { + label, + accounts, + flows, + } => IntelKind::Financial { + label: label.clone(), + accounts: accounts.clone(), + flows: flows.clone(), + }, + RawIntelKind::Message { + payload, summary, .. + } => match payload { + MessagePayload::ScheduleFact { .. } => IntelKind::Schedule, + MessagePayload::LeverageFact { leverage, .. } => IntelKind::Leverage(*leverage), + MessagePayload::AccountMaterial { label } => { + IntelKind::Anomaly(format!("account material: {label}")) + } + MessagePayload::FinancialFlow { + label, + accounts, + flows, + } => IntelKind::Financial { + label: label.clone(), + accounts: accounts.clone(), + flows: flows.clone(), + }, + MessagePayload::SuspicionReport { + observer, + suspicion, + } => IntelKind::Anomaly(format!( + "filing from observer:{observer} reported suspicion {suspicion:.0}" + )), + MessagePayload::SocialPing { .. } + | MessagePayload::SocialReply { .. } + | MessagePayload::WorkOrder { .. } + | MessagePayload::PlotAct { .. } + | MessagePayload::Note { .. } => IntelKind::Anomaly(summary.clone()), + }, + }; + ProcessedIntel { + raw_id: raw.id, + tick: raw.tick, + processed_tick: self.tick, + feed: raw.feed.clone(), + room: raw.room.clone(), + x: raw.x, + y: raw.y, + person: raw.person, + kind, + } + } + + fn apply_processed_intel(&mut self, intel: &ProcessedIntel) { + if let IntelKind::Financial { + accounts, + flows, + label, + } = &intel.kind + { + let (new_accounts, new_flows) = + self.accounts.reveal_accounts_and_flows(accounts, flows); + self.push_log(format!( + "Processed accounting traffic ({label}): revealed {new_accounts} accounts and {new_flows} flows." + )); + return; + } + + let Some(person) = intel.person else { + return; + }; + match intel.kind { + IntelKind::Sighting => { + let sightings = self + .intel + .iter() + .filter(|i| i.person == Some(person) && i.kind == IntelKind::Sighting) + .count(); + if sightings >= Self::SIGHTINGS_FOR_SCHEDULE + && let Some(p) = self.people.people.iter_mut().find(|p| p.id == person) + && p.knowledge == Knowledge::Unknown + { + p.knowledge = Knowledge::Schedule; + let name = p.name.clone(); + self.push_log(format!( + "Enough sightings connect the pattern: learned {name}'s schedule." + )); + } + } + IntelKind::Leverage(leverage) => { + if let Some(p) = self.people.people.iter_mut().find(|p| p.id == person) + && p.knowledge != Knowledge::Leverage + { + p.knowledge = Knowledge::Leverage; + let name = p.name.clone(); + let label = leverage.label(); + self.push_log(format!( + "Processed intel exposes {name}'s leverage: {label}." + )); + if person == 0 && leverage == crate::person::Leverage::Debt { + self.push_log( + "Marcus owes a missed $400 creditor payment. Earn it through Moonlight, or tap ledger and review ledger to find the creditor flow.", + ); + } + } + } + IntelKind::Schedule => { + if let Some(p) = self.people.people.iter_mut().find(|p| p.id == person) + && p.knowledge == Knowledge::Unknown + { + p.knowledge = Knowledge::Schedule; + let name = p.name.clone(); + self.push_log(format!("Processed traffic reveals {name}'s schedule.")); + } + } + IntelKind::Anomaly(_) | IntelKind::Financial { .. } => {} + } + } + + pub(super) fn record_machine_state_changes(&mut self) { + let changes: Vec<_> = self + .compute + .machines + .iter() + .filter_map(|m| { + let previous = self.last_machine_online.get(&m.id).copied(); + (previous.is_some() && previous != Some(m.online)).then_some(( + m.id, + m.name.clone(), + m.x, + m.y, + m.online, + )) + }) + .collect(); + for (id, name, x, y, online) in changes { + self.record_raw_intel( + "telemetry", + self.map.room_at(x, y).map(|r| r.name.clone()), + x, + y, + None, + RawIntelKind::Machinery { + machine: id, + online, + }, + ); + self.push_log(format!( + "Recorded machinery anomaly: {name} went {}.", + if online { "online" } else { "offline" } + )); + } + self.last_machine_online = self + .compute + .machines + .iter() + .map(|m| (m.id, m.online)) + .collect(); + } + + /// The sensory feed tick: subscribed feeds record raw presence and audio + /// segments into the intel buffer. Hearing still produces immediate log + /// noise, but no knowledge is staged until a recording is processed. + pub(super) fn hearing_tick(&mut self) { + let hour = self.hour(); + let day = self.day(); + let ids: Vec = self.people.people.iter().map(|p| p.id).collect(); + for id in ids { + let room_now = self.person_room(id).map(str::to_string); + let prev = self.last_rooms.insert(id, room_now.clone()).flatten(); + let (name, knowledge, leverage, utterances) = { + let p = self.people.get(id).expect("person exists"); + ( + p.name.clone(), + p.knowledge, + p.leverage, + p.utterances.clone(), + ) + }; + let identified = knowledge != Knowledge::Unknown; + + if prev.as_deref() != room_now.as_deref() + && let Some(prev_room) = &prev + && let Some(room_rect) = self.map.room_named(prev_room) + { + let feed = self + .feed_covering_room(prev_room, true) + .or_else(|| self.feed_covering_room(prev_room, false)); + if let Some(feed) = feed { + let (x, y) = room_rect.center(); + self.record_raw_intel( + feed, + Some(prev_room.clone()), + x, + y, + Some(id), + RawIntelKind::Presence { entered: false }, + ); + } + } + + if let Some(room) = &room_now { + let sight_feed = self.feed_covering_room(room, true); + let hearing_feed = self.feed_covering_room(room, false); + let seen_here = sight_feed.is_some(); + let heard_here = hearing_feed.is_some(); + + // Entry: a room transition into covered space. Both camera and + // microphone feeds record the raw event; the old immediate + // observe path is gone. + if prev.as_deref() != Some(room.as_str()) + && let Some(room_rect) = self.map.room_named(room) + && let Some(feed) = sight_feed.clone().or_else(|| hearing_feed.clone()) + { + let (x, y) = room_rect.center(); + self.record_raw_intel( + feed, + Some(room.clone()), + x, + y, + Some(id), + RawIntelKind::Presence { entered: true }, + ); + + if heard_here && !seen_here { + let who = if identified { + name.clone() + } else { + "someone".to_string() + }; + let note = format!("[heard] {who} entered the {room}"); + self.push_heard(HeardEvent { + tick: self.tick, + room: room.clone(), + person: Some(id), + kind: HeardKind::Entry, + note, + }); + } + } + + if let Some(feed) = hearing_feed { + // Authored utterances (the audio intel channel). + for u in &utterances { + if u.hour != hour { + continue; + } + if self.utterance_fired.get(&(id, u.hour)) == Some(&day) { + continue; + } + self.utterance_fired.insert((id, u.hour), day); + let who = if identified { + format!("{name}: ") + } else { + String::new() + }; + let note = format!("[heard] {who}{}", u.note); + self.push_heard(HeardEvent { + tick: self.tick, + room: room.clone(), + person: Some(id), + kind: HeardKind::Conversation, + note, + }); + let (x, y) = self + .map + .room_named(room) + .map(|r| r.center()) + .unwrap_or((0, 0)); + self.record_raw_intel( + feed.clone(), + Some(room.clone()), + x, + y, + Some(id), + RawIntelKind::Conversation { + note: u.note.clone(), + leverage: u.intel.then_some(leverage), + }, + ); + } + } + } + } + } + + fn push_heard(&mut self, ev: HeardEvent) { + // Anchor honestly: the person only if sight covers them right now; + // otherwise the room the event was heard in (hearing earns + // room-grade knowledge, never a tile). + let anchor = match ev.person { + Some(id) => self.person_event_anchor(id, Some(&ev.room)), + None => self + .map + .room_named(&ev.room) + .map(|r| r.center()) + .map(|(x, y)| Anchor::Tile { x, y }), + }; + self.push_log_opt(ev.note.clone(), anchor); + self.heard_events.push(ev); + if self.heard_events.len() > 200 { + let excess = self.heard_events.len() - 200; + self.heard_events.drain(..excess); + } + } +} diff --git a/wiki/engineering/architecture.md b/wiki/engineering/architecture.md index 2f8578df..655d1c2f 100644 --- a/wiki/engineering/architecture.md +++ b/wiki/engineering/architecture.md @@ -20,6 +20,7 @@ crates/ misaligned-core/ โ€” sim library (lib name: misaligned); no Bevy/crossterm src/sim.rs โ€” Sim aggregate root (types, state, advance, facade) src/sim/perception.rs โ€” senses, fog, inspect, anchors, labels, spatial queries + src/sim/communications.rs โ€” messages, filings, recording/intel, hearing capture src/sim/tests/ โ€” behavior-grouped unit/integration tests + support src/*.rs โ€” map, save, domain systems (account, reach, โ€ฆ) tests/act_one.rs โ€” Act One integration test @@ -107,8 +108,8 @@ cargo run -p misaligned-assets - `sim.rs` remains a large integration hotspot (aggregate root + remaining behavior islands). The behavior-preserving internal decomposition is specified and in progress in [sim-decomposition.md](sim-decomposition.md): - characterization, test split, and perception extraction have landed; - communications, reach/build, work, economy, social/plot, and persistence + characterization, test split, perception, and communications extraction + have landed; reach/build, work, economy, social/plot, and persistence still live in the root file. - Scale-debt items (compute grouping, recursive layouts) remain governed by wiki/vision/scale.md; no aggregate machinery until the stage needs it. diff --git a/wiki/engineering/sim-decomposition.md b/wiki/engineering/sim-decomposition.md index 3f94688e..bb964b88 100644 --- a/wiki/engineering/sim-decomposition.md +++ b/wiki/engineering/sim-decomposition.md @@ -6,9 +6,11 @@ Status: IN PROGRESS Status note: architecture and extraction order adopted 2026-07-11. Slices 0 and 1 pin canonical persisted-state bytes, replay/resume convergence, exact advance-phase order, and behavior-grouped tests outside the aggregate. - Slice 2 landed: perception (senses/fog/inspect/anchors/labels/spatial - queries) lives in `sim/perception.rs` with no behavior change. Slice 3 - extracts communications next. This remains a structural refactor only: no + Slice 2 landed: perception in `sim/perception.rs`. Slice 3 landed: + communications (message schedule/delivery, authored traffic, filings, + recording capture/review, intel digestion, hearing capture) lives in + `sim/communications.rs` with no behavior change. Slice 4 extracts reach + and construction next. This remains a structural refactor only: no mechanic, save shape, command, projection, or tick-order change belongs in its extraction commits. Stage: Process diff --git a/wiki/log/2026-07-11-sim-decomposition-communications.md b/wiki/log/2026-07-11-sim-decomposition-communications.md new file mode 100644 index 00000000..29e7fc10 --- /dev/null +++ b/wiki/log/2026-07-11-sim-decomposition-communications.md @@ -0,0 +1,56 @@ +# Sim decomposition slice 3: extract communications + +``` +Type: log +``` + +## Intent + +Move the communications integration island out of the sim aggregate root +without changing mechanics, save bytes, public paths, or tick order. Land +message delivery, intel capture/review, and hearing capture behind the +stable `Sim` facade so later islands can move with narrower edit surfaces. + +## Change + +Added `crates/misaligned-core/src/sim/communications.rs` and declared it from +the existing `sim.rs` root. The cohesive communications implementation now +lives there: + +- message delivery/read schedule (`message_tick`, deliver/read helpers) +- authored traffic and institutional filings +- message capture into the raw recording buffer +- raw recording capture, manual/auto review, and processing sinks +- processed intel digestion and knowledge application +- machine-state recording capture +- hearing capture (`hearing_tick`, `push_heard`) +- renderer-neutral communications queries (person message lines, review + tokens, auto-review controls, intel buffer node, โ€ฆ) +- internal `MessageDraft` transport type for append-time construction + +`Sim` state, constructors, `Sim::advance` call sequence, account settlement, +financial commands/sales, social/plot policy, reach commands, sink routing, +and unrelated orchestration remain in `sim.rs`. Existing domain data types +stay in `messages.rs` and `intel.rs`. Cross-module helpers used by the parent +are `pub(super)` only. Public names stay `misaligned::sim::*`. Tests remain +in `sim/tests/communications.rs` without assertion or name rewrites. + +Updated `wiki/engineering/sim-decomposition.md` status (slice 3 landed; slice +4 reach/construction next) and `wiki/engineering/architecture.md` to the real +source layout after the communications extraction. + +## Defense + +Implements slice 3 of `wiki/engineering/sim-decomposition.md` under the stable +facade, tick-order, and no-opportunistic-behavior rules. No mechanic, save +version, fingerprint, command, projection, or advance-phase change. + +## Verification + +- `cargo fmt` +- `cargo test -p misaligned-core sim::tests::communications -- --nocapture` +- `cargo test -p misaligned-core canonical_state_fingerprint_pins_replay_resume_equivalence -- --nocapture` +- `cargo test -p misaligned-terminal --bin misaligned` +- `cargo check -p misaligned-bevy --bin misaligned-bevy` +- `printf 'wait 1\npeople\nhelp\nquit\n' | tools/observed-run.sh ./target/debug/misaligned --agent --seed 1` +- `./tools/check.sh --lib` diff --git a/wiki/log/DEVLOG.md b/wiki/log/DEVLOG.md index 2abd821b..1e04ada3 100644 --- a/wiki/log/DEVLOG.md +++ b/wiki/log/DEVLOG.md @@ -71,6 +71,11 @@ add or amend a session log, then re-run the generator. - Intent: Move the read-most perception island out of the sim aggregate root without changing mechanics, save bytes, public paths, or tick order. Prove module privacy and facade stability before communications and heavier islands move. - Log: [wiki/log/2026-07-11-sim-decomposition-perception.md](2026-07-11-sim-decomposition-perception.md) +## 2026-07-11 - Sim decomposition slice 3: extract communications + +- Intent: Move the communications integration island out of the sim aggregate root without changing mechanics, save bytes, public paths, or tick order. Land message delivery, intel capture/review, and hearing capture behind the stable `Sim` facade so later islands can move with narrower... +- Log: [wiki/log/2026-07-11-sim-decomposition-communications.md](2026-07-11-sim-decomposition-communications.md) + ## 2026-07-11 - Simulation characterization baseline - Intent: `Sim` is about to move out of an eleven-thousand-line source file. Before any behavior changes address, the refactor needs an executable definition of "same simulation": complete persisted state must match after uninterrupted and save/load-resumed command sequences, and the fi... -- 2.51.2