feat(vision): add realtime intelligence pipeline

* docs: refresh distribution blowup status

* build: gate ppa publishing behind repo variable

* chore: bump sourcey to 3.3.10, fixes doxygen index links and node 24 compat

* build: default ppa target to 0state

* build: add temporary launchpad signing workflow

* fix: invoke ppa publish script with bash

* chore: remove temporary launchpad signing workflow

* fix: target noble for ppa publishing

* build: validate alpine and fedora packaging

* fix: accept shared llhttp targets for system builds

* build: source conan recipe from tagged releases

* build: add conan release pinning flow

* release: prepare 2.4.1

* fix: upstream package-manager compatibility followups

* release: prepare 2.4.2

* build: finalize 2.4.2 package metadata

* debian: force release build type for ppa builds

* debian: ignore vendored dependency artifacts in ppa builds

* debian: reuse accepted orig tarball for revision uploads

* debian: build revision uploads from accepted orig source

* build: refresh rpm changelog for 2.4.2

* packaging: prepare MacPorts and Spack recipes

* bump sourcey to 3.4.2

* bump sourcey to 3.4.1

* packaging: add package helper scripts

* add icey logo to docs

* packaging: refresh Homebrew and MacPorts deps

* packaging: sync conda-forge recipe

* packaging: add openSUSE seeds

* base: export shared-library symbols on Unix

* packaging: backport Unix export fix to conda-forge recipe

* distribution: sync package recipes and Unix exports

* av: support older CoreAudio SDKs

* ci: fix packaging and CoreAudio regressions

* docs: refresh icey docs and branding

* deps: bump sourcey to 3.4.4

* build: publish icey image on release

* build: fix downstream packaging compatibility

* release: prepare 2.4.3

* build: pin release archives for 2.4.3

* ci: dispatch finalized release publishing

* add webrtc receive jitter buffer

* update docs, add graft/speech/vision modules, bump sourcey

* docs: update changelog for jitter buffer, graft/speech/vision modules

* feat(vision): add realtime intelligence pipeline

---------

Co-authored-by: 0state OSS <oss@0state.com>

Author: auscaster

docs/concepts/packetstream.md

@@ -216,6 +216,51 @@ Use this when:
 
 This is the clean way to move work, not an excuse to make the whole graph vaguely asynchronous.
 
+## Branch Topology
+
+The real-time intelligence shape is not one long linear stream.
+
+It is:
+
+```text
+decoded ingress
+    |-> delivery branch -> encode/send -> browser
+    |-> detect branch   -> sample -> async clone boundary -> detector -> events
+```
+
+That means the two branches share one decoded source, but they do not share one latency budget.
+
+- the delivery branch stays close to the current transport hot path
+- the detect branch is allowed to sample, queue, and drop stale work
+- the first explicit `Cloned` or `Retained` adapter on the detect branch is the ownership handoff
+
+In practice the detect branch should cross that boundary before any worker-thread or slow detector stage:
+
+```text
+ingress PacketStream
+    |-> delivery PacketStream
+    |-> detect PacketStream -> AsyncPacketQueue -> detector
+```
+
+That keeps the contract honest:
+
+- decode once
+- branch from decoded packets, not from encoded sender output
+- keep browser delivery independent from detector backlog
+- make the async boundary visible in the graph instead of implicit in some callback
+
+### Shutdown Order
+
+The safe teardown order for that topology is:
+
+1. stop the detect branch
+2. stop the delivery branch
+3. stop the shared source
+
+Downstream branches can stop independently. The shared source should be the last thing to stop.
+
+Late packets after shutdown are expected around queue boundaries. They should be dropped cleanly after close, not dispatched into already-closed downstream sinks.
+
 ## Real Patterns
 
 ### Webcam to browser
