Gateway: keep finished threads in Recent until settled, with a full list and settle-all route
This commit is contained in:
1 parent
cf8995e63f
commit
2a127327bf
5 files changed
+109
-25
No files matched your search
+23
-4
@@ -58,13 +58,15 @@ GET /v1/feed
|
|||||||
"needsYou": [FeedItem],
|
"needsYou": [FeedItem],
|
||||||
"working": [FeedItem],
|
"working": [FeedItem],
|
||||||
"recent": [FeedItem],
|
"recent": [FeedItem],
|
||||||
|
"recentTotal": 12,
|
||||||
"hosts": [{ "id": "...", "label": "Mac mini", "status": "online", "error": null }]
|
"hosts": [{ "id": "...", "label": "Mac mini", "status": "online", "error": null }]
|
||||||
}
|
}
|
||||||
```
|
```
|
||||||
|
|
||||||
`needsYou` holds threads waiting for an approval or an answer, oldest first. `working` holds running
|
`needsYou` holds threads waiting for an approval or an answer, oldest first. `working` holds running
|
||||||
threads, most recently updated first. `recent` holds threads that completed or failed in the last 24
|
threads, most recently updated first. `recent` holds threads that completed or failed and are not
|
||||||
hours, at most 10. Host `status` is one of `connecting`, `online`, `offline`, `auth_invalid`, or
|
settled yet, however old, newest first and at most 10. `recentTotal` counts all of them. When it is
|
||||||
|
larger, the rest come from `GET /v1/recent`. Host `status` is one of `connecting`, `online`, `offline`, `auth_invalid`, or
|
||||||
`incompatible`.
|
`incompatible`.
|
||||||
|
|
||||||
```
|
```
|
||||||
@@ -145,8 +147,25 @@ POST /v1/threads/:hostId/:threadId/settle
|
|||||||
202
|
202
|
||||||
```
|
```
|
||||||
|
|
||||||
Settles a thread, as in T3. A settled thread that finished or failed leaves the feed. To settle all
|
Settles a thread, as in T3. A settled thread that finished or failed leaves the feed.
|
||||||
of Recent, settle each item.
|
|
||||||
|
```
|
||||||
|
GET /v1/recent
|
||||||
|
|
||||||
|
200 { "items": [FeedItem] }
|
||||||
|
```
|
||||||
|
|
||||||
|
Every unsettled finished thread, newest first. `lastMessage` is null for threads the gateway has not
|
||||||
|
read yet, since it does not fetch them all for this list.
|
||||||
|
|
||||||
|
```
|
||||||
|
POST /v1/recent/settle
|
||||||
|
|
||||||
|
200 { "settled": 11, "failed": 0 }
|
||||||
|
```
|
||||||
|
|
||||||
|
Settles every completed thread in Recent, including those past the first 10. Failed threads stay.
|
||||||
|
`failed` counts threads T3 refused or whose host was offline.
|
||||||
|
|
||||||
Thread detail fills `request` and `via` the same way as the feed when a subagent waits on you.
|
Thread detail fills `request` and `via` the same way as the feed when a subagent waits on you.
|
||||||
|
|
||||||
|
|||||||
+29
-1
@@ -12,6 +12,7 @@ import type {
|
|||||||
PushRegistration,
|
PushRegistration,
|
||||||
RespondRequest,
|
RespondRequest,
|
||||||
PushResult,
|
PushResult,
|
||||||
|
SettleRecentResponse,
|
||||||
SendMessageRequest,
|
SendMessageRequest,
|
||||||
StatusResponse,
|
StatusResponse,
|
||||||
} from "./contract.ts";
|
} from "./contract.ts";
|
||||||
@@ -21,7 +22,16 @@ import type { Pusher } from "./push.ts";
|
|||||||
import { HostError, type T3Host } from "./t3/host.ts";
|
import { HostError, type T3Host } from "./t3/host.ts";
|
||||||
import { RpcError } from "./t3/rpc.ts";
|
import { RpcError } from "./t3/rpc.ts";
|
||||||
import type { ApprovalDecision, ModelSelection, ThreadShell } from "./t3/types.ts";
|
import type { ApprovalDecision, ModelSelection, ThreadShell } from "./t3/types.ts";
|
||||||
import { buildFeed, buildSummary, buildThreadDetail, hostStatusDetail, hostSummary, type SummaryCache } from "./views.ts";
|
import {
|
||||||
|
buildFeed,
|
||||||
|
buildRecent,
|
||||||
|
buildSummary,
|
||||||
|
buildThreadDetail,
|
||||||
|
feedCandidates,
|
||||||
|
hostStatusDetail,
|
||||||
|
hostSummary,
|
||||||
|
type SummaryCache,
|
||||||
|
} from "./views.ts";
|
||||||
|
|
||||||
export interface ApiDeps {
|
export interface ApiDeps {
|
||||||
db: Db;
|
db: Db;
|
||||||
@@ -187,6 +197,24 @@ export function createRoutes({ db, hub, cache, pusher, startedAt, log = () => {}
|
|||||||
GET: authed(async () => json(await buildFeed(hub.list(), cache))),
|
GET: authed(async () => json(await buildFeed(hub.list(), cache))),
|
||||||
},
|
},
|
||||||
|
|
||||||
|
"/v1/recent": {
|
||||||
|
GET: authed(() => json(buildRecent(hub.list(), cache))),
|
||||||
|
},
|
||||||
|
|
||||||
|
/** Settles every completed thread in Recent. Failed ones stay for the user to look at. */
|
||||||
|
"/v1/recent/settle": {
|
||||||
|
POST: authed(async () => {
|
||||||
|
const completed = feedCandidates(hub.list()).recent.filter(({ phase }) => phase === "completed");
|
||||||
|
const results = await Promise.allSettled(
|
||||||
|
completed.map(({ host, thread }) =>
|
||||||
|
host.dispatch({ type: "thread.settle", commandId: crypto.randomUUID(), threadId: thread.id }),
|
||||||
|
),
|
||||||
|
);
|
||||||
|
const failed = results.filter((result) => result.status === "rejected").length;
|
||||||
|
return json({ settled: results.length - failed, failed } satisfies SettleRecentResponse);
|
||||||
|
}),
|
||||||
|
},
|
||||||
|
|
||||||
"/v1/summary": {
|
"/v1/summary": {
|
||||||
GET: authed(() => json(buildSummary(hub.list()))),
|
GET: authed(() => json(buildSummary(hub.list()))),
|
||||||
},
|
},
|
||||||
|
|||||||
+14
-1
@@ -126,11 +126,24 @@ export interface Feed {
|
|||||||
needsYou: FeedItem[];
|
needsYou: FeedItem[];
|
||||||
/** Starting or running. Most recently updated first. */
|
/** Starting or running. Most recently updated first. */
|
||||||
working: FeedItem[];
|
working: FeedItem[];
|
||||||
/** Completed or failed in the last 24 hours. Most recent first, at most 10. */
|
/** Completed or failed and not yet settled. Most recent first, at most 10. */
|
||||||
recent: FeedItem[];
|
recent: FeedItem[];
|
||||||
|
/** How many threads Recent holds in all. More than `recent.length` means `GET /v1/recent` has the rest. */
|
||||||
|
recentTotal: number;
|
||||||
hosts: HostSummary[];
|
hosts: HostSummary[];
|
||||||
}
|
}
|
||||||
|
|
||||||
|
export interface RecentResponse {
|
||||||
|
/** Every unsettled finished thread, most recent first. */
|
||||||
|
items: FeedItem[];
|
||||||
|
}
|
||||||
|
|
||||||
|
export interface SettleRecentResponse {
|
||||||
|
settled: number;
|
||||||
|
/** Threads T3 refused to settle, or whose host was offline. */
|
||||||
|
failed: number;
|
||||||
|
}
|
||||||
|
|
||||||
export interface Message {
|
export interface Message {
|
||||||
id: string;
|
id: string;
|
||||||
role: "user" | "assistant" | "system";
|
role: "user" | "assistant" | "system";
|
||||||
|
|||||||
+33
-16
@@ -11,6 +11,7 @@ import type {
|
|||||||
HostSummary,
|
HostSummary,
|
||||||
Message,
|
Message,
|
||||||
PendingRequest,
|
PendingRequest,
|
||||||
|
RecentResponse,
|
||||||
RouteKind,
|
RouteKind,
|
||||||
Summary,
|
Summary,
|
||||||
ThreadDetail,
|
ThreadDetail,
|
||||||
@@ -26,7 +27,6 @@ import {
|
|||||||
const LAST_MESSAGE_CHARS = 280;
|
const LAST_MESSAGE_CHARS = 280;
|
||||||
const MESSAGE_CHARS = 2000;
|
const MESSAGE_CHARS = 2000;
|
||||||
const MESSAGES_PER_THREAD = 20;
|
const MESSAGES_PER_THREAD = 20;
|
||||||
const RECENT_WINDOW_MS = 24 * 60 * 60_000;
|
|
||||||
const RECENT_LIMIT = 10;
|
const RECENT_LIMIT = 10;
|
||||||
/** How long a feed request waits for summaries it does not have yet. */
|
/** How long a feed request waits for summaries it does not have yet. */
|
||||||
const SUMMARY_WAIT_MS = 2_000;
|
const SUMMARY_WAIT_MS = 2_000;
|
||||||
@@ -284,14 +284,13 @@ export function buildSummary(hosts: T3Host[]): Summary {
|
|||||||
};
|
};
|
||||||
}
|
}
|
||||||
|
|
||||||
/** Builds the feed across every host, waiting briefly for summaries it has not read yet. */
|
/** Every thread the feed shows, in feed order, with subagent requests on their parents' cards. */
|
||||||
export async function buildFeed(hosts: T3Host[], cache: SummaryCache, now = Date.now()): Promise<Feed> {
|
export function feedCandidates(hosts: T3Host[]) {
|
||||||
const found: Candidate[] = [];
|
const found: Candidate[] = [];
|
||||||
for (const host of hosts) {
|
for (const host of hosts) {
|
||||||
for (const thread of host.threads.values()) {
|
for (const thread of host.threads.values()) {
|
||||||
const phase = threadPhase(thread);
|
const phase = threadPhase(thread);
|
||||||
if (phase === null || thread.archivedAt || thread.deletedAt) continue;
|
if (phase === null || thread.archivedAt || thread.deletedAt) continue;
|
||||||
if ((phase === "completed" || phase === "failed") && now - Date.parse(thread.updatedAt) > RECENT_WINDOW_MS) continue;
|
|
||||||
found.push({ host, thread, phase });
|
found.push({ host, thread, phase });
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -300,14 +299,25 @@ export async function buildFeed(hosts: T3Host[], cache: SummaryCache, now = Date
|
|||||||
const byNewest = (a: { thread: ThreadShell }, b: { thread: ThreadShell }) =>
|
const byNewest = (a: { thread: ThreadShell }, b: { thread: ThreadShell }) =>
|
||||||
b.thread.updatedAt.localeCompare(a.thread.updatedAt);
|
b.thread.updatedAt.localeCompare(a.thread.updatedAt);
|
||||||
const asker = (candidate: Candidate) => candidate.via?.thread ?? candidate.thread;
|
const asker = (candidate: Candidate) => candidate.via?.thread ?? candidate.thread;
|
||||||
const needsYou = candidates
|
return {
|
||||||
|
needsYou: candidates
|
||||||
.filter(({ phase }) => isWaiting(phase))
|
.filter(({ phase }) => isWaiting(phase))
|
||||||
.sort((a, b) => requestAge(asker(a)).localeCompare(requestAge(asker(b))));
|
.sort((a, b) => requestAge(asker(a)).localeCompare(requestAge(asker(b)))),
|
||||||
const working = candidates.filter(({ phase }) => phase === "starting" || phase === "running").sort(byNewest);
|
working: candidates.filter(({ phase }) => phase === "starting" || phase === "running").sort(byNewest),
|
||||||
const recent = candidates
|
/** Finished and failed threads not yet settled, however old. */
|
||||||
.filter(({ phase }) => phase === "completed" || phase === "failed")
|
recent: candidates.filter(({ phase }) => phase === "completed" || phase === "failed").sort(byNewest),
|
||||||
.sort(byNewest)
|
};
|
||||||
.slice(0, RECENT_LIMIT);
|
}
|
||||||
|
|
||||||
|
const toItem =
|
||||||
|
(cache: SummaryCache) =>
|
||||||
|
({ host, thread, phase, via }: Candidate): FeedItem =>
|
||||||
|
feedItem(host, thread, phase, cache.peek(host, thread), via && { ...via, summary: cache.peek(via.host, via.thread) });
|
||||||
|
|
||||||
|
/** Builds the feed across every host, waiting briefly for summaries it has not read yet. */
|
||||||
|
export async function buildFeed(hosts: T3Host[], cache: SummaryCache): Promise<Feed> {
|
||||||
|
const { needsYou, working, recent: allRecent } = feedCandidates(hosts);
|
||||||
|
const recent = allRecent.slice(0, RECENT_LIMIT);
|
||||||
|
|
||||||
const shown = [...needsYou, ...working, ...recent].flatMap((candidate) => (candidate.via ? [candidate, candidate.via] : [candidate]));
|
const shown = [...needsYou, ...working, ...recent].flatMap((candidate) => (candidate.via ? [candidate, candidate.via] : [candidate]));
|
||||||
const missing = shown
|
const missing = shown
|
||||||
@@ -315,16 +325,23 @@ export async function buildFeed(hosts: T3Host[], cache: SummaryCache, now = Date
|
|||||||
.map(({ host, thread }) => cache.get(host, thread).catch(() => undefined));
|
.map(({ host, thread }) => cache.get(host, thread).catch(() => undefined));
|
||||||
if (missing.length > 0) await Promise.race([Promise.all(missing), Bun.sleep(SUMMARY_WAIT_MS)]);
|
if (missing.length > 0) await Promise.race([Promise.all(missing), Bun.sleep(SUMMARY_WAIT_MS)]);
|
||||||
|
|
||||||
const toItem = ({ host, thread, phase, via }: Candidate) =>
|
|
||||||
feedItem(host, thread, phase, cache.peek(host, thread), via && { ...via, summary: cache.peek(via.host, via.thread) });
|
|
||||||
return {
|
return {
|
||||||
needsYou: needsYou.map(toItem),
|
needsYou: needsYou.map(toItem(cache)),
|
||||||
working: working.map(toItem),
|
working: working.map(toItem(cache)),
|
||||||
recent: recent.map(toItem),
|
recent: recent.map(toItem(cache)),
|
||||||
|
recentTotal: allRecent.length,
|
||||||
hosts: hosts.map(hostSummary),
|
hosts: hosts.map(hostSummary),
|
||||||
};
|
};
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Every unsettled finished thread, for the full Recent list. Uses only summaries already cached,
|
||||||
|
* since there can be many, so `lastMessage` is null for threads the gateway has not read.
|
||||||
|
*/
|
||||||
|
export function buildRecent(hosts: T3Host[], cache: SummaryCache): RecentResponse {
|
||||||
|
return { items: feedCandidates(hosts).recent.map(toItem(cache)) };
|
||||||
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* One thread with its recent messages. Fetches the summary if it is not cached. When the thread
|
* One thread with its recent messages. Fetches the summary if it is not cached. When the thread
|
||||||
* has no request of its own but a subagent it delegated to is waiting, shows that request.
|
* has no request of its own but a subagent it delegated to is waiting, shows that request.
|
||||||
|
|||||||
@@ -140,7 +140,6 @@ describe("routeKind", () => {
|
|||||||
});
|
});
|
||||||
|
|
||||||
describe("buildFeed", () => {
|
describe("buildFeed", () => {
|
||||||
const now = Date.parse("2026-11-01T09:30:00.000Z");
|
|
||||||
const host = (label: string, threads: ThreadShell[]) =>
|
const host = (label: string, threads: ThreadShell[]) =>
|
||||||
({
|
({
|
||||||
id: label,
|
id: label,
|
||||||
@@ -164,7 +163,6 @@ describe("buildFeed", () => {
|
|||||||
]),
|
]),
|
||||||
],
|
],
|
||||||
new SummaryCache(),
|
new SummaryCache(),
|
||||||
now,
|
|
||||||
);
|
);
|
||||||
expect(feed.working).toEqual([]);
|
expect(feed.working).toEqual([]);
|
||||||
expect(feed.needsYou.map((item) => [item.threadId, item.phase, item.via])).toEqual([
|
expect(feed.needsYou.map((item) => [item.threadId, item.phase, item.via])).toEqual([
|
||||||
@@ -182,11 +180,20 @@ describe("buildFeed", () => {
|
|||||||
]),
|
]),
|
||||||
],
|
],
|
||||||
new SummaryCache(),
|
new SummaryCache(),
|
||||||
now,
|
|
||||||
);
|
);
|
||||||
expect(feed.needsYou.map((item) => [item.threadId, item.via?.threadId ?? null])).toEqual([
|
expect(feed.needsYou.map((item) => [item.threadId, item.via?.threadId ?? null])).toEqual([
|
||||||
["parent", "first"],
|
["parent", "first"],
|
||||||
["later", null],
|
["later", null],
|
||||||
]);
|
]);
|
||||||
});
|
});
|
||||||
|
|
||||||
|
it("keeps finished threads in Recent until they are settled, however old", async () => {
|
||||||
|
const old = (id: string, day: number) => thread({ id, status: "completed", updatedAt: `2026-10-${day}T09:00:00.000Z` });
|
||||||
|
const feed = await buildFeed(
|
||||||
|
[host("Desktop", [...Array.from({ length: 11 }, (_, i) => old(`t${i}`, 10 + i)), { ...old("done", 29), settledOverride: "settled" }])],
|
||||||
|
new SummaryCache(),
|
||||||
|
);
|
||||||
|
expect(feed.recent.map((item) => item.threadId)).toEqual(["t10", "t9", "t8", "t7", "t6", "t5", "t4", "t3", "t2", "t1"]);
|
||||||
|
expect(feed.recentTotal).toBe(11);
|
||||||
|
});
|
||||||
});
|
});
|
||||||
Reference in new issue
Block a user