Something went wrong. Try again.
A local-first event pipeline for independent agents, built on Jazz.
Something went wrong. Try again.
5.9 kB · 148 lines
TypeScript
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149import { spawn } from "node:child_process";import fs from "node:fs/promises";import path from "node:path";import { afterEach, describe, expect, test } from "vitest";import { WebSocketServer } from "ws";import { temporaryProject, testDeclarationEnvironment } from "./helpers.js";
const roots: string[] = [];const servers: WebSocketServer[] = [];
afterEach(async () => { await Promise.all(servers.splice(0).map(async (server) => { for (const client of server.clients) client.terminate(); await new Promise<void>((resolve) => server.close(() => resolve())); })); await Promise.all(roots.splice(0).map((root) => fs.rm(root, { recursive: true, force: true })));});
describe("thought stream jetstream command", () => { test("runs against an explicit local fixture server, persists events, and exits at its message bound", async () => { const observedUrls: string[] = []; const server = new WebSocketServer({ host: "127.0.0.1", port: 0 }); servers.push(server); await new Promise<void>((resolve, reject) => { server.once("listening", resolve); server.once("error", reject); }); server.on("connection", (socket, request) => { observedUrls.push(request.url ?? ""); socket.send(JSON.stringify(commitMessage(1784042400000000, "one", "rev-one"))); socket.send(JSON.stringify(commitMessage(1784042401000000, "two", "rev-two"))); }); const address = server.address(); if (!address || typeof address === "string") throw new Error("Missing Jetstream CLI fixture server address"); const project = await temporaryProject(); roots.push(project); await fs.cp(path.join(process.cwd(), "agents"), path.join(project, "agents"), { recursive: true }); await fs.cp(path.join(process.cwd(), "prompts"), path.join(project, "prompts"), { recursive: true });
const result = await run([ "--import", "tsx", "src/cli.ts", "jetstream", "--source", "jetstream:cli-fixture", "--collections", "app.bsky.feed.post", "--max-runtime", "5", "--max-messages", "2", "--max-reconnects", "0", ], { THOUGHTSTREAM_ROOT: project, THOUGHTSTREAM_JETSTREAM_URL: `ws://127.0.0.1:${address.port}/subscribe`, });
expect(result).toMatchObject({ code: 0, stderr: "" }); const output = JSON.parse(result.stdout) as { subscription: { reason: string; messages: number; inserted: number; reconnects: number }; producerOnly: boolean; consumerRuns: number; }; expect(output).toEqual({ subscription: expect.objectContaining({ reason: "message-limit", messages: 2, inserted: 2, reconnects: 0 }), producerOnly: false, consumerRuns: expect.any(Number), }); expect(output.consumerRuns).toBeGreaterThanOrEqual(2); const requestUrl = new URL(observedUrls[0]!, `ws://127.0.0.1:${address.port}`); expect(requestUrl.pathname).toBe("/subscribe"); expect(requestUrl.searchParams.getAll("wantedCollections")).toEqual(["app.bsky.feed.post"]); expect(requestUrl.searchParams.has("cursor")).toBe(false); }, 15_000);
test("producer-only mode persists Jetstream events without loading or running declarations", async () => { const server = new WebSocketServer({ host: "127.0.0.1", port: 0 }); servers.push(server); await new Promise<void>((resolve, reject) => { server.once("listening", resolve); server.once("error", reject); }); server.on("connection", (socket) => { socket.send(JSON.stringify(commitMessage(1784042400000000, "producer-only", "rev-producer-only"))); }); const address = server.address(); if (!address || typeof address === "string") throw new Error("Missing producer-only fixture server address"); const project = await temporaryProject(); roots.push(project); await fs.mkdir(path.join(project, "agents")); await fs.writeFile(path.join(project, "agents", "must-not-load.yaml"), "invalid: [declaration\n");
const result = await run([ "--import", "tsx", "src/cli.ts", "jetstream", "--producer-only", "--source", "jetstream:producer-only-fixture", "--collections", "app.bsky.feed.post", "--max-runtime", "5", "--max-messages", "1", "--max-reconnects", "0", ], { THOUGHTSTREAM_ROOT: project, THOUGHTSTREAM_JETSTREAM_URL: `ws://127.0.0.1:${address.port}/subscribe`, });
expect(result).toMatchObject({ code: 0, stderr: "" }); expect(JSON.parse(result.stdout)).toEqual({ subscription: expect.objectContaining({ source: "jetstream:producer-only-fixture", reason: "message-limit", messages: 1, inserted: 1, }), producerOnly: true, consumerRuns: 0, }); }, 15_000);});
function commitMessage(timeUs: number, rkey: string, rev: string): Record<string, unknown> { return { did: "did:plc:alicefixture", time_us: timeUs, kind: "commit", commit: { rev, operation: "create", collection: "app.bsky.feed.post", rkey, cid: `bafyreifixture-${rkey}`, record: { $type: "app.bsky.feed.post", text: rkey, createdAt: "2026-07-14T15:20:00.000Z" }, }, };}
async function run(args: string[], extraEnv: Record<string, string>): Promise<{ code: number | null; stdout: string; stderr: string }> { return new Promise((resolve, reject) => { const child = spawn(process.execPath, args, { cwd: path.resolve(import.meta.dirname, ".."), env: { ...process.env, ...testDeclarationEnvironment, ...extraEnv }, stdio: ["ignore", "pipe", "pipe"], }); let stdout = ""; let stderr = ""; child.stdout.on("data", (chunk) => { stdout += String(chunk); }); child.stderr.on("data", (chunk) => { stderr += String(chunk); }); child.once("error", reject); child.once("exit", (code) => resolve({ code, stdout, stderr })); });}