--- name: bluesky-jetstream-10945feb8d55 description: Bluesky Jetstream Firehose Skill version: 1.0.0 --- # Bluesky Jetstream Firehose Skill **GF(3) Trit**: -1 (MINUS - validator/filter on incoming data stream) **Role**: Constrain and validate Bluesky firehose events before processing ## Overview Jetstream is Bluesky's simplified JSON firehose - a WebSocket streaming API that provides real-time access to all public activity on the Bluesky network. Unlike the full atproto firehose (which uses CBOR/CAR binary encoding), Jetstream delivers plain JSON, making it accessible for rapid prototyping. **Endpoints**: - Primary: `wss://jetstream2.us-east.bsky.network/subscribe` - Backup: `wss://jetstream1.us-west.bsky.network/subscribe` **Source**: [github.com/bluesky-social/jetstream](https://github.com/bluesky-social/jetstream) ## Query Parameters | Parameter | Type | Description | |-----------|------|-------------| | `wantedCollections` | string[] | Filter by NSID (repeatable) | | `wantedDids` | string[] | Filter by specific DIDs (repeatable) | | `compress` | boolean | Enable zstd compression | | `cursor` | integer | Resume from timestamp (microseconds since epoch) | Example URL: ``` wss://jetstream2.us-east.bsky.network/subscribe?wantedCollections=app.bsky.feed.post&wantedCollections=app.bsky.feed.like ``` ## Message Types ### commit Repository commit events (most common): ```json { "did": "did:plc:abc123...", "time_us": 1703123456789012, "kind": "commit", "commit": { "rev": "3k...", "operation": "create", "collection": "app.bsky.feed.post", "rkey": "3k...", "record": { "$type": "app.bsky.feed.post", "text": "Hello world!", "createdAt": "2024-12-21T10:00:00.000Z" }, "cid": "bafyrei..." } } ``` ### identity Identity updates: ```json { "did": "did:plc:abc123...", "time_us": 1703123456789012, "kind": "identity", "identity": { "did": "did:plc:abc123...", "handle": "alice.bsky.social", "seq": 12345 } } ``` ### account Account status changes: ```json { "did": "did:plc:abc123...", "time_us": 1703123456789012, "kind": "account", "account": { "active": true, "did": "did:plc:abc123...", "seq": 12345 } } ``` ## Collection NSIDs | NSID | Description | |------|-------------| | `app.bsky.feed.post` | Posts/skeets | | `app.bsky.feed.like` | Likes | | `app.bsky.feed.repost` | Reposts | | `app.bsky.graph.follow` | Follows | | `app.bsky.graph.block` | Blocks | | `app.bsky.graph.list` | Lists | | `app.bsky.graph.listitem` | List memberships | | `app.bsky.actor.profile` | Profile updates | | `app.bsky.feed.threadgate` | Thread gates | | `app.bsky.feed.postgate` | Post gates | ## JavaScript Example ```javascript const ws = new WebSocket( 'wss://jetstream2.us-east.bsky.network/subscribe?' + 'wantedCollections=app.bsky.feed.post' ); ws.onmessage = (event) => { const data = JSON.parse(event.data); if (data.kind === 'commit' && data.commit.operation === 'create') { console.log(`[${data.did}]: ${data.commit.record.text}`); } }; ``` ## Babashka/Clojure Example ```clojure (ns bluesky.jetstream (:require [clojure.data.json :as json] [org.httpkit.client :as http])) (def GAMMA 0x9E3779B9) (defn splitmix32 [x] (let [z (bit-and (+ x GAMMA) 0xFFFFFFFF) z (bit-and (* (bit-xor z (unsigned-bit-shift-right z 15)) 0x85EBCA6B) 0xFFFFFFFF) z (bit-and (* (bit-xor z (unsigned-bit-shift-right z 13)) 0xC2B2AE35) 0xFFFFFFFF)] (bit-xor z (unsigned-bit-shift-right z 16)))) (defn cid->trit [cid] (let [hash (reduce #(bit-xor (unchecked-multiply %1 31) (int %2)) 0 cid) hue (mod (splitmix32 hash) 360)] (cond (or (< hue 60) (>= hue 300)) +1 ; PLUS (warm) (< hue 180) 0 ; ERGODIC (neutral) :else -1))) ; MINUS (cold) (defn process-event [event] (when (and (= (:kind event) "commit") (= (get-in event [:commit :operation]) "create")) (let [cid (get-in event [:commit :cid]) text (get-in event [:commit :record :text]) trit (cid->trit cid)] {:did (:did event) :text text :cid cid :trit trit :trit-label (case trit 1 "PLUS" 0 "ERGODIC" -1 "MINUS")}))) ``` ## Integration with Agent-o-rama ### Memory Substrate Connection ```clojure ;; Feed Jetstream events into memory substrate (defn ingest-to-substrate [event memory-client] (when-let [processed (process-event event)] ;; Compress into control artifact (compress-trace memory-client [{:role :user :content (format "Bluesky post from %s: %s" (:did processed) (:text processed))}] {:tags ["bluesky" (:trit-label processed)] :source "jetstream" :cid (:cid processed)}))) ``` ### Decay Tracking Posts from Jetstream get the same decay epistemology: - **Half-life**: 10 hours (HALF_LIFE_MS = 36000000) - **Renewal**: User engagement (likes, replies) renews the artifact - **Expiration**: Unrenewed posts decay to trit=0 (neutral) ### GF(3) Conservation Each ingested post gets a trit assignment based on CID hash: ``` Σ(post trits) ≡ 0 (mod 3) ``` Over large samples, the distribution naturally balances due to SplitMix64's uniform properties. ## Rate Limits & Best Practices 1. **Use cursor for resumption**: Store `time_us` to resume after disconnect 2. **Exponential backoff**: 1s → 2s → 4s → ... → 60s max on reconnect 3. **Filter collections**: Request only what you need 4. **Handle compression**: Use zstd for bandwidth savings at scale 5. **Monitor lag**: Compare `time_us` to wall clock for lag detection ## DuckDB Schema ```sql CREATE TABLE IF NOT EXISTS bluesky_firehose ( seq_id BIGINT PRIMARY KEY, did VARCHAR NOT NULL, collection VARCHAR NOT NULL, rkey VARCHAR, cid VARCHAR, operation VARCHAR, -- create | update | delete record_json JSON, text VARCHAR, created_at TIMESTAMP, trit INTEGER, -- -1, 0, +1 color_hex VARCHAR, -- Gay.jl deterministic color cursor_us BIGINT, -- For resumption ingested_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, INDEX idx_did (did), INDEX idx_collection (collection), INDEX idx_trit (trit), INDEX idx_cursor (cursor_us) ); -- View for GF(3) balance check CREATE VIEW bluesky_gf3_balance AS SELECT DATE_TRUNC('hour', ingested_at) as hour, SUM(trit) as trit_sum, MOD(SUM(trit), 3) as gf3_remainder, COUNT(*) as event_count FROM bluesky_firehose GROUP BY 1 ORDER BY 1 DESC; ``` ## Savitch Connection The triadic processing of Jetstream events mirrors Savitch's reachability algorithm: ``` NSPACE(S(n)) ⊆ DSPACE(S(n)²) Firehose Event Processing: ├── MINUS (-1): Validate/filter incoming event ├── ERGODIC (0): Route to appropriate handler └── PLUS (+1): Generate processed artifact Each branch: O(S(n)) space Recursion depth: O(log n) for branching decisions Total: O(S(n)²) deterministic space simulating nondeterministic choices ``` ## Commands ```bash # Start Babashka consumer bb scripts/bluesky_jetstream.clj --collections app.bsky.feed.post --limit 1000 # Open HTML viewer open scripts/bluesky_viewer.html # Query stored events duckdb ~/.topos/ducklake.duckdb "SELECT * FROM bluesky_firehose ORDER BY cursor_us DESC LIMIT 10" # Check GF(3) balance duckdb ~/.topos/ducklake.duckdb "SELECT * FROM bluesky_gf3_balance LIMIT 24" ``` ## Related Skills - `duckdb-temporal-versioning` (+1): Store firehose with time-travel - `gay-mcp` (+1): Deterministic coloring for CIDs - `memory-substrate` (0): Ingest into Agent-o-rama memory ## References - [Jetstream Documentation](https://docs.bsky.app/blog/jetstream) - [Jetstream Source Code](https://github.com/bluesky-social/jetstream) - [ATProto Specification](https://atproto.com/specs/atp) - [Jazco Blog: Jetstream Design](https://jazco.dev/2024/09/24/jetstream/) --- **Skill Name**: bluesky-jetstream **Type**: Real-time data stream consumer **Trit**: -1 (MINUS - validator/filter) **GF(3) Triad**: bluesky-jetstream (-1) ⊗ duckdb-temporal (0) ⊗ gay-mcp (+1) = 0 ✓ ## Scientific Skill Interleaving This skill connects to the K-Dense-AI/claude-scientific-skills ecosystem: ### Graph Theory - **networkx** [○] via bicomodule - Universal graph hub ### Bibliography References - `general`: 734 citations in bib.duckdb ## Cat# Integration This skill maps to **Cat# = Comod(P)** as a bicomodule in the equipment structure: ``` Trit: 0 (ERGODIC) Home: Prof Poly Op: ⊗ Kan Role: Adj Color: #26D826 ``` ### GF(3) Naturality The skill participates in triads satisfying: ``` (-1) + (0) + (+1) ≡ 0 (mod 3) ``` This ensures compositional coherence in the Cat# equipment structure. --- Source attribution: plurigrid Original project: https://github.com/plurigrid/asi/blob/293ce9f76e3abcd5ba18db284688bc0d9ac6691d/plugins/asi/skills/bluesky-jetstream/SKILL.md License: MIT License information: https://opensource.org/license/mit Original upstream instructions; downloaded from Mega AI Hub. ## Upstream license notice MIT License Copyright (c) 2025 SkillCreator.ai Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions: The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software. THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. # Mega AI Hub resource metadata Creator / publisher: plurigrid Original project: https://github.com/plurigrid/asi/blob/293ce9f76e3abcd5ba18db284688bc0d9ac6691d/plugins/asi/skills/bluesky-jetstream/SKILL.md Attribution basis: Source repository record Fields of work: Development & Engineering, Marketing Roles: Backend Developer, Data Engineer, Growth Marketer Specialties: API, Backend Tools: Bluesky Country applicability: Country not specified Classification basis: Source catalog labels Instruction SHA-256: 16d2a0be28e89af78a800d91c39f8bc312497ee6106f85f97d805f9da7a8cfbb