@@ -277,4 +322,3 @@ That is what gives the library a coherent core instead of five separate async mo
 - [Runtime Contracts](runtime-contracts.md) for the loop, signal, and ownership rules around the pipeline
 - [WebRTC](../modules/webrtc.md) for the media send and receive layers built on top of it
 - [AV](../modules/av.md) for capture, decode, encode, and mux components
-

src/base/tests/basetests.cpp

@@ -1028,6 +1028,9 @@ int main(int argc, char** argv)
     describe("packet stream async queue clone boundary", new PacketStreamAsyncQueueCloneBoundaryTest);
     describe("packet stream retention contract", new PacketStreamRetentionContractTest);
     describe("packet stream shared source branch clone boundary", new PacketStreamSharedSourceBranchCloneBoundaryTest);
+    describe("packet stream branch fanout sequence", new PacketStreamBranchFanoutSequenceTest);
+    describe("packet stream branch teardown order", new PacketStreamBranchTeardownOrderTest);
+    describe("packet stream async late drop after close", new PacketStreamAsyncLateDropAfterCloseTest);
     // describe("multi packet stream", new MultiPacketStreamTest);
 
     test::runAll();

src/base/tests/basetests.h

@@ -36,7 +36,10 @@
 #include "icy/util.h"
 
 #include <cstring>
+#include <mutex>
 #include <sstream>
+#include <thread>
+#include <vector>
 
 
 using icy::test::Test;
@@ -476,6 +479,89 @@ struct MockBurstPacketSource : public PacketSource
     }
 };
 
+
+struct BranchLifecyclePacketSource : public PacketSource
+    , public basic::Startable
+{
+    PacketSignal emitter;
+    Thread runner;
+    std::atomic<bool> running{false};
+    std::atomic<bool> started{false};
+    std::atomic<int> sent{0};
+    std::vector<std::string>* lifecycle = nullptr;
+    std::mutex* lifecycleMutex = nullptr;
+
+    BranchLifecyclePacketSource(std::vector<std::string>& lifecycleLog,
+                                std::mutex& lifecycleLogMutex)
+        : PacketSource(emitter)
+        , lifecycle(&lifecycleLog)
+        , lifecycleMutex(&lifecycleLogMutex)
+    {
+    }
+
+    ~BranchLifecyclePacketSource() override
+    {
+        stop();
+    }
+
+    void start() override
+    {
+        if (running.exchange(true))
+            return;
+
+        started = true;
+        record("source:start");
+
+        runner.start([this]() {
+            while (running.load()) {
+                const std::string payload = "pkt" + std::to_string(sent.fetch_add(1));
+                RawPacket packet(payload.data(), payload.size());
+                emitter.emit(packet);
+                icy::sleep(2);
+            }
+        });
+    }
+
+    void stop() override
+    {
+        const bool wasRunning = running.exchange(false);
+        if (wasRunning)
+            record("source:stop");
+
+        if (started.exchange(false) && Thread::currentID() != runner.tid())
+            runner.join();
+    }
+
+    void record(const std::string& entry)
+    {
+        std::lock_guard<std::mutex> guard(*lifecycleMutex);
+        lifecycle->push_back(entry);
+    }
+};
+
+
+class GatedAsyncPacketQueue : public AsyncPacketQueue<>
+{
+public:
+    explicit GatedAsyncPacketQueue(int maxSize = 16)
+        : AsyncPacketQueue<>(maxSize)
+    {
+    }
+
+    std::atomic<bool> enteredDispatch{false};
+    std::atomic<bool> releaseDispatch{false};
+
+protected:
+    void dispatch(IPacket& packet) override
+    {
+        enteredDispatch = true;
+        while (!releaseDispatch.load())
+            icy::sleep(1);
+
+        AsyncPacketQueue<>::dispatch(packet);
+    }
+};
+
 class PacketStreamTest : public Test
 {
     int numPackets;
@@ -1191,6 +1277,199 @@ class PacketStreamSharedSourceBranchCloneBoundaryTest : public Test
 };
 
 
