feat(digsex): wire the dig-sex economic layer into the node

Cargo.toml

@@ -32,7 +32,7 @@ edition = "2021"
 # the ROOT manifest (`[workspace.package].version`), so it MUST be set here for a
 # release to fire (§3.6). The library crates (dig-node-core/dig-runtime/dig-wallet)
 # keep their own independent versions — only the released binary tracks the workspace version.
-version = "0.190.0"
+version = "0.194.0"
 
 # Release hardening, matching digstore: keep integer-overflow checks ON in release.
 # The node parses untrusted serialized input and does offset/length arithmetic over

crates/dig-node-core/Cargo.toml

@@ -30,7 +30,7 @@ name = "dig-node-core"
 # dig-node#276/#296). Changing a public return type is BREAKING for an out-of-workspace implementor;
 # this crate is consumed in-workspace only and is pre-1.0, so it is a MINOR bump under SemVer's 0.x
 # rule -- recorded here rather than letting the number imply the locator surface held still.
-version = "0.64.0"
+version = "0.65.0"
 edition = "2021"
 license = "GPL-2.0-only"
 description = "The canonical DIG node ENGINE library (crate `dig_node_core`): the JSON-RPC dispatch (`handle_rpc`, the same contract as rpc.dig.net), local-first content serve/fetch/redirect from LOCAL .dig store modules (via digstore_host::serve_blind), chain-anchored-root resolution, chain-watch + subscriptions + generation gap-fill, the LRU cache, and the full P2P stack. Shared UNCHANGED by both host shells: the `dig-node` OS-service binary (dig-node-service) and the DIG Browser's in-process cdylib (dig-runtime). Native Rust so the compiled-module serve path works."

crates/dig-node-core/src/download.rs

@@ -856,6 +856,15 @@ pub struct NodeContent {
     /// at all, because pool membership is its liveness gate. Folding them together would make one
     /// structure whose key, lifetime and eviction rule all mean two different things at once.
     ask_routing: AskRoutingState,
+    /// What this node has observed each pool peer DO, and the dial share that earns them (#268).
+    ///
+    /// Held beside [`Self::ask_routing`] because the two read the same verified identity and are
+    /// bounded by the same pool: routing ranks the peers worth asking FIRST, conduct decides which are
+    /// worth asking AT ALL. Node-local and never gossiped — a shared reputation channel would be a
+    /// defamation primitive.
+    conduct: crate::seams::dig_peer::conduct::ConductState,
+    /// The instant conduct ticks are measured from. Wall-clock-free and monotonic.
+    conduct_epoch: std::time::Instant,
 }
 
 /// What a holder search ESTABLISHED — the records it found AND whether an empty result is a fact.
@@ -1357,6 +1366,8 @@ impl NodeContent {
             ask_seen: AskSeenSet::new(),
             onion_relay: std::sync::atomic::AtomicBool::new(onion_relay_from_env()),
             ask_routing,
+            conduct: crate::seams::dig_peer::conduct::ConductState::new(),
+            conduct_epoch: std::time::Instant::now(),
         })
     }
 
