diff --git a/README.md b/README.md index c4980eb..4f21292 100644 --- a/README.md +++ b/README.md @@ -63,6 +63,7 @@ Runtime data is stored under `.thoughtstream/`. Set `THOUGHTSTREAM_ROOT` to use | `pnpm thought x-subscriptions-apply --source x: --confirm-plan-hash ` | Apply one unchanged source-owned subscription plan and verify convergence. | | `pnpm thought x-webhook-replay --source x: --from --to --confirm-webhook-id ` | Request one explicit replay within X's preceding 24-hour window. | | `pnpm thought x-webhook-delete --source x: --confirm-webhook-id ` | Delete an exact callback only after no live subscription still references it. | +| `pnpm thought x-user-lookup --source x: --username ` | Resolve one public handle to its exact X user id without mutation. | | `pnpm thought incidents --config ` | Project content-dark operational incidents, append the private JSONL ledger, and run the independent incident-alert policy. | | `pnpm thought fastmail-capture --file --source fastmail: --account-id ` | Ingest a captured JMAP response. | | `pnpm thought consume` | Run enabled consumers against Jazz subscriptions until interrupted. | @@ -224,6 +225,8 @@ On first activation, a dispatcher writes a durable activation event and ignores An `x-webhook` source owns one public callback path, one privacy lane, and exact event/user/tag subscriptions. The receiver binds only to loopback, answers CRC with the app consumer secret, verifies `x-twitter-webhooks-signature` over the raw POST body, and acknowledges only after durable settlement. It serializes writes through a bounded queue, returning retryable failure instead of accumulating unbounded work. It canonicalizes stable post content and references while dropping expanded profiles and mutable public metrics. Signed activity outside the source allowlist is durably counted and acknowledged without becoming a source event. +Tracked account lists live in `config/x-cameron-public.yaml` and `config/x-public-watch.yaml`. Each account has a readable handle and the immutable numeric user id used by X subscription filters. To add an account, run the read-only `x-user-lookup` command, verify the returned name and handle, add that pair to the appropriate file, then run `x-subscriptions-plan`. Editing the file alone neither contacts X nor changes a live subscription. Startup expands every account across the file's exact event types and fails closed on duplicates, path escape, symlinks, source mismatch, or unresolved ids. + Management uses a separate app-bearer environment. Status and plan are read-only. Register, apply, replay, and delete require hashes or ids copied from current readback, and none runs during receiver startup. The disabled `x-public-watch-hourly` batch and `x-public-watch` declaration show the first analysis path: at most one bounded public batch enters a dedicated output-only Letta Cloud agent on `letta/auto`; it has no X credential, tools, external actions, or route into the resident conversation. See [`spec/x-webhook.md`](spec/x-webhook.md). ## Consumer declarations diff --git a/config/x-cameron-public.yaml b/config/x-cameron-public.yaml new file mode 100644 index 0000000..1457491 --- /dev/null +++ b/config/x-cameron-public.yaml @@ -0,0 +1,8 @@ +version: 1 +source: x:cameron-public +eventTypes: + - post.create + - post.delete +accounts: + - handle: just_cameron + userId: "1232326955652931584" diff --git a/config/x-public-watch.yaml b/config/x-public-watch.yaml new file mode 100644 index 0000000..e204e5f --- /dev/null +++ b/config/x-public-watch.yaml @@ -0,0 +1,18 @@ +version: 1 +source: x:public-watch +eventTypes: + - post.create + - post.delete +accounts: + - handle: karpathy + userId: "33836629" + - handle: fchollet + userId: "68746721" + - handle: johnschulman2 + userId: "1388977636618080256" + - handle: lilianweng + userId: "96999384" + - handle: natolambert + userId: "2939913921" + - handle: srush_nlp + userId: "4558314927" diff --git a/spec/x-webhook.md b/spec/x-webhook.md index 0708dd3..8b901e5 100644 --- a/spec/x-webhook.md +++ b/spec/x-webhook.md @@ -35,6 +35,8 @@ Each `x-webhook` source declares: - one or more expected subscriptions with exact event type, numeric user id, nonsecret tag, privacy, and eventual webhook id; - a connector revision included in deterministic identity. +Expected subscriptions may be written inline or materialized from one project-relative `subscriptionFile`. The file is strict YAML containing one matching source id, one or both admitted public event types, and a bounded account list of exact `handle` plus numeric `userId` pairs. The handle is the operator-facing label; the stable user id remains the admission and subscription identity. Manifest loading refuses path escape, symlinks, source mismatch, duplicate case-insensitive handles, duplicate user ids, duplicate event types, oversized files, and a source that declares both forms. It deterministically expands every account/event pair into a source-owned tag before ordinary manifest validation, so receiver and management processes consume the same exact desired subscriptions without network resolution at startup. + The public URL has no credentials, query, fragment, or explicit port. Subscription tags are stable protocol labels, not prose. The receiver process receives only the consumer secret. Operator commands receive the bearer token in a separate credential compartment. No model consumer receives either credential. `X_CONSUMER_KEY` is not required by the public-event receiver and is not added merely because it exists. An `x-webhook` source fails startup when two routes share a public path, source id, or expected subscription tuple; when an event type conflicts with its declared privacy; when a personal-private route lacks the future user-grant contract; or when a public route names an event that X documents as user-context-only. diff --git a/src/connectors/x-client.ts b/src/connectors/x-client.ts index ec227f0..5670358 100644 --- a/src/connectors/x-client.ts +++ b/src/connectors/x-client.ts @@ -1,7 +1,7 @@ import { z } from "zod"; import { canonicalJson, sha256, type JsonObject } from "../core/json.js"; import type { XExpectedSubscription } from "./x-activity.js"; -import { xNumericIdSchema, xPublicActivityEventTypeSchema } from "./x-contract.js"; +import { xNumericIdSchema, xPublicActivityEventTypeSchema, xUsernameSchema } from "./x-contract.js"; const webhookSchema = z.object({ id: xNumericIdSchema, @@ -171,8 +171,7 @@ export class XApiClient { } async lookupUser(username: string, signal?: AbortSignal): Promise { - const normalized = username.trim().replace(/^@/, ""); - if (!/^[A-Za-z0-9_]{1,50}$/.test(normalized)) throw new Error("X username is invalid"); + const normalized = xUsernameSchema.parse(username.trim().replace(/^@/, "")); const response = await this.request("GET", `/2/users/by/username/${encodeURIComponent(normalized)}`, undefined, signal); const user = z.object({ data: userSchema }).passthrough().parse(response).data; return { id: user.id, username: user.username, name: user.name }; diff --git a/src/connectors/x-contract.ts b/src/connectors/x-contract.ts index 60f627c..258dbd4 100644 --- a/src/connectors/x-contract.ts +++ b/src/connectors/x-contract.ts @@ -10,6 +10,7 @@ export const X_SOURCE_LANES = ["personal-public", "public-watchlist"] as const; export const xPublicActivityEventTypeSchema = z.enum(X_PUBLIC_ACTIVITY_EVENT_TYPES); export const xSourceLaneSchema = z.enum(X_SOURCE_LANES); export const xNumericIdSchema = z.string().regex(/^[0-9]{1,19}$/); +export const xUsernameSchema = z.string().regex(/^[A-Za-z0-9_]{1,50}$/); const xReferenceSchema = z.object({ type: z.enum(["replied_to", "quoted", "retweeted"]), diff --git a/src/runtime/manifest.ts b/src/runtime/manifest.ts index 36dddd8..809515b 100644 --- a/src/runtime/manifest.ts +++ b/src/runtime/manifest.ts @@ -11,6 +11,7 @@ import { xNumericIdSchema, xPublicActivityEventTypeSchema, xSourceLaneSchema, + xUsernameSchema, } from "../connectors/x-contract.js"; const privacySchema = z.enum(["public-source", "private", "sensitive"]); @@ -141,6 +142,37 @@ const xExpectedSubscriptionSchema = z.object({ tag: z.string().min(1).max(200), }).strict(); +const xSubscriptionFileSchema = z.object({ + version: z.literal(1), + source: idSchema, + eventTypes: z.array(xPublicActivityEventTypeSchema).min(1).max(2), + accounts: z.array(z.object({ + handle: xUsernameSchema, + userId: xNumericIdSchema, + }).strict()).min(1).max(750), +}).strict().superRefine((value, context) => { + const eventTypes = new Set(); + for (const [index, eventType] of value.eventTypes.entries()) { + if (eventTypes.has(eventType)) { + context.addIssue({ code: "custom", path: ["eventTypes", index], message: `Duplicate X watch event type: ${eventType}` }); + } + eventTypes.add(eventType); + } + const handles = new Set(); + const userIds = new Set(); + for (const [index, account] of value.accounts.entries()) { + const handle = account.handle.toLowerCase(); + if (handles.has(handle)) { + context.addIssue({ code: "custom", path: ["accounts", index, "handle"], message: `Duplicate X watch handle: ${account.handle}` }); + } + handles.add(handle); + if (userIds.has(account.userId)) { + context.addIssue({ code: "custom", path: ["accounts", index, "userId"], message: `Duplicate X watch user id: ${account.userId}` }); + } + userIds.add(account.userId); + } +}); + const xWebhookSourceSchema = z.object({ ...sourceBase, kind: z.literal("x-webhook"), @@ -389,7 +421,8 @@ export async function loadThoughtStreamManifest( if (!stat.isFile()) throw new Error(`thought stream manifest is not a regular file: ${manifestPath}`); const raw = await fs.readFile(manifestPath, "utf8"); const parsed = manifestPath.endsWith(".json") ? JSON.parse(raw) : YAML.parse(raw); - const manifest = manifestSchema.parse(parsed); + const materialized = await materializeXSubscriptionFiles(projectRoot, parsed); + const manifest = manifestSchema.parse(materialized); const agentDirectory = resolveProjectPath(projectRoot, manifest.agents.directory); if (!isWithin(projectRoot, agentDirectory)) { throw new Error(`Agent directory escapes project root: ${manifest.agents.directory}`); @@ -397,6 +430,54 @@ export async function loadThoughtStreamManifest( return { path: manifestPath, manifest, hash: sha256(canonicalJson(asJsonObject(manifest))) }; } +async function materializeXSubscriptionFiles(projectRoot: string, value: unknown): Promise { + if (!isRecord(value) || !Array.isArray(value.sources)) return value; + const sources = await Promise.all(value.sources.map(async (source): Promise => { + if (!isRecord(source) || source.kind !== "x-webhook" || !("subscriptionFile" in source)) return source; + if ("expectedSubscriptions" in source) { + throw new Error("X webhook source may not declare both subscriptionFile and expectedSubscriptions"); + } + if (typeof source.subscriptionFile !== "string" || !source.subscriptionFile.trim()) { + throw new Error("X subscriptionFile must name one project-relative YAML file"); + } + const requestedPath = source.subscriptionFile; + if (path.isAbsolute(requestedPath) || !/\.ya?ml$/i.test(requestedPath)) { + throw new Error("X subscriptionFile must be a project-relative YAML file"); + } + const subscriptionPath = path.resolve(projectRoot, requestedPath); + if (!isWithin(projectRoot, subscriptionPath)) { + throw new Error(`X subscriptionFile escapes project root: ${requestedPath}`); + } + const stat = await fs.lstat(subscriptionPath).catch((error: NodeJS.ErrnoException) => { + if (error.code === "ENOENT") throw new Error(`X subscriptionFile not found: ${requestedPath}`); + throw error; + }); + if (stat.isSymbolicLink() || !stat.isFile()) { + throw new Error(`X subscriptionFile must be a regular nonsymlink file: ${requestedPath}`); + } + if (stat.size > 262_144) throw new Error(`X subscriptionFile exceeds 262144 bytes: ${requestedPath}`); + const [realRoot, realSubscriptionPath] = await Promise.all([fs.realpath(projectRoot), fs.realpath(subscriptionPath)]); + if (!isWithin(realRoot, realSubscriptionPath)) { + throw new Error(`X subscriptionFile resolves outside project root: ${requestedPath}`); + } + const config = xSubscriptionFileSchema.parse(YAML.parse(await fs.readFile(subscriptionPath, "utf8"))); + if (typeof source.id !== "string" || config.source !== source.id) { + throw new Error(`X subscriptionFile source does not match manifest source: ${requestedPath}`); + } + const expectedSubscriptions = config.accounts + .slice() + .sort((left, right) => left.handle.toLowerCase().localeCompare(right.handle.toLowerCase()) || left.userId.localeCompare(right.userId)) + .flatMap((account) => config.eventTypes.slice().sort().map((eventType) => ({ + eventType, + userId: account.userId, + tag: `thoughtstream:${config.source}:${account.handle.toLowerCase()}:${eventType.replaceAll(".", "-")}`, + }))); + const { subscriptionFile: _subscriptionFile, ...rest } = source; + return { ...rest, expectedSubscriptions }; + })); + return { ...value, sources }; +} + export function resolveProjectPath(projectRoot: string, value: string): string { return path.isAbsolute(value) ? path.normalize(value) : path.resolve(projectRoot, value); } @@ -406,6 +487,10 @@ function isWithin(root: string, candidate: string): boolean { return relative === "" || (!relative.startsWith("..") && !path.isAbsolute(relative)); } +function isRecord(value: unknown): value is Record { + return Boolean(value && typeof value === "object" && !Array.isArray(value)); +} + function asJsonObject(value: unknown): JsonObject { const normalized = JSON.parse(JSON.stringify(value)) as unknown; if (!normalized || typeof normalized !== "object" || Array.isArray(normalized)) { diff --git a/test/x-manifest.test.ts b/test/x-manifest.test.ts index e7f4594..e4c587f 100644 --- a/test/x-manifest.test.ts +++ b/test/x-manifest.test.ts @@ -14,7 +14,19 @@ afterEach(async () => { describe("X webhook manifest", () => { test("loads the disabled repository sources and Letta Auto watch declaration", async () => { const loaded = await loadThoughtStreamManifest(process.cwd()); - expect(loaded.manifest.sources.filter((source) => source.kind === "x-webhook")).toHaveLength(2); + const xSources = loaded.manifest.sources.filter((source) => source.kind === "x-webhook"); + expect(xSources).toHaveLength(2); + expect(xSources.find((source) => source.id === "x:cameron-public")?.expectedSubscriptions).toEqual([ + { eventType: "post.create", userId: "1232326955652931584", tag: "thoughtstream:x:cameron-public:just_cameron:post-create" }, + { eventType: "post.delete", userId: "1232326955652931584", tag: "thoughtstream:x:cameron-public:just_cameron:post-delete" }, + ]); + const publicWatch = xSources.find((source) => source.id === "x:public-watch"); + expect(publicWatch?.expectedSubscriptions).toHaveLength(12); + expect(publicWatch?.expectedSubscriptions).toContainEqual({ + eventType: "post.create", + userId: "33836629", + tag: "thoughtstream:x:public-watch:karpathy:post-create", + }); expect(loaded.manifest.batches.find((batch) => batch.id === "x-public-watch-hourly")).toMatchObject({ enabled: false, input: { eventTypes: ["stream.thought.source.x.activity"], sourceIds: ["x:public-watch"] }, @@ -77,6 +89,74 @@ describe("X webhook manifest", () => { await fs.writeFile(path.join(root, "thoughtstream.yaml"), `${manifestWith({})}${extras}`); await expect(loadThoughtStreamManifest(root)).rejects.toThrow("Too big"); }); + + test("loads one strict project-relative account file into exact subscriptions", async () => { + const root = await temporaryProject("thoughtstream-x-subscription-file-"); + roots.push(root); + await fs.mkdir(path.join(root, "config")); + await fs.writeFile(path.join(root, "config", "watch.yaml"), `version: 1 +source: x:fixture +eventTypes: [post.delete, post.create] +accounts: + - handle: Karpathy + userId: "33836629" + - handle: just_cameron + userId: "1232326955652931584" +`); + await fs.writeFile(path.join(root, "thoughtstream.yaml"), manifestWithSubscriptionFile("config/watch.yaml")); + const source = (await loadThoughtStreamManifest(root)).manifest.sources[0]; + expect(source?.kind).toBe("x-webhook"); + if (source?.kind !== "x-webhook") throw new Error("Expected X fixture source"); + expect(source.expectedSubscriptions).toEqual([ + { eventType: "post.create", userId: "1232326955652931584", tag: "thoughtstream:x:fixture:just_cameron:post-create" }, + { eventType: "post.delete", userId: "1232326955652931584", tag: "thoughtstream:x:fixture:just_cameron:post-delete" }, + { eventType: "post.create", userId: "33836629", tag: "thoughtstream:x:fixture:karpathy:post-create" }, + { eventType: "post.delete", userId: "33836629", tag: "thoughtstream:x:fixture:karpathy:post-delete" }, + ]); + }); + + test.each([ + ["path escape", "../watch.yaml", undefined, "escapes project root"], + ["source mismatch", "config/watch.yaml", `version: 1\nsource: x:other\neventTypes: [post.create]\naccounts: [{ handle: karpathy, userId: \"33836629\" }]\n`, "does not match"], + ["duplicate handle", "config/watch.yaml", `version: 1\nsource: x:fixture\neventTypes: [post.create]\naccounts: [{ handle: karpathy, userId: \"33836629\" }, { handle: Karpathy, userId: \"2\" }]\n`, "Duplicate X watch handle"], + ["duplicate user id", "config/watch.yaml", `version: 1\nsource: x:fixture\neventTypes: [post.create]\naccounts: [{ handle: karpathy, userId: \"33836629\" }, { handle: other, userId: \"33836629\" }]\n`, "Duplicate X watch user id"], + ["duplicate event type", "config/watch.yaml", `version: 1\nsource: x:fixture\neventTypes: [post.create, post.create]\naccounts: [{ handle: karpathy, userId: \"33836629\" }]\n`, "Duplicate X watch event type"], + ])("rejects subscription-file %s", async (_name, subscriptionFile, file, expected) => { + const root = await temporaryProject("thoughtstream-x-subscription-invalid-"); + roots.push(root); + await fs.mkdir(path.join(root, "config")); + if (file) await fs.writeFile(path.join(root, "config", "watch.yaml"), file); + await fs.writeFile(path.join(root, "thoughtstream.yaml"), manifestWithSubscriptionFile(subscriptionFile)); + await expect(loadThoughtStreamManifest(root)).rejects.toThrow(expected); + }); + + test("rejects a symlinked subscription file", async () => { + const root = await temporaryProject("thoughtstream-x-subscription-symlink-"); + roots.push(root); + await fs.mkdir(path.join(root, "config")); + await fs.writeFile(path.join(root, "outside.yaml"), `version: 1\nsource: x:fixture\neventTypes: [post.create]\naccounts: [{ handle: karpathy, userId: \"33836629\" }]\n`); + await fs.symlink(path.join(root, "outside.yaml"), path.join(root, "config", "watch.yaml")); + await fs.writeFile(path.join(root, "thoughtstream.yaml"), manifestWithSubscriptionFile("config/watch.yaml")); + await expect(loadThoughtStreamManifest(root)).rejects.toThrow("regular nonsymlink"); + }); + + test("rejects a source that declares both inline and file-backed subscriptions", async () => { + const root = await temporaryProject("thoughtstream-x-subscription-both-"); + roots.push(root); + await fs.mkdir(path.join(root, "config")); + await fs.writeFile(path.join(root, "config", "watch.yaml"), `version: 1\nsource: x:fixture\neventTypes: [post.create]\naccounts: [{ handle: karpathy, userId: \"33836629\" }]\n`); + await fs.writeFile(path.join(root, "thoughtstream.yaml"), `${manifestWithSubscriptionFile("config/watch.yaml")} expectedSubscriptions:\n - eventType: post.create\n userId: \"33836629\"\n tag: thoughtstream:x:fixture:karpathy:post-create\n`); + await expect(loadThoughtStreamManifest(root)).rejects.toThrow("may not declare both"); + }); + + test("rejects an oversized subscription file before parsing it", async () => { + const root = await temporaryProject("thoughtstream-x-subscription-oversized-"); + roots.push(root); + await fs.mkdir(path.join(root, "config")); + await fs.writeFile(path.join(root, "config", "watch.yaml"), " ".repeat(262_145)); + await fs.writeFile(path.join(root, "thoughtstream.yaml"), manifestWithSubscriptionFile("config/watch.yaml")); + await expect(loadThoughtStreamManifest(root)).rejects.toThrow("exceeds 262144 bytes"); + }); }); function manifestWith(mutation: Record): string { @@ -107,3 +187,24 @@ sources: tag: ${tag} `; } + +function manifestWithSubscriptionFile(subscriptionFile: string): string { + return `version: 1 +runtime: + revision: x-fixture +sources: + - id: x:fixture + kind: x-webhook + enabled: false + lane: public-watchlist + consumerSecretEnv: FIXTURE_X_SECRET + managementBearerTokenEnv: FIXTURE_X_BEARER + webhookUrl: https://thoughtstream.example/webhooks/x/fixture + webhookPath: /webhooks/x/fixture + listenHost: 127.0.0.1 + listenPort: 4320 + maxBodyBytes: 1048576 + requestTimeoutMs: 5000 + subscriptionFile: ${subscriptionFile} +`; +} diff --git a/thoughtstream.yaml b/thoughtstream.yaml index a9278e0..019525f 100644 --- a/thoughtstream.yaml +++ b/thoughtstream.yaml @@ -127,13 +127,7 @@ sources: maxBodyBytes: 2097152 requestTimeoutMs: 8000 pendingRequestLimit: 100 - expectedSubscriptions: - - eventType: post.create - userId: "123456789" - tag: thoughtstream:x:cameron-public:post-create - - eventType: post.delete - userId: "123456789" - tag: thoughtstream:x:cameron-public:post-delete + subscriptionFile: config/x-cameron-public.yaml - id: x:public-watch kind: x-webhook enabled: false @@ -147,13 +141,7 @@ sources: maxBodyBytes: 2097152 requestTimeoutMs: 8000 pendingRequestLimit: 100 - expectedSubscriptions: - - eventType: post.create - userId: "123456789" - tag: thoughtstream:x:public-watch:post-create-example - - eventType: post.delete - userId: "123456789" - tag: thoughtstream:x:public-watch:post-delete-example + subscriptionFile: config/x-public-watch.yaml - id: rss:example kind: rss enabled: false