+class PacketStreamBranchFanoutSequenceTest : public Test
+{
+    std::mutex receivedMutex;
+    std::vector<std::string> deliveryReceived;
+    std::vector<std::string> detectReceived;
+
+    static std::string packetString(IPacket& packet)
+    {
+        return std::string(packet.data(), packet.size());
+    }
+
+    void run()
+    {
+        PacketSignal ingress;
+        PacketStream deliveryBranch("delivery-branch");
+        PacketStream detectBranch("detect-branch");
+        auto detectQueue = std::make_shared<AsyncPacketQueue<>>(16);
+        const std::vector<std::string> expected = {"frame-0", "frame-1", "frame-2", "frame-3"};
+
+        deliveryBranch.attachSource(ingress);
+        detectBranch.attachSource(ingress);
+        detectBranch.attach(detectQueue, 0);
+
+        deliveryBranch.emitter += [&](IPacket& packet) {
+            std::lock_guard<std::mutex> guard(receivedMutex);
+            deliveryReceived.push_back(packetString(packet));
+        };
+        detectBranch.emitter += [&](IPacket& packet) {
+            std::lock_guard<std::mutex> guard(receivedMutex);
+            detectReceived.push_back(packetString(packet));
+        };
+
+        deliveryBranch.start();
+        detectBranch.start();
+
+        for (const auto& payload : expected) {
+            RawPacket packet(payload.data(), payload.size());
+            ingress.emit(packet);
+        }
+
+        expect(test::waitFor([&]() {
+            std::lock_guard<std::mutex> guard(receivedMutex);
+            return detectReceived.size() == expected.size();
+        }, 2000));
+
+        deliveryBranch.close();
+        detectBranch.close();
+
+        std::lock_guard<std::mutex> guard(receivedMutex);
+        expect(deliveryReceived == expected);
+        expect(detectReceived == expected);
+        expect(detectQueue->retention() == PacketRetention::Cloned);
+    }
+};
+
+
+class PacketStreamBranchTeardownOrderTest : public Test
+{
+    std::mutex lifecycleMutex;
+    std::vector<std::string> lifecycle;
+    std::atomic<int> deliveryReceived{0};
+    std::atomic<int> detectReceived{0};
+
+    void record(const std::string& entry)
+    {
+        std::lock_guard<std::mutex> guard(lifecycleMutex);
+        lifecycle.push_back(entry);
+    }
+
+    int indexOf(const std::string& entry)
+    {
+        std::lock_guard<std::mutex> guard(lifecycleMutex);
+        for (size_t index = 0; index < lifecycle.size(); ++index) {
+            if (lifecycle[index] == entry)
+                return static_cast<int>(index);
+        }
+
+        return -1;
+    }
+
+    void run()
+    {
+        PacketStream sourceStream("ingress-source");
+        auto source = std::make_shared<BranchLifecyclePacketSource>(lifecycle, lifecycleMutex);
+        PacketStream deliveryBranch("delivery-branch");
+        PacketStream detectBranch("detect-branch");
+        auto detectQueue = std::make_shared<AsyncPacketQueue<>>(16);
+
+        sourceStream.attachSource(source, true);
+        deliveryBranch.attachSource(sourceStream.emitter);
+        detectBranch.attachSource(sourceStream.emitter);
+        detectBranch.attach(detectQueue, 0);
+
+        deliveryBranch.emitter += [&](IPacket&) {
+            ++deliveryReceived;
+        };
+        detectBranch.emitter += [&](IPacket&) {
+            ++detectReceived;
+        };
+
+        deliveryBranch.StateChange += [&](void*, PacketStreamState& state, const PacketStreamState&) {
+            if (state.id() == PacketStreamState::Closed)
+                record("delivery:closed");
+        };
+        detectBranch.StateChange += [&](void*, PacketStreamState& state, const PacketStreamState&) {
+            if (state.id() == PacketStreamState::Closed)
+                record("detect:closed");
+        };
+
+        deliveryBranch.start();
+        detectBranch.start();
+        sourceStream.start();
+
+        expect(test::waitFor([&]() {
+            return deliveryReceived.load() > 0 && detectReceived.load() > 0;
+        }, 2000));
+
+        detectBranch.close();
+        expect(test::waitFor([&]() {
+            return detectBranch.closed();
+        }, 2000));
+
+        const int deliveryBefore = deliveryReceived.load();
+        expect(test::waitFor([&]() {
+            return deliveryReceived.load() > deliveryBefore;
+        }, 2000));
+        expect(source->running.load());
+        expect(indexOf("source:stop") == -1);
+
+        deliveryBranch.close();
+        expect(test::waitFor([&]() {
+            return deliveryBranch.closed();
+        }, 2000));
+        expect(source->running.load());
+        expect(indexOf("source:stop") == -1);
+
+        sourceStream.close();
+        expect(!source->running.load());
+
+        expect(indexOf("detect:closed") != -1);
+        expect(indexOf("delivery:closed") != -1);
+        expect(indexOf("source:stop") != -1);
+        expect(indexOf("detect:closed") < indexOf("source:stop"));
+        expect(indexOf("delivery:closed") < indexOf("source:stop"));
+    }
+};
+
+
+class PacketStreamAsyncLateDropAfterCloseTest : public Test
+{
+    std::atomic<int> received{0};
+
+    void onPacket(IPacket&)
+    {
+        ++received;
+    }
+
+    void run()
+    {
+        PacketStream stream("late-drop");
+        auto queue = std::make_shared<GatedAsyncPacketQueue>(16);
+
+        stream.attach(queue, 0);
+        stream.emitter += slot(this, &PacketStreamAsyncLateDropAfterCloseTest::onPacket);
+        stream.start();
+
+        stream.write(RawPacket("late", 4));
+
+        expect(test::waitFor([&]() {
+            return queue->enteredDispatch.load();
+        }, 2000));
+
+        std::atomic<bool> closeReturned{false};
+        std::thread closer([&]() {
+            stream.close();
+            closeReturned = true;
+        });
+
+        icy::sleep(25);
+        queue->releaseDispatch = true;
+
+        expect(test::waitFor([&]() {
+            return closeReturned.load();
+        }, 2000));
+
+        closer.join();
+
+        expect(received.load() == 0);
+        expect(stream.closed());
+    }
+};
+
+
 } // namespace icy
 
 

