refactor(web): unify form controls, one notification rule, dedupe i18n copy (#54)
Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -13,15 +13,22 @@
|
||||
* added to session-manager's active table (state idle).
|
||||
*/
|
||||
import path from "node:path";
|
||||
import { readdir } from "node:fs/promises";
|
||||
import { open, readdir } from "node:fs/promises";
|
||||
import {
|
||||
createAgent,
|
||||
isSessionMeta,
|
||||
parseTraceLines,
|
||||
readTraceTolerant,
|
||||
tracesDir,
|
||||
} from "@prismshadow/penguin-core";
|
||||
import type { SessionMetaMessage } from "@prismshadow/penguin-core";
|
||||
import type { ApprovalMode, SessionInfo, SessionSource } from "../api/types.js";
|
||||
import type {
|
||||
ApprovalMode,
|
||||
SessionCategory,
|
||||
SessionCategoryCounts,
|
||||
SessionInfo,
|
||||
SessionSource,
|
||||
} from "../api/types.js";
|
||||
import { HttpError, isMissingCredential, modelCredentialMissing } from "../http/errors.js";
|
||||
import { badRequest } from "../http/validate.js";
|
||||
import type { SessionRow, SessionsRepo } from "../db/repos/sessions.js";
|
||||
@@ -33,6 +40,39 @@ import type { ProjectConfigService } from "./project-config-service.js";
|
||||
const TRACE_FILE_RE = /^(.+)_(\d{3})\.jsonl$/;
|
||||
const SESSION_ID_TS_RE = /^session-(\d{4})-(\d{2})-(\d{2})-(\d{2})-(\d{2})-(\d{2})-[0-9a-f]{8}$/;
|
||||
|
||||
/** Head window for session_meta reads: generous for a long system prompt, far below a whole multi-MB shard. */
|
||||
const TRACE_HEAD_BYTES = 256 * 1024;
|
||||
|
||||
/**
|
||||
* Parse a Trace file's head window only. session_meta is the first line core writes to
|
||||
* every shard, so a bounded read finds it without pulling the whole file into memory —
|
||||
* category filtering / counts may need every Session's source in a single request.
|
||||
* The window is cut at its last newline (the tail fragment is incomplete); a first line
|
||||
* larger than the whole window falls back to the full tolerant read.
|
||||
*/
|
||||
async function readTraceHead(filePath: string) {
|
||||
const fh = await open(filePath, "r");
|
||||
let text: string;
|
||||
let truncated: boolean;
|
||||
try {
|
||||
// allocUnsafe: only subarray(0, bytesRead) is ever read, so the uninitialized tail never leaks.
|
||||
const { buffer, bytesRead } = await fh.read(
|
||||
Buffer.allocUnsafe(TRACE_HEAD_BYTES),
|
||||
0,
|
||||
TRACE_HEAD_BYTES,
|
||||
0,
|
||||
);
|
||||
text = buffer.subarray(0, bytesRead).toString("utf8");
|
||||
truncated = bytesRead === TRACE_HEAD_BYTES;
|
||||
} finally {
|
||||
await fh.close();
|
||||
}
|
||||
if (!truncated) return parseTraceLines(text);
|
||||
const nl = text.lastIndexOf("\n");
|
||||
if (nl === -1) return readTraceTolerant(filePath);
|
||||
return parseTraceLines(text.slice(0, nl + 1));
|
||||
}
|
||||
|
||||
/** Derives creation time from the local timestamp embedded in session_id; returns null if it doesn't match. */
|
||||
export function sessionIdCreatedAt(sessionId: string): string | null {
|
||||
const m = SESSION_ID_TS_RE.exec(sessionId);
|
||||
@@ -57,10 +97,15 @@ export class SessionService {
|
||||
/**
|
||||
* DB row -> SessionInfo (run status and pending approval count come from session-manager).
|
||||
* Async because `source` is derived from session_meta: a registry miss (Session predating
|
||||
* this process) falls back to reading the Trace head once (see sourceOf).
|
||||
* this process) falls back to reading the Trace head once (see sourceOf). `traces` is the
|
||||
* list flow's one-walk discovery result; without it a miss locates the shard itself.
|
||||
*/
|
||||
async toInfo(row: SessionRow, hasTrace: boolean): Promise<SessionInfo> {
|
||||
const source = await this.sourceOf(row, hasTrace);
|
||||
async toInfo(
|
||||
row: SessionRow,
|
||||
hasTrace: boolean,
|
||||
traces?: ReadonlyMap<string, TraceLocation>,
|
||||
): Promise<SessionInfo> {
|
||||
const source = await this.sourceOf(row, hasTrace, traces);
|
||||
return {
|
||||
sessionId: row.sessionId,
|
||||
projectId: row.projectId,
|
||||
@@ -86,11 +131,20 @@ export class SessionService {
|
||||
* session_meta is read once and cached. A Session with no Trace yet stays unknown and
|
||||
* is NOT cached negatively — its meta may appear with the first run.
|
||||
*/
|
||||
private async sourceOf(row: SessionRow, hasTrace: boolean): Promise<SessionSource | undefined> {
|
||||
private async sourceOf(
|
||||
row: SessionRow,
|
||||
hasTrace: boolean,
|
||||
traces?: ReadonlyMap<string, TraceLocation>,
|
||||
): Promise<SessionSource | undefined> {
|
||||
const known = this.deps.sources.get(row.sessionId);
|
||||
if (known !== undefined) return known ?? undefined;
|
||||
if (!hasTrace) return undefined;
|
||||
const meta = await this.readTraceMeta(row.projectId, row.agentId, row.sessionId);
|
||||
// Single-session paths carry no discovery map: locate this Session's earliest shard on demand.
|
||||
const location =
|
||||
traces?.get(row.sessionId) ??
|
||||
(await this.discoverTraces(row.projectId, row.agentId)).get(row.sessionId);
|
||||
if (!location) return undefined;
|
||||
const meta = await this.readTraceMeta(location);
|
||||
if (!meta) return undefined; // Unreadable/corrupt Trace: stay unknown, retry on the next list.
|
||||
// On-disk values are untrusted: only the exact known origins pass, junk = user-created.
|
||||
const source = asSessionSource(meta.payload.source) ?? null;
|
||||
@@ -100,8 +154,22 @@ export class SessionService {
|
||||
|
||||
/** Whether this Session already has a Trace record (a Task has been run). */
|
||||
async hasTrace(row: SessionRow): Promise<boolean> {
|
||||
const ids = await this.discoverTraceSessionIds(row.projectId, row.agentId);
|
||||
return ids.has(row.sessionId);
|
||||
return (await this.discoverTraces(row.projectId, row.agentId)).has(row.sessionId);
|
||||
}
|
||||
|
||||
/**
|
||||
* The list category of a row: archived wins (an explicit user action), then the
|
||||
* origin's bucket, and no/unknown source is `active` — the same precedence the
|
||||
* sidebar's partition applies to loaded rows, so server filtering and client
|
||||
* rendering can never disagree.
|
||||
*/
|
||||
private async categoryOf(
|
||||
row: SessionRow,
|
||||
hasTrace: boolean,
|
||||
traces?: ReadonlyMap<string, TraceLocation>,
|
||||
): Promise<SessionCategory> {
|
||||
if ((row.archivedAt ?? null) !== null) return "archived";
|
||||
return (await this.sourceOf(row, hasTrace, traces)) ?? "active";
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -110,29 +178,77 @@ export class SessionService {
|
||||
* more"); slicing happens before toInfo, so per-request source derivation (lazy
|
||||
* Trace-head reads) stays bounded by the page size. Discovery/adoption still scans
|
||||
* the whole directory — the union and global ordering need every id.
|
||||
*
|
||||
* `category` filters to one sidebar bucket **before** paging, so offset/limit page
|
||||
* within the category. Filtering needs each walked row's category (a possible
|
||||
* Trace-head read per row, cached in the sources registry); without `withCounts`
|
||||
* the walk stops as soon as the requested page is complete. `withCounts` classifies
|
||||
* every row and returns per-category totals over the whole list — plus the same
|
||||
* totals broken down by Workspace path — so the sidebar can label the collapsed
|
||||
* folders (and a workspace group can know its own share) without loading them.
|
||||
*/
|
||||
async listSessions(
|
||||
projectId: string,
|
||||
agentId: string,
|
||||
paging?: { offset: number; limit: number },
|
||||
): Promise<SessionInfo[]> {
|
||||
const traceIds = await this.discoverTraceSessionIds(projectId, agentId);
|
||||
opts: {
|
||||
paging?: { offset: number; limit: number };
|
||||
category?: SessionCategory;
|
||||
withCounts?: boolean;
|
||||
} = {},
|
||||
): Promise<{
|
||||
sessions: SessionInfo[];
|
||||
counts?: SessionCategoryCounts;
|
||||
workspaceCounts?: Record<string, SessionCategoryCounts>;
|
||||
}> {
|
||||
const { paging, category, withCounts } = opts;
|
||||
const traces = await this.discoverTraces(projectId, agentId);
|
||||
const rows = new Map(
|
||||
this.deps.sessions.listByAgent(projectId, agentId).map((r) => [r.sessionId, r]),
|
||||
);
|
||||
|
||||
// Unmanaged Trace Sessions: backfill an index row by reading the first line's session_meta.
|
||||
for (const sessionId of traceIds) {
|
||||
for (const [sessionId, location] of traces) {
|
||||
if (rows.has(sessionId)) continue;
|
||||
const discovered = await this.adoptTraceSession(projectId, agentId, sessionId);
|
||||
const discovered = await this.adoptTraceSession(projectId, agentId, sessionId, location);
|
||||
if (discovered) rows.set(sessionId, discovered);
|
||||
}
|
||||
|
||||
const sorted = [...rows.values()].sort(
|
||||
(a, b) => b.createdAt.localeCompare(a.createdAt) || b.sessionId.localeCompare(a.sessionId),
|
||||
);
|
||||
const page = paging ? sorted.slice(paging.offset, paging.offset + paging.limit) : sorted;
|
||||
return Promise.all(page.map((row) => this.toInfo(row, traceIds.has(row.sessionId))));
|
||||
const toPage = (page: SessionRow[]) =>
|
||||
Promise.all(page.map((row) => this.toInfo(row, traces.has(row.sessionId), traces)));
|
||||
|
||||
// No classification asked for: slice straight away (the pre-category behavior).
|
||||
if (category === undefined && !withCounts) {
|
||||
return {
|
||||
sessions: await toPage(
|
||||
paging ? sorted.slice(paging.offset, paging.offset + paging.limit) : sorted,
|
||||
),
|
||||
};
|
||||
}
|
||||
|
||||
const want = paging ? paging.offset + paging.limit : Infinity;
|
||||
const counts: SessionCategoryCounts = { active: 0, subagent: 0, schedule: 0, archived: 0 };
|
||||
const workspaceCounts: Record<string, SessionCategoryCounts> = {};
|
||||
const matched: SessionRow[] = [];
|
||||
for (const row of sorted) {
|
||||
if (!withCounts && matched.length >= want) break;
|
||||
const cat = await this.categoryOf(row, traces.has(row.sessionId), traces);
|
||||
counts[cat] += 1;
|
||||
if (withCounts) {
|
||||
const ws = (workspaceCounts[row.workspace] ??= {
|
||||
active: 0,
|
||||
subagent: 0,
|
||||
schedule: 0,
|
||||
archived: 0,
|
||||
});
|
||||
ws[cat] += 1;
|
||||
}
|
||||
if ((category === undefined || cat === category) && matched.length < want) matched.push(row);
|
||||
}
|
||||
const sessions = await toPage(paging ? matched.slice(paging.offset, want) : matched);
|
||||
return withCounts ? { sessions, counts, workspaceCounts } : { sessions };
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -273,45 +389,41 @@ export class SessionService {
|
||||
return this.toInfo(row, false);
|
||||
}
|
||||
|
||||
/** Scans the Trace directory to get the set of session_ids with records. */
|
||||
private async discoverTraceSessionIds(projectId: string, agentId: string): Promise<Set<string>> {
|
||||
const dir = tracesDir(this.deps.root, projectId, agentId);
|
||||
const ids = new Set<string>();
|
||||
for (const dateDir of await listDirsSafe(dir)) {
|
||||
for (const file of await listFilesSafe(path.join(dir, dateDir))) {
|
||||
const match = TRACE_FILE_RE.exec(file);
|
||||
if (match) ids.add(match[1]!);
|
||||
}
|
||||
}
|
||||
return ids;
|
||||
}
|
||||
|
||||
/**
|
||||
* session_meta from the earliest Trace shard of a Session (the shard whose head carries
|
||||
* the original meta); null when there is no readable Trace or it has no meta. Shared by
|
||||
* adoption backfill and lazy `source` resolution.
|
||||
* One walk over the Trace directory: session_id → its **earliest** shard (the shard
|
||||
* whose head carries the original session_meta). Discovery (which Sessions have
|
||||
* records) and the meta-read location come out of a single pass, so classifying every
|
||||
* row (`counts=1`) costs one directory walk total instead of one per Session.
|
||||
*/
|
||||
private async readTraceMeta(
|
||||
private async discoverTraces(
|
||||
projectId: string,
|
||||
agentId: string,
|
||||
sessionId: string,
|
||||
): Promise<SessionMetaMessage | null> {
|
||||
): Promise<Map<string, TraceLocation>> {
|
||||
const dir = tracesDir(this.deps.root, projectId, agentId);
|
||||
let earliest: { path: string; index: number } | null = null;
|
||||
const out = new Map<string, TraceLocation>();
|
||||
for (const dateDir of await listDirsSafe(dir)) {
|
||||
for (const file of await listFilesSafe(path.join(dir, dateDir))) {
|
||||
const match = TRACE_FILE_RE.exec(file);
|
||||
if (!match || match[1] !== sessionId) continue;
|
||||
if (!match) continue;
|
||||
const sessionId = match[1]!;
|
||||
const index = Number(match[2]);
|
||||
if (!earliest || index < earliest.index) {
|
||||
earliest = { path: path.join(dir, dateDir, file), index };
|
||||
const cur = out.get(sessionId);
|
||||
if (!cur || index < cur.index) {
|
||||
out.set(sessionId, { path: path.join(dir, dateDir, file), index });
|
||||
}
|
||||
}
|
||||
}
|
||||
if (!earliest) return null;
|
||||
return out;
|
||||
}
|
||||
|
||||
/**
|
||||
* session_meta from a located Trace shard head; null when unreadable or it has no
|
||||
* meta. Shared by adoption backfill and lazy `source` resolution.
|
||||
*/
|
||||
private async readTraceMeta(location: TraceLocation): Promise<SessionMetaMessage | null> {
|
||||
let messages;
|
||||
try {
|
||||
messages = await readTraceTolerant(earliest.path);
|
||||
messages = await readTraceHead(location.path);
|
||||
} catch {
|
||||
return null; // Corrupt file: skip (does not block the list)
|
||||
}
|
||||
@@ -323,8 +435,9 @@ export class SessionService {
|
||||
projectId: string,
|
||||
agentId: string,
|
||||
sessionId: string,
|
||||
location: TraceLocation,
|
||||
): Promise<SessionRow | null> {
|
||||
const meta = await this.readTraceMeta(projectId, agentId, sessionId);
|
||||
const meta = await this.readTraceMeta(location);
|
||||
if (!meta) return null;
|
||||
// An older Trace version's session_meta lacks provider (the model reference
|
||||
// wasn't split into separate fields yet): no backward compat, skip adoption
|
||||
@@ -352,6 +465,12 @@ export class SessionService {
|
||||
}
|
||||
}
|
||||
|
||||
/** A located Trace shard of one Session: absolute path plus its shard index. */
|
||||
interface TraceLocation {
|
||||
path: string;
|
||||
index: number;
|
||||
}
|
||||
|
||||
/** Local date as yyyy-mm-dd (matches the Trace date directory convention: core's internal formatLocalDate, not publicly exported). */
|
||||
function localDate(d: Date): string {
|
||||
const pad = (n: number) => (n < 10 ? `0${n}` : `${n}`);
|
||||
|
||||
Reference in New Issue
Block a user