diff --git a/plugins/thread-organizer/README.md b/plugins/thread-organizer/README.md index 4de7de7..84eab14 100644 --- a/plugins/thread-organizer/README.md +++ b/plugins/thread-organizer/README.md @@ -1,6 +1,6 @@ # Thread Organizer -Thread Organizer files new bb threads into existing work sections while leaving manual organization in control. It starts in observe mode and never creates sections, backfills old threads, or runs a hidden classification agent. +Thread Organizer turns BB’s pinned area into an inbox and files active work into semantic sections. Idle and failed threads rise to the inbox; resumed work returns to its preserved section. ![Thread Organizer settings in bb](docs/screenshot.png) @@ -12,16 +12,16 @@ bb plugin install git:https://github.com/brsbl/bb-plugins.git@plugin/thread-orga ## Use -Leave the plugin in its default `observe` mode first and inspect its decisions: +The plugin starts in `apply` mode and organizes threads automatically. Follow its decisions and lifecycle changes with: ```bash bb plugin logs thread-organizer -f ``` -When the proposals look right, enable changes: +To preview proposals without changing threads, switch to `observe` mode: ```bash -bb plugin config thread-organizer set mode apply +bb plugin config thread-organizer set mode observe ``` The setting takes effect immediately. @@ -34,22 +34,26 @@ The setting takes effect immediately. | `Writing` | Articles, positioning, copy, and editorial work | | `QA` | Never; reserved for manual phase management | -Threads that do not clear the confidence threshold stay unsectioned. Pinning remains independent priority state, and archiving remains completion state. +Threads that do not clear the confidence threshold stay unsectioned. The plugin never creates sections. ## Behavior -| Moment | Evaluation | +| Moment | Inbox and organization behavior | | --- | --- | +| Startup | Adopts existing visible root threads and reconciles their current state | | Creation | Uses project identity and the prompt-derived title fallback for an early section proposal | -| Activation | Retries prompt history briefly when the initial prompt was not yet available | +| Activation | Removes plugin-managed inbox pinning and returns the thread to its semantic section | +| Idle or failed | Pins the thread into the top inbox area without clearing its semantic section | | First completed turn | Refines the section and repairs a still-missing title | | Later turns | Re-evaluates at turns 5, 15, 25, and every ten completed turns thereafter | New placements require at least `0.85` confidence with a `0.20` lead over the next rule. Moving a plugin-managed thread requires `0.92` confidence, a `0.25` lead, and the same result at two consecutive scheduled evaluations. -Only visible, ordinary user root threads created while the plugin is loaded are tracked. Hidden workers, children, forks, side chats, plugin-originated threads, archived threads, deleted threads, and terminal error states are excluded. +Only visible, ordinary user root threads are tracked. Hidden workers, children, forks, side chats, plugin-originated threads, archived threads, and deleted threads are excluded. -An explicit creation-time title or section is locked. After the plugin writes a field, any different value is treated as a manual or external override and locks that field permanently. Native bb titles therefore win, and section changes are never cleared automatically. +An explicit creation-time title or section is locked. After the plugin writes a field, any different value is treated as a manual or external override and locks that field permanently. Native BB titles therefore win, and section changes are never cleared automatically. Pinning is reserved for the inbox lifecycle, so every active thread is unpinned even if it was pinned manually. + +Section collapse state belongs to each BB client and remains unchanged when a thread moves through the pinned inbox. ## Develop diff --git a/plugins/thread-organizer/core.ts b/plugins/thread-organizer/core.ts index 7142639..1062344 100644 --- a/plugins/thread-organizer/core.ts +++ b/plugins/thread-organizer/core.ts @@ -161,7 +161,7 @@ export function isSubstantiveText(value: string): boolean { return !/^(?:https?:\/\/\S+|@[a-z0-9:_-]+)$/i.test(normalized); } -export function isEligibleThread(thread: OrganizableThread): boolean { +export function isManageableThread(thread: OrganizableThread): boolean { return ( thread.visibility === "visible" && thread.parentThreadId === null && @@ -170,7 +170,13 @@ export function isEligibleThread(thread: OrganizableThread): boolean { thread.childOrigin === null && thread.originPluginId === null && thread.archivedAt === null && - thread.deletedAt === null && + thread.deletedAt === null + ); +} + +export function isEligibleThread(thread: OrganizableThread): boolean { + return ( + isManageableThread(thread) && thread.status !== "error" && thread.status !== "stopping" ); diff --git a/plugins/thread-organizer/package.json b/plugins/thread-organizer/package.json index 98d9c1f..7eecd30 100644 --- a/plugins/thread-organizer/package.json +++ b/plugins/thread-organizer/package.json @@ -1,7 +1,7 @@ { "name": "bb-plugin-thread-organizer", "version": "0.1.0", - "description": "Organizes new bb threads into existing sections while preserving manual changes.", + "description": "Uses pinned threads as an inbox and organizes active work into semantic sections.", "type": "module", "license": "UNLICENSED", "files": [ @@ -21,7 +21,7 @@ }, "bb": { "name": "Thread Organizer", - "description": "Organizes new bb threads into existing sections while preserving manual changes.", + "description": "Uses pinned threads as an inbox and organizes active work into semantic sections.", "branding": { "icon": "FolderTree" }, diff --git a/plugins/thread-organizer/server.test.ts b/plugins/thread-organizer/server.test.ts index 786edf0..bb24e1b 100644 --- a/plugins/thread-organizer/server.test.ts +++ b/plugins/thread-organizer/server.test.ts @@ -35,6 +35,8 @@ function completedEvent(seq: number) { } function createHarness(input?: { + archiveAfterList?: boolean; + existingThreads?: boolean; mode?: "apply" | "observe"; projectName?: string; prompt?: string; @@ -72,6 +74,22 @@ function createHarness(input?: { return thread; }, ); + const pin = vi.fn(async () => { + thread = makeThreadResponse({ + ...thread, + pinnedAt: thread.updatedAt + 1, + updatedAt: thread.updatedAt + 1, + }); + return thread; + }); + const unpin = vi.fn(async () => { + thread = makeThreadResponse({ + ...thread, + pinnedAt: null, + updatedAt: thread.updatedAt + 1, + }); + return thread; + }); const host = createFakePluginHost({ pluginId: "thread-organizer", settings: { mode: input?.mode ?? "observe" }, @@ -91,7 +109,20 @@ function createHarness(input?: { }, }, get: async () => thread, + list: async () => { + if (!input?.existingThreads) return []; + const listed = thread; + if (input.archiveAfterList) { + thread = makeThreadResponse({ + ...thread, + archivedAt: thread.updatedAt + 1, + }); + } + return [listed]; + }, + pin, promptHistory: async () => promptHistory, + unpin, update, }, }, @@ -116,23 +147,26 @@ function createHarness(input?: { ): void { thread = makeThreadResponse({ ...thread, ...changes }); }, + pin, + unpin, update, }; } describe("Thread Organizer plugin", () => { - it("registers a headless observe-mode lifecycle", async () => { + it("registers a headless apply-mode lifecycle", async () => { const { bb, harness } = createHarness(); plugin(bb); expect(harness.inspection.registrations.settingsDescriptors).toMatchObject({ - mode: { default: "observe", options: ["observe", "apply"] }, + mode: { default: "apply", options: ["observe", "apply"] }, }); expect(harness.inspection.registrations.threadEventHandlers).toMatchObject({ "thread.active": 1, "thread.archived": 1, "thread.created": 1, "thread.deleted": 1, + "thread.failed": 1, "thread.idle": 1, }); expect(harness.inspection.registrations.cli).toBeNull(); @@ -200,6 +234,165 @@ describe("Thread Organizer plugin", () => { await harness.lifecycle.dispose(); }); + it("uses the pinned area as an inbox while preserving the semantic section", async () => { + const organizer = createHarness({ mode: "apply" }); + plugin(organizer.bb); + await organizer.harness.behavior.emitThreadEvent("thread.created", { + thread: organizer.currentThread(), + }); + expect(organizer.currentThread().sectionId).toBe("sec_extensions"); + + organizer.setThread({ status: "idle" }); + await organizer.harness.behavior.emitThreadEvent("thread.idle", { + lastAssistantText: "Done.", + thread: organizer.currentThread(), + }); + + expect(organizer.pin).toHaveBeenCalledWith({ threadId: "thr_test" }); + expect(organizer.currentThread().pinnedAt).not.toBeNull(); + expect(organizer.currentThread().sectionId).toBe("sec_extensions"); + + organizer.setThread({ status: "active" }); + await organizer.harness.behavior.emitThreadEvent("thread.active", { + thread: organizer.currentThread(), + }); + + expect(organizer.unpin).toHaveBeenCalledWith({ threadId: "thr_test" }); + expect(organizer.currentThread().pinnedAt).toBeNull(); + expect(organizer.currentThread().sectionId).toBe("sec_extensions"); + await organizer.harness.lifecycle.dispose(); + }); + + it("adopts an existing idle thread into the inbox on startup", async () => { + const organizer = createHarness({ + existingThreads: true, + mode: "apply", + thread: { status: "idle" }, + }); + plugin(organizer.bb); + + await organizer.harness.behavior.runService("startup-reconciliation").done; + + expect(organizer.pin).toHaveBeenCalledWith({ threadId: "thr_test" }); + expect(organizer.currentThread().pinnedAt).not.toBeNull(); + await organizer.harness.lifecycle.dispose(); + }); + + it("adopts an existing failed thread into the inbox on startup", async () => { + const organizer = createHarness({ + existingThreads: true, + mode: "apply", + thread: { status: "error" }, + }); + plugin(organizer.bb); + + await organizer.harness.behavior.runService("startup-reconciliation").done; + + expect(organizer.pin).toHaveBeenCalledWith({ threadId: "thr_test" }); + await organizer.harness.lifecycle.dispose(); + }); + + it("skips a thread archived while startup reconciliation is running", async () => { + const organizer = createHarness({ + archiveAfterList: true, + existingThreads: true, + mode: "apply", + thread: { status: "idle" }, + }); + plugin(organizer.bb); + + await organizer.harness.behavior.runService("startup-reconciliation").done; + + expect(organizer.pin).not.toHaveBeenCalled(); + expect(await organizer.bb.storage.kv.list("thread:")).toEqual([]); + await organizer.harness.lifecycle.dispose(); + }); + + it("unpins an active thread even when it was pinned manually", async () => { + const organizer = createHarness({ + mode: "apply", + thread: { pinnedAt: 10 }, + }); + plugin(organizer.bb); + await organizer.harness.behavior.emitThreadEvent("thread.created", { + thread: organizer.currentThread(), + }); + + organizer.setThread({ status: "active" }); + await organizer.harness.behavior.emitThreadEvent("thread.active", { + thread: organizer.currentThread(), + }); + + expect(organizer.pin).not.toHaveBeenCalled(); + expect(organizer.unpin).toHaveBeenCalledWith({ threadId: "thr_test" }); + expect(organizer.currentThread().pinnedAt).toBeNull(); + await organizer.harness.lifecycle.dispose(); + }); + + it("unpins active work after a manual unpin and re-pin", async () => { + const organizer = createHarness({ mode: "apply" }); + plugin(organizer.bb); + await organizer.harness.behavior.emitThreadEvent("thread.created", { + thread: organizer.currentThread(), + }); + + organizer.setThread({ status: "idle" }); + await organizer.harness.behavior.emitThreadEvent("thread.idle", { + lastAssistantText: "Done.", + thread: organizer.currentThread(), + }); + organizer.setThread({ + pinnedAt: 100, + status: "active", + }); + await organizer.harness.behavior.emitThreadEvent("thread.active", { + thread: organizer.currentThread(), + }); + + expect(organizer.unpin).toHaveBeenCalledWith({ threadId: "thr_test" }); + expect(organizer.currentThread().pinnedAt).toBeNull(); + await organizer.harness.lifecycle.dispose(); + }); + + it("pins failed work into the inbox", async () => { + const organizer = createHarness({ mode: "apply" }); + plugin(organizer.bb); + await organizer.harness.behavior.emitThreadEvent("thread.created", { + thread: organizer.currentThread(), + }); + + organizer.setThread({ status: "error" }); + await organizer.harness.behavior.emitThreadEvent("thread.failed", { + error: "Provider failed", + thread: organizer.currentThread(), + }); + + expect(organizer.pin).toHaveBeenCalledWith({ threadId: "thr_test" }); + expect(organizer.currentThread().pinnedAt).not.toBeNull(); + await organizer.harness.lifecycle.dispose(); + }); + + it("reconciles the inbox immediately when apply mode is enabled", async () => { + const organizer = createHarness({ mode: "observe" }); + plugin(organizer.bb); + await organizer.harness.behavior.emitThreadEvent("thread.created", { + thread: organizer.currentThread(), + }); + organizer.setThread({ status: "idle" }); + await organizer.harness.behavior.emitThreadEvent("thread.idle", { + lastAssistantText: "Done.", + thread: organizer.currentThread(), + }); + expect(organizer.pin).not.toHaveBeenCalled(); + + await organizer.harness.behavior.setSettings({ mode: "apply" }); + + await vi.waitFor(() => { + expect(organizer.pin).toHaveBeenCalledWith({ threadId: "thr_test" }); + }); + await organizer.harness.lifecycle.dispose(); + }); + it("never changes an explicit creation-time section", async () => { const { bb, harness, currentThread, update } = createHarness({ mode: "apply", diff --git a/plugins/thread-organizer/server.ts b/plugins/thread-organizer/server.ts index 407322a..3a457ee 100644 --- a/plugins/thread-organizer/server.ts +++ b/plugins/thread-organizer/server.ts @@ -5,6 +5,7 @@ import { classifySection, deriveTaskTitle, isEligibleThread, + isManageableThread, isSubstantiveText, resolveSectionId, type SectionClassification, @@ -18,8 +19,10 @@ const MOVE_SECTION_CONFIDENCE = 0.92; const MOVE_SECTION_MARGIN = 0.25; const TITLE_CONFIDENCE = 0.9; const MAX_COMPLETED_EVENT_DRAIN = 100; +const THREAD_LIST_PAGE_SIZE = 100; type Thread = Awaited>; +type ThreadSeed = Pick; type EvaluationPhase = "active" | "created" | "settings" | "turn"; interface ThreadState { @@ -42,7 +45,7 @@ function stateKey(threadId: string): string { return `${STATE_PREFIX}${threadId}`; } -function initialState(thread: Thread): ThreadState { +function initialState(thread: ThreadSeed): ThreadState { return { completedTurns: 0, createdAt: thread.createdAt, @@ -151,12 +154,13 @@ export default function plugin(bb: BbPluginApi): void { type: "select", label: "Mode", description: - "Observe logs recommendations without changing threads. Apply enables high-confidence updates.", + "Apply organizes threads automatically. Observe only logs proposed changes.", options: ["observe", "apply"], - default: "observe", + default: "apply", }, }); const queues = new Map>(); + let acceptingWork = true; let disposed = false; async function readState(threadId: string): Promise { @@ -175,6 +179,7 @@ export default function plugin(bb: BbPluginApi): void { } function enqueue(threadId: string, work: () => Promise): Promise { + if (!acceptingWork) return Promise.resolve(); const previous = queues.get(threadId) ?? Promise.resolve(); const current = previous .catch(() => undefined) @@ -217,6 +222,43 @@ export default function plugin(bb: BbPluginApi): void { return loaded; } + async function reconcileInbox( + threadId: string, + state: ThreadState, + phase: "active" | "failed" | "idle", + ): Promise { + const { mode } = await settings.get(); + const thread = (await bb.sdk.threads.get({ threadId })) as Thread; + const shouldBeInInbox = phase !== "active"; + + if (shouldBeInInbox) { + if (thread.pinnedAt !== null) return; + if (mode !== "apply") { + bb.log.info( + `thread=${threadId} phase=${phase} mode=observe action=propose-inbox-pin`, + ); + return; + } + await bb.sdk.threads.pin({ threadId }); + bb.log.info( + `thread=${threadId} phase=${phase} mode=apply action=inbox-pinned`, + ); + return; + } + + if (thread.pinnedAt === null) return; + if (mode !== "apply") { + bb.log.info( + `thread=${threadId} phase=${phase} mode=observe action=propose-inbox-unpin`, + ); + return; + } + await bb.sdk.threads.unpin({ threadId }); + bb.log.info( + `thread=${threadId} phase=${phase} mode=apply action=inbox-unpinned`, + ); + } + async function applySection( thread: Thread, state: ThreadState, @@ -466,6 +508,57 @@ export default function plugin(bb: BbPluginApi): void { } } + async function reconcileExistingThreads(signal?: AbortSignal): Promise { + let offset = 0; + const threads: Awaited> = + []; + while (!disposed && !signal?.aborted) { + const page = await bb.sdk.threads.list({ + excludeSideChats: true, + limit: THREAD_LIST_PAGE_SIZE, + offset, + ...(signal === undefined ? {} : { signal }), + }); + threads.push(...page); + if (page.length < THREAD_LIST_PAGE_SIZE) break; + offset += THREAD_LIST_PAGE_SIZE; + } + if (signal?.aborted) return; + await Promise.all( + threads.map((thread) => + enqueue(thread.id, async () => { + if (!isManageableThread(thread)) return; + const fresh = (await bb.sdk.threads.get({ + threadId: thread.id, + })) as Thread; + if (!isManageableThread(fresh)) { + await bb.storage.kv.delete(stateKey(thread.id)); + return; + } + let state = await readState(thread.id); + const adopted = state === null; + if (state === null) { + state = initialState(fresh); + await saveState(thread.id, state); + } + await reconcileInbox( + thread.id, + state, + fresh.status === "idle" + ? "idle" + : fresh.status === "error" + ? "failed" + : "active", + ); + await saveState(thread.id, state); + if (adopted && isEligibleThread(fresh)) { + await evaluate(thread.id, "settings"); + } + }), + ), + ); + } + bb.events.on("thread.created", ({ thread }) => enqueue(thread.id, async () => { if (!isEligibleThread(thread)) return; @@ -477,7 +570,10 @@ export default function plugin(bb: BbPluginApi): void { bb.events.on("thread.active", ({ thread }) => enqueue(thread.id, async () => { - if ((await readState(thread.id)) === null) return; + const state = await readState(thread.id); + if (state === null) return; + await reconcileInbox(thread.id, state, "active"); + await saveState(thread.id, state); await evaluate(thread.id, "active"); }), ); @@ -501,11 +597,21 @@ export default function plugin(bb: BbPluginApi): void { state.completedTurns, ); } + await reconcileInbox(thread.id, state, "idle"); await saveState(thread.id, state); if (due) await evaluate(thread.id, "turn"); }), ); + bb.events.on("thread.failed", ({ thread }) => + enqueue(thread.id, async () => { + const state = await readState(thread.id); + if (state === null) return; + await reconcileInbox(thread.id, state, "failed"); + await saveState(thread.id, state); + }), + ); + const forget = (threadId: string) => enqueue(threadId, async () => { await bb.storage.kv.delete(stateKey(threadId)); @@ -522,9 +628,25 @@ export default function plugin(bb: BbPluginApi): void { .then((keys) => Promise.all( keys.map((key) => - enqueue(key.slice(STATE_PREFIX.length), () => - evaluate(key.slice(STATE_PREFIX.length), "settings"), - ), + enqueue(key.slice(STATE_PREFIX.length), async () => { + const threadId = key.slice(STATE_PREFIX.length); + const state = await readState(threadId); + if (state === null) return; + const thread = (await bb.sdk.threads.get({ + threadId, + })) as Thread; + await reconcileInbox( + threadId, + state, + thread.status === "idle" + ? "idle" + : thread.status === "error" + ? "failed" + : "active", + ); + await saveState(threadId, state); + await evaluate(threadId, "settings"); + }), ), ), ) @@ -534,7 +656,13 @@ export default function plugin(bb: BbPluginApi): void { }); }); - bb.onDispose(() => { + bb.background.service("startup-reconciliation", { + start: (signal) => reconcileExistingThreads(signal), + }); + + bb.onDispose(async () => { + acceptingWork = false; + await Promise.allSettled([...queues.values()]); disposed = true; }); void settings