src/vision/README.md

@@ -4,12 +4,21 @@ Low-latency video intelligence primitives for Icey.
 
 The first landing is intentionally narrow:
 
+- `types.h` defines the stable frame, detection, track, and region model.
+- `VisionEvent` serializes directly to `json::Value` so the server layer can
+  ship events over existing WebSocket signalling surfaces without another
+  transport format.
+- `eventemitter.h` provides the narrow event signal surface detectors publish
+  through.
 - `FrameSampler` keeps the branch zero-copy on the synchronous hot path.
-- `DetectionQueue` is the explicit async clone boundary.
-- `MotionDetector` provides a minimal built-in video detector without adding
-  external dependencies.
-- `VisionEvent` serializes directly to `json::Value` so it can flow into
-  existing Symple/WebSocket surfaces without another transport format.
+- `VisionFramePacket` is the detector-side frame contract carrying stable
+  metadata.
+- `FrameNormalizer` is the pre-inference seam. With default zero values it is a
+  no-op metadata/materialization step; if configured it can also resize or
+  normalize pixel format before inference.
+- `DetectionQueue` is the explicit bounded async clone boundary.
+- `MotionDetector` is the first built-in detector and now consumes
+  `VisionFramePacket` instead of raw decode packets directly.
 
 This module is video-only. Audio intelligence belongs in a future sibling
 module rather than a combined `ai` surface.