feat(chat): tier-3 reassembly accumulator in the chat service

Second half of tier-3 larger messages: the receive side. ChatService now buffers
the slices of a multi-transaction message (keyed by the shared message id) as
they arrive — across transactions and across scans — and reassembles the whole
ciphertext once every chunk is present, then decrypts and surfaces it as one
message. The buffer is NOT persisted: a rescan re-sends every slice, so it
re-derives from chain exactly like the control reducer, and survives a restart
mid-receive (the next full scan re-feeds the siblings and completes it).

Ingest's per-message build/decrypt/guard logic is refactored into one shared
lambda so a reassembled tier-3 message passes the identical block / on-chain-hide
/ sensitive-vault / tombstone checks as a single-tx message. A completed group is
retired after it surfaces (deduped across scans by its stable canonical txid —
the smallest sibling txid); a group withheld only because the sensitive vault is
locked is kept so a later unlock + scan completes it. Our own slices (sender ==
us) are skipped — the local echo records the outgoing message. The pending map is
bounded (256 groups) against a flood of never-completing partials.

Unit-tested (testHushChatTier3Reassembly): a ~9000-char message's first slice
alone surfaces nothing; once every slice has arrived it reassembles, decrypts to
the original and surfaces once; a re-scan dedups; our own slices never ingest.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

src/chat/chat_service.cpp

