Activity feed: one dock-shaped GET /api/activity (cross-chat processes)
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.
This commit is contained in:
@@ -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());
|
||||
}
|
||||
@@ -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 };
|
||||
}
|
||||
|
||||
@@ -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));
|
||||
}
|
||||
});
|
||||
Reference in New Issue
Block a user