From 8bd1a44fc7dde1d2ad4f7174e4d56b8e3d50c56f Mon Sep 17 00:00:00 2001 From: Tim Disney Date: Fri, 8 May 2026 15:46:32 -0700 Subject: [PATCH] clean up messages getting glued together --- skills/bluesky-feeds/SKILL.md | 48 +++++ skills/bluesky-feeds/scripts/bluesky-feeds.js | 183 ++++++++++++++++++ src/lanes.ts | 7 +- src/scheduler.ts | 12 +- src/util.ts | 39 +++- 5 files changed, 277 insertions(+), 12 deletions(-) create mode 100644 skills/bluesky-feeds/SKILL.md create mode 100644 skills/bluesky-feeds/scripts/bluesky-feeds.js diff --git a/skills/bluesky-feeds/SKILL.md b/skills/bluesky-feeds/SKILL.md new file mode 100644 index 0000000..8f95927 --- /dev/null +++ b/skills/bluesky-feeds/SKILL.md @@ -0,0 +1,48 @@ +--- +name: bluesky-feeds +description: Fetch and summarize pinned Bluesky feeds. Use when the user asks to read, browse, or summarize their Bluesky feeds — especially the "For You" feed or any other pinned feed. Fetches recent posts and summarizes themes, links, and highlights. Requires BLUESKY_HANDLE and BLUESKY_APP_PASSWORD environment variables. +--- + +# Bluesky Feeds + +Fetch posts from the user's pinned Bluesky feeds and summarize them. + +## Setup + +No install needed — uses Node.js built-in `fetch`. + +Requires environment variables: +- `BLUESKY_HANDLE` — e.g. `yourname.bsky.social` +- `BLUESKY_APP_PASSWORD` — from Settings → App Passwords on bsky.app + +## Usage + +### List pinned feeds + +```bash +node /app/skills/bluesky-feeds/scripts/bluesky-feeds.js list +``` + +### Fetch posts from a feed (by name or URI) + +```bash +node /app/skills/bluesky-feeds/scripts/bluesky-feeds.js fetch "For You" +node /app/skills/bluesky-feeds/scripts/bluesky-feeds.js fetch "PKM" --limit 50 +node /app/skills/bluesky-feeds/scripts/bluesky-feeds.js fetch at://did:plc:.../app.bsky.feed.generator/... +``` + +Default limit is 30 posts. Use `--limit N` to fetch more. + +## Summarization workflow + +When the user asks to summarize a feed: + +1. Run `list` to confirm available feeds (skip if already known) +2. Run `fetch ""` with an appropriate limit (default 30, or more if user wants a thorough summary) +3. Read the output and produce a summary covering: + - **Main themes** — what topics are dominating the feed right now + - **Notable posts** — stand-out posts by engagement or content quality + - **Interesting links** — any URLs or articles worth highlighting + - **People to note** — active or interesting authors in this batch + +Keep the summary conversational and scannable. Use headers and bullets. diff --git a/skills/bluesky-feeds/scripts/bluesky-feeds.js b/skills/bluesky-feeds/scripts/bluesky-feeds.js new file mode 100644 index 0000000..6075dc9 --- /dev/null +++ b/skills/bluesky-feeds/scripts/bluesky-feeds.js @@ -0,0 +1,183 @@ +#!/usr/bin/env node +/** + * Bluesky Feeds Helper + * Usage: + * node bluesky-feeds.js list # List all pinned feeds + * node bluesky-feeds.js fetch # Fetch recent posts from a feed + * node bluesky-feeds.js fetch --limit 50 + * + * Credentials via env vars: + * BLUESKY_HANDLE - your handle, e.g. yourname.bsky.social + * BLUESKY_APP_PASSWORD - an app password from Settings > App Passwords + */ + +const BASE = "https://bsky.social/xrpc"; + +async function createSession(identifier, password) { + const res = await fetch(`${BASE}/com.atproto.server.createSession`, { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ identifier, password }), + }); + if (!res.ok) throw new Error(`Auth failed (${res.status}): ${await res.text()}`); + return res.json(); +} + +async function getPinnedFeeds(accessJwt) { + const res = await fetch(`${BASE}/app.bsky.actor.getPreferences`, { + headers: { Authorization: `Bearer ${accessJwt}` }, + }); + if (!res.ok) throw new Error(`getPreferences failed (${res.status}): ${await res.text()}`); + const data = await res.json(); + + const pref = + data.preferences?.find((p) => p.$type === "app.bsky.actor.defs#savedFeedsPrefV2") || + data.preferences?.find((p) => p.$type === "app.bsky.actor.defs#savedFeedsPref"); + + if (!pref) return []; + + const items = pref.items || []; + return items + .map((i) => ({ uri: i.value || i, pinned: i.pinned !== false })) + .filter((i) => typeof i.uri === "string" && i.uri.startsWith("at://")); +} + +async function getFeedGenerators(accessJwt, uris) { + const params = uris.map((u) => `feeds=${encodeURIComponent(u)}`).join("&"); + const res = await fetch(`${BASE}/app.bsky.feed.getFeedGenerators?${params}`, { + headers: { Authorization: `Bearer ${accessJwt}` }, + }); + if (!res.ok) throw new Error(`getFeedGenerators failed (${res.status}): ${await res.text()}`); + const data = await res.json(); + return data.feeds || []; +} + +async function getFeedPosts(accessJwt, feedUri, limit = 30) { + const allPosts = []; + let cursor; + do { + const url = new URL(`${BASE}/app.bsky.feed.getFeed`); + url.searchParams.set("feed", feedUri); + url.searchParams.set("limit", String(Math.min(limit - allPosts.length, 100))); + if (cursor) url.searchParams.set("cursor", cursor); + + const res = await fetch(url, { + headers: { Authorization: `Bearer ${accessJwt}` }, + }); + if (!res.ok) throw new Error(`getFeed failed (${res.status}): ${await res.text()}`); + const data = await res.json(); + allPosts.push(...(data.feed || [])); + cursor = data.cursor; + } while (cursor && allPosts.length < limit); + + return allPosts.slice(0, limit); +} + +function formatPost(item) { + const post = item.post; + const author = `@${post.author?.handle}`; + const displayName = post.author?.displayName ? `${post.author.displayName} (${author})` : author; + const text = post.record?.text || ""; + const likes = post.likeCount ?? 0; + const reposts = post.repostCount ?? 0; + const replies = post.replyCount ?? 0; + const indexedAt = post.indexedAt ? new Date(post.indexedAt).toLocaleString("en-US", { timeZone: "America/Los_Angeles" }) : ""; + + const lines = [ + `Author: ${displayName}`, + `Time: ${indexedAt}`, + `Text: ${text}`, + `Stats: ❤️ ${likes} 🔁 ${reposts} 💬 ${replies}`, + ]; + + // Include reason (repost/like reason) + if (item.reason) { + const reasonType = item.reason.$type || ""; + if (reasonType.includes("repost")) { + lines.push(`Via: reposted by @${item.reason.by?.handle}`); + } + } + + // Include external link if present + const embed = post.record?.embed; + if (embed?.external?.uri) { + lines.push(`Link: ${embed.external.uri}`); + if (embed.external.title) lines.push(` "${embed.external.title}"`); + } + + return lines.join("\n"); +} + +async function main() { + const handle = process.env.BLUESKY_HANDLE; + const appPassword = process.env.BLUESKY_APP_PASSWORD; + + if (!handle || !appPassword) { + console.error("Error: Set BLUESKY_HANDLE and BLUESKY_APP_PASSWORD environment variables."); + process.exit(1); + } + + const args = process.argv.slice(2); + const command = args[0]; + const arg = args[1]; + const limitFlag = args.indexOf("--limit"); + const limit = limitFlag !== -1 ? parseInt(args[limitFlag + 1], 10) : 30; + + const session = await createSession(handle, appPassword); + + if (!command || command === "list") { + const items = await getPinnedFeeds(session.accessJwt); + if (items.length === 0) { console.log("No pinned feeds found."); return; } + + const feeds = await getFeedGenerators(session.accessJwt, items.map((i) => i.uri)); + console.log(`${feeds.length} pinned feed(s) for @${session.handle}:\n`); + feeds.forEach((f, i) => { + console.log(`${i + 1}. ${f.displayName} (@${f.creator?.handle})`); + console.log(` URI: ${f.uri}`); + if (f.description) console.log(` ${f.description.split("\n")[0]}`); + console.log(` ❤️ ${f.likeCount ?? "?"} likes`); + console.log(); + }); + + } else if (command === "fetch") { + if (!arg) { + console.error("Usage: bluesky-feeds.js fetch [--limit N]"); + process.exit(1); + } + + let feedUri = arg; + let feedName = arg; + + if (!arg.startsWith("at://")) { + const items = await getPinnedFeeds(session.accessJwt); + const feeds = await getFeedGenerators(session.accessJwt, items.map((i) => i.uri)); + const match = feeds.find((f) => f.displayName.toLowerCase() === arg.toLowerCase()); + if (!match) { + console.error(`No pinned feed named "${arg}". Run 'list' to see available feeds.`); + process.exit(1); + } + feedUri = match.uri; + feedName = match.displayName; + console.log(`Fetching feed: "${feedName}" (${feedUri})\n`); + } + + const posts = await getFeedPosts(session.accessJwt, feedUri, limit); + if (posts.length === 0) { console.log("No posts found."); return; } + + console.log(`--- ${posts.length} posts from "${feedName}" ---\n`); + posts.forEach((item, i) => { + console.log(`[${i + 1}]`); + console.log(formatPost(item)); + console.log(); + }); + + } else { + console.error(`Unknown command: ${command}. Use 'list' or 'fetch '.`); + process.exit(1); + } +} + +main().catch((e) => { + console.error(e.message); + process.exit(1); +}); diff --git a/src/lanes.ts b/src/lanes.ts index e267f10..2a1a7a4 100644 --- a/src/lanes.ts +++ b/src/lanes.ts @@ -1,6 +1,7 @@ import type { AgentSession } from "@earendil-works/pi-coding-agent"; import type { Client } from "discord.js"; import { type SharedDeps, createSession } from "./agent.js"; +import { createAssistantTextCollector } from "./util.js"; interface Lane { session: AgentSession; @@ -63,13 +64,13 @@ export class LaneManager { await lane.session.steer(message); return { steered: true }; } - let buffer = ""; + const collector = createAssistantTextCollector(); const unsubscribe = lane.session.subscribe((event) => { + collector.handleEvent(event); if ( event.type === "message_update" && event.assistantMessageEvent.type === "text_delta" ) { - buffer += event.assistantMessageEvent.delta; opts.onDelta?.(event.assistantMessageEvent.delta); } if (event.type === "tool_execution_start") { @@ -78,7 +79,7 @@ export class LaneManager { }); try { await lane.session.prompt(message); - return { steered: false, output: buffer }; + return { steered: false, output: collector.getOutput() }; } finally { unsubscribe(); } diff --git a/src/scheduler.ts b/src/scheduler.ts index 496c293..21b389c 100644 --- a/src/scheduler.ts +++ b/src/scheduler.ts @@ -8,6 +8,7 @@ import { import { type ScheduleFile, chunkForDiscord, + createAssistantTextCollector, listScheduleFiles, updateLastRun, } from "./util.js"; @@ -114,14 +115,9 @@ export class Scheduler { private async runAndPost(sched: ScheduleFile): Promise { const session = await createSession(this.opts.deps, {}, this.opts.client); - let buffer = ""; + const collector = createAssistantTextCollector(); const unsub = session.subscribe((event) => { - if ( - event.type === "message_update" && - event.assistantMessageEvent.type === "text_delta" - ) { - buffer += event.assistantMessageEvent.delta; - } + collector.handleEvent(event); }); try { await session.prompt(sched.prompt); @@ -130,7 +126,7 @@ export class Scheduler { session.dispose(); } - await this.postResult(sched, buffer); + await this.postResult(sched, collector.getOutput()); await updateLastRun(sched.filePath).catch((err) => console.error(`scheduler: failed to update last_run for ${sched.name}:`, err) ); diff --git a/src/util.ts b/src/util.ts index 045f975..657e0ee 100644 --- a/src/util.ts +++ b/src/util.ts @@ -1,6 +1,9 @@ import { readFile, writeFile, readdir, mkdir, unlink } from "node:fs/promises"; import { join } from "node:path"; -import { parseFrontmatter } from "@earendil-works/pi-coding-agent"; +import { + parseFrontmatter, + type AgentSessionEvent, +} from "@earendil-works/pi-coding-agent"; export const DISCORD_LIMIT = 1900; @@ -21,6 +24,40 @@ export function chunkForDiscord(text: string, limit = DISCORD_LIMIT): string[] { return out; } +function assistantMessageSeparator(buffer: string): string { + if (!buffer.trim()) return ""; + if (buffer.endsWith("\n\n")) return ""; + if (buffer.endsWith("\n")) return "\n"; + return "\n\n"; +} + +export function createAssistantTextCollector() { + let buffer = ""; + let assistantMessages = 0; + + return { + handleEvent(event: AgentSessionEvent) { + if (event.type === "message_start" && event.message.role === "assistant") { + if (assistantMessages > 0) { + buffer += assistantMessageSeparator(buffer); + } + assistantMessages += 1; + return; + } + + if ( + event.type === "message_update" && + event.assistantMessageEvent.type === "text_delta" + ) { + buffer += event.assistantMessageEvent.delta; + } + }, + getOutput() { + return buffer; + }, + }; +} + export interface ScheduleFrontmatter { name: string; cron: string; -- 2.51.2