feat: session source in session_meta and an Automated sidebar folder (#40)
Co-authored-by: Alice <alice@prismshadow.com> Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
30 files changed
+1023
-74
No files matched your search
@@ -454,7 +454,7 @@ export interface SessionInfo {
|
||||
approvalMode: ApprovalMode;
|
||||
/** Short title auto-generated by the model after the first turn; unset until generated (frontend shows "New Chat"). */
|
||||
title?: string;
|
||||
/** Session source (for list badges); unset for user-created sessions. */
|
||||
/** Session source (for list badges/folders), derived from core session_meta — the single source of truth (not stored in the DB); unset for user-created sessions. */
|
||||
source?: SessionSource;
|
||||
createdAt: string;
|
||||
status: SessionStatus;
|
||||
|
||||
@@ -50,6 +50,7 @@ import { agentSessionsRoutes, sessionsRoutes } from "./http/routes/sessions.js";
|
||||
import { ChannelHub } from "./runtime/channel.js";
|
||||
import { ErrorRecorder } from "./runtime/error-recorder.js";
|
||||
import { createCoreSessionLoader, SessionManager } from "./runtime/session-manager.js";
|
||||
import { SessionSources } from "./runtime/session-sources.js";
|
||||
import type { SessionLoader } from "./runtime/session-manager.js";
|
||||
import { Scheduler } from "./runtime/scheduler.js";
|
||||
import { TitleGenerator } from "./runtime/title-generator.js";
|
||||
@@ -91,6 +92,8 @@ export interface AppDeps {
|
||||
scheduler: Scheduler;
|
||||
channels: ChannelHub;
|
||||
manager: SessionManager;
|
||||
/** Session-origin registry derived from session_meta (single source of truth; no DB column). */
|
||||
sessionSources: SessionSources;
|
||||
/** Error persistence (shared by app.onError and various background capture points; the process-level fallback is in index.ts). */
|
||||
errors: ErrorRecorder;
|
||||
/** Request log output (minimal one-liner); tests inject a noop. */
|
||||
@@ -149,10 +152,15 @@ export function buildAppDeps(config: ServerConfig, overrides: BuildDepsOverrides
|
||||
const titles =
|
||||
overrides.titles ??
|
||||
new TitleGenerator({ sessions: sessionsRepo, channels, recorder, errors, log });
|
||||
// Session-origin registry: session_meta is the single source of truth (no DB column);
|
||||
// shared by the manager (subagent registration), the loader (self-heal rebuild) and
|
||||
// SessionService (creation / adoption / lazy list resolution).
|
||||
const sessionSources = new SessionSources();
|
||||
const manager = new SessionManager({
|
||||
sessions: sessionsRepo,
|
||||
channels,
|
||||
loader: overrides.loader ?? createCoreSessionLoader(config.root),
|
||||
loader: overrides.loader ?? createCoreSessionLoader(config.root, sessionSources),
|
||||
sources: sessionSources,
|
||||
recorder,
|
||||
errors,
|
||||
titles,
|
||||
@@ -194,6 +202,7 @@ export function buildAppDeps(config: ServerConfig, overrides: BuildDepsOverrides
|
||||
sessions: sessionsRepo,
|
||||
manager,
|
||||
projectConfig: projectConfigService,
|
||||
sources: sessionSources,
|
||||
});
|
||||
// Schedule scheduler: active only while the server is running. Only
|
||||
// assembled here; start() is called in index.ts (tests drive it via tickOnce, no real timer).
|
||||
@@ -232,6 +241,7 @@ export function buildAppDeps(config: ServerConfig, overrides: BuildDepsOverrides
|
||||
scheduler,
|
||||
channels,
|
||||
manager,
|
||||
sessionSources,
|
||||
errors,
|
||||
log,
|
||||
};
|
||||
|
||||
@@ -19,8 +19,8 @@ export interface SessionRow {
|
||||
title: string | null;
|
||||
/** Archive timestamp, ISO; NULL = not archived (omitting on insert defaults to NULL). */
|
||||
archivedAt?: string | null;
|
||||
/** Session origin: NULL = user-created; schedule = triggered by a Schedule; subagent = registered as a subagent session. */
|
||||
source?: "schedule" | "subagent" | null;
|
||||
// The Session origin (schedule / subagent) is deliberately NOT a row field: core
|
||||
// session_meta in the Trace is the single source of truth (runtime/session-sources.ts).
|
||||
createdAt: string;
|
||||
}
|
||||
|
||||
@@ -35,7 +35,6 @@ function mapRow(r: Record<string, unknown>): SessionRow {
|
||||
approvalMode: r.approval_mode as ApprovalMode,
|
||||
title: (r.title as string | null) ?? null,
|
||||
archivedAt: (r.archived_at as string | null) ?? null,
|
||||
source: (r.source as "schedule" | "subagent" | null) ?? null,
|
||||
createdAt: r.created_at as string,
|
||||
};
|
||||
}
|
||||
@@ -46,8 +45,8 @@ export class SessionsRepo {
|
||||
insert(row: SessionRow): void {
|
||||
this.db
|
||||
.prepare(
|
||||
`INSERT INTO sessions (session_id, project_id, agent_id, provider, model_id, workspace, approval_mode, title, source, created_at)
|
||||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
|
||||
`INSERT INTO sessions (session_id, project_id, agent_id, provider, model_id, workspace, approval_mode, title, created_at)
|
||||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`,
|
||||
)
|
||||
.run(
|
||||
row.sessionId,
|
||||
@@ -58,7 +57,6 @@ export class SessionsRepo {
|
||||
row.workspace,
|
||||
row.approvalMode,
|
||||
row.title,
|
||||
row.source ?? null,
|
||||
row.createdAt,
|
||||
);
|
||||
}
|
||||
@@ -67,8 +65,8 @@ export class SessionsRepo {
|
||||
insertOrIgnore(row: SessionRow): void {
|
||||
this.db
|
||||
.prepare(
|
||||
`INSERT OR IGNORE INTO sessions (session_id, project_id, agent_id, provider, model_id, workspace, approval_mode, title, source, created_at)
|
||||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
|
||||
`INSERT OR IGNORE INTO sessions (session_id, project_id, agent_id, provider, model_id, workspace, approval_mode, title, created_at)
|
||||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`,
|
||||
)
|
||||
.run(
|
||||
row.sessionId,
|
||||
@@ -79,7 +77,6 @@ export class SessionsRepo {
|
||||
row.workspace,
|
||||
row.approvalMode,
|
||||
row.title,
|
||||
row.source ?? null,
|
||||
row.createdAt,
|
||||
);
|
||||
}
|
||||
|
||||
@@ -48,9 +48,8 @@ CREATE TABLE IF NOT EXISTS sessions (
|
||||
approval_mode TEXT NOT NULL DEFAULT 'allow-all', -- allow-all|deny-all|read-only|always-ask
|
||||
title TEXT, -- auto-generated by the model after the first exchange; NULL=not yet (frontend shows "New chat")
|
||||
archived_at TEXT, -- archive time; NULL=not archived (shown by default; archived ones move under "Archived")
|
||||
source TEXT, -- session origin: NULL=user-created | schedule (scheduled task) | subagent (child session)
|
||||
created_at TEXT NOT NULL
|
||||
);
|
||||
); -- session origin is NOT stored: session_meta in the Trace is the single source of truth (see runtime/session-sources.ts)
|
||||
CREATE TABLE IF NOT EXISTS usage_records (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
ts TEXT NOT NULL,
|
||||
|
||||
@@ -30,6 +30,7 @@ import { sseEndpoint } from "../sse.js";
|
||||
import {
|
||||
badRequest,
|
||||
optionalEnum,
|
||||
optionalPagingQuery,
|
||||
optionalString,
|
||||
paginationQuery,
|
||||
pathParam,
|
||||
@@ -96,7 +97,14 @@ export function agentSessionsRoutes(deps: AppDeps): Hono<AppEnv> {
|
||||
const agentId = requireValidId(c, "agentId");
|
||||
deps.projectService.requireProjectAccess(c.var.user.userId, projectId);
|
||||
await deps.agentConfigService.requireExists(projectId, agentId);
|
||||
const sessions = await deps.sessionService.listSessions(projectId, agentId);
|
||||
// Optional paging (absent = full list, the pre-paging contract): the sidebar requests
|
||||
// limit+1 and shows limit, detecting "has more" without a response-envelope change.
|
||||
const paging = optionalPagingQuery(c);
|
||||
const sessions = await deps.sessionService.listSessions(
|
||||
projectId,
|
||||
agentId,
|
||||
...(paging ? [paging] : []),
|
||||
);
|
||||
return c.json({ sessions } satisfies SessionsResponse);
|
||||
});
|
||||
|
||||
@@ -167,7 +175,9 @@ export function sessionsRoutes(deps: AppDeps): Hono<AppEnv> {
|
||||
app.get("/:sessionId", async (c) => {
|
||||
const row = resolveSession(c);
|
||||
const hasTrace = await deps.sessionService.hasTrace(row);
|
||||
return c.json({ session: deps.sessionService.toInfo(row, hasTrace) } satisfies SessionResponse);
|
||||
return c.json({
|
||||
session: await deps.sessionService.toInfo(row, hasTrace),
|
||||
} satisfies SessionResponse);
|
||||
});
|
||||
|
||||
app.patch("/:sessionId", async (c) => {
|
||||
@@ -216,7 +226,7 @@ export function sessionsRoutes(deps: AppDeps): Hono<AppEnv> {
|
||||
}
|
||||
const hasTrace = await deps.sessionService.hasTrace(updated);
|
||||
return c.json({
|
||||
session: deps.sessionService.toInfo(updated, hasTrace),
|
||||
session: await deps.sessionService.toInfo(updated, hasTrace),
|
||||
} satisfies SessionResponse);
|
||||
});
|
||||
|
||||
@@ -244,6 +254,9 @@ export function sessionsRoutes(deps: AppDeps): Hono<AppEnv> {
|
||||
{ recursive: true, force: true },
|
||||
);
|
||||
deps.sessionsRepo.deleteById(row.sessionId);
|
||||
// Drop the derived-origin entry along with the Session (bulk Agent/Project deletion
|
||||
// may leave stale entries; session ids are never reused, so they are never matched).
|
||||
deps.sessionSources.delete(row.sessionId);
|
||||
} finally {
|
||||
deps.manager.endSessionDeletion(row.sessionId);
|
||||
}
|
||||
|
||||
@@ -64,6 +64,28 @@ export function paginationQuery(c: Context): { offset: number; limit: number } {
|
||||
return { offset, limit };
|
||||
}
|
||||
|
||||
/**
|
||||
* Parse OPTIONAL list paging query params: both absent = null (caller returns the full
|
||||
* list — the pre-paging behavior stays intact for existing callers). When paging, `limit`
|
||||
* is required (1-1000) and `offset` optional (>= 0, default 0) — an offset alone would
|
||||
* silently return the full list shifted, which no caller ever means.
|
||||
*/
|
||||
export function optionalPagingQuery(c: Context): { offset: number; limit: number } | null {
|
||||
const rawLimit = c.req.query("limit");
|
||||
const rawOffset = c.req.query("offset");
|
||||
if (rawLimit === undefined && rawOffset === undefined) return null;
|
||||
if (rawLimit === undefined) throw badRequest("offset requires limit.");
|
||||
const limit = Number.parseInt(rawLimit, 10);
|
||||
if (!Number.isInteger(limit) || limit < 1 || limit > 1000) {
|
||||
throw badRequest("limit must be an integer between 1 and 1000.");
|
||||
}
|
||||
const offset = Number.parseInt(rawOffset ?? "0", 10);
|
||||
if (!Number.isInteger(offset) || offset < 0) {
|
||||
throw badRequest("offset must be a non-negative integer.");
|
||||
}
|
||||
return { offset, limit };
|
||||
}
|
||||
|
||||
/** Read the JSON request body (parse failure / non-object -> 400). */
|
||||
export async function readJson(c: Context): Promise<Record<string, unknown>> {
|
||||
let body: unknown;
|
||||
|
||||
@@ -48,6 +48,8 @@ import { ApprovalRegistry, makeApprove } from "./approvals.js";
|
||||
import type { PendingApproval } from "./approvals.js";
|
||||
import type { ChannelHub } from "./channel.js";
|
||||
import type { ErrorSink } from "./error-recorder.js";
|
||||
import { asSessionSource } from "./session-sources.js";
|
||||
import type { SessionSources } from "./session-sources.js";
|
||||
import { StreamErrorWatcher } from "./stream-error-watcher.js";
|
||||
import type { TitleNotifier } from "./title-generator.js";
|
||||
import type { UsageContext } from "./usage-recorder.js";
|
||||
@@ -96,8 +98,13 @@ export interface SessionLoader {
|
||||
load(row: SessionRow): Promise<RuntimeSession>;
|
||||
}
|
||||
|
||||
/** Production loader: the core SDK's resumeSession / createSession. */
|
||||
export function createCoreSessionLoader(root: string): SessionLoader {
|
||||
/**
|
||||
* Production loader: the core SDK's resumeSession / createSession. `sources` (when given)
|
||||
* lets the no-Trace self-heal rebuild re-record a known origin into the fresh session_meta;
|
||||
* with no registry entry (e.g. the process restarted and no Trace was ever written) the
|
||||
* rebuilt Session is unsourced — session_meta is the single source of truth, and none survived.
|
||||
*/
|
||||
export function createCoreSessionLoader(root: string, sources?: SessionSources): SessionLoader {
|
||||
return {
|
||||
async load(row: SessionRow): Promise<RuntimeSession> {
|
||||
const agent = await createAgent({
|
||||
@@ -136,11 +143,14 @@ export function createCoreSessionLoader(root: string): SessionLoader {
|
||||
`This Session's Workspace no longer exists: ${row.workspace}, so it cannot continue. Create a new Session.`,
|
||||
);
|
||||
}
|
||||
const knownSource = sources?.get(row.sessionId);
|
||||
try {
|
||||
return await agent.createSession({
|
||||
workspaceDir: row.workspace,
|
||||
modelId: row.modelId,
|
||||
provider: row.provider,
|
||||
// The rebuilt Session re-records a known origin in its fresh session_meta.
|
||||
...(knownSource != null ? { source: knownSource } : {}),
|
||||
});
|
||||
} catch (err) {
|
||||
if (isMissingCredential(err)) throw modelCredentialMissing(row.modelId);
|
||||
@@ -168,6 +178,8 @@ export interface SessionManagerDeps {
|
||||
sessions: SessionsRepo;
|
||||
channels: ChannelHub;
|
||||
loader: SessionLoader;
|
||||
/** Session-origin registry (session_meta is the single source of truth; subagent registration records the forwarded meta's source here). */
|
||||
sources: SessionSources;
|
||||
recorder: UsageRecorderLike;
|
||||
/** Automatic Session title generation (optional: not injected in tests or when disabled). */
|
||||
titles?: TitleNotifier;
|
||||
@@ -879,6 +891,12 @@ export class SessionManager {
|
||||
const p = msg.payload as SessionMetaPayload;
|
||||
const agentId = path.basename(path.dirname(p.agent_state));
|
||||
if (!agentId || agentId === "." || agentId === "..") return null;
|
||||
// The forwarded session_meta records the origin at the source (core's spawn site); fall
|
||||
// back to inferring "subagent" from the registration path for older metas (narrowed —
|
||||
// a junk value also falls back). It goes into the in-process registry only — the index
|
||||
// row deliberately stores no source column.
|
||||
const source = asSessionSource(p.source) ?? "subagent";
|
||||
this.deps.sources.set(childSid, source);
|
||||
this.deps.sessions.insertOrIgnore({
|
||||
sessionId: childSid,
|
||||
projectId: entry.projectId,
|
||||
@@ -890,7 +908,6 @@ export class SessionManager {
|
||||
// inserted with defaults (matches the convention for Sessions discovered by the CLI).
|
||||
approvalMode: "allow-all",
|
||||
title: null,
|
||||
source: "subagent",
|
||||
createdAt: new Date().toISOString(),
|
||||
});
|
||||
// Make the subagent appear immediately in the sidebar: notify via the parent
|
||||
@@ -900,7 +917,7 @@ export class SessionManager {
|
||||
projectId: entry.projectId,
|
||||
agentId,
|
||||
sessionId: childSid,
|
||||
source: "subagent",
|
||||
source,
|
||||
});
|
||||
const child: ChildSession = {
|
||||
sessionId: childSid,
|
||||
|
||||
@@ -0,0 +1,41 @@
|
||||
/**
|
||||
* In-process registry of Session origins, derived from core `session_meta` — the single
|
||||
* source of truth for a Session's origin (the DB stores no `source` column).
|
||||
*
|
||||
* Populated wherever the server actually has the meta in hand: Session creation
|
||||
* (SessionService reads the just-created core Session's meta), subagent registration
|
||||
* (SessionManager reads the forwarded child meta), and Trace adoption / lazy list
|
||||
* resolution (SessionService reads the Trace head's session_meta). `null` records a
|
||||
* **known** user-created Session (meta seen, no source) so the Trace is not re-read on
|
||||
* every list; an absent entry means "unknown" and the list path resolves it from the
|
||||
* Trace once per process lifetime.
|
||||
*/
|
||||
import type { SessionSource } from "../api/types.js";
|
||||
|
||||
/**
|
||||
* Narrows an untrusted value (on-disk Trace JSON / forwarded meta) to a SessionSource:
|
||||
* only the exact known origins pass; anything else — including junk written by third
|
||||
* parties — is treated as absent rather than cast through.
|
||||
*/
|
||||
export function asSessionSource(v: unknown): SessionSource | undefined {
|
||||
return v === "schedule" || v === "subagent" ? v : undefined;
|
||||
}
|
||||
|
||||
export class SessionSources {
|
||||
private readonly map = new Map<string, SessionSource | null>();
|
||||
|
||||
/** Records a Session's origin as read from session_meta (`null` = meta seen, user-created). */
|
||||
set(sessionId: string, source: SessionSource | null): void {
|
||||
this.map.set(sessionId, source);
|
||||
}
|
||||
|
||||
/** Known origin, `null` for a known user-created Session, `undefined` when this process has not seen the meta. */
|
||||
get(sessionId: string): SessionSource | null | undefined {
|
||||
return this.map.get(sessionId);
|
||||
}
|
||||
|
||||
/** Drops a deleted Session's entry (bulk Agent/Project deletion may leave stale entries; they are never matched again). */
|
||||
delete(sessionId: string): void {
|
||||
this.map.delete(sessionId);
|
||||
}
|
||||
}
|
||||
@@ -20,11 +20,14 @@ import {
|
||||
readTraceTolerant,
|
||||
tracesDir,
|
||||
} from "@prismshadow/penguin-core";
|
||||
import type { ApprovalMode, SessionInfo } from "../api/types.js";
|
||||
import type { SessionMetaMessage } from "@prismshadow/penguin-core";
|
||||
import type { ApprovalMode, 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";
|
||||
import type { SessionManager } from "../runtime/session-manager.js";
|
||||
import { asSessionSource } from "../runtime/session-sources.js";
|
||||
import type { SessionSources } from "../runtime/session-sources.js";
|
||||
import type { ProjectConfigService } from "./project-config-service.js";
|
||||
|
||||
const TRACE_FILE_RE = /^(.+)_(\d{3})\.jsonl$/;
|
||||
@@ -44,13 +47,20 @@ export interface SessionServiceDeps {
|
||||
sessions: SessionsRepo;
|
||||
manager: SessionManager;
|
||||
projectConfig: ProjectConfigService;
|
||||
/** In-process origin registry derived from session_meta (the DB stores no source column). */
|
||||
sources: SessionSources;
|
||||
}
|
||||
|
||||
export class SessionService {
|
||||
constructor(private readonly deps: SessionServiceDeps) {}
|
||||
|
||||
/** DB row -> SessionInfo (run status and pending approval count come from session-manager). */
|
||||
toInfo(row: SessionRow, hasTrace: boolean): SessionInfo {
|
||||
/**
|
||||
* 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).
|
||||
*/
|
||||
async toInfo(row: SessionRow, hasTrace: boolean): Promise<SessionInfo> {
|
||||
const source = await this.sourceOf(row, hasTrace);
|
||||
return {
|
||||
sessionId: row.sessionId,
|
||||
projectId: row.projectId,
|
||||
@@ -60,7 +70,7 @@ export class SessionService {
|
||||
workspace: row.workspace,
|
||||
approvalMode: row.approvalMode,
|
||||
...(row.title !== null ? { title: row.title } : {}),
|
||||
...(row.source != null ? { source: row.source } : {}),
|
||||
...(source !== undefined ? { source } : {}),
|
||||
createdAt: row.createdAt,
|
||||
status: this.deps.manager.statusOf(row.sessionId),
|
||||
pendingApprovalCount: this.deps.manager.pendingApprovalCount(row.sessionId),
|
||||
@@ -69,14 +79,43 @@ export class SessionService {
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* A Session's origin, with session_meta as the single source of truth: the in-process
|
||||
* registry answers first (populated at creation / subagent registration / adoption);
|
||||
* on a miss (a Session created before this process started) the earliest Trace shard's
|
||||
* 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> {
|
||||
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);
|
||||
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;
|
||||
this.deps.sources.set(row.sessionId, source);
|
||||
return source ?? undefined;
|
||||
}
|
||||
|
||||
/** 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);
|
||||
}
|
||||
|
||||
/** List: DB ∪ Trace directory discovery, sorted by createdAt descending. */
|
||||
async listSessions(projectId: string, agentId: string): Promise<SessionInfo[]> {
|
||||
/**
|
||||
* List: DB ∪ Trace directory discovery, sorted by createdAt descending. Optional
|
||||
* `paging` returns just that slice (the sidebar pages with limit+1 to detect "has
|
||||
* 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.
|
||||
*/
|
||||
async listSessions(
|
||||
projectId: string,
|
||||
agentId: string,
|
||||
paging?: { offset: number; limit: number },
|
||||
): Promise<SessionInfo[]> {
|
||||
const traceIds = await this.discoverTraceSessionIds(projectId, agentId);
|
||||
const rows = new Map(
|
||||
this.deps.sessions.listByAgent(projectId, agentId).map((r) => [r.sessionId, r]),
|
||||
@@ -89,11 +128,11 @@ export class SessionService {
|
||||
if (discovered) rows.set(sessionId, discovered);
|
||||
}
|
||||
|
||||
return [...rows.values()]
|
||||
.sort(
|
||||
(a, b) => b.createdAt.localeCompare(a.createdAt) || b.sessionId.localeCompare(a.sessionId),
|
||||
)
|
||||
.map((row) => this.toInfo(row, traceIds.has(row.sessionId)));
|
||||
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))));
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -196,6 +235,8 @@ export class SessionService {
|
||||
modelId,
|
||||
provider,
|
||||
...(args.workspace !== undefined ? { workspaceDir: args.workspace } : {}),
|
||||
// The origin is also recorded in core session_meta (Trace), not just the index row.
|
||||
...(args.source !== undefined ? { source: args.source } : {}),
|
||||
});
|
||||
} catch (err) {
|
||||
// A missing credential is its own category (the frontend shows localized text
|
||||
@@ -208,6 +249,14 @@ export class SessionService {
|
||||
err instanceof Error ? err.message : String(err),
|
||||
);
|
||||
}
|
||||
// The origin is derived from the just-created core Session's session_meta (the single
|
||||
// source of truth) rather than echoing args.source back: what the registry serves is
|
||||
// exactly what the Trace will record.
|
||||
const metaMsg = session.metaMessage;
|
||||
this.deps.sources.set(
|
||||
session.sessionId,
|
||||
isSessionMeta(metaMsg) ? (asSessionSource(metaMsg.payload.source) ?? null) : null,
|
||||
);
|
||||
const row: SessionRow = {
|
||||
sessionId: session.sessionId,
|
||||
projectId: args.projectId,
|
||||
@@ -218,7 +267,6 @@ export class SessionService {
|
||||
approvalMode: args.approvalMode ?? "allow-all",
|
||||
title: null,
|
||||
createdAt: new Date().toISOString(),
|
||||
...(args.source !== undefined ? { source: args.source } : {}),
|
||||
};
|
||||
this.deps.sessions.insert(row);
|
||||
this.deps.manager.adopt(row, session);
|
||||
@@ -238,12 +286,16 @@ export class SessionService {
|
||||
return ids;
|
||||
}
|
||||
|
||||
/** Adopts a Session that exists only in the Trace directory: reads session_meta from the first line of the earliest index file. */
|
||||
private async adoptTraceSession(
|
||||
/**
|
||||
* 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.
|
||||
*/
|
||||
private async readTraceMeta(
|
||||
projectId: string,
|
||||
agentId: string,
|
||||
sessionId: string,
|
||||
): Promise<SessionRow | null> {
|
||||
): Promise<SessionMetaMessage | null> {
|
||||
const dir = tracesDir(this.deps.root, projectId, agentId);
|
||||
let earliest: { path: string; index: number } | null = null;
|
||||
for (const dateDir of await listDirsSafe(dir)) {
|
||||
@@ -263,13 +315,25 @@ export class SessionService {
|
||||
} catch {
|
||||
return null; // Corrupt file: skip (does not block the list)
|
||||
}
|
||||
const meta = messages.find(isSessionMeta);
|
||||
return messages.find(isSessionMeta) ?? null;
|
||||
}
|
||||
|
||||
/** Adopts a Session that exists only in the Trace directory: reads session_meta from the first line of the earliest index file. */
|
||||
private async adoptTraceSession(
|
||||
projectId: string,
|
||||
agentId: string,
|
||||
sessionId: string,
|
||||
): Promise<SessionRow | null> {
|
||||
const meta = await this.readTraceMeta(projectId, agentId, sessionId);
|
||||
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
|
||||
// (core will give a clear error on resume; the product hasn't launched yet, so
|
||||
// old data can simply be deleted and recreated).
|
||||
if (typeof meta.payload.provider !== "string") return null;
|
||||
// The adoption read already has the meta in hand: record the origin (single source of
|
||||
// truth); on-disk values are narrowed — junk counts as user-created.
|
||||
this.deps.sources.set(sessionId, asSessionSource(meta.payload.source) ?? null);
|
||||
const row: SessionRow = {
|
||||
sessionId,
|
||||
projectId,
|
||||
|
||||
@@ -40,6 +40,7 @@ describe("scheduler", () => {
|
||||
workspace?: string;
|
||||
provider?: string;
|
||||
modelId?: string;
|
||||
source?: "schedule";
|
||||
}>;
|
||||
let events: Array<{ userId: string; event: ScheduleServerEvent }>;
|
||||
let errors: ErrorRecordArgs[];
|
||||
@@ -288,13 +289,15 @@ describe("scheduler", () => {
|
||||
nowMs = T0 + 6 * MIN;
|
||||
await scheduler.tickOnce();
|
||||
expect(created).toHaveLength(2);
|
||||
// The file's model reference is passed straight through as a pair.
|
||||
// The file's model reference is passed straight through as a pair; every schedule-opened
|
||||
// session is marked with its origin (SessionService stores it on the row and in core session_meta).
|
||||
expect(created[0]).toMatchObject({
|
||||
projectId: P,
|
||||
agentId: A,
|
||||
workspace: "/tmp/ws",
|
||||
provider: "custom",
|
||||
modelId: "m-bench",
|
||||
source: "schedule",
|
||||
});
|
||||
expect(started.map((s) => s.sessionId)).toEqual(["session-new-1", "session-new-2"]);
|
||||
});
|
||||
|
||||
@@ -103,6 +103,140 @@ describe("session-index", () => {
|
||||
expect(list.sessions.map((s) => s.sessionId)).toContain(session.sessionId);
|
||||
});
|
||||
|
||||
it("schedule-created Session: source derives from session_meta (registry), never from the DB row; user sessions carry none", async () => {
|
||||
await configureModels();
|
||||
// The scheduler goes through SessionService.createSession directly (no HTTP route exposes source).
|
||||
const info = await t.deps.sessionService.createSession({
|
||||
projectId,
|
||||
agentId: "default_agent",
|
||||
source: "schedule",
|
||||
});
|
||||
expect(info.source).toBe("schedule");
|
||||
// The index row stores no origin: session_meta is the single source of truth.
|
||||
const row = t.deps.sessionsRepo.findById(info.sessionId);
|
||||
expect(row && "source" in row).toBe(false);
|
||||
expect(t.deps.sessionSources.get(info.sessionId)).toBe("schedule");
|
||||
|
||||
// A user-created session (HTTP) has no source, and the list surfaces both accordingly.
|
||||
const res = await api.post(base(), {});
|
||||
expect(res.status).toBe(201);
|
||||
const { session: plain } = (await res.json()) as SessionCreateResponse;
|
||||
const list = (await (await api.get(base())).json()) as SessionsResponse;
|
||||
expect(list.sessions.find((s) => s.sessionId === info.sessionId)?.source).toBe("schedule");
|
||||
expect(list.sessions.find((s) => s.sessionId === plain.sessionId)?.source).toBeUndefined();
|
||||
});
|
||||
|
||||
it("source survives a restart via the Trace head: an indexed row unknown to this process derives it lazily from session_meta", async () => {
|
||||
await configureModels();
|
||||
// Simulate a Session created by a previous process: the index row exists, but the
|
||||
// in-process registry has never seen it — only its Trace's session_meta knows the origin.
|
||||
const sid = "session-2026-07-02-09-00-00-feedc0de";
|
||||
t.deps.sessionsRepo.insert({
|
||||
sessionId: sid,
|
||||
projectId,
|
||||
agentId: "default_agent",
|
||||
provider: "custom",
|
||||
modelId: "m-x",
|
||||
workspace: "/tmp/w-restart",
|
||||
approvalMode: "allow-all",
|
||||
title: null,
|
||||
createdAt: "2026-07-02T09:00:00.000Z",
|
||||
});
|
||||
const meta: SessionMetaPayload = {
|
||||
session_id: sid,
|
||||
model_id: "m-x",
|
||||
provider: "custom",
|
||||
model_context_window: 1000,
|
||||
system_prompt: "",
|
||||
tools: [],
|
||||
thinking_level: "default",
|
||||
agent_state: "/tmp/a",
|
||||
workspace: "/tmp/w-restart",
|
||||
source: "subagent",
|
||||
};
|
||||
await writeTraceFile(t.root, projectId, "default_agent", "2026-07-02", sid, 1, [
|
||||
sessionMeta(meta),
|
||||
userText("child work"),
|
||||
]);
|
||||
const list = (await (await api.get(base())).json()) as SessionsResponse;
|
||||
expect(list.sessions.find((s) => s.sessionId === sid)?.source).toBe("subagent");
|
||||
// The single-session endpoint derives it the same way (and the second read hits the registry).
|
||||
const single = (await (await api.get(`/api/sessions/${sid}`)).json()) as SessionResponse;
|
||||
expect(single.session.source).toBe("subagent");
|
||||
});
|
||||
|
||||
it("adoption derives source from the Trace meta, narrowing junk values to user-created", async () => {
|
||||
await configureModels();
|
||||
// Discovered (no index row) with a valid origin: adoption records it.
|
||||
const adopted = "session-2026-07-03-10-00-00-0badf00d";
|
||||
const sourced: SessionMetaPayload = {
|
||||
session_id: adopted,
|
||||
model_id: "m-cli",
|
||||
provider: "custom",
|
||||
model_context_window: 1000,
|
||||
system_prompt: "",
|
||||
tools: [],
|
||||
thinking_level: "default",
|
||||
agent_state: "/tmp/a",
|
||||
workspace: "/tmp/w-cli",
|
||||
source: "schedule",
|
||||
};
|
||||
await writeTraceFile(t.root, projectId, "default_agent", "2026-07-03", adopted, 1, [
|
||||
sessionMeta(sourced),
|
||||
userText("adopted"),
|
||||
]);
|
||||
// Discovered with a junk source (untrusted on-disk data): narrowed to user-created.
|
||||
const junk = "session-2026-07-03-11-00-00-0badf00e";
|
||||
const junkMeta = {
|
||||
...sourced,
|
||||
session_id: junk,
|
||||
source: "weird-origin",
|
||||
} as unknown as SessionMetaPayload;
|
||||
await writeTraceFile(t.root, projectId, "default_agent", "2026-07-03", junk, 1, [
|
||||
sessionMeta(junkMeta),
|
||||
userText("junk"),
|
||||
]);
|
||||
const list = (await (await api.get(base())).json()) as SessionsResponse;
|
||||
expect(list.sessions.find((s) => s.sessionId === adopted)?.source).toBe("schedule");
|
||||
expect(list.sessions.find((s) => s.sessionId === junk)?.source).toBeUndefined();
|
||||
});
|
||||
|
||||
it("list paging: limit/offset slice the newest-first list; absent params keep the full list; invalid values 400", async () => {
|
||||
await configureModels();
|
||||
// Three sessions with distinct createdAt ordering (insert directly for deterministic times).
|
||||
const mk = (n: number) => ({
|
||||
sessionId: `session-2026-07-0${n}-08-00-00-aaaa000${n}`,
|
||||
projectId,
|
||||
agentId: "default_agent",
|
||||
provider: "custom",
|
||||
modelId: "m-page",
|
||||
workspace: `/tmp/w-${n}`,
|
||||
approvalMode: "allow-all" as const,
|
||||
title: null,
|
||||
createdAt: `2026-07-0${n}T08:00:00.000Z`,
|
||||
});
|
||||
for (const n of [1, 2, 3]) t.deps.sessionsRepo.insert(mk(n));
|
||||
|
||||
const ids = async (qs: string) => {
|
||||
const res = await api.get(`${base()}${qs}`);
|
||||
expect(res.status).toBe(200);
|
||||
const body = (await res.json()) as SessionsResponse;
|
||||
return body.sessions.map((s) => s.sessionId);
|
||||
};
|
||||
const all = await ids("");
|
||||
expect(all).toEqual([mk(3).sessionId, mk(2).sessionId, mk(1).sessionId]); // newest first, unpaged
|
||||
expect(await ids("?limit=2")).toEqual(all.slice(0, 2));
|
||||
expect(await ids("?limit=2&offset=2")).toEqual(all.slice(2));
|
||||
expect(await ids("?limit=2&offset=9")).toEqual([]); // past the end: empty page, not an error
|
||||
// The sidebar's limit+1 trick: one extra row answers "has more" without an envelope change.
|
||||
expect((await ids("?limit=3")).length).toBe(3);
|
||||
|
||||
for (const bad of ["?limit=0", "?limit=-1", "?limit=abc", "?limit=1001", "?offset=1"]) {
|
||||
expect((await api.get(`${base()}${bad}`)).status, bad).toBe(400);
|
||||
}
|
||||
expect((await api.get(`${base()}?limit=2&offset=-1`)).status).toBe(400);
|
||||
});
|
||||
|
||||
it("half a model reference is 400: the missing half is never inferred", async () => {
|
||||
await configureModels();
|
||||
// Only modelId: even though it names the one configured model, the provider is never
|
||||
|
||||
@@ -10,6 +10,7 @@ import { afterEach, beforeEach, describe, expect, it } from "vitest";
|
||||
import { createAgent, saveProjectConfig, sessionMeta, userText } from "@prismshadow/penguin-core";
|
||||
import type { SessionMetaPayload } from "@prismshadow/penguin-core";
|
||||
import { createCoreSessionLoader } from "../src/runtime/session-manager.js";
|
||||
import { SessionSources } from "../src/runtime/session-sources.js";
|
||||
import type { SessionRow } from "../src/db/repos/sessions.js";
|
||||
import { HttpError } from "../src/http/errors.js";
|
||||
import { makeTempRoot, writeTraceFile } from "./helpers.js";
|
||||
@@ -99,4 +100,38 @@ describe("session-loader", () => {
|
||||
expect((err as HttpError).status).toBe(409);
|
||||
expect((err as HttpError).code).toBe("workspace_missing");
|
||||
});
|
||||
|
||||
it("self-heal rebuild re-records a registry-known origin in the fresh session_meta; unknown stays absent", async () => {
|
||||
// The anthropic pair constructs without a credential (the same pair session-index
|
||||
// creates over HTTP); custom/m1 would demand a key at client construction.
|
||||
await saveProjectConfig(root, PROJECT, {
|
||||
default_model: { provider: "anthropic", model_id: "claude-sonnet-4-6" },
|
||||
models: [{ provider: "anthropic", model_id: "claude-sonnet-4-6", context_window: 1000 }],
|
||||
});
|
||||
const ws = path.join(root, "ws-heal");
|
||||
await fs.mkdir(ws, { recursive: true });
|
||||
const healRow: SessionRow = {
|
||||
...row(ws),
|
||||
modelId: "claude-sonnet-4-6",
|
||||
provider: "anthropic",
|
||||
};
|
||||
|
||||
// Origin known to this process (e.g. a schedule-created Session rebuilt after adoption):
|
||||
// the fresh session_meta must re-record it — meta is the single source of truth, and the
|
||||
// rebuilt Session's Trace is the only place the origin can survive.
|
||||
const sources = new SessionSources();
|
||||
sources.set(SID, "schedule");
|
||||
const known = await createCoreSessionLoader(root, sources).load(healRow);
|
||||
const knownMeta = (known as unknown as { metaMessage: { payload: { source?: string } } })
|
||||
.metaMessage;
|
||||
expect(knownMeta.payload.source).toBe("schedule");
|
||||
(known as unknown as { dispose(): void }).dispose();
|
||||
|
||||
// No registry entry (e.g. the process restarted and no Trace was ever written): the
|
||||
// rebuilt Session is unsourced — no source key is invented.
|
||||
const unknown = await createCoreSessionLoader(root, new SessionSources()).load(healRow);
|
||||
const unknownMeta = (unknown as unknown as { metaMessage: { payload: object } }).metaMessage;
|
||||
expect("source" in unknownMeta.payload).toBe(false);
|
||||
(unknown as unknown as { dispose(): void }).dispose();
|
||||
});
|
||||
});
|
||||
@@ -33,6 +33,7 @@ import type { ChannelEvent } from "../src/runtime/channel.js";
|
||||
import type { ErrorRecordArgs, ErrorSink } from "../src/runtime/error-recorder.js";
|
||||
import { SessionManager } from "../src/runtime/session-manager.js";
|
||||
import type { RuntimeSession, SessionLoader } from "../src/runtime/session-manager.js";
|
||||
import { SessionSources } from "../src/runtime/session-sources.js";
|
||||
import type { TitleRequest } from "../src/runtime/title-generator.js";
|
||||
import type { UsageContext } from "../src/runtime/usage-recorder.js";
|
||||
import { waitFor } from "./helpers.js";
|
||||
@@ -78,6 +79,7 @@ describe("session-manager", () => {
|
||||
let db: DatabaseSync;
|
||||
let sessions: SessionsRepo;
|
||||
let channels: ChannelHub;
|
||||
let sources: SessionSources;
|
||||
let recorded: OmniMessage[];
|
||||
let recordedCtx: UsageContext[];
|
||||
|
||||
@@ -85,6 +87,7 @@ describe("session-manager", () => {
|
||||
new SessionManager({
|
||||
sessions,
|
||||
channels,
|
||||
sources,
|
||||
loader,
|
||||
recorder: {
|
||||
record: async (ctx, msg) => {
|
||||
@@ -114,6 +117,7 @@ describe("session-manager", () => {
|
||||
sessions = new SessionsRepo(db);
|
||||
sessions.insert(ROW);
|
||||
channels = new ChannelHub();
|
||||
sources = new SessionSources();
|
||||
recorded = [];
|
||||
recordedCtx = [];
|
||||
});
|
||||
@@ -287,6 +291,7 @@ describe("session-manager", () => {
|
||||
const manager = new SessionManager({
|
||||
sessions,
|
||||
channels,
|
||||
sources,
|
||||
loader: loaderOf(fake),
|
||||
recorder: { record: async () => {} },
|
||||
titles: {
|
||||
@@ -301,6 +306,11 @@ describe("session-manager", () => {
|
||||
expect(child?.agentId).toBe("child_agent");
|
||||
expect(child?.modelId).toBe("m-child");
|
||||
expect(child?.workspace).toBe("/tmp/w-child");
|
||||
// The origin lands in the in-process registry (session_meta is the single source of
|
||||
// truth; the row stores no source column) — "subagent" even when the forwarded meta
|
||||
// predates the source field (this fake omits it): the registration path is the fallback.
|
||||
expect(sources.get("child-1")).toBe("subagent");
|
||||
expect(child && "source" in child).toBe(false);
|
||||
// Title left blank: produced by the title generator from the sub-session's own conversation (falls back to the prompt's first line on failure).
|
||||
expect(child?.title).toBeNull();
|
||||
|
||||
@@ -577,6 +587,7 @@ describe("session-manager", () => {
|
||||
const manager = new SessionManager({
|
||||
sessions,
|
||||
channels,
|
||||
sources,
|
||||
loader: loaderOf(plainSession),
|
||||
recorder: { record: async () => {} },
|
||||
titles: {
|
||||
@@ -629,6 +640,7 @@ describe("session-manager", () => {
|
||||
const manager = new SessionManager({
|
||||
sessions,
|
||||
channels,
|
||||
sources,
|
||||
loader: loaderOf(longSession),
|
||||
recorder: { record: async () => {} },
|
||||
titles: {
|
||||
@@ -695,6 +707,7 @@ describe("session-manager", () => {
|
||||
const manager = new SessionManager({
|
||||
sessions,
|
||||
channels,
|
||||
sources,
|
||||
loader: loaderOf(delegating),
|
||||
recorder: {
|
||||
record: async (_ctx, msg) => {
|
||||
|
||||
Reference in new issue
Block a user