@@ -39,25 +39,14 @@ int ChatService::ingest(const std::vector<HushChatTransactionMetadata>& metadata
 
     int added = 0;
     std::vector<ChatMessage> fresh;   // persisted once after the loop (one transaction)
-    for (const auto& meta : metadata) {
-        // A self-addressed control signal (delete / restore / ...). Reduce it and move on: it is never a
-        // message, and it must be processed even though WE are its sender — unlike ordinary self-sent memos,
-        // which the guard below skips. The label (peer hint) rides the payload.
-        if (meta.type == HushChatHeaderType::Control) {
-            const auto ctlTimeIt = txTimestamps.find(meta.txid);
-            const std::int64_t ctlRef = ctlTimeIt != txTimestamps.end() ? ctlTimeIt->second : fallbackTimestamp;
-            applyControl(meta.conversation_id, meta.op, meta.payload_memo,
-                         meta.sent_at > 0 ? meta.sent_at : ctlRef, meta.txid);
-            continue;
-        }
+
+    // Build, decrypt (Message) and store one piece of message metadata — a single-transaction message,
+    // or a tier-3 message reassembled from its slices below. Shared so both apply the same guards exactly.
+    auto processMessage = [&](const HushChatTransactionMetadata& meta) {
         // Already stored: append would refuse it. Every scan re-sends the whole history, so this skips the
-        // decrypt for all but the new messages.
-        if (store_.contains(meta.txid, meta.payload_position)) continue;
-        // A memo whose sender is our OWN identity is something we sent (only we hold our key). The local
-        // echo already records it as outgoing — ingesting it as incoming would duplicate it as a phantom
-        // "from peer" message. (This also collapses same-seed self-chat, where the "peer" wallet shares
-        // our identity, so every message would otherwise loop back.)
-        if (!myPubKey.empty() && meta.sender_public_key_hex == myPubKey) continue;
+        // decrypt for all but the new messages. For a reassembled tier-3 message this keys off its stable
+        // canonical txid, deduping it across scans.
+        if (store_.contains(meta.txid, meta.payload_position)) return;
 
         ChatMessage message;
         message.direction = ChatDirection::Incoming;
@@ -86,13 +75,13 @@ int ChatService::ingest(const std::vector<HushChatTransactionMetadata>& metadata
         // is dropped outright (old + future messages) until unblocked; a tombstoned (txid,position) was
         // deleted with "revive on new message", so only that exact message is skipped — a new message
         // in the same conversation has a different txid and flows through normally.
-        if (blocked_pred_ && blocked_pred_(message.conversation_id)) continue;
-        if (conversationHiddenOnChain(message.conversation_id)) continue;  // hidden by an on-chain delete
+        if (blocked_pred_ && blocked_pred_(message.conversation_id)) return;
+        if (conversationHiddenOnChain(message.conversation_id)) return;  // hidden by an on-chain delete
         // Withhold a sensitive conversation's messages while its local vault key is absent (locked vault or
         // a fresh seed import). Fail-closed: if we can't store it under the vault key we don't surface it.
         // They re-import and persist under the vault key once it is unlocked and a scan runs.
-        if (isConversationSensitive(message.conversation_id) && !(db_ && db_->hasSensitiveKey())) continue;
-        if (db_ && db_->isTombstoned(message)) continue;
+        if (isConversationSensitive(message.conversation_id) && !(db_ && db_->hasSensitiveKey())) return;
+        if (db_ && db_->isTombstoned(message)) return;
 
         if (meta.type == HushChatHeaderType::ContactRequest) {
             message.kind = ChatMessageKind::ContactRequest;
@@ -103,7 +92,7 @@ int ChatService::ingest(const std::vector<HushChatTransactionMetadata>& metadata
             // different header with this payload gets its own try.
             const std::string failKey = ChatStore::messageKey(meta.txid, meta.payload_position) + "|" +
                                         meta.sender_public_key_hex + "|" + meta.secretstream_header_hex;
-            if (undecryptable_.count(failKey)) continue;
+            if (undecryptable_.count(failKey)) return;
             ++decrypt_attempts_;
             std::string plaintext;
             const ChatCryptoStatus status = decryptIncoming(
@@ -114,7 +103,7 @@ int ChatService::ingest(const std::vector<HushChatTransactionMetadata>& metadata
                     if (undecryptable_.size() >= 4096) undecryptable_.clear();
                     undecryptable_.insert(failKey);
                 }
-                continue;
+                return;
             }
             message.body = std::move(plaintext);
         }
@@ -129,7 +118,84 @@ int ChatService::ingest(const std::vector<HushChatTransactionMetadata>& metadata
             // relying on a seen-watermark delta (which block-time vs wall-clock skew can swallow).
             if (newIncomingCids) newIncomingCids->push_back(message.conversation_id);
         }
+    };
+
+    for (const auto& meta : metadata) {
+        // A self-addressed control signal (delete / restore / ...). Reduce it and move on: it is never a
+        // message, and it must be processed even though WE are its sender — unlike ordinary self-sent memos,
+        // which the guard below skips. The label (peer hint) rides the payload.
+        if (meta.type == HushChatHeaderType::Control) {
+            const auto ctlTimeIt = txTimestamps.find(meta.txid);
+            const std::int64_t ctlRef = ctlTimeIt != txTimestamps.end() ? ctlTimeIt->second : fallbackTimestamp;
+            applyControl(meta.conversation_id, meta.op, meta.payload_memo,
+                         meta.sent_at > 0 ? meta.sent_at : ctlRef, meta.txid);
+            continue;
+        }
+        // A memo whose sender is our OWN identity is something we sent (only we hold our key). The local
+        // echo already records it as outgoing — ingesting it as incoming would duplicate it as a phantom
+        // "from peer" message (this also collapses same-seed self-chat). Skips our own tier-3 slices too.
+        if (!myPubKey.empty() && meta.sender_public_key_hex == myPubKey) continue;
+
+        // Larger-messages tier 3: a slice of a multi-transaction message. Accumulate it by message id;
+        // the whole message is reassembled below once every chunk has arrived (across txs and scans).
+        if (!meta.message_id.empty()) {
+            if (meta.total_parts <= 0 || meta.total_parts > kHushChatMaxMessageChunks) continue;  // sanity
+            // Bound the buffer against a flood of never-completing partials (re-fed from chain, so a clear
+            // only delays, never loses, a genuine message).
+            if (tier3_pending_.size() > 256 && !tier3_pending_.count(meta.message_id)) tier3_pending_.clear();
+            Tier3Pending& g = tier3_pending_[meta.message_id];
+            if (g.chunks.empty()) {   // first slice seen: capture the shared header fields
+                g.total_parts = meta.total_parts;
+                g.secretstream_header_hex = meta.secretstream_header_hex;
+                g.sender_public_key_hex = meta.sender_public_key_hex;
+                g.conversation_id = meta.conversation_id;
+                g.reply_zaddr = meta.reply_zaddr;
+                g.sent_at = meta.sent_at;
+                g.canonical_txid = meta.txid;
+            }
+            if (meta.txid < g.canonical_txid) g.canonical_txid = meta.txid;   // deterministic stable key
+            for (const auto& c : meta.partial_chunks)
+                if (c.first >= 0 && c.first < g.total_parts) g.chunks[c.first] = c.second;
+            continue;
+        }
+
+        processMessage(meta);
     }
+
+    // Finalize every tier-3 message whose chunks have all arrived: reassemble the one ciphertext and run
+    // it through the shared message path (decrypt + the block/hide/sensitive guards). Retire the group
+    // afterwards unless it was withheld only because the sensitive vault is still locked.
+    {
+        std::vector<std::string> retire;
+        for (auto& entry : tier3_pending_) {
+            Tier3Pending& g = entry.second;
+            if (g.total_parts <= 0 || static_cast<int>(g.chunks.size()) != g.total_parts) continue;
+            bool complete = true;
+            for (int i = 0; i < g.total_parts; ++i) if (!g.chunks.count(i)) { complete = false; break; }
+            if (!complete) continue;
+            if (store_.contains(g.canonical_txid, 0)) { retire.push_back(entry.first); continue; }  // already have it
+
+            HushChatTransactionMetadata synth;
+            synth.type = HushChatHeaderType::Message;
+            synth.txid = g.canonical_txid;
+            synth.payload_position = 0;
+            synth.conversation_id = g.conversation_id;
+            synth.reply_zaddr = g.reply_zaddr;
+            synth.sender_public_key_hex = g.sender_public_key_hex;
+            synth.secretstream_header_hex = g.secretstream_header_hex;
+            synth.sent_at = g.sent_at;
+            std::string full;
+            for (int i = 0; i < g.total_parts; ++i) full += g.chunks[i];
+            synth.payload_memo = std::move(full);
+            processMessage(synth);
+
+            const bool withheldSensitive =
+                isConversationSensitive(g.conversation_id) && !(db_ && db_->hasSensitiveKey());
+            if (!withheldSensitive) retire.push_back(entry.first);
+        }
+        for (const auto& mid : retire) tier3_pending_.erase(mid);
+    }
+
     if (db_ && !fresh.empty()) db_->appendBatch(fresh);
     // Enforce on-chain deletions against the store view: erase any conversation whose latest control op is
     // "del". This catches rows appended earlier in this very batch (before their delete signal was seen in

src/chat/chat_service.h

@@ -153,6 +153,22 @@ private:
     };
     std::unordered_map<std::string, ControlState> control_;            // hide axis: del / res
     std::unordered_map<std::string, ControlState> sensitive_control_;  // sensitive axis: sensitive / unsensitive
+
+    // Larger-messages tier 3: a big message split across several transactions arrives one slice at a
+    // time (siblings share a message id). We accumulate the slices here until every chunk is present,
+    // then reassemble + decrypt + emit the whole message. Not persisted — a rescan re-sends every slice
+    // so the buffer re-derives from chain (like the control reducer); a completed group is retired.
+    struct Tier3Pending {
+        int total_parts = 0;
+        std::string secretstream_header_hex;        // "e", shared by every slice
+        std::string sender_public_key_hex;          // "p", shared
+        std::string conversation_id;
+        std::string reply_zaddr;
+        std::int64_t sent_at = 0;
+        std::string canonical_txid;                 // smallest sibling txid — the reassembled message's stable key
+        std::unordered_map<int, std::string> chunks;  // global index -> ciphertext-hex chunk
+    };
+    std::unordered_map<std::string, Tier3Pending> tier3_pending_;  // message id -> accumulating slices
     // Monotonic reduce: route the op to its axis (hide vs sensitive), ignore unknown ops (forward-compat),
     // and within an axis a later (timestamp, txid) wins so a replay of an older op never flips the state.
     void applyControl(const std::string& cid, const std::string& op, const std::string& label,

tests/test_phase4.cpp

@@ -15827,6 +15827,65 @@ void testHushChatLargeMessageTier3()
                                      cid, tooHuge, over) == ChatComposeStatus::TooLong);
 }
 
+// Tier 3 through the ChatService: slices arriving across separate ingests (separate scans) accumulate
+// until the message is whole, then it surfaces once; a re-scan of every slice dedups it.
+void testHushChatTier3Reassembly()
+{
+    using namespace dragonx::chat;
+    ChatKeyPair alice, bob;
+    ChatIdentityResult ra = deriveChatIdentityFromSecret("t3svc-alice", alice, true);
+    ChatIdentityResult rb = deriveChatIdentityFromSecret("t3svc-bob", bob, true);
+    EXPECT_TRUE(ra.status == ChatIdentityStatus::Ready && rb.status == ChatIdentityStatus::Ready);
+    const std::string cid = "conv-t3svc";
+
+    std::string big;
+    for (int i = 0; i < 9000; ++i) big += static_cast<char>('a' + (i % 26));
+
+    std::vector<OutgoingChatMemos> txs;
+    EXPECT_TRUE(splitOutgoingMessage(alice, ra.public_key_hex, "zs-alice", rb.public_key_hex, "zs-bob",
+                                     cid, big, txs) == ChatComposeStatus::Ok);
+    EXPECT_TRUE((int)txs.size() >= 3);
+
+    // Extract each transaction into partial metadata (outputs reversed to mimic the daemon shuffle).
+    std::vector<HushChatTransactionMetadata> all;
+    for (std::size_t t = 0; t < txs.size(); ++t) {
+        std::vector<std::string> wire{ txs[t].headerMemo };
+        for (const auto& pm : txs[t].payloadMemos) wire.push_back(pm);
+        std::reverse(wire.begin(), wire.end());
+        HushChatTransactionInput tx; tx.txid = "t3svc-" + std::to_string(t);
+        for (std::size_t i = 0; i < wire.size(); ++i) tx.outputs.push_back({i, wire[i]});
+        auto md = extractHushChatTransactionMetadata(tx, true).metadata;
+        for (auto& m : md) all.push_back(std::move(m));
+    }
+    EXPECT_TRUE(all.size() == txs.size());   // one partial metadata per transaction
+
+    ChatService svc;
+    svc.setIdentity(bob);
+
+    // Stage 1: only the FIRST transaction's slice has arrived — the message is incomplete, nothing surfaces.
+    std::vector<HushChatTransactionMetadata> firstOnly(all.begin(), all.begin() + 1);
+    EXPECT_EQ(svc.ingest(firstOnly, {}, 1000), 0);
+    EXPECT_EQ((int)svc.store().conversation(cid).size(), 0);
+
+    // Stage 2: the whole history re-arrives (every slice) — reassembled, decrypted and surfaced once.
+    EXPECT_EQ(svc.ingest(all, {}, 1000), 1);
+    std::vector<ChatMessage> conv = svc.store().conversation(cid);
+    EXPECT_EQ((int)conv.size(), 1);
+    EXPECT_EQ(conv[0].body, big);
+    EXPECT_TRUE(conv[0].direction == ChatDirection::Incoming);
+    EXPECT_EQ(conv[0].peer_public_key_hex, ra.public_key_hex);
+
+    // A re-scan of the same slices does not duplicate the message.
+    EXPECT_EQ(svc.ingest(all, {}, 1000), 0);
+    EXPECT_EQ((int)svc.store().conversation(cid).size(), 1);
+
+    // Our OWN tier-3 slices (sender == us) are never ingested as incoming — the echo records them.
+    ChatService selfSvc;
+    selfSvc.setIdentity(alice);
+    EXPECT_EQ(selfSvc.ingest(all, {}, 1000), 0);
+    EXPECT_EQ((int)selfSvc.store().conversation(cid).size(), 0);
+}
+
 void testHushChatOutgoing()
 {
     using namespace dragonx::chat;
@@ -20438,6 +20497,7 @@ int main()
     testHushChatOutgoing();
     testHushChatLargeMessage();
     testHushChatLargeMessageTier3();
+    testHushChatTier3Reassembly();
     testHushChatTransport();
     testHushChatShuffledReceive();
     testWalletFileProbe();