@@ -1842,15 +1853,22 @@ impl NodeContent {
             .filter_map(|(peer, _)| RoutedPeer::from_pool_key(peer))
             .collect();
 
+        // CONDUCT gates who is dialable at all (#268, SPEC 8.3); routing then ranks what remains.
+        // The order matters: ranking a peer this node has PROVEN dishonest would still spend a dial on
+        // it whenever the ranking happened to favour it. A peer excluded here has a verifiable fault
+        // against it — a lie or a self-contradiction — and never merely a slow or silent history,
+        // which `dial_share` floors above zero precisely so distress cannot evict an honest holder.
+        let dialable = self.conduct.dialable(&routable, self.conduct_ticks());
+
         // Ranked by what THIS node has observed, not by the pool `HashMap`'s arbitrary order — and the
-        // observations of peers no longer in `routable` are dropped in the same call, so a cycled-away
+        // observations of peers no longer in `dialable` are dropped in the same call, so a cycled-away
         // peer leaves this node's memory when it leaves the pool.
         let decision = self.ask_routing.decide(
             &config,
             asker,
             budget.remaining(config.hop_cap),
             me,
-            &routable,
+            &dialable,
             self.relay_rate_limiter.check(requestor),
         );
 
@@ -1938,6 +1956,30 @@ impl NodeContent {
             // The ONLY writer of this node's routing memory, fed an outcome this node classified from
             // an exchange it issued and saw complete (dig_ecosystem#3129).
             self.ask_routing.record(routed, &outcome, started.elapsed());
+            // The SAME exchange, classified for conduct (#268). An outcome where the peer answered —
+            // including an honest "I do not have it" — is an `HonestAnswer`, because SPEC 8.2A
+            // requires that answering is never worse than staying silent. A refusal, a timeout and an
+            // unreachable peer are `NonPerformance`: unverifiable, decaying, and floored, since none
+            // of them can be distinguished from distress an attacker induced in an honest peer.
+            //
+            // Neither VERIFIABLE class is produced here, and deliberately so. A `ProvenLie` needs
+            // bytes that failed verification against the anchor, attributed to the peer that supplied
+            // them; that attribution happens inside `dig-download`'s engine against `chunk_hashes` and
+            // is not surfaced per-peer to this node (see the report on #268). Claiming one from a
+            // transport error would brand an honest peer on unverifiable evidence, which is the exact
+            // conflation SPEC 8.2A exists to prevent.
+            self.conduct.observe(
+                routed,
+                match outcome {
+                    AskOutcome::Answered(_) | AskOutcome::AnsweredInconclusive(_) => {
+                        dig_sex::ConductEvidence::HonestAnswer
+                    }
+                    AskOutcome::Refused | AskOutcome::TimedOut | AskOutcome::Unreachable => {
+                        dig_sex::ConductEvidence::NonPerformance
+                    }
+                },
+                self.conduct_ticks(),
+            );
             match outcome {
                 AskOutcome::Answered(records) => answers.records.extend(records),
                 // The peer answered and told us its OWN subtree did not finish. Its records are
@@ -1974,6 +2016,16 @@ impl NodeContent {
             .collect()
     }
 
+    /// This node's monotonic conduct clock, in seconds since process start.
+    ///
+    /// Conduct decay is measured in elapsed ticks and nothing else, so the clock must advance on its
+    /// own — a counter incremented per exchange would mean a peer nobody dials never ages out of its
+    /// penalty, and the recovery SPEC 8.2A requires would be unreachable for exactly the peer being
+    /// punished.
+    fn conduct_ticks(&self) -> u64 {
+        self.conduct_epoch.elapsed().as_secs()
+    }
+
     /// This node's routing memory, so a test can drive the forwarded ask and then ask what the ask
     /// LEFT BEHIND. Without it the recording leg would only be observable through its effect on a
     /// later round, and a test that could not see the write directly could not tell a missing write

crates/dig-node-core/src/lib.rs

@@ -386,6 +386,12 @@ pub struct Node {
     /// registry. The registry's distinct-generation cap ([`crate::seams::dig_peer::DEFAULT_MAX_CONCURRENT_WARMS`])
     /// therefore bounds concurrent acquisitions across BOTH legs, not each in isolation.
     capsule_acquisition: Arc<crate::seams::dig_peer::WarmRegistry>,
+    /// Inbound admission for the mTLS peer surface (dig-sex SPEC 8.5, #269).
+    ///
+    /// One meter for the whole NODE, not one per connection or per responder: the ceiling it enforces
+    /// is node-wide, and a per-connection meter would let a peer buy more allowance simply by opening
+    /// more connections.
+    peer_admission: Arc<crate::seams::dig_peer::admission::PeerAdmission>,
     /// A WEAK self-reference, installed by the standalone peer-network bring-up (which holds the
     /// `Arc<Node>`), so a `&self` read handler can spawn a detached background task that needs an owned
     /// `Arc<Node>` — the capsule backfill (§14.3). `Weak` (not `Arc`) so the node's refcount is
@@ -3920,7 +3926,11 @@ impl Node {
     /// a per-request walk of the whole cache is a cost amplifier a peer controls.
     ///
     /// The batch is CAPPED at [`MAX_AVAILABILITY_ITEMS`] — the item count is caller-controlled — with
-    /// the excess simply not answered (the result array is aligned to the answered prefix).
+    /// the excess simply not answered (the result array is aligned to the answered prefix). That
+    /// truncation is the LAST line of defence, reached only by in-process callers: on the peer
+    /// surface an oversized batch never gets here, because
+    /// `NodeResponder::handle_availability` (peer.rs) meters the requested item
+    /// count against the same limit and refuses the whole request at the boundary (#269).
     ///
     /// `requestor` keys the per-item not-held → DHT `find_providers` enrichment against its
     /// per-requestor miss-lookup budget (dig_ecosystem#2007), so a large batch of not-held items from
@@ -4429,6 +4439,11 @@ impl Node {
         self.capsule_acquisition.clone()
     }
 
+    /// The node-wide inbound admission meter (dig-sex SPEC 8.5, #269).
+    pub(crate) fn peer_admission(&self) -> &Arc<crate::seams::dig_peer::admission::PeerAdmission> {
+        &self.peer_admission
+    }
+
     /// Build a node from the environment (cache dir/cap, §21 identity, upstream).
     /// Used by both the standalone bin's [`run`] and the in-process `dig-runtime`.
     pub fn from_env() -> Arc<Node> {
@@ -4506,6 +4521,7 @@ impl Node {
             content_cache: std::sync::Mutex::new(ContentCache::default()),
             inventory_refresher: OnceLock::new(),
             capsule_acquisition: Arc::new(crate::seams::dig_peer::WarmRegistry::new()),
+            peer_admission: Arc::new(crate::seams::dig_peer::admission::PeerAdmission::default()),
             verification_ledger: verification_ledger::VerificationLedger::new(),
             self_ref: OnceLock::new(),
             gossip: OnceLock::new(),
@@ -4797,6 +4813,7 @@ pub(crate) mod test_support {
             content_cache: std::sync::Mutex::new(ContentCache::default()),
             inventory_refresher: OnceLock::new(),
             capsule_acquisition: Arc::new(crate::seams::dig_peer::WarmRegistry::new()),
+            peer_admission: Arc::new(crate::seams::dig_peer::admission::PeerAdmission::default()),
             verification_ledger: verification_ledger::VerificationLedger::new(),
             self_ref: OnceLock::new(),
             gossip: OnceLock::new(),
@@ -5585,6 +5602,7 @@ mod tests {
             content_cache: std::sync::Mutex::new(ContentCache::default()),
             inventory_refresher: OnceLock::new(),
             capsule_acquisition: Arc::new(crate::seams::dig_peer::WarmRegistry::new()),
+            peer_admission: Arc::new(crate::seams::dig_peer::admission::PeerAdmission::default()),
             verification_ledger: verification_ledger::VerificationLedger::new(),
             self_ref: OnceLock::new(),
             gossip: OnceLock::new(),
@@ -5717,6 +5735,7 @@ mod tests {
             content_cache: std::sync::Mutex::new(ContentCache::default()),
             inventory_refresher: OnceLock::new(),
             capsule_acquisition: Arc::new(crate::seams::dig_peer::WarmRegistry::new()),
+            peer_admission: Arc::new(crate::seams::dig_peer::admission::PeerAdmission::default()),
             verification_ledger: verification_ledger::VerificationLedger::new(),
             self_ref: OnceLock::new(),
             gossip: OnceLock::new(),
@@ -5783,6 +5802,7 @@ mod tests {
             content_cache: std::sync::Mutex::new(ContentCache::default()),
             inventory_refresher: OnceLock::new(),
             capsule_acquisition: Arc::new(crate::seams::dig_peer::WarmRegistry::new()),
+            peer_admission: Arc::new(crate::seams::dig_peer::admission::PeerAdmission::default()),
             verification_ledger: verification_ledger::VerificationLedger::new(),
             self_ref: OnceLock::new(),
             gossip: OnceLock::new(),
@@ -5874,6 +5894,9 @@ mod tests {
                 content_cache: std::sync::Mutex::new(ContentCache::default()),
                 inventory_refresher: OnceLock::new(),
                 capsule_acquisition: Arc::new(crate::seams::dig_peer::WarmRegistry::new()),
+                peer_admission: Arc::new(
+                    crate::seams::dig_peer::admission::PeerAdmission::default(),
+                ),
                 verification_ledger: verification_ledger::VerificationLedger::new(),
                 self_ref: OnceLock::new(),
                 gossip: OnceLock::new(),
@@ -5947,6 +5970,9 @@ mod tests {
                 content_cache: std::sync::Mutex::new(ContentCache::default()),
                 inventory_refresher: OnceLock::new(),
                 capsule_acquisition: Arc::new(crate::seams::dig_peer::WarmRegistry::new()),
+                peer_admission: Arc::new(
+                    crate::seams::dig_peer::admission::PeerAdmission::default(),
+                ),
                 verification_ledger: verification_ledger::VerificationLedger::new(),
                 self_ref: OnceLock::new(),
                 gossip: OnceLock::new(),
@@ -8622,6 +8648,7 @@ mod tests {
             content_cache: std::sync::Mutex::new(ContentCache::default()),
             inventory_refresher: OnceLock::new(),
             capsule_acquisition: Arc::new(crate::seams::dig_peer::WarmRegistry::new()),
+            peer_admission: Arc::new(crate::seams::dig_peer::admission::PeerAdmission::default()),
             verification_ledger: verification_ledger::VerificationLedger::new(),
             self_ref: OnceLock::new(),
             gossip: OnceLock::new(),
@@ -10243,6 +10270,10 @@ mod tests {
         assert_eq!(arr[2]["available"], false, "unknown capsule is a miss");
     }
 
+    /// Pins the IN-PROCESS truncation only. A peer-surface batch this size is refused whole at
+    /// admission long before it reaches here, and this test cannot see that: it calls the batch
+    /// BELOW the responder that decides. The responder-level pair lives in `peer.rs`
+    /// (`the_responder_serves_a_batch_at_the_advertised_limit_and_refuses_one_past_it`).
     #[tokio::test]
     async fn availability_batch_caps_the_item_count() {
         let (node, _td) = test_node(None);