From f8ba5b16282a0bdb222f63699016caad481e4550 Mon Sep 17 00:00:00 2001 From: amosblomqvist Date: Fri, 11 Sep 2026 02:54:31 +0300 Subject: [PATCH] Activity feed: one dock-shaped GET /api/activity (cross-chat processes) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The dock is visible in every view, so it needs every running process — sub-agent jobs across ALL chats (not just the open one) plus the global extraction runs — in one poll. /api/jobs stays chat-scoped and untouched: it is the transcript machinery's poll and carries turnInFlight for the re-attach seam (chat-transport), the seam we spent three live rounds stabilizing; making it polymorphic would fork its response shape by mode and put regression risk on exactly the wrong path. The dock is a different consumer with a different shape (cross-chat, running-only, cost-bearing), served by a thin route over server/observability.ts: - activityFeed(): running jobs from every chat (title-joined, oldest-first on last activity) + extraction runs + busyChats (chats with a teacher turn in flight — the biggest spender; data made available, rendering is a dock decision). - GET /api/activity — the dock's single poll everywhere. - Cost math stays client-side (core/prices.ts folds); raw usage rides the records and the transport stays thin. Decided with the user over the polymorphic-/api/jobs and separate-all-jobs-endpoint alternatives. --- app/api/activity/route.ts | 14 +++++++ src/server/observability.ts | 74 ++++++++++++++++++++++++++++++++++--- tests/activity-feed.test.ts | 54 +++++++++++++++++++++++++++ 3 files changed, 136 insertions(+), 6 deletions(-) create mode 100644 app/api/activity/route.ts create mode 100644 tests/activity-feed.test.ts diff --git a/app/api/activity/route.ts b/app/api/activity/route.ts new file mode 100644 index 0000000..e225725 --- /dev/null +++ b/app/api/activity/route.ts @@ -0,0 +1,14 @@ +/** + * GET /api/activity — the dock's one poll: every running background + * process, every view (chat, home, everywhere). Composition only — the + * feed lives in server/observability.ts; cost folds client-side via + * core/prices.ts. See that module's header for why /api/jobs stays + * chat-scoped and untouched. + */ +import { activityFeed } from "@/src/server"; + +export const dynamic = "force-dynamic"; + +export async function GET() { + return Response.json(activityFeed()); +} diff --git a/src/server/observability.ts b/src/server/observability.ts index e92ebdf..c1ef08b 100644 --- a/src/server/observability.ts +++ b/src/server/observability.ts @@ -2,11 +2,73 @@ * Server-side observability surface — the barrel-approved exports the * transport layer (app/api/*) reads for background-process status. * - * The dock (2026-09 observability design) needs: extraction-pipeline runs - * (from the pipeline's module-level registry) and the store handles for - * title joins. app/ routes import ONLY from @/src/core and @/src/server — - * this file is where the deep imports surface, so routes stay clean. + * THE ACTIVITY FEED (2026-09 observability design): the bottom-right dock is + * visible in every view, so it needs the complete running-process picture — + * sub-agent jobs across ALL chats (not just the open one) plus the global + * extraction runs — in one poll. /api/jobs stays chat-scoped untouched: it + * is the transcript machinery's poll and carries turnInFlight for the + * re-attach seam (chat-transport); the dock is a different consumer with a + * different shape, served here. Single sources of truth are unchanged: the + * job store, the pipeline registry, the turn-liveness registry — this file + * only composes them. */ -export { extractionStatuses, type ExtractionPhase } from "../agent/subagents/extraction.ts"; -export { chatStore } from "../store/chats.ts"; +import { extractionStatuses, type ExtractionPhase } from "../agent/subagents/extraction.ts"; +import { chatStore } from "../store/chats.ts"; +import { isTurnRunning } from "./turn-liveness.ts"; +import { listJobs } from "./jobs.ts"; +import type { JobRecord } from "../core/contracts.ts"; + +export { extractionStatuses, type ExtractionPhase }; +export { chatStore }; + +/** A running sub-agent job, joined with its chat's title for the dock. */ +export interface ActivityJob extends JobRecord { + title: string; +} + +/** One extraction-pipeline run (title joined), oldest first. */ +export interface ActivityExtraction { + chatId: string; + title: string; + phase: ExtractionPhase; + startedAt: number; +} + +/** A chat with a teacher turn in flight right now (the wake-turn process — + * the biggest spender; data made available, rendering it is a dock + * decision). */ +export interface ActivityChatTurn { + chatId: string; + title: string; +} + +/** The dock's whole world in one payload: every running process, every view. + * Jobs are RUNNING-ONLY — finished jobs belong to their chat's transcript; + * the blob disappears when the process ends. Cost math is NOT done here: + * raw usage rides on the records and the client folds it with the pure + * helpers in core/prices.ts (transport stays thin). */ +export function activityFeed(): { jobs: ActivityJob[]; extractions: ActivityExtraction[]; busyChats: ActivityChatTurn[] } { + const chats = chatStore.listChats(); + const jobs: ActivityJob[] = []; + const busyChats: ActivityChatTurn[] = []; + for (const chat of chats) { + for (const job of listJobs(chat.id)) { + if (job.status === "running") { + jobs.push({ ...job, title: chat.title ?? chat.id }); + } + } + if (isTurnRunning(chat.id)) { + busyChats.push({ chatId: chat.id, title: chat.title ?? chat.id }); + } + } + const extractions: ActivityExtraction[] = extractionStatuses().map((run) => ({ + chatId: run.chatId, + title: chatStore.getChat(run.chatId)?.title ?? run.chatId, + phase: run.phase, + startedAt: run.startedAt, + })); + // oldest-first on last activity: the stable blob order (ISO strings sort chronologically) + jobs.sort((a, b) => a.updatedAt.localeCompare(b.updatedAt)); + return { jobs, extractions: extractions.sort((a, b) => a.startedAt - b.startedAt), busyChats }; +} diff --git a/tests/activity-feed.test.ts b/tests/activity-feed.test.ts new file mode 100644 index 0000000..33b90be --- /dev/null +++ b/tests/activity-feed.test.ts @@ -0,0 +1,54 @@ +/** + * Activity-feed tests — the dock's composition (server/observability.ts). + * + * Real stores with fixture chats/jobs, cleaned up in a top-level test. + * Verifies the cross-chat gathering, running-only filtering, title joins, + * stable oldest-first ordering, and busy-chat reporting. + */ +import assert from "node:assert/strict"; +import test from "node:test"; +import * as fs from "node:fs"; +import * as path from "node:path"; + +import { chatStore } from "../src/store/chats.ts"; +import { jobStore } from "../src/store/jobs.ts"; +import { PATHS } from "../config.ts"; +import { activityFeed } from "../src/server/observability.ts"; + +test("activityFeed: running jobs across ALL chats, running-only, title-joined, oldest first", () => { + const chatA = chatStore.createChat("feed chat A"); + const chatB = chatStore.createChat("feed chat B"); + try { + const jobA = jobStore.createJob({ kind: "research", chatId: chatA.id, toolCallId: "tc_a" }); + const jobB = jobStore.createJob({ kind: "demo", chatId: chatB.id, toolCallId: "tc_b" }); + const jobDone = jobStore.createJob({ kind: "research", chatId: chatA.id, toolCallId: "tc_done" }); + jobStore.updateJob(jobDone.id, { status: "done" }); + const jobB2 = jobStore.createJob({ kind: "research", chatId: chatB.id, toolCallId: "tc_b2" }); + // make jobB2 the MOST recently active → should still sort AFTER jobA/jobB (oldest-first). + // Explicit later timestamp: the real bump can land in the same ms as creation. + jobStore.updateJob(jobB2.id, { activity: "second step", updatedAt: new Date(Date.now() + 10).toISOString() }); + + const feed = activityFeed(); + const feedJobs = feed.jobs.filter((j) => [jobA.id, jobB.id, jobB2.id].includes(j.id)); + assert.equal(feedJobs.length, 3); // done job filtered out + for (const j of feedJobs) { + assert.ok(j.title.length > 0, "title joined onto every job"); + assert.equal(typeof j.chatId, "string"); + } + // title join is the chat's title + assert.equal(feedJobs.find((j) => j.id === jobA.id)?.title, "feed chat A"); + // oldest-first on updatedAt: jobA/jobB created before jobB2's bump + const order = feedJobs.map((j) => j.id); + assert.equal(order.indexOf(jobB2.id), order.length - 1); + + // extraction runs are absent when the pipeline is idle (registry empty) + assert.deepEqual(feed.extractions, []); + // busyChats reports turn-liveness (none in flight here) + assert.deepEqual(feed.busyChats, []); + } finally { + fs.unlinkSync(`${PATHS.chats}/${chatA.id}.json`); + fs.unlinkSync(`${PATHS.chats}/${chatB.id}.json`); + const jobsDir = path.join(PATHS.data, "jobs"); + for (const f of fs.readdirSync(jobsDir)) fs.unlinkSync(path.join(jobsDir, f)); + } +});