diff --git a/docs/configuration.md b/docs/configuration.md index 408018c7f..d07e3ef6b 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -43,6 +43,10 @@ Run `devspace init` to create both files. `devspace config set publicBaseUrl "ui": { "enabled": true, }, + "activity": { + "enabled": true, + "maxTurns": 200, + }, "artifacts": { "enabled": false, "maxFileBytes": 104857600, @@ -107,6 +111,18 @@ This avoids rendering an iframe for every read, edit, search, or command call. Setting `ui.enabled` to `false` removes the metadata but does not remove the `show_changes` tool. +## Workspace activity + +DevSpace records each workspace tool call, with its arguments and result, in the +local state database (`storage.stateDir`, readable only by your user). Calls are +grouped into turns that close at each review checkpoint. Strings are truncated +and common secret patterns are redacted, but file contents and command output +are kept as the tools returned them. + +`activity.maxTurns` (default 200) caps the closed turns kept per workspace; +older turns and their tool calls are deleted. Set `activity.enabled` to `false` +to stop recording; turns already recorded stay until their workspace is removed. + ## Skills and subagents DevSpace discovers standard Agent Skills from `~/.agents/skills`, project diff --git a/schema/v1/devspace.schema.json b/schema/v1/devspace.schema.json index cf9ce8f84..0159a530f 100644 --- a/schema/v1/devspace.schema.json +++ b/schema/v1/devspace.schema.json @@ -114,6 +114,25 @@ }, "additionalProperties": false }, + "activity": { + "default": {}, + "type": "object", + "properties": { + "enabled": { + "default": true, + "description": "Record workspace tool calls, with their arguments and results, in the local state database.", + "type": "boolean" + }, + "maxTurns": { + "default": 200, + "description": "Closed turns kept per workspace. Older turns and their tool calls are deleted.", + "type": "integer", + "exclusiveMinimum": 0, + "maximum": 9007199254740991 + } + }, + "additionalProperties": false + }, "artifacts": { "default": {}, "type": "object", diff --git a/src/config-schema.ts b/src/config-schema.ts index 95141fbd9..ed74b155b 100644 --- a/src/config-schema.ts +++ b/src/config-schema.ts @@ -35,6 +35,13 @@ const artifactsConfigSchema = z.object({ maxFileBytes: z.number().int().positive().default(100 * 1024 * 1024), }).strict().prefault({}); +const activityConfigSchema = z.object({ + enabled: z.boolean().default(true) + .describe("Record workspace tool calls, with their arguments and results, in the local state database."), + maxTurns: z.number().int().positive().default(200) + .describe("Closed turns kept per workspace. Older turns and their tool calls are deleted."), +}).strict().prefault({}); + const skillsConfigSchema = z.object({ enabled: z.boolean().default(true), paths: z.array(z.string().trim().min(1)).default([]), @@ -76,6 +83,7 @@ export const devspaceConfigSchema = z.object({ storage: storageConfigSchema, tools: toolsConfigSchema, ui: uiConfigSchema, + activity: activityConfigSchema, artifacts: artifactsConfigSchema, skills: skillsConfigSchema, subagents: subagentsConfigSchema.default({ diff --git a/src/config.test.ts b/src/config.test.ts index ddb2fc1c3..0690fb6d0 100644 --- a/src/config.test.ts +++ b/src/config.test.ts @@ -20,6 +20,7 @@ try { assert.deepEqual(defaults.allowedHosts, ["localhost", "127.0.0.1", "::1"]); assert.equal(defaults.toolMode, "codex"); assert.equal(defaults.uiEnabled, true); + assert.deepEqual(defaults.activity, { enabled: true, maxTurns: 200 }); assert.equal(defaults.skillsEnabled, true); assert.equal(defaults.artifactsEnabled, false); assert.deepEqual(defaults.subagents, { diff --git a/src/config.ts b/src/config.ts index 34fcdfc25..bf1c878ef 100644 --- a/src/config.ts +++ b/src/config.ts @@ -18,6 +18,7 @@ export interface ServerConfig { publicBaseUrl: string; toolMode: ToolMode; uiEnabled: boolean; + activity: { enabled: boolean; maxTurns: number }; stateDir: string; worktreeRoot: string; artifactsEnabled: boolean; @@ -67,6 +68,7 @@ export function loadConfig(env: NodeJS.ProcessEnv = process.env): ServerConfig { publicBaseUrl, toolMode: stored.tools.mode, uiEnabled: stored.ui.enabled, + activity: stored.activity, stateDir: normalizePath(stored.storage.stateDir), worktreeRoot: normalizePath(stored.workspaces.worktreeRoot), artifactsEnabled: stored.artifacts.enabled, diff --git a/src/db/migrations.ts b/src/db/migrations.ts index 8bb1f7e4f..9b2fa4dda 100644 --- a/src/db/migrations.ts +++ b/src/db/migrations.ts @@ -47,6 +47,11 @@ const migrations: Migration[] = [ name: "local-agent-turns", up: migrateLocalAgentTurns, }, + { + version: 9, + name: "workspace-activity", + up: migrateWorkspaceActivity, + }, ]; export function migrateDatabase(sqlite: Database.Database): void { @@ -289,6 +294,52 @@ function migrateLocalAgentTurns(sqlite: Database.Database): void { `); } +function migrateWorkspaceActivity(sqlite: Database.Database): void { + sqlite.exec(` + create table if not exists workspace_turns ( + id integer primary key autoincrement, + workspace_session_id text not null, + sequence integer not null, + status text not null default 'open', + review_ref text, + started_at text not null, + closed_at text, + foreign key (workspace_session_id) + references workspace_sessions(id) + on delete cascade + ); + + create unique index if not exists workspace_turns_sequence_idx + on workspace_turns(workspace_session_id, sequence); + + create unique index if not exists workspace_turns_open_idx + on workspace_turns(workspace_session_id) + where status = 'open'; + + create table if not exists workspace_turn_tool_calls ( + id integer primary key autoincrement, + turn_id integer not null, + origin text not null default 'model', + conversation_scope_id text, + request_id text, + tool_name text not null, + arguments_json text not null, + result_json text, + error_json text, + is_error integer not null default 0, + started_at text not null, + completed_at text not null, + duration_ms integer not null, + foreign key (turn_id) + references workspace_turns(id) + on delete cascade + ); + + create index if not exists workspace_turn_tool_calls_turn_idx + on workspace_turn_tool_calls(turn_id, started_at); + `); +} + function addColumnIfMissing( sqlite: Database.Database, table: "workspace_sessions" | "local_agent_sessions", diff --git a/src/db/schema.ts b/src/db/schema.ts index 77b234368..abe0064df 100644 --- a/src/db/schema.ts +++ b/src/db/schema.ts @@ -1,4 +1,4 @@ -import { index, integer, primaryKey, sqliteTable, text } from "drizzle-orm/sqlite-core"; +import { index, integer, primaryKey, sqliteTable, text, uniqueIndex } from "drizzle-orm/sqlite-core"; export const workspaceSessions = sqliteTable( "workspace_sessions", @@ -56,6 +56,48 @@ export const workspaceConversationBindings = sqliteTable( ], ); +export const workspaceTurns = sqliteTable( + "workspace_turns", + { + id: integer("id").primaryKey({ autoIncrement: true }), + workspaceSessionId: text("workspace_session_id") + .notNull() + .references(() => workspaceSessions.id, { onDelete: "cascade" }), + sequence: integer("sequence").notNull(), + status: text("status").notNull().default("open"), + reviewRef: text("review_ref"), + startedAt: text("started_at").notNull(), + closedAt: text("closed_at"), + }, + (table) => [ + uniqueIndex("workspace_turns_sequence_idx").on(table.workspaceSessionId, table.sequence), + ], +); + +export const workspaceTurnToolCalls = sqliteTable( + "workspace_turn_tool_calls", + { + id: integer("id").primaryKey({ autoIncrement: true }), + turnId: integer("turn_id") + .notNull() + .references(() => workspaceTurns.id, { onDelete: "cascade" }), + origin: text("origin").notNull().default("model"), + conversationScopeId: text("conversation_scope_id"), + requestId: text("request_id"), + toolName: text("tool_name").notNull(), + argumentsJson: text("arguments_json").notNull(), + resultJson: text("result_json"), + errorJson: text("error_json"), + isError: integer("is_error").notNull().default(0), + startedAt: text("started_at").notNull(), + completedAt: text("completed_at").notNull(), + durationMs: integer("duration_ms").notNull(), + }, + (table) => [ + index("workspace_turn_tool_calls_turn_idx").on(table.turnId, table.startedAt), + ], +); + export const oauthClients = sqliteTable( "oauth_clients", { @@ -123,5 +165,7 @@ export type LoadedAgentFileRow = typeof loadedAgentFiles.$inferSelect; export type NewLoadedAgentFileRow = typeof loadedAgentFiles.$inferInsert; export type WorkspaceConversationBindingRow = typeof workspaceConversationBindings.$inferSelect; export type NewWorkspaceConversationBindingRow = typeof workspaceConversationBindings.$inferInsert; +export type WorkspaceTurnRow = typeof workspaceTurns.$inferSelect; +export type WorkspaceTurnToolCallRow = typeof workspaceTurnToolCalls.$inferSelect; export type LocalAgentSessionRow = typeof localAgentSessions.$inferSelect; export type NewLocalAgentSessionRow = typeof localAgentSessions.$inferInsert; diff --git a/src/oauth-store.test.ts b/src/oauth-store.test.ts index 90a8ade25..565d6786e 100644 --- a/src/oauth-store.test.ts +++ b/src/oauth-store.test.ts @@ -52,6 +52,7 @@ async function testDatabaseConfiguration(stateDir: string): Promise { { version: 6, name: "local-agent-effort-rename" }, { version: 7, name: "workspace-recovery-state" }, { version: 8, name: "local-agent-turns" }, + { version: 9, name: "workspace-activity" }, ]); } finally { database.close(); diff --git a/src/review-checkpoints.test.ts b/src/review-checkpoints.test.ts index 0a6dfce81..b58d494c8 100644 --- a/src/review-checkpoints.test.ts +++ b/src/review-checkpoints.test.ts @@ -101,6 +101,26 @@ test("historical review refs survive later reviews and manager restarts", async assert.doesNotMatch(restoredFirst.patch, /\+second/); }); +test("pinned checkpoints stay reviewable after the baseline history is rewritten", async (t) => { + const root = await committedRepository(t); + const manager = createReviewCheckpointManager(); + await manager.initializeWorkspace({ workspaceId: "ws_pinned", root }); + + await writeFile(join(root, "README.md"), "hello\nfirst\n"); + const first = await manager.reviewChanges({ workspaceId: "ws_pinned", root }); + + // Losing the baseline makes the next review restart from workspace open, so + // the first checkpoint is no longer an ancestor of the baseline. + await deleteReviewRef(root, "ws_pinned", "baseline"); + const restarted = createReviewCheckpointManager(); + await writeFile(join(root, "README.md"), "hello\nfirst\nsecond\n"); + await restarted.reviewChanges({ workspaceId: "ws_pinned", root }); + + const restored = await restarted.reviewByRef({ workspaceId: "ws_pinned", root, reviewRef: first.reviewRef }); + assert.equal(restored.patch, first.patch); + assert.equal((await readReviewRef(root, first.reviewRef)).patch, first.patch); +}); + test("review refs are scoped to the workspace review history", async (t) => { const root = await committedRepository(t); const manager = createReviewCheckpointManager(); diff --git a/src/review-checkpoints.ts b/src/review-checkpoints.ts index 1c43778ca..3fe500fee 100644 --- a/src/review-checkpoints.ts +++ b/src/review-checkpoints.ts @@ -36,6 +36,7 @@ interface WorkspaceReviewState { gitRoot?: string; openRef: string; baselineRef: string; + checkpointRefPrefix: string; openRefAvailable: boolean; baselineRefAvailable: boolean; diagnostic?: string; @@ -123,6 +124,10 @@ export function createReviewCheckpointManager(): ReviewCheckpointManager { const review = await readReviewBetween(state.gitRoot, baseline, current); if (markReviewed) { + // The baseline only reaches the latest checkpoint, and a workspace-open + // fallback rewrites its history. Pin each checkpoint so past turns stay + // reviewable instead of becoming garbage-collectable. + await git(state.gitRoot, ["update-ref", `${state.checkpointRefPrefix}/${current}`, current]); await git(state.gitRoot, ["update-ref", state.baselineRef, current]); state.baselineRefAvailable = true; } @@ -159,6 +164,12 @@ export function createReviewCheckpointManager(): ReviewCheckpointManager { commitForRef(state.gitRoot, state.baselineRef), resolveReviewCommitOrUndefined(state.gitRoot, reviewRef), ]); + if ( + reviewCommit + && await commitForRef(state.gitRoot, `${state.checkpointRefPrefix}/${reviewCommit}`) === reviewCommit + ) { + return readReviewCommit(state.gitRoot, reviewCommit); + } if ( !openCommit || !baselineCommit @@ -274,11 +285,12 @@ async function commitForRef(gitRoot: string, ref: string): Promise { +): Pick { const segment = safeWorkspaceRefSegment(workspaceId); return { openRef: `${REVIEW_REF_PREFIX}/${segment}/open`, baselineRef: `${REVIEW_REF_PREFIX}/${segment}/baseline`, + checkpointRefPrefix: `${REVIEW_REF_PREFIX}/${segment}/checkpoints`, }; } @@ -368,6 +380,7 @@ async function isKnownReviewCommit(gitRoot: string, reviewCommit: string): Promi for (const line of refs.split("\n")) { const [ref, commit] = line.split("\t"); if (!ref || !commit) continue; + if (commit === reviewCommit && ref.endsWith(`/checkpoints/${reviewCommit}`)) return true; const match = ref.match(/^refs\/devspace\/review\/(.+)\/(open|baseline)$/); if (!match) continue; diff --git a/src/server.test.ts b/src/server.test.ts index 59bcc5d37..0a21004c3 100644 --- a/src/server.test.ts +++ b/src/server.test.ts @@ -18,6 +18,14 @@ import { createMcpServer, createServer } from "./server.js"; import { SqliteWorkspaceStore } from "./workspace-store.js"; import { WorkspaceRegistry } from "./workspaces.js"; import { writeTestDevspaceConfig } from "./test-support/config.test.js"; +import { + listActivityWorkspaces, + listTurnToolCalls, + listWorkspaceTurns, + readTurnDiff, +} from "./workspace-activity.js"; +import { WorkspaceActivityJournal } from "./workspace-activity-journal.js"; +import { WorkspaceActivityStore } from "./workspace-activity-store.js"; const execFileAsync = promisify(execFile); @@ -312,6 +320,75 @@ test("show_changes can reopen a historical review without advancing the checkpoi ); }); +test("workspace activity records turns between review checkpoints", async (t) => { + const context = await fixture(t, { git: true, captureActivity: true }); + const activity = context.activity!; + const workspaceId = structuredContent( + await callOpen(context.client, context.project, "activity-turns"), + ).workspace_id as string; + const patch = (from: string, to: string) => context.client.callTool({ + name: "apply_patch", + arguments: { + workspace_id: workspaceId, + patch: `*** Begin Patch\n*** Update File: README.md\n@@\n-${from}\n+${to}\n*** End Patch`, + }, + }); + const showChanges = async () => structuredContent(await context.client.callTool({ + name: "show_changes", + arguments: { workspace_id: workspaceId }, + })).review_ref as string; + + await patch("hello", "hello turn one"); + const firstRef = await showChanges(); + await patch("hello turn one", "hello turn two"); + await context.client.callTool({ + name: "exec_command", + arguments: { workspace_id: workspaceId, cmd: "printf activity" }, + }); + const secondRef = await showChanges(); + await context.client.callTool({ + name: "show_changes", + arguments: { workspace_id: workspaceId }, + _meta: { "devspace/reviewRef": firstRef }, + } as Parameters[0]); + + const turns = listWorkspaceTurns(activity, workspaceId); + assert.deepEqual( + turns.map(({ sequence, status, reviewRef }) => ({ sequence, status, reviewRef })), + [ + { sequence: 3, status: "open", reviewRef: undefined }, + { sequence: 2, status: "closed", reviewRef: secondRef }, + { sequence: 1, status: "closed", reviewRef: firstRef }, + ], + ); + assert.deepEqual( + listTurnToolCalls(activity, workspaceId, 1)?.map((call) => call.toolName), + ["open_workspace", "apply_patch", "show_changes"], + ); + assert.deepEqual( + listTurnToolCalls(activity, workspaceId, 2)?.map((call) => call.toolName), + ["apply_patch", "exec_command", "show_changes"], + ); + assert.equal(listTurnToolCalls(activity, workspaceId, 3)?.[0]?.origin, "card"); + + const secondDiff = await readTurnDiff(activity, workspaceId, 2); + assert.equal(secondDiff?.status, "available"); + if (secondDiff?.status === "available") { + assert.match(secondDiff.review.patch, /-hello turn one\n\+hello turn two/); + assert.doesNotMatch(secondDiff.review.patch, /^-hello$/m); + } + assert.deepEqual(await readTurnDiff(activity, workspaceId, 3), { status: "open" }); + + const pinned = await execFileAsync("git", ["for-each-ref", "--format=%(refname)", "refs/devspace/review"], { + cwd: context.project, + }); + for (const ref of [firstRef, secondRef]) assert.match(pinned.stdout, new RegExp(`/checkpoints/${ref}$`, "m")); + + const [listed] = await listActivityWorkspaces(activity); + assert.equal(listed?.workspaceId, workspaceId); + assert.equal(listed?.turns, 3); +}); + test("open_workspace keeps lifecycle flags out of model output and preserves complete card metadata", async (t) => { const providerNote = "available"; const context = await fixture(t, { @@ -640,6 +717,7 @@ test("server shutdown waits for an active MCP tool call", async (t) => { interface ServerFixture { client: Client; project: string; + activity?: WorkspaceActivityStore; } function schemaPropertyPaths( @@ -722,6 +800,7 @@ async function fixture( subagents?: SubagentsConfig; toolMode?: ToolMode; uiEnabled?: boolean; + captureActivity?: boolean; } = {}, ): Promise { const root = await mkdtemp(join(tmpdir(), "devspace-server-test-")); @@ -791,6 +870,10 @@ async function fixture( resolveProviderAvailability(), ); const store = new SqliteWorkspaceStore(stateDir); + const activity = options.captureActivity + ? new WorkspaceActivityStore(stateDir) + : undefined; + const activityJournal = activity ? new WorkspaceActivityJournal(activity) : undefined; const workspaces = new WorkspaceRegistry(config, store); const server = createMcpServer( config, @@ -799,6 +882,8 @@ async function fixture( new ProcessSessionManager(), resolveLocalAgentProviders, [], + undefined, + activityJournal, ); const [clientTransport, serverTransport] = InMemoryTransport.createLinkedPair(); const client = new Client({ name: "devspace-test-client", version: "1.0.0" }); @@ -813,6 +898,7 @@ async function fixture( closed = true; await client.close(); await server.close(); + activity?.close(); store.close(); }; @@ -821,7 +907,7 @@ async function fixture( await rm(root, { recursive: true, force: true }); }); - return { client, project }; + return { client, project, ...(activity ? { activity } : {}) }; } async function git(cwd: string, args: string[]): Promise { diff --git a/src/server.ts b/src/server.ts index 0a2792f3f..a8771addb 100644 --- a/src/server.ts +++ b/src/server.ts @@ -46,6 +46,8 @@ import { shutdownHttpServer } from "./server-shutdown.js"; import { formatPathForPrompt } from "./skills.js"; import { DEVSPACE_VERSION } from "./version.js"; import { createWorkspaceStore } from "./workspace-store.js"; +import { WorkspaceActivityJournal } from "./workspace-activity-journal.js"; +import { WorkspaceActivityStore } from "./workspace-activity-store.js"; import { formatAgentsPath, WorkspaceRegistry } from "./workspaces.js"; import { getLocalAgentProviderAvailabilitySnapshot, @@ -313,6 +315,7 @@ export function createMcpServer( resolveLocalAgentProviders: () => LocalAgentProviderStatus[], incomingArtifactAdapters: readonly IncomingArtifactAdapter[], trackToolActivity?: TrackToolActivity, + workspaceActivityJournal?: WorkspaceActivityJournal, ): McpServer { const toolSurface = getToolSurface(config.toolMode); const server = new McpServer( @@ -331,6 +334,7 @@ export function createMcpServer( resolveLocalAgentProviders, incomingArtifactAdapters, trackToolActivity, + workspaceActivityJournal, ); return server; } @@ -344,9 +348,10 @@ function registerMcpSurface( resolveLocalAgentProviders: () => LocalAgentProviderStatus[], incomingArtifactAdapters: readonly IncomingArtifactAdapter[], trackToolActivity?: TrackToolActivity, + workspaceActivityJournal?: WorkspaceActivityJournal, ): void { - const registrationTarget = trackToolActivity - ? withTrackedToolHandlers(server, trackToolActivity) + const registrationTarget = trackToolActivity || workspaceActivityJournal + ? withObservedToolHandlers(server, { trackToolActivity, workspaceActivityJournal }) : server; const toolSurface = getToolSurface(config.toolMode); @@ -771,18 +776,33 @@ function registerMcpSurface( } } -function withTrackedToolHandlers( +function withObservedToolHandlers( server: McpRegistrationTarget, - trackToolActivity: TrackToolActivity, + options: { + trackToolActivity?: TrackToolActivity; + workspaceActivityJournal?: WorkspaceActivityJournal; + }, ): McpRegistrationTarget { return { registerTool: ((...args: unknown[]) => { + const toolName = args[0] as string; const handler = args.at(-1) as (...handlerArgs: unknown[]) => unknown; return (server.registerTool as (...callArgs: unknown[]) => unknown)( ...args.slice(0, -1), - (...handlerArgs: unknown[]) => trackToolActivity( - () => Promise.resolve(handler(...handlerArgs)), - ), + (...handlerArgs: unknown[]) => { + const operation = () => Promise.resolve(handler(...handlerArgs)); + const observedOperation = options.workspaceActivityJournal + ? () => options.workspaceActivityJournal!.capture({ + toolName, + arguments: handlerArgs[0], + extra: (handlerArgs[1] ?? {}) as Record, + operation, + }) + : operation; + return options.trackToolActivity + ? options.trackToolActivity(observedOperation) + : observedOperation(); + }, ); }) as McpRegistrationTarget["registerTool"], registerResource: server.registerResource.bind(server), @@ -819,6 +839,14 @@ export function createServer( const reviewCheckpoints = createReviewCheckpointManager(); const processSessions = new ProcessSessionManager(); const toolActivities = new ToolActivityTracker(); + const workspaceActivity = new WorkspaceActivityStore(config.stateDir, config.activity.maxTurns); + const workspaceActivityJournal = config.activity.enabled + ? new WorkspaceActivityJournal(workspaceActivity, (error) => { + logEvent(config.logging, "warn", "workspace_activity_journal_error", { + error: error instanceof Error ? error.message : String(error), + }); + }) + : undefined; const localAgentProviders = buildLocalAgentProviderStatuses( config.subagents, getLocalAgentProviderAvailabilitySnapshot(process.env, config.subagents), @@ -838,6 +866,7 @@ export function createServer( resolveLocalAgentProviders, incomingArtifactAdapters, toolActivities.track, + workspaceActivityJournal, ); }); const logMcpHandlerError = (error: Error) => logEvent( @@ -974,6 +1003,7 @@ export function createServer( }); } await toolActivities.waitForIdle(); + workspaceActivity.close(); processSessions.shutdown(); oauthProvider.close(); workspaceStore.close?.(); diff --git a/src/workspace-activity-journal.test.ts b/src/workspace-activity-journal.test.ts new file mode 100644 index 000000000..21abc30d3 --- /dev/null +++ b/src/workspace-activity-journal.test.ts @@ -0,0 +1,212 @@ +import assert from "node:assert/strict"; +import { mkdtemp, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import test, { type TestContext } from "node:test"; +import { WorkspaceActivityJournal, journalJson } from "./workspace-activity-journal.js"; +import { WorkspaceActivityStore } from "./workspace-activity-store.js"; +import { SqliteWorkspaceStore } from "./workspace-store.js"; + +const REVIEW_REF = "a".repeat(40); + +test("tool calls belong to the open turn until a model show_changes closes it", async (t) => { + const { journal, activity } = await journalFixture(t, ["ws_turns"]); + + await call(journal, "read", { workspace_id: "ws_turns", path: "README.md" }); + await call(journal, "show_changes", { workspace_id: "ws_turns" }, { + result: { structuredContent: { workspace_id: "ws_turns", review_ref: REVIEW_REF } }, + }); + await call(journal, "exec_command", { workspace_id: "ws_turns", cmd: "ls" }); + + const [second, first] = activity.listTurns("ws_turns"); + assert.deepEqual( + { sequence: first?.sequence, status: first?.status, reviewRef: first?.reviewRef, calls: first?.toolCalls }, + { sequence: 1, status: "closed", reviewRef: REVIEW_REF, calls: 2 }, + ); + assert.deepEqual( + { sequence: second?.sequence, status: second?.status, calls: second?.toolCalls }, + { sequence: 2, status: "open", calls: 1 }, + ); + assert.deepEqual(activity.listToolCalls(first!.id).map((entry) => entry.toolName), ["read", "show_changes"]); +}); + +test("closing a turn deletes closed turns beyond maxTurns with their calls", async (t) => { + const { journal, activity } = await journalFixture(t, ["ws_cap"], undefined, 2); + + for (let index = 0; index < 3; index += 1) { + await call(journal, "read", { workspace_id: "ws_cap", path: "README.md" }); + await call(journal, "show_changes", { workspace_id: "ws_cap" }, { + result: { structuredContent: { workspace_id: "ws_cap", review_ref: REVIEW_REF } }, + }); + } + + assert.deepEqual(activity.listTurns("ws_cap").map((turn) => turn.sequence), [3, 2]); +}); + +test("card replays are recorded without closing the turn", async (t) => { + const { journal, activity } = await journalFixture(t, ["ws_card"]); + + await call(journal, "show_changes", { workspace_id: "ws_card" }, { + meta: { "devspace/reviewRef": REVIEW_REF }, + result: { structuredContent: { workspace_id: "ws_card", review_ref: REVIEW_REF } }, + }); + + const [turn] = activity.listTurns("ws_card"); + assert.equal(turn?.status, "open"); + assert.equal(activity.listToolCalls(turn!.id)[0]?.origin, "card"); +}); + +test("open_workspace is recorded once its workspace exists, and failed opens leave nothing", async (t) => { + const { journal, activity, workspaces } = await journalFixture(t, []); + + await assert.rejects(call(journal, "open_workspace", { path: "/outside" }, { + error: new Error("Path is outside allowed roots"), + })); + await call(journal, "open_workspace", { path: "/tmp/project" }, { + meta: { "openai/session": "conversation-1" }, + result: () => { + workspaces.createSession({ id: "ws_opened", root: "/tmp/project" }); + return { structuredContent: { workspace_id: "ws_opened" } }; + }, + }); + + assert.deepEqual(activity.listWorkspaces().map((workspace) => workspace.workspaceId), ["ws_opened"]); + const [turn] = activity.listTurns("ws_opened"); + const [opened] = activity.listToolCalls(turn!.id); + assert.equal(opened?.toolName, "open_workspace"); + assert.equal(opened?.conversationScopeId, "conversation-1"); + assert.equal(opened?.requestId, "7"); +}); + +test("journal redacts secrets, caps payloads, and keeps errors", async (t) => { + const { journal, activity } = await journalFixture(t, ["ws_redact"]); + + await call(journal, "exec_command", { + workspace_id: "ws_redact", + cmd: "curl -H 'Authorization: Bearer abcdefghijklmnopqrstuvwxyz' https://example.test", + env: { GITHUB_TOKEN: "ghp_shouldnotpersist", max_output_tokens: 100 }, + }, { + result: { content: [{ type: "text", text: "x".repeat(20_000) }], _meta: { card: { patch: "large" } } }, + }); + await assert.rejects(call(journal, "read", { workspace_id: "ws_redact", path: "missing" }, { + error: new Error("ENOENT"), + })); + + const [turn] = activity.listTurns("ws_redact"); + const [exec, failed] = activity.listToolCalls(turn!.id); + const serialized = JSON.stringify(exec); + assert.doesNotMatch(serialized, /abcdefghijklmnop|ghp_should/); + assert.match(serialized, /\[redacted\]/); + assert.deepEqual((exec?.arguments as { env: unknown }).env, { GITHUB_TOKEN: "[redacted]", max_output_tokens: 100 }); + assert.ok(serialized.length < 10_000); + assert.equal(serialized.includes("_meta"), false); + assert.deepEqual( + { isError: failed?.isError, error: failed?.error }, + { isError: true, error: { name: "Error", message: "ENOENT" } }, + ); + assert.equal(turn?.failedToolCalls, 1); +}); + +test("journal redacts credentials in commands and outputs", () => { + const redacted = (value: string): string => JSON.parse(journalJson(value)) as string; + const secrets = [ + "API_KEY=abc123def", + "export GITHUB_TOKEN=ghx_notavendorshape", + 'DB_PASSWORD="hunter two"', + "AWS_SECRET_ACCESS_KEY='abc/def+ghi'", + "curl --password hunter2 https://example.test", + "docker login --password=hunter2", + "tool --api-key abc123", + "curl -H 'Authorization: Bearer abc.def' https://example.test", + 'curl -H "authorization: basic dXNlcjpwYXNzd29yZA==" https://example.test', + "Authorization: Basic dXNlcjpwYXNzd29yZA==", + "token Bearer abcdefghijklmnopqrstuvwxyz0123", + "git clone https://user:pass@github.com/acme/repo.git", + "postgres://app:s3cret@db.internal:5432/app", + ]; + for (const secret of secrets) { + const output = redacted(`run: ${secret}`); + assert.match(output, /\[redacted\]/, secret); + assert.doesNotMatch(output, /abc123|hunter|ghx_|abc\/def|abc\.def|dXNlcj|s3cret|:pass@|abcdefghij/, secret); + } + + assert.equal(redacted("export API_KEY=abc123 && run"), "export API_KEY=[redacted] && run"); + assert.equal(redacted("curl https://u:p@host/x"), "curl https://[redacted]@host/x"); +}); + +test("journal leaves ordinary prose and code alone", () => { + const unchanged = [ + "Basic functionality and basic configuration of the internationalization layer", + "Authorization: required for write access", + "bearer authentication is supported", + "MAX_TOKENS=4096", + "TOKEN_LIMIT=100 npm test", + "export TOKEN=$(cat token.txt)", + "API_KEY=$API_KEY node server.js", + "docker login --password-stdin", + "const tokenCount = 5; // password: string", + "git clone https://github.com/acme/repo.git", + "git clone ssh://git@github.com/acme/repo.git", + "open http://localhost:3000/@user", + "curl https://example.com:8080/path", + ]; + for (const text of unchanged) { + assert.equal(JSON.parse(journalJson(text)), text); + } +}); + +test("journal failures never replace the tool result", async (t) => { + const { journal, activity } = await journalFixture(t, ["ws_closed"], () => { + throw new Error("reporter failed too"); + }); + + activity.close(); + const result = await journal.capture({ + toolName: "read", + arguments: { workspace_id: "ws_closed" }, + extra: {}, + operation: async () => "tool result", + }); + + assert.equal(result, "tool result"); +}); + +async function journalFixture( + t: TestContext, + workspaceIds: string[], + onError?: (error: unknown) => void, + maxTurns?: number, +) { + const stateDir = await mkdtemp(join(tmpdir(), "devspace-activity-journal-test-")); + const workspaces = new SqliteWorkspaceStore(stateDir); + for (const id of workspaceIds) workspaces.createSession({ id, root: "/tmp/project" }); + const activity = new WorkspaceActivityStore(stateDir, maxTurns); + const journal = new WorkspaceActivityJournal(activity, onError); + t.after(async () => { + try { + activity.close(); + } catch { + // Some tests close the store early. + } + workspaces.close(); + await rm(stateDir, { recursive: true, force: true }); + }); + return { journal, activity, workspaces }; +} + +function call( + journal: WorkspaceActivityJournal, + toolName: string, + args: Record, + options: { meta?: Record; result?: unknown; error?: Error } = {}, +): Promise { + return journal.capture({ + toolName, + arguments: args, + extra: { _meta: options.meta, requestId: 7 }, + operation: async () => { + if (options.error) throw options.error; + return typeof options.result === "function" ? (options.result as () => unknown)() : options.result ?? {}; + }, + }); +} diff --git a/src/workspace-activity-journal.ts b/src/workspace-activity-journal.ts new file mode 100644 index 000000000..2d7f5645a --- /dev/null +++ b/src/workspace-activity-journal.ts @@ -0,0 +1,254 @@ +import { conversationScopeIdFromRequestMeta } from "./request-meta.js"; +import type { + WorkspaceActivityStore, + WorkspaceToolCallOrigin, + WorkspaceTurn, +} from "./workspace-activity-store.js"; + +type ToolHandlerExtra = { + _meta?: Record; + requestId?: unknown; +}; + +const MAX_STRING_LENGTH = 8 * 1024; +const MAX_RECORD_LENGTH = 64 * 1024; +const REDACTED = "[redacted]"; +const SECRET_KEY_PATTERN = /(token|secret|password|passwd|authorization|cookie|credential|api[_-]?key|private[_-]?key)/i; +// Value patterns are deliberately shape- or context-bound (an Authorization +// header, an env-var name, a credential flag) so ordinary prose and source code +// such as "Basic functionality" or `MAX_TOKENS=4096` survive intact. A pattern's +// replacement may keep its leading context via $1. +const SECRET_VALUE_PATTERNS: Array<[RegExp, string]> = [ + [/\b(Authorization["']?\s*[:=]\s*["']?)(?:Bearer|Basic|Token|Digest)\s+[^\s"']+/gi, `$1${REDACTED}`], + [/\bBearer\s+[A-Za-z0-9._~+/=-]{20,}/g, `Bearer ${REDACTED}`], + // ALL_CAPS env assignment whose name ends in a secret word: API_KEY=…, export GITHUB_TOKEN="…". + // MAX_TOKENS=… and values that defer to another variable ($VAR, $(cmd)) are left alone. + [/\b([A-Z0-9_]*(?:TOKEN|SECRET|PASSWORD|PASSWD|API_?KEY|ACCESS_?KEY|SECRET_?KEY|PRIVATE_?KEY|CREDENTIALS?)=)(?!\$)(?:"[^"]*"|'[^']*'|[^\s"'`;&|)]+)/g, `$1${REDACTED}`], + // CLI credential flags: --password hunter2, --api-key=abc. --password-stdin takes no value. + [/(--(?:[a-z0-9]+-)*(?:password|passwd|token|secret|api-?key|access-?key|secret-?key|private-?key)(?:\s+|=))(?![-$])(?:"[^"]*"|'[^']*'|[^\s"'`;&|)]+)/gi, `$1${REDACTED}`], + // Credentials embedded in a URL: https://user:pass@host. + [/\b([a-z][a-z0-9+.-]*:\/\/)[^\s:@/?#]+:[^\s@/?#]+@/gi, `$1${REDACTED}@`], + [/\b(sk|pk|rk)-[A-Za-z0-9_-]{16,}/g, REDACTED], + [/\bgh[pousr]_[A-Za-z0-9]{20,}/g, REDACTED], + [/\bgithub_pat_[A-Za-z0-9_]{20,}/g, REDACTED], + [/\bxox[abprs]-[A-Za-z0-9-]{10,}/g, REDACTED], + [/\bAKIA[0-9A-Z]{16}\b/g, REDACTED], + [/-----BEGIN [A-Z ]*PRIVATE KEY-----[\s\S]*?-----END [A-Z ]*PRIVATE KEY-----/g, REDACTED], +]; + +/** + * Records every registered MCP tool call into the open turn of its workspace. + * + * A turn is the work between two review checkpoints: a model-initiated + * show_changes closes the open turn with its review_ref, and the next call + * starts a new one. Calls are persisted only once their workspace exists, so a + * failed open_workspace leaves nothing behind. Journal failures are reported to + * `onError` and never change the tool outcome. + */ +export class WorkspaceActivityJournal { + constructor( + private readonly store: WorkspaceActivityStore, + private readonly onError: (error: unknown) => void = () => {}, + ) {} + + async capture(input: { + toolName: string; + arguments: unknown; + extra: ToolHandlerExtra; + operation: () => Promise; + }): Promise { + const startedAt = new Date(); + const startedAtMs = performance.now(); + const argumentWorkspaceId = workspaceIdFromValue(input.arguments); + // Calls belong to the turn that was open when they started. + const turnId = argumentWorkspaceId + ? this.safely(() => this.openTurnId(argumentWorkspaceId, startedAt)) + : undefined; + + let result: T; + try { + result = await input.operation(); + } catch (error) { + this.safely(() => this.record(input, { + workspaceId: argumentWorkspaceId, + turnId, + startedAt, + startedAtMs, + error: serializeError(error), + })); + throw error; + } + + this.safely(() => this.record(input, { + workspaceId: workspaceIdFromValue(result) ?? argumentWorkspaceId, + turnId, + startedAt, + startedAtMs, + result, + })); + return result; + } + + /** + * Hook point for subagent activity: the turn that events for `workspaceId` + * should attach to right now. Subagent events (#379–384) are not journaled + * yet; their app-owned turnId is a subagent turn, not this workspace turn. + */ + currentTurn(workspaceId: string): WorkspaceTurn | undefined { + return this.safely(() => this.store.currentTurn(workspaceId)); + } + + private openTurnId(workspaceId: string, now: Date): number | undefined { + return this.store.hasWorkspace(workspaceId) + ? this.store.openTurn(workspaceId, now).id + : undefined; + } + + private record( + input: { toolName: string; arguments: unknown; extra: ToolHandlerExtra }, + outcome: { + workspaceId?: string; + turnId?: number; + startedAt: Date; + startedAtMs: number; + result?: unknown; + error?: { name: string; message: string }; + }, + ): void { + if (!outcome.workspaceId || !this.store.hasWorkspace(outcome.workspaceId)) return; + + const completedAt = new Date(); + const origin = toolCallOrigin(input.extra._meta); + const reviewRef = origin === "model" && input.toolName === "show_changes" && !outcome.error + ? reviewRefFromValue(outcome.result) + : undefined; + const turnId = reviewRef + ? this.store.closeTurn(outcome.workspaceId, reviewRef, completedAt).id + : outcome.turnId ?? this.store.openTurn(outcome.workspaceId, completedAt).id; + const resultSummary = outcome.result === undefined ? undefined : summarizeResult(outcome.result); + + this.store.recordToolCall({ + turnId, + origin, + conversationScopeId: conversationScopeIdFromRequestMeta(input.extra._meta), + requestId: requestIdFromExtra(input.extra.requestId), + toolName: input.toolName, + argumentsJson: journalJson(input.arguments ?? {}), + ...(resultSummary === undefined ? {} : { resultJson: journalJson(resultSummary) }), + ...(outcome.error ? { errorJson: journalJson(outcome.error) } : {}), + isError: outcome.error !== undefined || asRecord(outcome.result)?.isError === true, + startedAt: outcome.startedAt.toISOString(), + completedAt: completedAt.toISOString(), + durationMs: Math.round(performance.now() - outcome.startedAtMs), + }); + } + + private safely(operation: () => R): R | undefined { + try { + return operation(); + } catch (error) { + try { + this.onError(error); + } catch { + // Error reporting must not affect the tool call either. + } + return undefined; + } + } +} + +/** Serializes a value for the journal with secrets redacted and size capped. */ +export function journalJson(value: unknown): string { + const json = JSON.stringify(redact(value, 0)) ?? "null"; + if (json.length <= MAX_RECORD_LENGTH) return json; + return JSON.stringify({ + truncated: true, + length: json.length, + preview: json.slice(0, MAX_STRING_LENGTH), + }); +} + +/** + * Keeps what a reader needs to understand a tool result: its text, structured + * content, and error flag. Card metadata (`_meta`) is dropped because it + * duplicates large payloads such as full review patches. + */ +function summarizeResult(result: unknown): unknown { + const record = asRecord(result); + if (!record || !Array.isArray(record.content)) return result; + + const text = record.content + .map((block) => asRecord(block)) + .filter((block) => block?.type === "text" && typeof block.text === "string") + .map((block) => block!.text as string) + .join("\n"); + return { + ...(text ? { text } : {}), + ...(record.structuredContent !== undefined ? { structuredContent: record.structuredContent } : {}), + ...(record.isError === true ? { isError: true } : {}), + }; +} + +function redact(value: unknown, depth: number): unknown { + if (typeof value === "string") return redactString(value); + if (value === null || typeof value !== "object") return value; + if (depth > 8) return "[nested]"; + if (Array.isArray(value)) return value.map((item) => redact(item, depth + 1)); + + return Object.fromEntries( + Object.entries(value as Record).map(([key, item]) => [ + key, + SECRET_KEY_PATTERN.test(key) && typeof item === "string" && item !== "" + ? REDACTED + : redact(item, depth + 1), + ]), + ); +} + +function redactString(value: string): string { + let redacted = value; + for (const [pattern, replacement] of SECRET_VALUE_PATTERNS) { + redacted = redacted.replace(pattern, replacement); + } + return redacted.length > MAX_STRING_LENGTH + ? `${redacted.slice(0, MAX_STRING_LENGTH)}… [truncated ${redacted.length - MAX_STRING_LENGTH} chars]` + : redacted; +} + +/** The workspace card replays historical reviews through show_changes with this marker. */ +function toolCallOrigin(meta: Record | undefined): WorkspaceToolCallOrigin { + return typeof meta?.["devspace/reviewRef"] === "string" ? "card" : "model"; +} + +function workspaceIdFromValue(value: unknown): string | undefined { + const record = asRecord(value); + const direct = record?.workspace_id; + if (typeof direct === "string" && direct.length > 0) return direct; + + const structured = asRecord(record?.structuredContent); + const nested = structured?.workspace_id; + return typeof nested === "string" && nested.length > 0 ? nested : undefined; +} + +function reviewRefFromValue(value: unknown): string | undefined { + const structured = asRecord(asRecord(value)?.structuredContent); + const reviewRef = structured?.review_ref; + return typeof reviewRef === "string" && reviewRef.length > 0 ? reviewRef : undefined; +} + +function requestIdFromExtra(value: unknown): string | undefined { + return typeof value === "string" || typeof value === "number" ? String(value) : undefined; +} + +function serializeError(error: unknown): { name: string; message: string } { + if (error instanceof Error) { + return { name: error.name, message: error.message }; + } + return { name: typeof error, message: String(error) }; +} + +function asRecord(value: unknown): Record | undefined { + return value !== null && typeof value === "object" && !Array.isArray(value) + ? value as Record + : undefined; +} diff --git a/src/workspace-activity-store.ts b/src/workspace-activity-store.ts new file mode 100644 index 000000000..81dc0d982 --- /dev/null +++ b/src/workspace-activity-store.ts @@ -0,0 +1,320 @@ +import { and, asc, count, desc, eq, max, sql } from "drizzle-orm"; +import { openDatabase, type DatabaseHandle } from "./db/client.js"; +import { + workspaceSessions, + workspaceTurns, + workspaceTurnToolCalls, + type WorkspaceSessionRow, + type WorkspaceTurnRow, + type WorkspaceTurnToolCallRow, +} from "./db/schema.js"; +import type { WorkspaceMode, WorkspaceStatus } from "./workspace-store.js"; + +export type WorkspaceTurnStatus = "open" | "closed"; +export type WorkspaceToolCallOrigin = "model" | "card"; + +export interface ActivityWorkspaceRecord { + workspaceId: string; + root: string; + mode: WorkspaceMode; + status: WorkspaceStatus; + sourceRoot?: string; + createdAt: string; + lastUsedAt: string; + lastActivityAt?: string; + turns: number; +} + +export interface WorkspaceTurn { + id: number; + workspaceId: string; + /** 1-based position of the turn within its workspace. */ + sequence: number; + status: WorkspaceTurnStatus; + /** Review checkpoint that closed the turn; absent while the turn is open. */ + reviewRef?: string; + startedAt: string; + closedAt?: string; + toolCalls: number; + failedToolCalls: number; +} + +export interface WorkspaceToolCall { + id: number; + turnId: number; + origin: WorkspaceToolCallOrigin; + conversationScopeId?: string; + requestId?: string; + toolName: string; + arguments: unknown; + result?: unknown; + error?: unknown; + isError: boolean; + startedAt: string; + completedAt: string; + durationMs: number; +} + +export interface RecordWorkspaceToolCall { + turnId: number; + origin: WorkspaceToolCallOrigin; + conversationScopeId?: string; + requestId?: string; + toolName: string; + argumentsJson: string; + resultJson?: string; + errorJson?: string; + isError: boolean; + startedAt: string; + completedAt: string; + durationMs: number; +} + +/** + * SQLite persistence for workspace turns and the tool calls made in them. + * A workspace has at most one open turn; a review checkpoint closes it. + */ +export class WorkspaceActivityStore { + private readonly database: DatabaseHandle; + + constructor(stateDir: string, private readonly maxTurns = Number.POSITIVE_INFINITY) { + this.database = openDatabase(stateDir); + } + + hasWorkspace(workspaceId: string): boolean { + return this.database.db + .select({ id: workspaceSessions.id }) + .from(workspaceSessions) + .where(eq(workspaceSessions.id, workspaceId)) + .get() !== undefined; + } + + /** Returns the workspace's open turn, starting the next one when none is open. */ + openTurn(workspaceId: string, now = new Date()): WorkspaceTurn { + const open = this.database.sqlite.transaction(() => { + const existing = this.findOpenTurnRow(workspaceId); + if (existing) return existing; + + const previous = this.database.db + .select({ sequence: max(workspaceTurns.sequence) }) + .from(workspaceTurns) + .where(eq(workspaceTurns.workspaceSessionId, workspaceId)) + .get(); + return this.database.db + .insert(workspaceTurns) + .values({ + workspaceSessionId: workspaceId, + sequence: (previous?.sequence ?? 0) + 1, + status: "open", + startedAt: now.toISOString(), + }) + .returning() + .get(); + }); + return this.withCallCounts(open.immediate()); + } + + currentTurn(workspaceId: string): WorkspaceTurn | undefined { + const row = this.findOpenTurnRow(workspaceId); + return row ? this.withCallCounts(row) : undefined; + } + + /** Closes the open turn with its review checkpoint and returns it. */ + closeTurn(workspaceId: string, reviewRef: string, now = new Date()): WorkspaceTurn { + const turn = this.openTurn(workspaceId, now); + this.database.db + .update(workspaceTurns) + .set({ status: "closed", reviewRef, closedAt: now.toISOString() }) + .where(eq(workspaceTurns.id, turn.id)) + .run(); + this.pruneClosedTurns(workspaceId); + return { ...turn, status: "closed", reviewRef, closedAt: now.toISOString() }; + } + + recordToolCall(input: RecordWorkspaceToolCall): number { + const inserted = this.database.db + .insert(workspaceTurnToolCalls) + .values({ + turnId: input.turnId, + origin: input.origin, + conversationScopeId: input.conversationScopeId ?? null, + requestId: input.requestId ?? null, + toolName: input.toolName, + argumentsJson: input.argumentsJson, + resultJson: input.resultJson ?? null, + errorJson: input.errorJson ?? null, + isError: input.isError ? 1 : 0, + startedAt: input.startedAt, + completedAt: input.completedAt, + durationMs: input.durationMs, + }) + .run(); + return Number(inserted.lastInsertRowid); + } + + /** Lists workspaces, most recently active first. */ + listWorkspaces(): ActivityWorkspaceRecord[] { + return this.workspaceQuery().all().map(rowToWorkspace); + } + + getWorkspace(workspaceId: string): ActivityWorkspaceRecord | undefined { + const row = this.workspaceQuery().where(eq(workspaceSessions.id, workspaceId)).get(); + return row ? rowToWorkspace(row) : undefined; + } + + private workspaceQuery() { + const activity = this.database.db + .select({ + workspaceId: workspaceTurns.workspaceSessionId, + turns: sql`count(distinct ${workspaceTurns.id})`.as("turns"), + lastActivityAt: max(workspaceTurnToolCalls.completedAt).as("last_activity_at"), + }) + .from(workspaceTurns) + .leftJoin(workspaceTurnToolCalls, eq(workspaceTurnToolCalls.turnId, workspaceTurns.id)) + .groupBy(workspaceTurns.workspaceSessionId) + .as("activity"); + + return this.database.db + .select({ + session: workspaceSessions, + turns: activity.turns, + lastActivityAt: activity.lastActivityAt, + }) + .from(workspaceSessions) + .leftJoin(activity, eq(activity.workspaceId, workspaceSessions.id)) + .orderBy(desc(sql`coalesce(${activity.lastActivityAt}, ${workspaceSessions.lastUsedAt})`)) + .$dynamic(); + } + + /** Lists a workspace's turns, latest first. */ + listTurns(workspaceId: string): WorkspaceTurn[] { + return this.turnQuery() + .where(eq(workspaceTurns.workspaceSessionId, workspaceId)) + .orderBy(desc(workspaceTurns.sequence)) + .all() + .map(rowToTurn); + } + + getTurn(workspaceId: string, sequence: number): WorkspaceTurn | undefined { + const row = this.turnQuery() + .where(and( + eq(workspaceTurns.workspaceSessionId, workspaceId), + eq(workspaceTurns.sequence, sequence), + )) + .get(); + return row ? rowToTurn(row) : undefined; + } + + /** Lists the tool calls of one turn in the order they started. */ + listToolCalls(turnId: number): WorkspaceToolCall[] { + return this.database.db + .select() + .from(workspaceTurnToolCalls) + .where(eq(workspaceTurnToolCalls.turnId, turnId)) + .orderBy(asc(workspaceTurnToolCalls.startedAt), asc(workspaceTurnToolCalls.id)) + .all() + .map(rowToToolCall); + } + + close(): void { + this.database.close(); + } + + /** Keeps the newest `maxTurns` closed turns; their tool calls cascade. */ + private pruneClosedTurns(workspaceId: string): void { + if (!Number.isFinite(this.maxTurns)) return; + this.database.sqlite.prepare(` + delete from workspace_turns + where workspace_session_id = ? and status = 'closed' and id not in ( + select id from workspace_turns + where workspace_session_id = ? and status = 'closed' + order by sequence desc + limit ? + ) + `).run(workspaceId, workspaceId, this.maxTurns); + } + + private findOpenTurnRow(workspaceId: string): WorkspaceTurnRow | undefined { + return this.database.db + .select() + .from(workspaceTurns) + .where(and( + eq(workspaceTurns.workspaceSessionId, workspaceId), + eq(workspaceTurns.status, "open"), + )) + .get(); + } + + private withCallCounts(row: WorkspaceTurnRow): WorkspaceTurn { + const counts = this.turnQuery().where(eq(workspaceTurns.id, row.id)).get(); + return counts ? rowToTurn(counts) : rowToTurn({ turn: row, toolCalls: 0, failedToolCalls: 0 }); + } + + private turnQuery() { + return this.database.db + .select({ + turn: workspaceTurns, + toolCalls: count(workspaceTurnToolCalls.id), + failedToolCalls: sql`coalesce(sum(${workspaceTurnToolCalls.isError}), 0)`, + }) + .from(workspaceTurns) + .leftJoin(workspaceTurnToolCalls, eq(workspaceTurnToolCalls.turnId, workspaceTurns.id)) + .groupBy(workspaceTurns.id) + .$dynamic(); + } +} + +function rowToWorkspace({ session, turns, lastActivityAt }: { + session: WorkspaceSessionRow; + turns: number | null; + lastActivityAt: string | null; +}): ActivityWorkspaceRecord { + return { + workspaceId: session.id, + root: session.root, + mode: session.mode as WorkspaceMode, + status: session.status as WorkspaceStatus, + ...(session.sourceRoot ? { sourceRoot: session.sourceRoot } : {}), + createdAt: session.createdAt, + lastUsedAt: session.lastUsedAt, + ...(lastActivityAt ? { lastActivityAt } : {}), + turns: turns ?? 0, + }; +} + +function rowToTurn(row: { + turn: WorkspaceTurnRow; + toolCalls: number; + failedToolCalls: number; +}): WorkspaceTurn { + const { turn } = row; + return { + id: turn.id, + workspaceId: turn.workspaceSessionId, + sequence: turn.sequence, + status: turn.status as WorkspaceTurnStatus, + ...(turn.reviewRef ? { reviewRef: turn.reviewRef } : {}), + startedAt: turn.startedAt, + ...(turn.closedAt ? { closedAt: turn.closedAt } : {}), + toolCalls: row.toolCalls, + failedToolCalls: Number(row.failedToolCalls), + }; +} + +function rowToToolCall(row: WorkspaceTurnToolCallRow): WorkspaceToolCall { + return { + id: row.id, + turnId: row.turnId, + origin: row.origin as WorkspaceToolCallOrigin, + ...(row.conversationScopeId ? { conversationScopeId: row.conversationScopeId } : {}), + ...(row.requestId ? { requestId: row.requestId } : {}), + toolName: row.toolName, + arguments: JSON.parse(row.argumentsJson) as unknown, + ...(row.resultJson ? { result: JSON.parse(row.resultJson) as unknown } : {}), + ...(row.errorJson ? { error: JSON.parse(row.errorJson) as unknown } : {}), + isError: row.isError === 1, + startedAt: row.startedAt, + completedAt: row.completedAt, + durationMs: row.durationMs, + }; +} diff --git a/src/workspace-activity.ts b/src/workspace-activity.ts new file mode 100644 index 000000000..a62ca73a3 --- /dev/null +++ b/src/workspace-activity.ts @@ -0,0 +1,90 @@ +import { git } from "./git.js"; +import { readReviewRef, type ReviewChangesResult } from "./review-checkpoints.js"; +import type { + ActivityWorkspaceRecord, + WorkspaceActivityStore, + WorkspaceToolCall, + WorkspaceTurn, +} from "./workspace-activity-store.js"; + +// Read API for workspace observability. A turn is the work between two review +// checkpoints: turn N holds the tool calls made after checkpoint N-1 and the +// diff from checkpoint N-1 to checkpoint N. + +export type { WorkspaceToolCall, WorkspaceTurn } from "./workspace-activity-store.js"; + +export interface ActivityWorkspace extends ActivityWorkspaceRecord { + /** Current branch of the workspace root; absent when detached or unreadable. */ + branch?: string; +} + +export type TurnDiff = + | { status: "available"; review: ReviewChangesResult } + | { status: "open" } + | { status: "unavailable"; reason: string }; + +export async function listActivityWorkspaces( + store: WorkspaceActivityStore, +): Promise { + return Promise.all(store.listWorkspaces().map(withBranch)); +} + +export async function getActivityWorkspace( + store: WorkspaceActivityStore, + workspaceId: string, +): Promise { + const workspace = store.getWorkspace(workspaceId); + return workspace ? withBranch(workspace) : undefined; +} + +/** Lists a workspace's turns, latest first. */ +export function listWorkspaceTurns( + store: WorkspaceActivityStore, + workspaceId: string, +): WorkspaceTurn[] { + return store.listTurns(workspaceId); +} + +/** Lists the tool calls of turn `sequence`, or undefined when the turn does not exist. */ +export function listTurnToolCalls( + store: WorkspaceActivityStore, + workspaceId: string, + sequence: number, +): WorkspaceToolCall[] | undefined { + const turn = store.getTurn(workspaceId, sequence); + return turn ? store.listToolCalls(turn.id) : undefined; +} + +/** + * Reads the diff of turn `sequence`: checkpoint N-1 to checkpoint N. Each + * checkpoint snapshot is a commit whose parent is the previous checkpoint, so + * this is exactly the diff show_changes reported when it closed the turn. + */ +export async function readTurnDiff( + store: WorkspaceActivityStore, + workspaceId: string, + sequence: number, +): Promise { + const workspace = store.getWorkspace(workspaceId); + const turn = store.getTurn(workspaceId, sequence); + if (!workspace || !turn) return undefined; + if (!turn.reviewRef) return { status: "open" }; + + try { + return { status: "available", review: await readReviewRef(workspace.root, turn.reviewRef) }; + } catch (error) { + return { + status: "unavailable", + reason: error instanceof Error ? error.message : String(error), + }; + } +} + +async function withBranch(workspace: ActivityWorkspaceRecord): Promise { + try { + const branch = (await git(workspace.root, ["symbolic-ref", "--quiet", "--short", "HEAD"])).stdout.trim(); + return branch ? { ...workspace, branch } : workspace; + } catch { + return workspace; + } +}