diff --git a/package-lock.json b/package-lock.json index 2a16263..c5e53ee 100644 --- a/package-lock.json +++ b/package-lock.json @@ -6760,6 +6760,7 @@ "better-sqlite3": "^12.10.0", "cron-parser": "^5.5.0", "hono": "^4.11.9", + "jsdom": "^29.0.1", "typescript": "^5.7.0", "vitest": "^4.1.8", "zod": "^4.3.6" diff --git a/plugins/thread-organizer/README.md b/plugins/thread-organizer/README.md index 4de7de7..9782271 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 inboxMode observe ``` The setting takes effect immediately. @@ -29,27 +29,38 @@ The setting takes effect immediately. | Existing section | Automatic use | | --- | --- | | `bb` or `bb quick fixes` | bb engineering work | -| `Extensions` | Plugins, skills, automations, and agent tooling | +| `bb Extensions` or `Extensions` | Plugins, skills, automations, and agent tooling | | `Design` | Design systems, UI patterns, information architecture, and API-surface work | +| `moss` | Moss product and engineering work that is not more specifically design, writing, QA, or extension work | +| `QA` | Explicit code review, regression, test, coverage, and release-audit work | | `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 and every five seconds | Adopts 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 | Applies any missing semantic section before removing an organizer-owned inbox pin | +| Idle or failed | Pins the thread into the top inbox area without clearing its semantic section | +| Manual pin | Preserves the pin until the user removes it | +| Manual unpin | Keeps the idle thread out of the inbox until its next run completes and collapses its destination 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. + +Pin ownership is tracked separately from pin visibility. Thread Organizer only removes pins it created; a manual pin or re-pin is preserved. Recurring reconciliation also detects manual unpins that do not emit a thread lifecycle event. + +Section decisions are persisted with the classifier version, completed-turn count, confidence, margin, reasons, and evaluation time. Pinning and unpinning reuse that record instead of rerunning categorization; new conversational evidence and classifier upgrades are the only automatic reevaluation triggers. New decisions are written to the plugin log. + +Every sidebar section, including the inbox, starts collapsed. If BB expands a destination while moving an unpinned thread, the plugin collapses it again; a section the user deliberately expands remains open. ## Develop diff --git a/plugins/thread-organizer/app.ts b/plugins/thread-organizer/app.ts new file mode 100644 index 0000000..d1a7bf1 --- /dev/null +++ b/plugins/thread-organizer/app.ts @@ -0,0 +1,10 @@ +import { definePluginApp } from "@bb/plugin-sdk/app"; + +import { mountInboxSectionCollapser } from "./sidebar-controller.js"; + +export default definePluginApp((app) => { + app.experimental_contentScripts.register({ + id: "collapse-unpinned-destination", + mount: ({ signal }) => mountInboxSectionCollapser({ signal }), + }); +}); diff --git a/plugins/thread-organizer/core.test.ts b/plugins/thread-organizer/core.test.ts index 3f380b4..8b648c4 100644 --- a/plugins/thread-organizer/core.test.ts +++ b/plugins/thread-organizer/core.test.ts @@ -74,6 +74,32 @@ describe("section classification", () => { texts: ["Rewrite the doctrine docs."], title: "Rewrite doctrine docs", }, + { + expected: "moss", + projectName: "moss", + texts: ["Update the Claude Code version shipped in Moss."], + title: "Moss engineering work", + }, + { + expected: "moss", + projectName: "Personal", + texts: ["Debug sync persistence in the Moss desktop app."], + title: "Personal Moss work", + }, + { + expected: "qa", + projectName: "bb plugins", + texts: [ + "Review-only: audit the plugin PR for regressions and missing tests.", + ], + title: "Explicit QA plugin work", + }, + { + expected: "bb", + projectName: "Personal", + texts: ["Prototype a nicer JSON timeline display in bb."], + title: "Personal bb product work", + }, { expected: "design", projectName: "UI Pattern Atlas", @@ -123,16 +149,16 @@ describe("section classification", () => { ).toBeNull(); }); - it("surfaces a low margin instead of hiding mixed bb design intent", () => { + it("lets explicit design intent override a repository default", () => { const decision = classifySection({ projectName: "bb", texts: ["Design a durable UI pattern system."], }); expect(decision).toMatchObject({ - runnerUp: "bb", + runnerUp: null, target: "design", }); - expect(decision?.margin).toBeCloseTo(0.05); + expect(decision?.margin).toBeCloseTo(0.95); }); }); @@ -161,10 +187,12 @@ describe("existing section resolution", () => { ).toBeNull(); }); - it("never treats QA as a target alias", () => { - expect( - resolveSectionId([{ id: "sec_qa", name: "QA" }], "design"), - ).toBeNull(); + it.each([ + ["extensions", "bb Extensions", "sec_extensions"], + ["moss", "moss", "sec_moss"], + ["qa", "QA", "sec_qa"], + ] as const)("resolves %s through the live %s section", (target, name, id) => { + expect(resolveSectionId([{ id, name }], target)).toBe(id); }); }); diff --git a/plugins/thread-organizer/core.ts b/plugins/thread-organizer/core.ts index 7142639..23647b5 100644 --- a/plugins/thread-organizer/core.ts +++ b/plugins/thread-organizer/core.ts @@ -2,6 +2,8 @@ export const SECTION_TARGETS = [ "bb", "extensions", "design", + "moss", + "qa", "writing", ] as const; @@ -40,7 +42,9 @@ export interface TitleCandidate { const SECTION_ALIASES: Record = { bb: ["bb", "bb quick fixes"], design: ["design"], - extensions: ["extensions"], + extensions: ["extensions", "bb extensions"], + moss: ["moss"], + qa: ["qa", "quality assurance"], writing: ["writing"], }; @@ -161,7 +165,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 +174,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" ); @@ -231,26 +241,88 @@ export function classifySection(input: { } if ( - ["bb plugins", "prompt shaper"].includes(project) || - (project !== "bb" && - matches( - corpus, - /\b(bb\s+plugin|plugin|skill|automation|agent tool|agent tooling)\b/i, - )) + matches( + corpus, + /\b(review-only|code review|qa|quality assurance|regression (?:review|test|coverage)|test audit|coverage audit|audit (?:a |the )?(?:pull request|pr|release|tests?))\b/i, + ) + ) { + setScore(scores, "qa", 0.98, "explicit verification intent"); + } + + const hasCrossCuttingIntent = + scores.has("design") || scores.has("qa") || scores.has("writing"); + if ( + !hasCrossCuttingIntent && + ([ + "bb plugins", + "design doctrine", + "loop-machine", + "moss-skills", + "ottonomous", + "prompt shaper", + ].includes(project) || + (project !== "bb" && + matches( + corpus, + /\b(bb\s+plugin|plugin|skill|automation|agent tool|agent tooling)\b/i, + ))) ) { setScore( scores, "extensions", - project === "bb plugins" || project === "prompt shaper" ? 0.98 : 0.96, - project === "bb plugins" || project === "prompt shaper" + [ + "bb plugins", + "design doctrine", + "loop-machine", + "moss-skills", + "ottonomous", + "prompt shaper", + ].includes(project) + ? 0.98 + : 0.96, + [ + "bb plugins", + "design doctrine", + "loop-machine", + "moss-skills", + "ottonomous", + "prompt shaper", + ].includes(project) ? "extension project identity" : "explicit extension intent", ); - } else if (project === "design doctrine") { - setScore(scores, "extensions", 0.9, "extension project identity"); } - if (project === "bb") { + const hasSpecificIntent = [...scores.keys()].some((target) => + ["design", "extensions", "qa", "writing"].includes(target), + ); + if ( + !hasSpecificIntent && + (["moss", "moss collab recovery", "moss-collab"].includes(project) || + matches(corpus, /\bmoss(?:-collab)?\b/i)) + ) { + setScore( + scores, + "moss", + ["moss", "moss collab recovery", "moss-collab"].includes(project) + ? 0.97 + : 0.94, + ["moss", "moss collab recovery", "moss-collab"].includes(project) + ? "moss project identity" + : "explicit moss product intent", + ); + } + + if ( + project !== "bb" && + !hasSpecificIntent && + !scores.has("extensions") && + matches(corpus, /\bbb\b/i) + ) { + setScore(scores, "bb", 0.94, "explicit bb product intent"); + } + + if (project === "bb" && !hasCrossCuttingIntent) { setScore(scores, "bb", 0.9, "bb project identity"); if ( matches( diff --git a/plugins/thread-organizer/package.json b/plugins/thread-organizer/package.json index 98d9c1f..b046758 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,11 +21,12 @@ }, "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" }, "server": "./server.ts", + "app": "./app.ts", "skills": [] }, "devDependencies": { @@ -35,6 +36,7 @@ "better-sqlite3": "^12.10.0", "cron-parser": "^5.5.0", "hono": "^4.11.9", + "jsdom": "^29.0.1", "typescript": "^5.7.0", "vitest": "^4.1.8", "zod": "^4.3.6" diff --git a/plugins/thread-organizer/server.test.ts b/plugins/thread-organizer/server.test.ts index 786edf0..c013f42 100644 --- a/plugins/thread-organizer/server.test.ts +++ b/plugins/thread-organizer/server.test.ts @@ -9,7 +9,8 @@ import plugin from "./server.js"; const sections = [ { id: "sec_bb", name: "bb" }, { id: "sec_design", name: "Design" }, - { id: "sec_extensions", name: "Extensions" }, + { id: "sec_extensions", name: "bb Extensions" }, + { id: "sec_moss", name: "moss" }, { id: "sec_qa", name: "QA" }, { id: "sec_writing", name: "Writing" }, ]; @@ -35,6 +36,10 @@ function completedEvent(seq: number) { } function createHarness(input?: { + archiveAfterList?: boolean; + existingThreads?: boolean; + getFailures?: number; + legacyModeOnly?: boolean; mode?: "apply" | "observe"; projectName?: string; prompt?: string; @@ -52,6 +57,14 @@ function createHarness(input?: { ...input?.thread, }); const events: ReturnType[] = []; + type ThreadChangedCallback = (event: { + changes: readonly ("pin-state-changed" | "status-changed")[]; + entity: "thread"; + id?: string; + type: "changed"; + }) => void; + const threadChangedCallbacks = new Set(); + let remainingGetFailures = input?.getFailures ?? 0; let promptHistory = [promptEntry(prompt)]; const update = vi.fn( async (args: { @@ -72,10 +85,35 @@ 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" }, + settings: input?.legacyModeOnly + ? { mode: input?.mode ?? "observe" } + : { inboxMode: input?.mode ?? "observe" }, sdk: { + subscribe: ({ callback: realtimeCallback }) => { + const callback = realtimeCallback as ThreadChangedCallback; + threadChangedCallbacks.add(callback); + return () => { + threadChangedCallbacks.delete(callback); + }; + }, projects: { get: async () => ({ id: "proj_test", @@ -90,8 +128,27 @@ function createHarness(input?: { return events.find((event) => event.seq > after) ?? null; }, }, - get: async () => thread, + get: async () => { + if (remainingGetFailures > 0) { + remainingGetFailures -= 1; + throw new Error("transient get failure"); + } + return 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, }, }, @@ -106,6 +163,18 @@ function createHarness(input?: { currentThread() { return thread; }, + emitThreadChanged( + ...changes: ("pin-state-changed" | "status-changed")[] + ): void { + for (const callback of threadChangedCallbacks) { + callback({ + changes, + entity: "thread", + id: thread.id, + type: "changed", + }); + } + }, setPromptHistory(...texts: string[]): void { promptHistory = texts.map((text, index) => promptEntry(text, index + 1), @@ -116,23 +185,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"] }, + inboxMode: { 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(); @@ -140,6 +212,32 @@ describe("Thread Organizer plugin", () => { await harness.lifecycle.dispose(); }); + it("uses the new apply default instead of a stored legacy observe value", async () => { + const organizer = createHarness({ + existingThreads: true, + legacyModeOnly: true, + mode: "observe", + thread: { status: "idle" }, + }); + plugin(organizer.bb); + + const service = organizer.harness.behavior.runService( + "inbox-reconciliation", + ); + await vi.waitFor(() => { + expect(organizer.pin).toHaveBeenCalledWith({ threadId: "thr_test" }); + }); + service.controller.abort(); + await service.done; + + expect(organizer.harness.inspection.logEntries).toContainEqual( + expect.objectContaining({ + message: expect.stringContaining("Thread Organizer loaded mode=apply"), + }), + ); + await organizer.harness.lifecycle.dispose(); + }); + it("logs a recommendation without changing a thread in observe mode", async () => { const { bb, harness, currentThread, update } = createHarness(); plugin(bb); @@ -200,6 +298,837 @@ 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("reuses its durable section decision during pin lifecycle changes", async () => { + const organizer = createHarness({ mode: "apply" }); + plugin(organizer.bb); + await organizer.harness.behavior.emitThreadEvent("thread.created", { + thread: organizer.currentThread(), + }); + const promptHistoryCalls = + organizer.harness.inspection.sdk.callsTo("threads.promptHistory").length; + const projectCalls = + organizer.harness.inspection.sdk.callsTo("projects.get").length; + const sectionListCalls = + organizer.harness.inspection.sdk.callsTo("threadSections.list").length; + + organizer.setThread({ status: "idle" }); + await organizer.harness.behavior.emitThreadEvent("thread.idle", { + lastAssistantText: "Done.", + thread: organizer.currentThread(), + }); + organizer.setThread({ status: "active" }); + await organizer.harness.behavior.emitThreadEvent("thread.active", { + thread: organizer.currentThread(), + }); + + expect( + organizer.harness.inspection.sdk.callsTo("threads.promptHistory"), + ).toHaveLength(promptHistoryCalls); + expect( + organizer.harness.inspection.sdk.callsTo("projects.get"), + ).toHaveLength(projectCalls); + expect( + organizer.harness.inspection.sdk.callsTo("threadSections.list"), + ).toHaveLength(sectionListCalls); + expect( + await organizer.bb.storage.kv.get("thread:v1:thr_test"), + ).toMatchObject({ + sectionClassification: { + decision: { target: "extensions" }, + }, + }); + await organizer.harness.lifecycle.dispose(); + }); + + it("remembers organizer-owned pins across reloads", 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(), + }); + expect( + await organizer.bb.storage.kv.get("thread:v1:thr_test"), + ).toMatchObject({ + inboxManagedPinnedAt: organizer.currentThread().pinnedAt, + }); + + const reloaded = await organizer.harness.lifecycle.reload(plugin); + organizer.setThread({ status: "active" }); + await reloaded.harness.behavior.emitThreadEvent("thread.active", { + thread: organizer.currentThread(), + }); + + expect(organizer.unpin).toHaveBeenCalledWith({ threadId: "thr_test" }); + expect(organizer.currentThread().pinnedAt).toBeNull(); + await reloaded.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); + + const service = organizer.harness.behavior.runService( + "inbox-reconciliation", + ); + await vi.waitFor(() => { + expect(organizer.pin).toHaveBeenCalledWith({ threadId: "thr_test" }); + }); + service.controller.abort(); + await service.done; + expect(organizer.currentThread().pinnedAt).not.toBeNull(); + expect( + organizer.harness.inspection.sdk.callsTo("threads.list")[0]?.[0], + ).toMatchObject({ + archived: false, + excludeSideChats: true, + hasParent: false, + }); + 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); + + const service = organizer.harness.behavior.runService( + "inbox-reconciliation", + ); + await vi.waitFor(() => { + expect(organizer.pin).toHaveBeenCalledWith({ threadId: "thr_test" }); + }); + service.controller.abort(); + await service.done; + 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); + + const service = organizer.harness.behavior.runService( + "inbox-reconciliation", + ); + await vi.waitFor(() => { + expect( + organizer.harness.inspection.sdk.callsTo("threads.get"), + ).toHaveLength(1); + }); + service.controller.abort(); + await service.done; + expect(organizer.pin).not.toHaveBeenCalled(); + expect(await organizer.bb.storage.kv.list("thread:")).toEqual([]); + await organizer.harness.lifecycle.dispose(); + }); + + it("uses realtime changes instead of rescanning on the old five-second interval", async () => { + vi.useFakeTimers(); + try { + const organizer = createHarness({ + existingThreads: true, + mode: "apply", + thread: { status: "idle" }, + }); + plugin(organizer.bb); + + const service = organizer.harness.behavior.runService( + "inbox-reconciliation", + ); + let settled = false; + void service.done.then(() => { + settled = true; + }); + await vi.advanceTimersByTimeAsync(0); + expect(settled).toBe(false); + expect(organizer.pin).toHaveBeenCalledWith({ threadId: "thr_test" }); + const listCalls = + organizer.harness.inspection.sdk.callsTo("threads.list").length; + + organizer.setThread({ status: "active" }); + await vi.advanceTimersByTimeAsync(5_000); + + expect( + organizer.harness.inspection.sdk.callsTo("threads.list"), + ).toHaveLength(listCalls); + expect(organizer.unpin).not.toHaveBeenCalled(); + + organizer.emitThreadChanged("status-changed"); + await vi.advanceTimersByTimeAsync(0); + + expect(organizer.unpin).toHaveBeenCalledWith({ + threadId: "thr_test", + }); + service.controller.abort(); + await service.done; + await organizer.harness.lifecycle.dispose(); + } finally { + vi.useRealTimers(); + } + }); + + it("snoozes a manually unpinned idle thread until its next completed run", async () => { + const organizer = createHarness({ + existingThreads: true, + mode: "apply", + thread: { status: "idle" }, + }); + plugin(organizer.bb); + + const service = organizer.harness.behavior.runService( + "inbox-reconciliation", + ); + await vi.waitFor(() => { + expect(organizer.pin).toHaveBeenCalledTimes(1); + }); + + organizer.setThread({ pinnedAt: null, status: "idle" }); + organizer.emitThreadChanged("pin-state-changed"); + await vi.waitFor(async () => { + expect( + await organizer.bb.storage.kv.get("thread:v1:thr_test"), + ).toMatchObject({ inboxSnoozed: true }); + }); + + expect(organizer.pin).toHaveBeenCalledTimes(1); + expect(organizer.currentThread().pinnedAt).toBeNull(); + + organizer.setThread({ status: "active" }); + await organizer.harness.behavior.emitThreadEvent("thread.active", { + thread: organizer.currentThread(), + }); + organizer.setThread({ status: "idle" }); + await organizer.harness.behavior.emitThreadEvent("thread.idle", { + lastAssistantText: "Done.", + thread: organizer.currentThread(), + }); + + expect(organizer.pin).toHaveBeenCalledTimes(2); + expect(organizer.currentThread().pinnedAt).not.toBeNull(); + service.controller.abort(); + await service.done; + await organizer.harness.lifecycle.dispose(); + }); + + it("retries a transient reconciliation failure without a reload", async () => { + const organizer = createHarness({ + existingThreads: true, + getFailures: 1, + mode: "apply", + thread: { status: "idle" }, + }); + plugin(organizer.bb); + + const service = organizer.harness.behavior.runService( + "inbox-reconciliation", + ); + await vi.waitFor(() => { + expect(organizer.pin).toHaveBeenCalledWith({ threadId: "thr_test" }); + }); + service.controller.abort(); + await service.done; + + expect(organizer.harness.inspection.logEntries).toContainEqual( + expect.objectContaining({ + message: expect.stringContaining( + "action=reconciliation-retry attempt=1 error=transient get failure", + ), + }), + ); + await organizer.harness.lifecycle.dispose(); + }); + + it("surfaces an exhausted reconciliation failure to the service host", async () => { + vi.useFakeTimers(); + try { + const organizer = createHarness({ + existingThreads: true, + getFailures: 3, + mode: "apply", + thread: { status: "idle" }, + }); + plugin(organizer.bb); + + const service = organizer.harness.behavior.runService( + "inbox-reconciliation", + ); + const outcome = service.done.catch((error: unknown) => error); + await vi.advanceTimersByTimeAsync(600); + + await expect(outcome).resolves.toEqual( + expect.objectContaining({ message: "transient get failure" }), + ); + expect(organizer.pin).not.toHaveBeenCalled(); + await organizer.harness.lifecycle.dispose(); + } finally { + vi.useRealTimers(); + } + }); + + it("bounds reconciliation workers and admits no mutation after abort", async () => { + const threads = Array.from({ length: 8 }, (_, index) => + makeThreadResponse({ + id: `thr_${index}`, + projectId: "proj_test", + status: "error", + titleFallback: `Thread ${index}`, + }), + ); + let releaseGets: () => void = () => undefined; + const getGate = new Promise((resolve) => { + releaseGets = resolve; + }); + let getCalls = 0; + const pin = vi.fn(); + const host = createFakePluginHost({ + pluginId: "thread-organizer", + settings: { inboxMode: "apply" }, + sdk: { + threads: { + get: async ({ threadId }) => { + getCalls += 1; + await getGate; + return threads.find((thread) => thread.id === threadId)!; + }, + list: async () => threads, + pin, + }, + }, + }); + plugin(host.bb); + + const service = host.harness.behavior.runService( + "inbox-reconciliation", + ); + await vi.waitFor(() => { + expect(getCalls).toBe(4); + }); + service.controller.abort(); + releaseGets(); + await service.done; + + expect(getCalls).toBe(4); + expect(pin).not.toHaveBeenCalled(); + await host.harness.lifecycle.dispose(); + }); + + it("repeats discovery to close a pin-order pagination gap", async () => { + const threads = Array.from({ length: 101 }, (_, index) => + makeThreadResponse({ + id: `thr_${index}`, + projectId: "proj_test", + status: "error", + titleFallback: `Thread ${index}`, + }), + ); + const byId = new Map(threads.map((thread) => [thread.id, thread])); + const pinned = new Set(); + let pass = 0; + const host = createFakePluginHost({ + pluginId: "thread-organizer", + settings: { inboxMode: "apply" }, + sdk: { + threads: { + get: async ({ threadId }) => byId.get(threadId)!, + list: async (args) => { + const offset = args?.offset ?? 0; + if (offset === 0) { + return pass === 0 + ? threads.slice(0, 100) + : threads.slice(1, 101); + } + const duplicate = threads[pass === 0 ? 99 : 100]!; + pass += 1; + return [duplicate]; + }, + pin: async ({ threadId }) => { + pinned.add(threadId); + const thread = byId.get(threadId)!; + const updated = makeThreadResponse({ + ...thread, + pinnedAt: thread.updatedAt + 1, + }); + byId.set(threadId, updated); + return updated; + }, + }, + }, + }); + plugin(host.bb); + + const service = host.harness.behavior.runService( + "inbox-reconciliation", + ); + await vi.waitFor(() => { + expect(pinned.size).toBe(101); + }); + service.controller.abort(); + await service.done; + + expect(pinned).toEqual(new Set(threads.map((thread) => thread.id))); + await host.harness.lifecycle.dispose(); + }); + + it("preserves a manually pinned active thread", 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).not.toHaveBeenCalled(); + expect(organizer.currentThread().pinnedAt).toBe(10); + await organizer.harness.lifecycle.dispose(); + }); + + it("keeps a same-run manual unpin snoozed until another run starts", async () => { + const organizer = createHarness({ + mode: "apply", + thread: { pinnedAt: 10, status: "active" }, + }); + plugin(organizer.bb); + await organizer.harness.behavior.emitThreadEvent("thread.created", { + thread: organizer.currentThread(), + }); + await organizer.harness.behavior.emitThreadEvent("thread.active", { + thread: organizer.currentThread(), + }); + + organizer.setThread({ pinnedAt: null }); + organizer.emitThreadChanged("pin-state-changed"); + await vi.waitFor(async () => { + expect( + await organizer.bb.storage.kv.get("thread:v1:thr_test"), + ).toMatchObject({ + inboxLastPhase: "active", + inboxSnoozed: true, + }); + }); + + organizer.setThread({ status: "idle" }); + await organizer.harness.behavior.emitThreadEvent("thread.idle", { + lastAssistantText: "Done.", + thread: organizer.currentThread(), + }); + expect(organizer.pin).not.toHaveBeenCalled(); + + organizer.setThread({ status: "active" }); + await organizer.harness.behavior.emitThreadEvent("thread.active", { + thread: organizer.currentThread(), + }); + organizer.setThread({ status: "idle" }); + await organizer.harness.behavior.emitThreadEvent("thread.idle", { + lastAssistantText: "Done again.", + thread: organizer.currentThread(), + }); + + expect(organizer.pin).toHaveBeenCalledTimes(1); + expect(organizer.currentThread().pinnedAt).not.toBeNull(); + await organizer.harness.lifecycle.dispose(); + }); + + it("treats a coalesced prior-run unpin followed by chat as a new run", 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(), + }); + expect(organizer.pin).toHaveBeenCalledTimes(1); + + organizer.setThread({ pinnedAt: null, status: "active" }); + await organizer.harness.behavior.emitThreadEvent("thread.active", { + thread: organizer.currentThread(), + }); + expect( + await organizer.bb.storage.kv.get("thread:v1:thr_test"), + ).toMatchObject({ + inboxLastPhase: "active", + inboxSnoozed: false, + }); + + organizer.setThread({ status: "idle" }); + await organizer.harness.behavior.emitThreadEvent("thread.idle", { + lastAssistantText: "Done again.", + thread: organizer.currentThread(), + }); + expect(organizer.pin).toHaveBeenCalledTimes(2); + expect(organizer.currentThread().pinnedAt).not.toBeNull(); + await organizer.harness.lifecycle.dispose(); + }); + + it("snoozes an unpin that happens after the active run starts", 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(), + }); + expect(organizer.pin).toHaveBeenCalledTimes(1); + + let signalGetStarted: () => void = () => undefined; + const getStarted = new Promise((resolve) => { + signalGetStarted = resolve; + }); + let releaseGet: () => void = () => undefined; + const getGate = new Promise((resolve) => { + releaseGet = resolve; + }); + let gateNextGet = true; + organizer.harness.inspection.sdk.stub("threads.get", async () => { + if (gateNextGet) { + gateNextGet = false; + signalGetStarted(); + await getGate; + } + return organizer.currentThread(); + }); + + organizer.setThread({ status: "active" }); + const activeSnapshot = organizer.currentThread(); + const activeEvent = + organizer.harness.behavior.emitThreadEvent("thread.active", { + thread: activeSnapshot, + }); + await getStarted; + + organizer.setThread({ pinnedAt: null }); + releaseGet(); + await activeEvent; + expect( + await organizer.bb.storage.kv.get("thread:v1:thr_test"), + ).toMatchObject({ + inboxLastPhase: "active", + inboxSnoozed: true, + }); + + organizer.setThread({ status: "idle" }); + await organizer.harness.behavior.emitThreadEvent("thread.idle", { + lastAssistantText: "Still done.", + thread: organizer.currentThread(), + }); + expect(organizer.pin).toHaveBeenCalledTimes(1); + await organizer.harness.lifecycle.dispose(); + }); + + it("does not adopt a manual pin after an organizer pin request rejects", async () => { + const organizer = createHarness({ mode: "apply" }); + plugin(organizer.bb); + await organizer.harness.behavior.emitThreadEvent("thread.created", { + thread: organizer.currentThread(), + }); + organizer.pin.mockRejectedValueOnce(new Error("pin rejected")); + + organizer.setThread({ status: "idle" }); + await organizer.harness.behavior.emitThreadEvent("thread.idle", { + lastAssistantText: "Done.", + thread: organizer.currentThread(), + }); + expect( + await organizer.bb.storage.kv.get("thread:v1:thr_test"), + ).toMatchObject({ + inboxManagedPinnedAt: null, + inboxPendingPin: false, + }); + + organizer.setThread({ pinnedAt: 100 }); + organizer.emitThreadChanged("pin-state-changed"); + await vi.waitFor(async () => { + expect( + await organizer.bb.storage.kv.get("thread:v1:thr_test"), + ).toMatchObject({ + inboxManagedPinnedAt: null, + inboxObservedPinned: true, + }); + }); + organizer.setThread({ status: "active" }); + await organizer.harness.behavior.emitThreadEvent("thread.active", { + thread: organizer.currentThread(), + }); + + expect(organizer.unpin).not.toHaveBeenCalled(); + expect(organizer.currentThread().pinnedAt).toBe(100); + await organizer.harness.lifecycle.dispose(); + }); + + it("snoozes a manual unpin after an organizer unpin request rejects", 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.unpin.mockRejectedValueOnce(new Error("unpin rejected")); + + organizer.setThread({ status: "active" }); + await organizer.harness.behavior.emitThreadEvent("thread.active", { + thread: organizer.currentThread(), + }); + expect( + await organizer.bb.storage.kv.get("thread:v1:thr_test"), + ).toMatchObject({ inboxPendingUnpin: false }); + + organizer.setThread({ pinnedAt: null }); + organizer.emitThreadChanged("pin-state-changed"); + await vi.waitFor(async () => { + expect( + await organizer.bb.storage.kv.get("thread:v1:thr_test"), + ).toMatchObject({ + inboxLastPhase: "active", + inboxSnoozed: true, + }); + }); + + organizer.setThread({ status: "idle" }); + await organizer.harness.behavior.emitThreadEvent("thread.idle", { + lastAssistantText: "Still done.", + thread: organizer.currentThread(), + }); + expect(organizer.pin).toHaveBeenCalledTimes(1); + await organizer.harness.lifecycle.dispose(); + }); + + it("uses fresh status when lifecycle events arrive out of order", async () => { + const organizer = createHarness({ mode: "apply" }); + plugin(organizer.bb); + await organizer.harness.behavior.emitThreadEvent("thread.created", { + thread: organizer.currentThread(), + }); + + organizer.setThread({ status: "active" }); + await organizer.harness.behavior.emitThreadEvent("thread.idle", { + lastAssistantText: "Stale idle event.", + thread: makeThreadResponse({ + ...organizer.currentThread(), + status: "idle", + }), + }); + expect(organizer.pin).not.toHaveBeenCalled(); + + organizer.setThread({ status: "idle" }); + await organizer.harness.behavior.emitThreadEvent("thread.active", { + thread: makeThreadResponse({ + ...organizer.currentThread(), + status: "active", + }), + }); + expect(organizer.pin).toHaveBeenCalledTimes(1); + await organizer.harness.lifecycle.dispose(); + }); + + it("recovers organizer pin ownership after the post-pin state save fails", async () => { + const organizer = createHarness({ mode: "apply" }); + plugin(organizer.bb); + await organizer.harness.behavior.emitThreadEvent("thread.created", { + thread: organizer.currentThread(), + }); + + const originalSet = organizer.bb.storage.kv.set.bind( + organizer.bb.storage.kv, + ); + let failPostPinSave = true; + vi.spyOn(organizer.bb.storage.kv, "set").mockImplementation( + async (key, value) => { + if ( + failPostPinSave && + organizer.currentThread().pinnedAt !== null + ) { + failPostPinSave = false; + throw new Error("post-pin save failed"); + } + await originalSet(key, value); + }, + ); + + organizer.setThread({ status: "idle" }); + await organizer.harness.behavior.emitThreadEvent("thread.idle", { + lastAssistantText: "Done.", + thread: organizer.currentThread(), + }); + expect(organizer.currentThread().pinnedAt).not.toBeNull(); + expect( + await organizer.bb.storage.kv.get("thread:v1:thr_test"), + ).toMatchObject({ inboxPendingPin: true }); + + const reloaded = await organizer.harness.lifecycle.reload(plugin); + organizer.emitThreadChanged("pin-state-changed"); + await vi.waitFor(async () => { + expect( + await reloaded.bb.storage.kv.get("thread:v1:thr_test"), + ).toMatchObject({ + inboxManagedPinnedAt: organizer.currentThread().pinnedAt, + inboxPendingPin: false, + }); + }); + + organizer.setThread({ status: "active" }); + await reloaded.harness.behavior.emitThreadEvent("thread.active", { + thread: organizer.currentThread(), + }); + expect(organizer.unpin).toHaveBeenCalledWith({ + threadId: "thr_test", + }); + await reloaded.harness.lifecycle.dispose(); + }); + + it("preserves a manual re-pin of an organizer-managed thread", 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: null, + status: "idle", + }); + await organizer.harness.behavior.emitThreadEvent("thread.idle", { + lastAssistantText: "Still done.", + thread: organizer.currentThread(), + }); + organizer.setThread({ + pinnedAt: 100, + status: "active", + }); + await organizer.harness.behavior.emitThreadEvent("thread.active", { + thread: organizer.currentThread(), + }); + + expect(organizer.unpin).not.toHaveBeenCalled(); + expect(organizer.currentThread().pinnedAt).toBe(100); + 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({ inboxMode: "apply" }); + + await vi.waitFor(() => { + expect(organizer.pin).toHaveBeenCalledWith({ threadId: "thr_test" }); + }); + await organizer.harness.lifecycle.dispose(); + }); + + it("removes archived stored state before applying inbox changes", async () => { + const organizer = createHarness({ mode: "observe" }); + plugin(organizer.bb); + await organizer.harness.behavior.emitThreadEvent("thread.created", { + thread: organizer.currentThread(), + }); + expect( + await organizer.bb.storage.kv.list("thread:"), + ).toHaveLength(1); + + organizer.setThread({ archivedAt: 100, status: "idle" }); + await organizer.harness.behavior.setSettings({ inboxMode: "apply" }); + await vi.waitFor(async () => { + expect(await organizer.bb.storage.kv.list("thread:")).toEqual([]); + }); + + expect(organizer.pin).not.toHaveBeenCalled(); + 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..5955a2e 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,21 +19,42 @@ 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; +const RECONCILIATION_CONCURRENCY = 4; +const RECONCILIATION_INTERVAL_MS = 5 * 60_000; +const RECONCILIATION_RETRY_DELAYS_MS = [100, 500] as const; +const SECTION_CLASSIFIER_VERSION = 2; type Thread = Awaited>; +type ThreadSeed = Pick; type EvaluationPhase = "active" | "created" | "settings" | "turn"; +interface SectionClassificationCache { + classifierVersion: number; + completedTurns: number; + contextAvailable: boolean; + decision: SectionClassification | null; + evaluatedAt: number; +} + interface ThreadState { completedTurns: number; createdAt: number; hasAppliedSection: boolean; hasAppliedTitle: boolean; + inboxManagedPinnedAt: number | null; + inboxObservedPinned: boolean; + inboxPendingPin: boolean; + inboxPendingUnpin: boolean; + inboxLastPhase: "active" | "failed" | "idle" | null; + inboxSnoozed: boolean; lastAppliedSectionId: string | null; lastAppliedTitle: string | null; lastCompletedSeq: number; nextEvaluationTurn: number; pendingSectionId: string | null; pendingSectionStreak: number; + sectionClassification: SectionClassificationCache | null; sectionLocked: boolean; titleLocked: boolean; version: 1; @@ -42,18 +64,25 @@ 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, hasAppliedSection: false, hasAppliedTitle: false, + inboxManagedPinnedAt: null, + inboxObservedPinned: false, + inboxPendingPin: false, + inboxPendingUnpin: false, + inboxLastPhase: null, + inboxSnoozed: false, lastAppliedSectionId: null, lastAppliedTitle: null, lastCompletedSeq: 0, nextEvaluationTurn: 1, pendingSectionId: null, pendingSectionStreak: 0, + sectionClassification: null, sectionLocked: thread.sectionId !== null, titleLocked: thread.title !== null, version: 1, @@ -69,6 +98,22 @@ function isThreadState(value: unknown): value is ThreadState { typeof state.createdAt === "number" && typeof state.hasAppliedSection === "boolean" && typeof state.hasAppliedTitle === "boolean" && + (typeof state.inboxManagedPinnedAt === "number" || + state.inboxManagedPinnedAt === null || + state.inboxManagedPinnedAt === undefined) && + (typeof state.inboxObservedPinned === "boolean" || + state.inboxObservedPinned === undefined) && + (typeof state.inboxPendingPin === "boolean" || + state.inboxPendingPin === undefined) && + (typeof state.inboxPendingUnpin === "boolean" || + state.inboxPendingUnpin === undefined) && + (state.inboxLastPhase === "active" || + state.inboxLastPhase === "failed" || + state.inboxLastPhase === "idle" || + state.inboxLastPhase === null || + state.inboxLastPhase === undefined) && + (typeof state.inboxSnoozed === "boolean" || + state.inboxSnoozed === undefined) && (typeof state.lastAppliedSectionId === "string" || state.lastAppliedSectionId === null) && (typeof state.lastAppliedTitle === "string" || @@ -78,11 +123,33 @@ function isThreadState(value: unknown): value is ThreadState { (typeof state.pendingSectionId === "string" || state.pendingSectionId === null) && typeof state.pendingSectionStreak === "number" && + (state.sectionClassification === undefined || + state.sectionClassification === null || + (typeof state.sectionClassification === "object" && + typeof state.sectionClassification.classifierVersion === "number" && + typeof state.sectionClassification.completedTurns === "number" && + typeof state.sectionClassification.contextAvailable === "boolean" && + (state.sectionClassification.decision === null || + typeof state.sectionClassification.decision === "object") && + typeof state.sectionClassification.evaluatedAt === "number")) && typeof state.sectionLocked === "boolean" && typeof state.titleLocked === "boolean" ); } +function normalizeThreadState(state: ThreadState): ThreadState { + return { + ...state, + inboxManagedPinnedAt: state.inboxManagedPinnedAt ?? null, + inboxObservedPinned: state.inboxObservedPinned ?? false, + inboxPendingPin: state.inboxPendingPin ?? false, + inboxPendingUnpin: state.inboxPendingUnpin ?? false, + inboxLastPhase: state.inboxLastPhase ?? null, + inboxSnoozed: state.inboxSnoozed ?? false, + sectionClassification: state.sectionClassification ?? null, + }; +} + function syncManualLocks(state: ThreadState, thread: Thread): boolean { let changed = false; if (!state.titleLocked) { @@ -145,24 +212,42 @@ function delay(milliseconds: number): Promise { return new Promise((resolve) => setTimeout(resolve, milliseconds)); } +function abortableDelay( + milliseconds: number, + signal: AbortSignal, +): Promise { + if (signal.aborted) return Promise.resolve(); + return new Promise((resolve) => { + const timeout = setTimeout(finish, milliseconds); + signal.addEventListener("abort", finish, { once: true }); + + function finish(): void { + clearTimeout(timeout); + signal.removeEventListener("abort", finish); + resolve(); + } + }); +} + export default function plugin(bb: BbPluginApi): void { const settings = bb.settings.define({ - mode: { + inboxMode: { 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 { const stored = await bb.storage.kv.get(stateKey(threadId)); if (stored === undefined) return null; - if (isThreadState(stored)) return stored; + if (isThreadState(stored)) return normalizeThreadState(stored); bb.log.warn(`thread=${threadId} action=ignore-invalid-state`); return null; } @@ -174,17 +259,28 @@ export default function plugin(bb: BbPluginApi): void { await bb.storage.kv.set(stateKey(threadId), state); } - function enqueue(threadId: string, work: () => Promise): Promise { + function enqueue( + threadId: string, + work: () => Promise, + containErrors = true, + ): Promise { + if (!acceptingWork) return Promise.resolve(); const previous = queues.get(threadId) ?? Promise.resolve(); - const current = previous + const workPromise = previous .catch(() => undefined) .then(async () => { if (!disposed) await work(); - }) - .catch((error: unknown) => { - const message = error instanceof Error ? error.message : String(error); - bb.log.error(`thread=${threadId} action=queue-failed error=${message}`); - }) + }); + const contained = containErrors + ? workPromise.catch((error: unknown) => { + const message = + error instanceof Error ? error.message : String(error); + bb.log.error( + `thread=${threadId} action=queue-failed error=${message}`, + ); + }) + : workPromise; + const current = contained .finally(() => { if (queues.get(threadId) === current) queues.delete(threadId); }); @@ -217,6 +313,172 @@ export default function plugin(bb: BbPluginApi): void { return loaded; } + async function reconcileInbox( + thread: Thread, + state: ThreadState, + phase: "active" | "failed" | "idle", + signal?: AbortSignal, + runStartPinnedAt?: number | null, + ): Promise { + const { inboxMode } = await settings.get(); + if (signal?.aborted) return; + const threadId = thread.id; + const startedNewRun = + phase === "active" && state.inboxLastPhase !== "active"; + if ( + startedNewRun && + thread.pinnedAt === null && + state.inboxObservedPinned && + runStartPinnedAt === undefined + ) { + bb.log.debug( + `thread=${threadId} phase=${phase} action=await-run-start-pin-observation`, + ); + return; + } + state.inboxLastPhase = phase; + if (startedNewRun) state.inboxSnoozed = false; + + if (state.inboxPendingPin && thread.pinnedAt !== null) { + state.inboxManagedPinnedAt = thread.pinnedAt; + state.inboxObservedPinned = true; + state.inboxPendingPin = false; + bb.log.info( + `thread=${threadId} phase=${phase} action=inbox-pin-adopted`, + ); + } + + let recoveredManagedUnpin = false; + if (state.inboxPendingUnpin && thread.pinnedAt === null) { + state.inboxManagedPinnedAt = null; + state.inboxObservedPinned = false; + state.inboxPendingUnpin = false; + recoveredManagedUnpin = true; + bb.log.info( + `thread=${threadId} phase=${phase} action=inbox-unpin-adopted`, + ); + } + + if ( + thread.pinnedAt === null && + state.inboxObservedPinned && + !recoveredManagedUnpin + ) { + state.inboxManagedPinnedAt = null; + state.inboxObservedPinned = false; + state.inboxPendingPin = false; + state.inboxPendingUnpin = false; + if (startedNewRun && runStartPinnedAt === null) { + bb.log.info( + `thread=${threadId} phase=${phase} action=prior-run-unpin-observed`, + ); + } else { + state.inboxSnoozed = true; + bb.log.info( + `thread=${threadId} phase=${phase} action=inbox-snoozed`, + ); + } + } + + if (phase !== "active") { + if (thread.pinnedAt !== null) { + if ( + state.inboxManagedPinnedAt !== null && + state.inboxManagedPinnedAt !== thread.pinnedAt + ) { + state.inboxManagedPinnedAt = null; + } + state.inboxObservedPinned = true; + return; + } + if (state.inboxSnoozed) return; + if (inboxMode !== "apply") { + bb.log.info( + `thread=${threadId} phase=${phase} mode=observe action=propose-inbox-pin`, + ); + return; + } + if (signal?.aborted) return; + if (!state.inboxPendingPin) { + state.inboxPendingPin = true; + await saveState(threadId, state); + } + if (signal?.aborted) return; + let pinned: Thread; + try { + pinned = await bb.sdk.threads.pin({ threadId }); + } catch (error: unknown) { + const fresh = (await bb.sdk.threads.get({ threadId })) as Thread; + state.inboxPendingPin = false; + if (fresh.pinnedAt === null) { + state.inboxManagedPinnedAt = null; + state.inboxObservedPinned = false; + } else { + state.inboxManagedPinnedAt = null; + state.inboxObservedPinned = true; + } + await saveState(threadId, state); + throw error; + } + state.inboxManagedPinnedAt = pinned.pinnedAt; + state.inboxObservedPinned = true; + state.inboxPendingPin = false; + bb.log.info( + `thread=${threadId} phase=${phase} mode=apply action=inbox-pinned`, + ); + return; + } + + if (thread.pinnedAt === null) { + state.inboxManagedPinnedAt = null; + state.inboxObservedPinned = false; + state.inboxPendingPin = false; + state.inboxPendingUnpin = false; + return; + } + state.inboxObservedPinned = true; + if (state.inboxManagedPinnedAt !== thread.pinnedAt) { + state.inboxManagedPinnedAt = null; + return; + } + if (inboxMode !== "apply") { + bb.log.info( + `thread=${threadId} phase=${phase} mode=observe action=propose-inbox-unpin`, + ); + return; + } + if (signal?.aborted) return; + if (!state.inboxPendingUnpin) { + state.inboxPendingUnpin = true; + await saveState(threadId, state); + } + if (signal?.aborted) return; + try { + await bb.sdk.threads.unpin({ threadId }); + } catch (error: unknown) { + const fresh = (await bb.sdk.threads.get({ threadId })) as Thread; + state.inboxPendingUnpin = false; + if (fresh.pinnedAt === null) { + state.inboxManagedPinnedAt = null; + state.inboxObservedPinned = false; + state.inboxSnoozed = true; + } else { + if (fresh.pinnedAt !== state.inboxManagedPinnedAt) { + state.inboxManagedPinnedAt = null; + } + state.inboxObservedPinned = true; + } + await saveState(threadId, state); + throw error; + } + state.inboxManagedPinnedAt = null; + state.inboxObservedPinned = false; + state.inboxPendingUnpin = false; + bb.log.info( + `thread=${threadId} phase=${phase} mode=apply action=inbox-unpinned`, + ); + } + async function applySection( thread: Thread, state: ThreadState, @@ -337,10 +599,13 @@ export default function plugin(bb: BbPluginApi): void { async function evaluate( threadId: string, phase: EvaluationPhase, + signal?: AbortSignal, ): Promise { const state = await readState(threadId); if (state === null) return; + if (signal?.aborted) return; const thread = (await bb.sdk.threads.get({ threadId })) as Thread; + if (signal?.aborted) return; const locksChanged = syncManualLocks(state, thread); if (locksChanged) { bb.log.info( @@ -356,8 +621,31 @@ export default function plugin(bb: BbPluginApi): void { return; } - const attempts = phase === "active" ? 3 : 1; - const historyTexts = await loadContextTexts(thread, attempts); + const { inboxMode } = await settings.get(); + if (signal?.aborted) return; + const movingManagedSection = + state.hasAppliedSection && thread.sectionId !== null; + const canManageSection = + !state.sectionLocked && + (!movingManagedSection || phase === "turn"); + const cachedClassification = state.sectionClassification; + const needsClassification = + canManageSection && + (cachedClassification === null || + cachedClassification.classifierVersion !== SECTION_CLASSIFIER_VERSION || + (phase === "active" && + (!cachedClassification.contextAvailable || + cachedClassification.decision === null)) || + (phase === "turn" && + cachedClassification.completedTurns < state.completedTurns)); + const needsHistory = needsClassification || phase === "turn"; + const historyTexts = needsHistory + ? await loadContextTexts( + thread, + phase === "active" || phase === "created" ? 3 : 1, + ) + : []; + if (signal?.aborted) return; const texts = [ ...(thread.title === null ? [] : [thread.title]), ...(thread.titleFallback === null ? [] : [thread.titleFallback]), @@ -372,27 +660,40 @@ export default function plugin(bb: BbPluginApi): void { ? [] : [latestPromptText] : texts; - const { mode } = await settings.get(); - const movingManagedSection = - state.hasAppliedSection && thread.sectionId !== null; - if ( - !state.sectionLocked && - (!movingManagedSection || phase === "turn") - ) { + if (canManageSection) { try { - const projectName = - thread.projectId === PERSONAL_PROJECT_ID - ? "Personal" - : ( - await bb.sdk.projects.get({ - projectId: thread.projectId, - }) - ).name; - const decision = classifySection({ - projectName, - texts: sectionTexts, - }); + let decision = cachedClassification?.decision ?? null; + if (needsClassification) { + const projectName = + thread.projectId === PERSONAL_PROJECT_ID + ? "Personal" + : ( + await bb.sdk.projects.get({ + projectId: thread.projectId, + }) + ).name; + decision = classifySection({ + projectName, + texts: sectionTexts, + }); + state.sectionClassification = { + classifierVersion: SECTION_CLASSIFIER_VERSION, + completedTurns: state.completedTurns, + contextAvailable: sectionTexts.some(isSubstantiveText), + decision, + evaluatedAt: Date.now(), + }; + bb.log.info( + decision === null + ? `thread=${threadId} phase=${phase} action=section-classified target=none` + : `thread=${threadId} phase=${phase} action=section-classified ${classificationSummary(decision)}`, + ); + } else { + bb.log.debug( + `thread=${threadId} phase=${phase} action=section-cache-hit target=${decision?.target ?? "none"}`, + ); + } if (decision !== null) { const sectionId = resolveSectionId( await bb.sdk.threadSections.list(), @@ -409,7 +710,7 @@ export default function plugin(bb: BbPluginApi): void { phase, decision, sectionId, - mode, + inboxMode, ); } } else { @@ -425,7 +726,7 @@ export default function plugin(bb: BbPluginApi): void { } try { - await applyTitle(thread, state, phase, historyTexts, mode); + await applyTitle(thread, state, phase, historyTexts, inboxMode); } catch (error: unknown) { const message = error instanceof Error ? error.message : String(error); bb.log.warn( @@ -466,6 +767,128 @@ export default function plugin(bb: BbPluginApi): void { } } + function inboxPhase(thread: Thread): "active" | "failed" | "idle" { + if (thread.status === "idle") return "idle"; + if (thread.status === "error") return "failed"; + return "active"; + } + + async function reconcileManagedThread( + threadId: string, + signal?: AbortSignal, + ): Promise { + if (signal?.aborted) return; + const fresh = (await bb.sdk.threads.get({ threadId })) as Thread; + if (signal?.aborted) return; + if (!isManageableThread(fresh)) { + if (!signal?.aborted) { + await bb.storage.kv.delete(stateKey(threadId)); + } + return; + } + let state = await readState(threadId); + if (signal?.aborted) return; + const adopted = state === null; + if (state === null) { + state = initialState(fresh); + } + if (signal?.aborted) return; + await reconcileInbox(fresh, state, inboxPhase(fresh), signal); + if (signal?.aborted) return; + await saveState(threadId, state); + if ( + isEligibleThread(fresh) && + (adopted || + state.sectionClassification?.classifierVersion !== + SECTION_CLASSIFIER_VERSION) + ) { + await evaluate(threadId, "settings", signal); + } + } + + async function discoverThreadIds(signal: AbortSignal): Promise { + const discovered = new Set(); + let foundNewIds = true; + while (foundNewIds && !disposed && !signal.aborted) { + foundNewIds = false; + let offset = 0; + while (!disposed && !signal.aborted) { + const page = await bb.sdk.threads.list({ + archived: false, + excludeSideChats: true, + hasParent: false, + limit: THREAD_LIST_PAGE_SIZE, + offset, + signal, + }); + for (const thread of page) { + if (!discovered.has(thread.id)) { + discovered.add(thread.id); + foundNewIds = true; + } + } + if (page.length < THREAD_LIST_PAGE_SIZE) break; + offset += THREAD_LIST_PAGE_SIZE; + } + } + return [...discovered]; + } + + async function reconcileWithRetry( + threadId: string, + signal: AbortSignal, + ): Promise { + for ( + let attempt = 0; + attempt <= RECONCILIATION_RETRY_DELAYS_MS.length; + attempt += 1 + ) { + if (signal.aborted) return; + try { + await enqueue( + threadId, + () => reconcileManagedThread(threadId, signal), + false, + ); + return; + } catch (error: unknown) { + if (signal.aborted) return; + const retryDelay = RECONCILIATION_RETRY_DELAYS_MS[attempt]; + if (retryDelay === undefined) throw error; + const message = + error instanceof Error ? error.message : String(error); + bb.log.warn( + `thread=${threadId} action=reconciliation-retry attempt=${attempt + 1} error=${message}`, + ); + await abortableDelay(retryDelay, signal); + } + } + } + + async function reconcileExistingThreads(signal: AbortSignal): Promise { + const threadIds = await discoverThreadIds(signal); + let nextIndex = 0; + let firstError: unknown; + const workerCount = Math.min( + RECONCILIATION_CONCURRENCY, + threadIds.length, + ); + const workers = Array.from({ length: workerCount }, async () => { + while (!signal.aborted && firstError === undefined) { + const threadId = threadIds[nextIndex]; + nextIndex += 1; + if (threadId === undefined || signal.aborted) return; + try { + await reconcileWithRetry(threadId, signal); + } catch (error: unknown) { + firstError ??= error; + } + } + }); + await Promise.all(workers); + if (firstError !== undefined) throw firstError; + } + bb.events.on("thread.created", ({ thread }) => enqueue(thread.id, async () => { if (!isEligibleThread(thread)) return; @@ -477,8 +900,24 @@ export default function plugin(bb: BbPluginApi): void { bb.events.on("thread.active", ({ thread }) => enqueue(thread.id, async () => { - if ((await readState(thread.id)) === null) return; await evaluate(thread.id, "active"); + const state = await readState(thread.id); + if (state === null) 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; + } + await reconcileInbox( + fresh, + state, + inboxPhase(fresh), + undefined, + thread.pinnedAt, + ); + await saveState(thread.id, state); }), ); @@ -501,11 +940,35 @@ export default function plugin(bb: BbPluginApi): void { state.completedTurns, ); } + const fresh = (await bb.sdk.threads.get({ + threadId: thread.id, + })) as Thread; + if (!isManageableThread(fresh)) { + await bb.storage.kv.delete(stateKey(thread.id)); + return; + } + await reconcileInbox(fresh, state, inboxPhase(fresh)); 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; + const fresh = (await bb.sdk.threads.get({ + threadId: thread.id, + })) as Thread; + if (!isManageableThread(fresh)) { + await bb.storage.kv.delete(stateKey(thread.id)); + return; + } + await reconcileInbox(fresh, state, inboxPhase(fresh)); + await saveState(thread.id, state); + }), + ); + const forget = (threadId: string) => enqueue(threadId, async () => { await bb.storage.kv.delete(stateKey(threadId)); @@ -513,33 +976,81 @@ export default function plugin(bb: BbPluginApi): void { bb.events.on("thread.archived", ({ thread }) => forget(thread.id)); bb.events.on("thread.deleted", ({ thread }) => forget(thread.id)); + let unsubscribeThreadChanges: () => void = () => undefined; + try { + unsubscribeThreadChanges = bb.sdk.subscribe({ + event: "thread:changed", + callback(event) { + const threadId = event.id; + if ( + threadId === undefined || + (!event.changes.includes("pin-state-changed") && + !event.changes.includes("status-changed")) + ) { + return; + } + void enqueue(threadId, () => reconcileManagedThread(threadId)); + }, + }); + } catch (error: unknown) { + const message = error instanceof Error ? error.message : String(error); + bb.log.warn(`action=realtime-subscribe-failed error=${message}`); + } + settings.onChange((next, previous) => { - if (next.mode === previous.mode) return; - bb.log.info(`action=mode-changed previous=${previous.mode} next=${next.mode}`); - if (next.mode !== "apply") return; + if (next.inboxMode === previous.inboxMode) return; + bb.log.info( + `action=mode-changed previous=${previous.inboxMode} next=${next.inboxMode}`, + ); + if (next.inboxMode !== "apply") return; void bb.storage.kv .list(STATE_PREFIX) - .then((keys) => - Promise.all( - keys.map((key) => - enqueue(key.slice(STATE_PREFIX.length), () => - evaluate(key.slice(STATE_PREFIX.length), "settings"), - ), - ), - ), - ) + .then(async (keys) => { + for (const key of keys) { + const threadId = key.slice(STATE_PREFIX.length); + await enqueue(threadId, async () => { + const state = await readState(threadId); + if (state === null) return; + const thread = (await bb.sdk.threads.get({ + threadId, + })) as Thread; + if (!isManageableThread(thread)) { + await bb.storage.kv.delete(stateKey(threadId)); + return; + } + await reconcileInbox(thread, state, inboxPhase(thread)); + await saveState(threadId, state); + await evaluate(threadId, "settings"); + }); + } + }) .catch((error: unknown) => { const message = error instanceof Error ? error.message : String(error); bb.log.error(`action=apply-mode-evaluation-failed error=${message}`); }); }); - bb.onDispose(() => { + bb.background.service("inbox-reconciliation", { + async start(signal) { + while (!signal.aborted) { + await reconcileExistingThreads(signal); + if (signal.aborted) return; + await abortableDelay(RECONCILIATION_INTERVAL_MS, signal); + } + }, + }); + + bb.onDispose(async () => { + acceptingWork = false; + unsubscribeThreadChanges(); + await Promise.allSettled([...queues.values()]); disposed = true; }); void settings .get() - .then(({ mode }) => bb.log.info(`Thread Organizer loaded mode=${mode}`)) + .then(({ inboxMode }) => + bb.log.info(`Thread Organizer loaded mode=${inboxMode}`), + ) .catch((error: unknown) => { const message = error instanceof Error ? error.message : String(error); bb.log.warn(`action=mode-read-failed error=${message}`); diff --git a/plugins/thread-organizer/sidebar-controller.test.ts b/plugins/thread-organizer/sidebar-controller.test.ts new file mode 100644 index 0000000..354ee27 --- /dev/null +++ b/plugins/thread-organizer/sidebar-controller.test.ts @@ -0,0 +1,421 @@ +// @vitest-environment jsdom +import { afterEach, describe, expect, it, vi } from "vitest"; + +import { mountInboxSectionCollapser } from "./sidebar-controller.js"; + +function section( + label: string, + expanded: boolean, + threadIds: string[], + commit: "sync" | "microtask" = "sync", + hasFullRowToggle = true, +): HTMLElement { + const group = document.createElement("div"); + group.dataset.sidebarStickyGroup = ""; + const button = document.createElement("button"); + const rowToggle = document.createElement("button"); + rowToggle.setAttribute("aria-hidden", "true"); + rowToggle.tabIndex = -1; + const setExpanded = (nextExpanded: boolean) => { + button.setAttribute("aria-expanded", String(nextExpanded)); + button.setAttribute( + "aria-label", + `${nextExpanded ? "Collapse" : "Expand"} ${label} section`, + ); + }; + setExpanded(expanded); + const toggleExpanded = () => { + const nextExpanded = button.getAttribute("aria-expanded") !== "true"; + if (commit === "microtask") { + queueMicrotask(() => setExpanded(nextExpanded)); + } else { + setExpanded(nextExpanded); + } + }; + button.addEventListener("click", () => { + toggleExpanded(); + }); + rowToggle.addEventListener("click", () => { + toggleExpanded(); + }); + group.append(button); + if (hasFullRowToggle) group.append(rowToggle); + for (const id of threadIds) { + const row = document.createElement("a"); + row.dataset.sidebarThreadId = id; + group.append(row); + } + return group; +} + +function sidebar(...groups: HTMLElement[]): HTMLElement { + const root = document.createElement("aside"); + root.dataset.sidebar = "sidebar"; + root.append(...groups); + return root; +} + +function setup() { + const pinned = section("Pinned", true, ["thr_active"], "sync", false); + const destination = section("Engineering", true, []); + const root = sidebar(pinned, destination); + document.body.append(root); + const controller = new AbortController(); + mountInboxSectionCollapser({ document, signal: controller.signal }); + return { controller, destination, pinned, sidebar: root }; +} + +function toggle(group: Element): HTMLButtonElement { + return group.querySelector("button[aria-expanded]")!; +} + +async function mutationsSettled(): Promise { + await new Promise((resolve) => setTimeout(resolve, 0)); +} + +afterEach(() => { + document.body.replaceChildren(); + vi.restoreAllMocks(); +}); + +describe("inbox section collapser", () => { + it("collapses sections on mount without collapsing the native Pinned inbox", () => { + const { controller, destination, pinned } = setup(); + + expect(toggle(destination).getAttribute("aria-expanded")).toBe("false"); + expect(toggle(pinned).getAttribute("aria-expanded")).toBe("true"); + controller.abort(); + }); + + it("collapses the destination section after a thread moves between groups", async () => { + const { controller, destination, pinned } = setup(); + const collapse = toggle(destination); + const click = vi.spyOn(collapse, "click"); + const row = pinned.querySelector( + '[data-sidebar-thread-id="thr_active"]', + )!; + + collapse.setAttribute("aria-expanded", "true"); + collapse.setAttribute("aria-label", "Collapse Engineering section"); + destination.append(row); + + await vi.waitFor(() => expect(click).toHaveBeenCalledOnce()); + expect(collapse.getAttribute("aria-expanded")).toBe("false"); + controller.abort(); + }); + + it("never collapses the native Pinned inbox after a thread moves into it", async () => { + const { controller, destination, pinned } = setup(); + const pinnedToggle = toggle(pinned); + const click = vi.spyOn(pinnedToggle, "click"); + const row = pinned.querySelector( + '[data-sidebar-thread-id="thr_active"]', + )!; + + destination.append(row); + await mutationsSettled(); + pinned.append(row); + await mutationsSettled(); + + expect(click).not.toHaveBeenCalled(); + expect(pinnedToggle.getAttribute("aria-expanded")).toBe("true"); + controller.abort(); + }); + + it("collapses a top-level custom section named Pinned", async () => { + const nativePinned = section( + "Pinned", + true, + ["thr_active"], + "sync", + false, + ); + const customPinned = section("Pinned", true, [], "sync", false); + const action = document.createElement("button"); + action.setAttribute("aria-label", "New thread in Pinned"); + customPinned.append(action); + const root = sidebar(nativePinned, customPinned); + document.body.append(root); + const controller = new AbortController(); + mountInboxSectionCollapser({ document, signal: controller.signal }); + const nativeToggle = toggle(nativePinned); + const customToggle = toggle(customPinned); + + expect(nativeToggle.getAttribute("aria-expanded")).toBe("true"); + expect(customToggle.getAttribute("aria-expanded")).toBe("false"); + + const customClick = vi.spyOn(customToggle, "click"); + const row = nativePinned.querySelector( + '[data-sidebar-thread-id="thr_active"]', + )!; + customToggle.setAttribute("aria-expanded", "true"); + customToggle.setAttribute("aria-label", "Collapse Pinned section"); + customPinned.append(row); + + await vi.waitFor(() => expect(customClick).toHaveBeenCalledOnce()); + expect(customToggle.getAttribute("aria-expanded")).toBe("false"); + expect(nativeToggle.getAttribute("aria-expanded")).toBe("true"); + controller.abort(); + }); + + it("remembers a thread while collapsed section contents are unmounted", async () => { + const { controller, destination, pinned } = setup(); + const collapse = toggle(destination); + const click = vi.spyOn(collapse, "click"); + const row = pinned.querySelector( + '[data-sidebar-thread-id="thr_active"]', + )!; + + row.remove(); + await mutationsSettled(); + collapse.setAttribute("aria-expanded", "true"); + collapse.setAttribute("aria-label", "Collapse Engineering section"); + destination.append(row); + + await vi.waitFor(() => expect(click).toHaveBeenCalledOnce()); + expect(collapse.getAttribute("aria-expanded")).toBe("false"); + controller.abort(); + }); + + it("preserves a section the user deliberately expands with its chevron", async () => { + const { controller, destination } = setup(); + const sectionToggle = toggle(destination); + sectionToggle.click(); + const click = vi.spyOn(sectionToggle, "click"); + const row = document.createElement("a"); + row.dataset.sidebarThreadId = "thr_new"; + + destination.append(row); + await mutationsSettled(); + + expect(click).not.toHaveBeenCalled(); + expect(sectionToggle.getAttribute("aria-expanded")).toBe("true"); + controller.abort(); + }); + + it("preserves a deliberate expansion when the native state commit is deferred", async () => { + const destination = section("Engineering", false, [], "microtask"); + const root = sidebar(destination); + document.body.append(root); + const controller = new AbortController(); + mountInboxSectionCollapser({ document, signal: controller.signal }); + const sectionToggle = toggle(destination); + + sectionToggle.click(); + await mutationsSettled(); + + expect(sectionToggle.getAttribute("aria-expanded")).toBe("true"); + controller.abort(); + }); + + it("records a deliberate expansion when native handling runs first", async () => { + const destination = section("Engineering", false, []); + const originalToggle = toggle(destination); + const sectionToggle = originalToggle.cloneNode(true) as HTMLButtonElement; + originalToggle.replaceWith(sectionToggle); + const root = sidebar(destination); + root.addEventListener( + "click", + () => { + sectionToggle.setAttribute("aria-expanded", "true"); + sectionToggle.setAttribute( + "aria-label", + "Collapse Engineering section", + ); + }, + true, + ); + document.body.append(root); + const controller = new AbortController(); + mountInboxSectionCollapser({ document, signal: controller.signal }); + + sectionToggle.click(); + await mutationsSettled(); + + expect(sectionToggle.getAttribute("aria-expanded")).toBe("true"); + controller.abort(); + }); + + it("preserves a section the user deliberately expands with the native full row", async () => { + const { controller, destination } = setup(); + const sectionToggle = toggle(destination); + const rowToggle = destination.querySelector( + 'button[aria-hidden="true"][tabindex="-1"]', + )!; + rowToggle.click(); + const click = vi.spyOn(sectionToggle, "click"); + const row = document.createElement("a"); + row.dataset.sidebarThreadId = "thr_new"; + + destination.append(row); + await mutationsSettled(); + + expect(click).not.toHaveBeenCalled(); + expect(sectionToggle.getAttribute("aria-expanded")).toBe("true"); + controller.abort(); + }); + + it("keeps a deliberately expanded destination open when a known thread moves into it", async () => { + const { controller, destination, pinned } = setup(); + const sectionToggle = toggle(destination); + sectionToggle.click(); + await mutationsSettled(); + const click = vi.spyOn(sectionToggle, "click"); + const row = pinned.querySelector( + '[data-sidebar-thread-id="thr_active"]', + )!; + + destination.append(row); + await mutationsSettled(); + + expect(click).not.toHaveBeenCalled(); + expect(sectionToggle.getAttribute("aria-expanded")).toBe("true"); + controller.abort(); + }); + + it("does not preserve expansion intent when the native row click is suppressed", async () => { + const { controller, destination } = setup(); + const sectionToggle = toggle(destination); + const nativeRowToggle = destination.querySelector( + 'button[aria-hidden="true"][tabindex="-1"]', + )!; + const suppressedRowToggle = nativeRowToggle.cloneNode( + true, + ) as HTMLButtonElement; + nativeRowToggle.replaceWith(suppressedRowToggle); + await mutationsSettled(); + + suppressedRowToggle.click(); + await mutationsSettled(); + expect(sectionToggle.getAttribute("aria-expanded")).toBe("false"); + + const click = vi.spyOn(sectionToggle, "click"); + sectionToggle.setAttribute("aria-expanded", "true"); + sectionToggle.setAttribute( + "aria-label", + "Collapse Engineering section", + ); + + await vi.waitFor(() => expect(click).toHaveBeenCalledOnce()); + expect(sectionToggle.getAttribute("aria-expanded")).toBe("false"); + controller.abort(); + }); + + it("does not mistake a section action for a deliberate expansion", async () => { + const { controller, destination } = setup(); + const sectionToggle = toggle(destination); + const action = document.createElement("button"); + action.textContent = "Section action"; + destination.append(action); + + action.click(); + sectionToggle.setAttribute("aria-expanded", "true"); + sectionToggle.setAttribute( + "aria-label", + "Collapse Engineering section", + ); + + await vi.waitFor(() => + expect(sectionToggle.getAttribute("aria-expanded")).toBe("false"), + ); + controller.abort(); + }); + + it("tracks duplicate labels as independent section elements", async () => { + const first = section("Engineering", true, []); + const second = section("Engineering", true, []); + const root = sidebar(first, second); + document.body.append(root); + const controller = new AbortController(); + mountInboxSectionCollapser({ document, signal: controller.signal }); + const firstToggle = toggle(first); + const secondToggle = toggle(second); + + firstToggle.click(); + await mutationsSettled(); + secondToggle.setAttribute("aria-expanded", "true"); + secondToggle.setAttribute("aria-label", "Collapse Engineering section"); + + await vi.waitFor(() => + expect(secondToggle.getAttribute("aria-expanded")).toBe("false"), + ); + expect(firstToggle.getAttribute("aria-expanded")).toBe("true"); + controller.abort(); + }); + + it("does not give a custom section named Pinned special treatment", () => { + const customPinned = section("Pinned", true, []); + const root = sidebar(customPinned); + document.body.append(root); + const controller = new AbortController(); + mountInboxSectionCollapser({ document, signal: controller.signal }); + + expect(toggle(customPinned).getAttribute("aria-expanded")).toBe("false"); + controller.abort(); + }); + + it("retries a controller collapse once when drag-click suppression swallows it", async () => { + const destination = section("Engineering", true, []); + let suppressNextClick = true; + destination.addEventListener( + "click", + (event) => { + if (!suppressNextClick) return; + suppressNextClick = false; + event.preventDefault(); + event.stopPropagation(); + }, + true, + ); + const root = sidebar(destination); + document.body.append(root); + const sectionToggle = toggle(destination); + const click = vi.spyOn(sectionToggle, "click"); + const controller = new AbortController(); + + mountInboxSectionCollapser({ document, signal: controller.signal }); + + await vi.waitFor(() => + expect(sectionToggle.getAttribute("aria-expanded")).toBe("false"), + ); + expect(click).toHaveBeenCalledTimes(2); + controller.abort(); + }); + + it("mounts replacement sidebars and stops handling the detached root", async () => { + const { controller, destination, sidebar: oldSidebar } = setup(); + const oldToggle = toggle(destination); + oldSidebar.remove(); + + const replacementGroup = section("Product", true, []); + const replacement = sidebar(replacementGroup); + document.body.append(replacement); + + await vi.waitFor(() => + expect(toggle(replacementGroup).getAttribute("aria-expanded")).toBe( + "false", + ), + ); + + oldToggle.setAttribute("aria-expanded", "true"); + oldToggle.setAttribute("aria-label", "Collapse Engineering section"); + destination.append(document.createElement("span")); + await mutationsSettled(); + + expect(oldToggle.getAttribute("aria-expanded")).toBe("true"); + controller.abort(); + }); + + it("stops observing when the content script is aborted", async () => { + const { controller, destination } = setup(); + const collapse = toggle(destination); + const click = vi.spyOn(collapse, "click"); + + controller.abort(); + collapse.setAttribute("aria-expanded", "true"); + destination.append(document.createElement("span")); + await mutationsSettled(); + + expect(click).not.toHaveBeenCalled(); + }); +}); diff --git a/plugins/thread-organizer/sidebar-controller.ts b/plugins/thread-organizer/sidebar-controller.ts new file mode 100644 index 0000000..8214126 --- /dev/null +++ b/plugins/thread-organizer/sidebar-controller.ts @@ -0,0 +1,261 @@ +const SIDEBAR_SELECTOR = '[data-sidebar="sidebar"]'; +const STICKY_GROUP_SELECTOR = "[data-sidebar-sticky-group]"; +const THREAD_SELECTOR = "[data-sidebar-thread-id]"; +const SECTION_TOGGLE_SELECTOR = + 'button[aria-expanded][aria-label$=" section"]'; +const SECTION_ROW_TOGGLE_SELECTOR = + 'button[aria-hidden="true"][tabindex="-1"]'; + +interface MountInboxSectionCollapserOptions { + document?: Document; + signal: AbortSignal; +} + +function groupToggle(group: Element): HTMLButtonElement | null { + for (const button of group.querySelectorAll( + SECTION_TOGGLE_SELECTOR, + )) { + if (button.closest(STICKY_GROUP_SELECTOR) === group) return button; + } + return null; +} + +function isNativePinnedGroup(group: Element): boolean { + const toggle = groupToggle(group); + if ( + !/^(?:Expand|Collapse) Pinned section$/.test( + toggle?.getAttribute("aria-label") ?? "", + ) + ) { + return false; + } + for (const button of group.querySelectorAll( + `${SECTION_ROW_TOGGLE_SELECTOR}, button[aria-label="Pinned section actions"], button[aria-label="New thread in Pinned"]`, + )) { + if (button.closest(STICKY_GROUP_SELECTOR) === group) return false; + } + return true; +} + +function visibleThreadGroups(sidebar: Element): Map { + const groups = new Map(); + for (const row of sidebar.querySelectorAll(THREAD_SELECTOR)) { + const id = row.dataset.sidebarThreadId; + const group = row.closest(STICKY_GROUP_SELECTOR); + if (id && group) groups.set(id, group); + } + return groups; +} + +function addExpandedGroupAndAncestors( + controls: Set, + group: Element, + userExpandedGroups: WeakSet, + pendingUserExpandedGroups: WeakSet, +): void { + let current: Element | null = group; + while (current) { + const toggle = groupToggle(current); + if ( + !isNativePinnedGroup(current) && + toggle?.getAttribute("aria-expanded") === "true" && + !userExpandedGroups.has(current) && + !pendingUserExpandedGroups.has(current) + ) { + controls.add(toggle); + } + current = + current.parentElement?.closest(STICKY_GROUP_SELECTOR) ?? null; + } +} + +function mountSidebarCollapser( + sidebar: Element, + signal: AbortSignal, +): () => void { + const expandedByGroup = new WeakMap(); + const userExpandedGroups = new WeakSet(); + const pendingUserExpandedGroups = new WeakSet(); + const controllerCollapseControls = new WeakSet(); + const pendingExpansionTimers = new Set>(); + const knownThreadGroups = visibleThreadGroups(sidebar); + let scheduled = false; + + const reconcile = () => { + scheduled = false; + if (signal.aborted || !sidebar.isConnected) return; + + const currentThreadGroups = visibleThreadGroups(sidebar); + const controls = new Set(); + + for (const group of sidebar.querySelectorAll(STICKY_GROUP_SELECTOR)) { + const toggle = groupToggle(group); + if (toggle === null || isNativePinnedGroup(group)) continue; + const expanded = toggle.getAttribute("aria-expanded") === "true"; + const wasExpanded = expandedByGroup.get(group); + if (!expanded) userExpandedGroups.delete(group); + if ( + expanded && + !userExpandedGroups.has(group) && + !pendingUserExpandedGroups.has(group) && + (wasExpanded === undefined || wasExpanded === false) + ) { + controls.add(toggle); + } + } + + for (const [id, currentGroup] of currentThreadGroups) { + const previousGroup = knownThreadGroups.get(id); + if (previousGroup && previousGroup !== currentGroup) { + addExpandedGroupAndAncestors( + controls, + currentGroup, + userExpandedGroups, + pendingUserExpandedGroups, + ); + } + knownThreadGroups.set(id, currentGroup); + } + + for (const control of controls) { + if ( + control.isConnected && + control.getAttribute("aria-expanded") === "true" + ) { + const collapse = (remainingAttempts: number) => { + if ( + signal.aborted || + !control.isConnected || + control.getAttribute("aria-expanded") !== "true" + ) { + return; + } + controllerCollapseControls.add(control); + try { + control.click(); + } finally { + controllerCollapseControls.delete(control); + } + if ( + remainingAttempts > 1 && + control.getAttribute("aria-expanded") === "true" + ) { + queueMicrotask(() => collapse(remainingAttempts - 1)); + } + }; + collapse(2); + } + } + + for (const group of sidebar.querySelectorAll(STICKY_GROUP_SELECTOR)) { + const toggle = groupToggle(group); + if (toggle !== null) { + expandedByGroup.set( + group, + toggle.getAttribute("aria-expanded") === "true", + ); + } + } + }; + + const recordUserExpansion = (event: Event) => { + const target = event.target; + if (!(target instanceof Element)) return; + const button = target.closest( + `${SECTION_TOGGLE_SELECTOR}, ${SECTION_ROW_TOGGLE_SELECTOR}`, + ); + if (button === null) return; + if (controllerCollapseControls.has(button)) return; + const group = button.closest(STICKY_GROUP_SELECTOR); + if (group === null) return; + const toggle = groupToggle(group); + if (toggle?.getAttribute("aria-expanded") === "true") { + userExpandedGroups.add(group); + return; + } + if (toggle !== null) { + pendingUserExpandedGroups.add(group); + const timer = setTimeout(() => { + pendingExpansionTimers.delete(timer); + pendingUserExpandedGroups.delete(group); + if ( + !signal.aborted && + group.isConnected && + groupToggle(group)?.getAttribute("aria-expanded") === "true" + ) { + userExpandedGroups.add(group); + } + scheduleReconcile(); + }, 0); + pendingExpansionTimers.add(timer); + } + }; + sidebar.addEventListener("click", recordUserExpansion, true); + + const scheduleReconcile = () => { + if (scheduled || signal.aborted) return; + scheduled = true; + queueMicrotask(reconcile); + }; + const observer = new MutationObserver(scheduleReconcile); + observer.observe(sidebar, { + attributeFilter: ["aria-expanded", "aria-label"], + attributes: true, + childList: true, + subtree: true, + }); + reconcile(); + signal.addEventListener( + "abort", + () => { + observer.disconnect(); + for (const timer of pendingExpansionTimers) clearTimeout(timer); + pendingExpansionTimers.clear(); + sidebar.removeEventListener("click", recordUserExpansion, true); + }, + { once: true }, + ); + + return () => { + observer.disconnect(); + for (const timer of pendingExpansionTimers) clearTimeout(timer); + pendingExpansionTimers.clear(); + sidebar.removeEventListener("click", recordUserExpansion, true); + }; +} + +export function mountInboxSectionCollapser({ + document: targetDocument = document, + signal, +}: MountInboxSectionCollapserOptions): () => void { + const disposers = new Map void>(); + + const mountSidebars = () => { + for (const [sidebar, stop] of disposers) { + if (!sidebar.isConnected) { + stop(); + disposers.delete(sidebar); + } + } + for (const sidebar of targetDocument.querySelectorAll(SIDEBAR_SELECTOR)) { + if (!disposers.has(sidebar)) { + disposers.set(sidebar, mountSidebarCollapser(sidebar, signal)); + } + } + }; + + mountSidebars(); + const discoveryObserver = new MutationObserver(mountSidebars); + discoveryObserver.observe(targetDocument.documentElement, { + childList: true, + subtree: true, + }); + + const dispose = () => { + discoveryObserver.disconnect(); + for (const stop of disposers.values()) stop(); + disposers.clear(); + }; + signal.addEventListener("abort", dispose, { once: true }); + return dispose; +} diff --git a/plugins/thread-organizer/tsconfig.json b/plugins/thread-organizer/tsconfig.json index db529e7..1fcd2ab 100644 --- a/plugins/thread-organizer/tsconfig.json +++ b/plugins/thread-organizer/tsconfig.json @@ -2,6 +2,8 @@ "compilerOptions": { "target": "ES2022", "lib": [ + "DOM", + "DOM.Iterable", "ES2022" ], "module": "ESNext",