From 23122ccbc04e8e84f5b5bd1e08ab2b83b48b831c Mon Sep 17 00:00:00 2001 From: weishu Date: Thu, 29 Oct 2026 08:28:24 +0800 Subject: [PATCH] feat(hub): persist agent turns and report them through hapi usage Whether a session's agent is thinking lived only in the session cache, so once a turn ended nothing could say how long the agent had worked, and a hub restart forgot even the turn in progress. Tools that track time (Weft, for one) had no way to see when agents on other machines were busy. Schema v28 adds session_turns: a row opens when a session starts thinking (a prompt queued through the hub, the agent reporting it, or a patch) and closes when it stops, the session ends, or its keepalives expire, at the last keepalive. Rows are only written on those transitions, never per keepalive. Turns move with a session when it is merged away. At startup the hub closes turns left open by the previous process at the session's newest message. If the agent carries on thinking within 10 minutes of the restart, the same turn continues, so the work between that message and the restart isn't lost. The deadline counts from the restart, and a keepalive or patch reporting idle, the session ending, or a new prompt ends the chance: those start a new turn. GET /api/activity returns, for the caller's namespace, each session's turns in a window (default 24 h, at most 31 days) with what identifies it (flavor, agent session id, machine, path, title) and the permission prompts and questions it waited on, taken from agent_state. `hapi usage [--since] [--until] [--json]` prints a summary or a hapi.activity.v1 report for other tools, with distinct exit codes for bad arguments (2), missing or refused credentials (3) and a hub too old to answer (4). The summary counts only the part of each turn inside the window and subtracts overlapping waits once. It reuses the peer commands' token exchange. Co-Authored-By: Claude Opus 5.5 --- cli/README.md | 1 + cli/src/commands/help.ts | 1 + cli/src/commands/registry.ts | 4 +- cli/src/commands/usage.test.ts | 18 ++ cli/src/commands/usage.ts | 103 +++++++++++ cli/src/modules/activity/activity.test.ts | 159 +++++++++++++++++ cli/src/modules/activity/activity.ts | 199 ++++++++++++++++++++++ cli/src/modules/pingPeer/pingPeer.ts | 8 +- docs/api/client-contract/rest.md | 8 + hub/README.md | 4 + hub/src/store/index.ts | 64 ++++++- hub/src/store/migration-v13.test.ts | 6 +- hub/src/store/migration-v15.test.ts | 6 +- hub/src/store/migration-v18.test.ts | 2 +- hub/src/store/migration-v19.test.ts | 2 +- hub/src/store/migration-v20.test.ts | 2 +- hub/src/store/migration-v23.test.ts | 2 +- hub/src/store/migration-v24.test.ts | 4 +- hub/src/store/migration-v25.test.ts | 2 +- hub/src/store/migration-v26.test.ts | 2 +- hub/src/store/migration-v28.test.ts | 42 +++++ hub/src/store/turnStore.ts | 47 +++++ hub/src/store/turns.test.ts | 130 ++++++++++++++ hub/src/store/turns.ts | 147 ++++++++++++++++ hub/src/sync/activityService.ts | 90 ++++++++++ hub/src/sync/sessionCache.ts | 82 ++++++++- hub/src/sync/sessionCache.turns.test.ts | 166 ++++++++++++++++++ hub/src/sync/syncEngine.ts | 1 + hub/src/web/routes/activity.test.ts | 102 +++++++++++ hub/src/web/routes/activity.ts | 38 +++++ hub/src/web/server.ts | 2 + shared/src/apiTypes.ts | 40 +++++ shared/src/sessionSummary.ts | 4 +- 33 files changed, 1465 insertions(+), 23 deletions(-) create mode 100644 cli/src/commands/usage.test.ts create mode 100644 cli/src/commands/usage.ts create mode 100644 cli/src/modules/activity/activity.test.ts create mode 100644 cli/src/modules/activity/activity.ts create mode 100644 hub/src/store/migration-v28.test.ts create mode 100644 hub/src/store/turnStore.ts create mode 100644 hub/src/store/turns.test.ts create mode 100644 hub/src/store/turns.ts create mode 100644 hub/src/sync/activityService.ts create mode 100644 hub/src/sync/sessionCache.turns.test.ts create mode 100644 hub/src/web/routes/activity.test.ts create mode 100644 hub/src/web/routes/activity.ts diff --git a/cli/README.md b/cli/README.md index 668c1644..d5fbb39b 100644 --- a/cli/README.md +++ b/cli/README.md @@ -43,6 +43,7 @@ Choose a supported coding agent from your terminal and control its sessions remo DSH is remote-only and its ACP server must be configured separately. - `hapi resume [sessionId]` - List resumable sessions for this machine or resume one locally. - `hapi ping-peer ` - Resume (if needed) and message another session. Prefer this or MCP `ping_peer` / `list_peers` over reinventing JWT+curl. Also `--message-file` / `--list`. +- `hapi usage [--since 24h|7d||] [--until …] [--json]` - When agents were working across every machine on the hub, from `GET /api/activity`. `--json` prints one `hapi.activity.v1` report for other tools (time trackers); errors go to stderr with exit codes 2 (arguments), 3 (no token or refused), 4 (hub too old), 1 (other). - `hapi inspect-peer ` - Read-only peer metadata + recent message text (no resume). Prefer this or MCP `inspect_peer` when a user cites `[title](/sessions/)` or Copy-reference `See session "…" (/sessions/) for context`. `/sessions/` is a hub path, not a local file. Optional `--limit`. - `hapi set-mode ` - Change the permission mode of an existing session, like the web permission picker. Active sessions switch immediately; inactive ones use it on next resume. An invalid mode prints the modes allowed for that agent. diff --git a/cli/src/commands/help.ts b/cli/src/commands/help.ts index 612df48c..83fe2b7c 100644 --- a/cli/src/commands/help.ts +++ b/cli/src/commands/help.ts @@ -25,6 +25,7 @@ Commands: hapi ping-peer List peers or message another session hapi inspect-peer Read another session's metadata and messages hapi set-mode Change a session's permission mode + hapi usage [--json] When agents were working, across machines hapi mcp Start the MCP stdio bridge Options: diff --git a/cli/src/commands/registry.ts b/cli/src/commands/registry.ts index ae439d27..fe49283f 100644 --- a/cli/src/commands/registry.ts +++ b/cli/src/commands/registry.ts @@ -21,6 +21,7 @@ import { notifyCommand } from './notify' import { hubCommand } from './hub' import { pingPeerCommand } from './pingPeer' import { inspectPeerCommand } from './inspectPeer' +import { usageCommand } from './usage' import { setModeCommand } from './setMode' import type { CommandContext, CommandDefinition } from './types' @@ -64,7 +65,8 @@ const COMMANDS: CommandDefinition[] = [ notifyCommand, pingPeerCommand, inspectPeerCommand, - setModeCommand + setModeCommand, + usageCommand ] const commandMap = new Map() diff --git a/cli/src/commands/usage.test.ts b/cli/src/commands/usage.test.ts new file mode 100644 index 00000000..2dc3b340 --- /dev/null +++ b/cli/src/commands/usage.test.ts @@ -0,0 +1,18 @@ +import { describe, expect, it } from 'vitest' +import { ActivityError } from '@/modules/activity/activity' +import { parseUsageArgs } from './usage' + +describe('parseUsageArgs', () => { + it('reads the window and output format', () => { + expect(parseUsageArgs([])).toEqual({ help: false, json: false }) + expect(parseUsageArgs(['--since', '7d', '--until=2026-10-08', '--json'])).toEqual({ + help: false, json: true, since: '7d', until: '2026-10-08' + }) + expect(parseUsageArgs(['-h']).help).toBe(true) + }) + + it('rejects anything it doesn\'t know', () => { + expect(() => parseUsageArgs(['--since'])).toThrow(ActivityError) + expect(() => parseUsageArgs(['today'])).toThrow(ActivityError) + }) +}) diff --git a/cli/src/commands/usage.ts b/cli/src/commands/usage.ts new file mode 100644 index 00000000..e893cf34 --- /dev/null +++ b/cli/src/commands/usage.ts @@ -0,0 +1,103 @@ +import chalk from 'chalk' +import { initializeToken } from '@/ui/tokenInit' +import { + ActivityError, + exitCodeForActivityError, + fetchActivity, + formatActivityReport, + parseTimeArg +} from '@/modules/activity/activity' +import type { CommandDefinition } from './types' + +type ParsedUsageArgs = { + help: boolean + json: boolean + since?: string + until?: string +} + +function showHelp(): void { + console.log(` +${chalk.bold('hapi usage')} - When agents were working, across every machine on this hub + +${chalk.bold('Usage:')} + hapi usage Last 24 hours + hapi usage --since 7d + hapi usage --since 2026-10-01 --until 2026-10-08 + hapi usage --json Machine-readable (schema hapi.activity.v1) + +${chalk.bold('Notes:')} + A turn runs from the agent starting to think until it stops; waits for + permission or answers are listed separately. Windows are at most 31 days. + --since/--until take 30m, 24h, 7d, an ISO date or epoch milliseconds. + +${chalk.bold('Exit codes:')} + 2 bad arguments · 3 not signed in or refused · 4 hub too old · 1 other + +${chalk.bold('Env:')} + HAPI_API_URL / CLI_API_TOKEN (or ~/.hapi/settings.json via \`hapi auth login\`) +`) +} + +export function parseUsageArgs(args: string[]): ParsedUsageArgs { + const result: ParsedUsageArgs = { help: false, json: false } + for (let i = 0; i < args.length; i++) { + const arg = args[i]! + if (arg === '--help' || arg === '-h') { + result.help = true + } else if (arg === '--json') { + result.json = true + } else if (arg === '--since' || arg === '--until') { + const value = args[++i] + if (!value) throw new ActivityError('bad_args', `${arg} requires a value`) + result[arg === '--since' ? 'since' : 'until'] = value + } else if (arg.startsWith('--since=')) { + result.since = arg.slice('--since='.length) + } else if (arg.startsWith('--until=')) { + result.until = arg.slice('--until='.length) + } else { + throw new ActivityError('bad_args', `unexpected argument: ${arg}`) + } + } + return result +} + +export async function handleUsageCommand(args: string[]): Promise { + const parsed = parseUsageArgs(args) + if (parsed.help) { + showHelp() + return + } + const now = Date.now() + const until = parsed.until ? parseTimeArg(parsed.until, now) : now + const since = parsed.since ? parseTimeArg(parsed.since, now) : until - 24 * 60 * 60 * 1000 + + try { + await initializeToken() + } catch (error) { + throw new ActivityError('auth_failed', error instanceof Error ? error.message : String(error)) + } + const report = await fetchActivity({ since, until }) + console.log(parsed.json ? JSON.stringify(report) : formatActivityReport(report)) +} + +export const usageCommand: CommandDefinition = { + name: 'usage', + requiresRuntimeAssets: false, + run: async ({ commandArgs }) => { + try { + await handleUsageCommand(commandArgs) + } catch (error) { + // Errors go to stderr so --json output on stdout stays parseable. + if (error instanceof ActivityError) { + console.error(chalk.red('hapi usage:'), error.message) + process.exit(exitCodeForActivityError(error)) + } + console.error(chalk.red('hapi usage:'), error instanceof Error ? error.message : 'Unknown error') + if (process.env.DEBUG) { + console.error(error) + } + process.exit(1) + } + } +} diff --git a/cli/src/modules/activity/activity.test.ts b/cli/src/modules/activity/activity.test.ts new file mode 100644 index 00000000..3068c5c8 --- /dev/null +++ b/cli/src/modules/activity/activity.test.ts @@ -0,0 +1,159 @@ +import { describe, expect, it, vi } from 'vitest' +import type { AxiosInstance } from 'axios' +import type { AgentActivitySession } from '@hapi/protocol/apiTypes' +import { + ACTIVITY_REPORT_SCHEMA, + ActivityError, + exitCodeForActivityError, + fetchActivity, + formatActivityReport, + parseTimeArg, + workedMs +} from './activity' + +vi.mock('@/persistence', () => ({ + readSettings: vi.fn(async () => ({ machineId: 'local-machine' })) +})) + +type MockResponse = { status: number; data: unknown } + +function createHttp(activity: MockResponse, auth: MockResponse = { status: 200, data: { token: 'jwt' } }) { + return { + post: vi.fn(async () => auth), + get: vi.fn(async (_url: string, _config?: { params?: Record; headers?: Record }) => activity) + } +} + +const session: AgentActivitySession = { + id: 'session-1', + flavor: 'codex', + agentSessionId: 'thread-1', + machineId: 'remote-machine', + host: 'build-box', + machineName: 'Build box', + path: '/srv/project', + title: 'Port the runner', + turns: [ + { start: 0, end: 600_000, endReason: 'idle', startedBy: 'hub_prompt' }, + { start: 900_000, end: null, endReason: null, startedBy: 'agent' } + ], + waits: [{ start: 100_000, end: 160_000, kind: 'permission', tool: 'Bash' }] +} + +describe('parseTimeArg', () => { + const now = Date.UTC(2026, 9, 29, 12) + + it('reads relative times, ISO dates and epoch milliseconds', () => { + expect(parseTimeArg('30m', now)).toBe(now - 30 * 60_000) + expect(parseTimeArg('24h', now)).toBe(now - 24 * 3_600_000) + expect(parseTimeArg('7d', now)).toBe(now - 7 * 86_400_000) + expect(parseTimeArg('2026-10-01T00:00:00Z', now)).toBe(Date.UTC(2026, 9, 1)) + expect(parseTimeArg('1793000000000', now)).toBe(1_793_000_000_000) + }) + + it('rejects anything else', () => { + expect(() => parseTimeArg('yesterday', now)).toThrow(ActivityError) + }) +}) + +describe('fetchActivity', () => { + it('wraps the hub response in a versioned report', async () => { + const http = createHttp({ status: 200, data: { since: 1, until: 2, generatedAt: 3, sessions: [session] } }) + const report = await fetchActivity({ + since: 1, until: 2, apiUrl: 'https://hub.example', accessToken: 'token', + http: http as unknown as AxiosInstance, now: () => 42 + }) + + expect(report).toEqual({ + schema: ACTIVITY_REPORT_SCHEMA, + generatedAt: 42, + hubUrl: 'https://hub.example', + localMachineId: 'local-machine', + since: 1, + until: 2, + sessions: [session] + }) + const [url, config] = http.get.mock.calls[0]! + expect(url).toBe('https://hub.example/api/activity') + expect(config?.params).toEqual({ since: 1, until: 2 }) + expect(config?.headers?.Authorization).toBe('Bearer jwt') + }) + + it('says when the hub is too old, the token is refused, or the request fails', async () => { + const run = (activity: MockResponse, auth?: MockResponse) => fetchActivity({ + since: 1, until: 2, apiUrl: 'https://hub.example', accessToken: 'token', + http: createHttp(activity, auth) as unknown as AxiosInstance + }).catch((error: ActivityError) => error.code) + + expect(await run({ status: 404, data: 'Not Found' })).toBe('hub_too_old') + expect(await run({ status: 200, data: {} }, { status: 401, data: { error: 'Invalid token' } })).toBe('auth_failed') + expect(await run({ status: 400, data: { error: 'The window can be at most 31 days' } })).toBe('request_failed') + }) + + it('rejects an empty window before asking the hub', async () => { + const http = createHttp({ status: 200, data: { sessions: [] } }) + await expect(fetchActivity({ since: 5, until: 5, http: http as unknown as AxiosInstance })).rejects.toThrow(ActivityError) + expect(http.post).not.toHaveBeenCalled() + }) + + it('maps errors to documented exit codes', () => { + expect(exitCodeForActivityError(new ActivityError('bad_args', ''))).toBe(2) + expect(exitCodeForActivityError(new ActivityError('auth_failed', ''))).toBe(3) + expect(exitCodeForActivityError(new ActivityError('hub_too_old', ''))).toBe(4) + expect(exitCodeForActivityError(new ActivityError('request_failed', ''))).toBe(1) + }) +}) + +describe('reporting', () => { + it('counts turns minus waits, open turns up to the end of the window', () => { + // 600s turn minus a 60s wait, plus an open turn from 900s to 1000s. + expect(workedMs(session, { since: 0, until: 1_000_000 })).toBe(540_000 + 100_000) + }) + + it('only counts the part of a turn inside the window', () => { + const long: AgentActivitySession = { + ...session, + turns: [{ start: 0, end: 3 * 3_600_000, endReason: 'idle', startedBy: 'agent' }], + waits: [{ start: 3_600_000 - 600_000, end: 3_600_000 + 600_000, kind: 'permission', tool: 'Bash' }] + } + // One hour out of a three-hour turn, minus the ten minutes of the wait inside it. + expect(workedMs(long, { since: 3_600_000, until: 2 * 3_600_000 })).toBe(3_600_000 - 600_000) + // An open turn in an earlier window stops at that window's end. + const open: AgentActivitySession = { ...session, turns: [{ start: 0, end: null, endReason: null, startedBy: 'agent' }], waits: [] } + expect(workedMs(open, { since: 0, until: 60_000 })).toBe(60_000) + }) + + it('subtracts overlapping waits once', () => { + const minute = 60_000 + const parallel: AgentActivitySession = { + ...session, + turns: [{ start: 0, end: 10 * minute, endReason: 'idle', startedBy: 'agent' }], + waits: [ + { start: 2 * minute, end: 6 * minute, kind: 'permission', tool: 'Bash' }, + { start: 4 * minute, end: 8 * minute, kind: 'input', tool: 'AskUserQuestion' } + ] + } + expect(workedMs(parallel, { since: 0, until: 20 * minute })).toBe(4 * minute) + }) + + it('lists sessions busiest first, saying where they ran', () => { + const local: AgentActivitySession = { + ...session, id: 'session-2', machineId: 'local-machine', title: 'Short one', + turns: [{ start: 0, end: 60_000, endReason: 'idle', startedBy: 'agent' }], waits: [] + } + const text = formatActivityReport({ + schema: ACTIVITY_REPORT_SCHEMA, generatedAt: 1_000_000, hubUrl: 'https://hub.example', + localMachineId: 'local-machine', since: 0, until: 1_000_000, sessions: [local, session] + }) + const lines = text.split('\n') + expect(lines[0]).toBe('Agents worked 12m across 2 sessions:') + expect(lines[1]).toContain('Port the runner · codex · Build box · 2 turns (working now)') + expect(lines[2]).toContain('Short one · codex · this machine · 1 turn') + }) + + it('says so when nothing happened', () => { + expect(formatActivityReport({ + schema: ACTIVITY_REPORT_SCHEMA, generatedAt: 0, hubUrl: '', localMachineId: null, since: 0, until: 1, sessions: [] + })).toBe('No agent activity in this period.') + }) +}) diff --git a/cli/src/modules/activity/activity.ts b/cli/src/modules/activity/activity.ts new file mode 100644 index 00000000..969c6032 --- /dev/null +++ b/cli/src/modules/activity/activity.ts @@ -0,0 +1,199 @@ +/** + * When agents were working, from the hub's GET /api/activity, for + * `hapi usage`. Other tools (e.g. time trackers) read the `--json` output, so + * its shape is versioned by `ACTIVITY_REPORT_SCHEMA`. + */ + +import axios, { type AxiosInstance } from 'axios' +import type { AgentActivitySession } from '@hapi/protocol/apiTypes' +import { readSettings } from '@/persistence' +import { + PingPeerError, + authHeaders, + exchangeJwt, + resolveAccessToken, + resolveApiUrl +} from '@/modules/pingPeer/pingPeer' + +/** Bump when the JSON output changes incompatibly. */ +export const ACTIVITY_REPORT_SCHEMA = 'hapi.activity.v1' + +export type ActivityReport = { + schema: typeof ACTIVITY_REPORT_SCHEMA + generatedAt: number + hubUrl: string + /** This machine's id, so a reader can tell local sessions from remote ones. */ + localMachineId: string | null + since: number + until: number + sessions: AgentActivitySession[] +} + +export type ActivityErrorCode = 'bad_args' | 'auth_failed' | 'hub_too_old' | 'request_failed' + +export class ActivityError extends Error { + readonly code: ActivityErrorCode + + constructor(code: ActivityErrorCode, message: string) { + super(message) + this.name = 'ActivityError' + this.code = code + } +} + +export function exitCodeForActivityError(error: ActivityError): number { + switch (error.code) { + case 'bad_args': return 2 + case 'auth_failed': return 3 + case 'hub_too_old': return 4 + default: return 1 + } +} + +const UNIT_MS: Record = { m: 60_000, h: 3_600_000, d: 86_400_000 } + +/** `30m`, `24h`, `7d` (before `now`), an ISO date, or epoch milliseconds. */ +export function parseTimeArg(value: string, now: number): number { + const trimmed = value.trim() + const relative = /^(\d+)([mhd])$/.exec(trimmed) + if (relative) { + return now - Number(relative[1]) * UNIT_MS[relative[2]!]! + } + if (/^\d{10,15}$/.test(trimmed)) { + return Number(trimmed) + } + const parsed = Date.parse(trimmed) + if (Number.isNaN(parsed)) { + throw new ActivityError('bad_args', `can't read '${value}' as a time; use 24h, 7d, an ISO date or epoch milliseconds`) + } + return parsed +} + +export type FetchActivityOptions = { + since: number + until: number + apiUrl?: string + accessToken?: string + http?: AxiosInstance + now?: () => number +} + +export async function fetchActivity(options: FetchActivityOptions): Promise { + if (options.since >= options.until) { + throw new ActivityError('bad_args', '--since must be before --until') + } + const http = options.http ?? axios + let apiUrl: string + let jwt: string + try { + apiUrl = resolveApiUrl(options.apiUrl) + jwt = await exchangeJwt(apiUrl, resolveAccessToken(options.accessToken), http) + } catch (error) { + if (error instanceof PingPeerError) { + throw new ActivityError(error.code === 'auth_failed' ? 'auth_failed' : 'bad_args', error.message) + } + throw error + } + + let response + try { + response = await http.get(`${apiUrl}/api/activity`, { + params: { since: options.since, until: options.until }, + headers: authHeaders(jwt), + timeout: 30_000, + validateStatus: () => true + }) + } catch (error) { + throw new ActivityError('request_failed', `could not reach ${apiUrl}: ${error instanceof Error ? error.message : String(error)}`) + } + if (response.status === 404) { + throw new ActivityError('hub_too_old', `the hub at ${apiUrl} doesn't report agent activity yet; update it`) + } + if (response.status === 401 || response.status === 403) { + throw new ActivityError('auth_failed', `the hub at ${apiUrl} refused the request (HTTP ${response.status})`) + } + const body = response.data as { error?: unknown; sessions?: unknown; since?: unknown; until?: unknown } | undefined + if (response.status < 200 || response.status >= 300 || !Array.isArray(body?.sessions)) { + const detail = typeof body?.error === 'string' ? body.error : `HTTP ${response.status}` + throw new ActivityError('request_failed', `the hub at ${apiUrl} could not report activity: ${detail}`) + } + + const settings = await readSettings() + return { + schema: ACTIVITY_REPORT_SCHEMA, + generatedAt: (options.now ?? Date.now)(), + hubUrl: apiUrl, + localMachineId: settings.machineId ?? null, + since: typeof body.since === 'number' ? body.since : options.since, + until: typeof body.until === 'number' ? body.until : options.until, + sessions: body.sessions as AgentActivitySession[] + } +} + +type Interval = { start: number; end: number } + +/** Sorted, non-overlapping intervals covering the same time. */ +function mergeIntervals(intervals: Interval[]): Interval[] { + const merged: Interval[] = [] + for (const interval of [...intervals].sort((a, b) => a.start - b.start)) { + const last = merged[merged.length - 1] + if (last && interval.start <= last.end) { + last.end = Math.max(last.end, interval.end) + } else { + merged.push({ ...interval }) + } + } + return merged +} + +/** + * Time a session's agent spent working within `window`: its turns, minus + * the time it waited on the user. The hub returns whole turns that overlap + * the window, so turns and waits are cut to it first; a turn still open + * runs to the end of the window. + */ +export function workedMs(session: AgentActivitySession, window: { since: number; until: number }): number { + const clamp = (start: number, end: number | null): Interval | null => { + const clamped = { start: Math.max(start, window.since), end: Math.min(end ?? window.until, window.until) } + return clamped.end > clamped.start ? clamped : null + } + const turns = mergeIntervals(session.turns.flatMap((turn) => clamp(turn.start, turn.end) ?? [])) + const waits = mergeIntervals(session.waits.flatMap((wait) => clamp(wait.start, wait.end) ?? [])) + + let total = 0 + for (const turn of turns) { + let waited = 0 + for (const wait of waits) { + const overlap = Math.min(turn.end, wait.end) - Math.max(turn.start, wait.start) + if (overlap > 0) waited += overlap + } + total += turn.end - turn.start - waited + } + return total +} + +function formatDuration(ms: number): string { + const minutes = Math.round(ms / 60_000) + if (minutes < 60) return `${minutes}m` + return minutes % 60 === 0 ? `${minutes / 60}h` : `${Math.floor(minutes / 60)}h ${minutes % 60}m` +} + +/** One line per session, busiest first. */ +export function formatActivityReport(report: ActivityReport): string { + if (report.sessions.length === 0) { + return 'No agent activity in this period.' + } + const window = { since: report.since, until: Math.min(report.until, report.generatedAt) } + const rows = report.sessions + .map((session) => ({ session, worked: workedMs(session, window) })) + .sort((a, b) => b.worked - a.worked) + const total = rows.reduce((sum, row) => sum + row.worked, 0) + const lines = rows.map(({ session, worked }) => { + const title = session.title ?? session.path ?? session.id.slice(0, 8) + const where = session.machineId && session.machineId === report.localMachineId ? 'this machine' : (session.machineName ?? session.host ?? 'unknown machine') + const live = session.turns.some((turn) => turn.end === null) ? ' (working now)' : '' + const turns = `${session.turns.length} turn${session.turns.length === 1 ? '' : 's'}` + return ` ${formatDuration(worked).padStart(7)} ${title} · ${session.flavor ?? 'agent'} · ${where} · ${turns}${live}` + }) + return [`Agents worked ${formatDuration(total)} across ${rows.length} session${rows.length === 1 ? '' : 's'}:`, ...lines].join('\n') +} diff --git a/cli/src/modules/pingPeer/pingPeer.ts b/cli/src/modules/pingPeer/pingPeer.ts index fe6b2ea3..234aa99b 100644 --- a/cli/src/modules/pingPeer/pingPeer.ts +++ b/cli/src/modules/pingPeer/pingPeer.ts @@ -125,7 +125,7 @@ const AUTH_RECOVERY_HINT = 'or run `hapi auth login` to save the token. Inside a HAPI session prefer MCP ' + '`list_peers` / `ping_peer` / `inspect_peer`, which use the session CLI credentials.' -function resolveApiUrl(apiUrl?: string): string { +export function resolveApiUrl(apiUrl?: string): string { const raw = (apiUrl ?? configuration.apiUrl).trim().replace(/\/+$/, '') if (!raw) { throw new PingPeerError( @@ -138,7 +138,7 @@ function resolveApiUrl(apiUrl?: string): string { return raw } -function resolveAccessToken(accessToken?: string): string { +export function resolveAccessToken(accessToken?: string): string { let token = '' try { token = (accessToken ?? getAuthToken()).trim() @@ -158,7 +158,7 @@ function authFailedMessage(apiUrl: string, detail: string): string { return `failed to exchange access token for JWT (${detail}). Hub URL: ${apiUrl}. ${AUTH_RECOVERY_HINT}` } -async function exchangeJwt( +export async function exchangeJwt( apiUrl: string, accessToken: string, http: AxiosInstance @@ -192,7 +192,7 @@ async function exchangeJwt( } } -function authHeaders(jwt: string): Record { +export function authHeaders(jwt: string): Record { return buildHubRequestHeaders({ Authorization: `Bearer ${jwt}`, 'Content-Type': 'application/json' diff --git a/docs/api/client-contract/rest.md b/docs/api/client-contract/rest.md index c16be5b2..8f2ed0f4 100644 --- a/docs/api/client-contract/rest.md +++ b/docs/api/client-contract/rest.md @@ -260,6 +260,14 @@ Source: `hub/src/web/routes/usage.ts`, `hub/src/web/routes/storage.ts`. Both `40 | `GET /api/usage/summary` | Query: `range=7d\|30d\|all` (default 7d), `timeZone` (IANA, validated) | `UsageSummaryResponse` `{range, totals, daily[], byAgent[], byModel[], updatedAt}` | | `GET /api/storage/sqlite` | — | `{path, databaseBytes, walBytes, shmBytes, totalBytes}` | +### Agent activity + +Source: `hub/src/web/routes/activity.ts`. Scoped to the caller's namespace (not owner-only). + +| Method & path | Request | Response | +|---|---|---| +| `GET /api/activity` | Query: `since`, `until` (epoch ms; default the 24 h before `until`, which defaults to now; at most 31 days; else `400`) | `AgentActivityResponse` `{since, until, generatedAt, sessions[]}`. Each session: `{id, flavor, agentSessionId, machineId, host, machineName, path, title, turns[], waits[]}`. A turn is `{start, end, endReason, startedBy}` with `end: null` while the agent is still working; turns are persisted when `thinking` flips, so they survive hub restarts. Waits are permission prompts and questions raised during the turns, `{start, end, kind: 'permission'\|'input', tool}`, with `end: null` while still pending. | + ### Devices (Android / iOS push) Source: `hub/src/web/routes/devices.ts`; full push contract in [`native-companion-contract.md`](../native-companion-contract.md). diff --git a/hub/README.md b/hub/README.md index dbe1c1e9..ca35a284 100644 --- a/hub/README.md +++ b/hub/README.md @@ -158,6 +158,10 @@ for request/response shapes and error semantics, and `src/web/routes/` for all e - `GET /api/usage/summary` - Get cache-aware token usage for the owner namespace (`range=7d|30d|all`). +### Activity (`src/web/routes/activity.ts`) + +- `GET /api/activity` - When agents in the caller's namespace were working (`since`/`until` epoch ms, default last 24 h, max 31 days): each session's turns from the `session_turns` table, plus the permission prompts and questions it waited on. Used by `hapi usage`. + ### Storage (`src/web/routes/storage.ts`) - `GET /api/storage/sqlite` - SQLite database/WAL/SHM sizes for the owner namespace. diff --git a/hub/src/store/index.ts b/hub/src/store/index.ts index 5efe5dd9..d05657d1 100644 --- a/hub/src/store/index.ts +++ b/hub/src/store/index.ts @@ -11,6 +11,7 @@ import { FcmStore } from './fcmStore' import { ScratchlistStore } from './scratchlistStore' import { SessionStore } from './sessionStore' import { UserStore } from './userStore' +import { TurnStore } from './turnStore' import { UsageStore } from './usageStore' import { WorkGraphStore } from './workGraphStore' @@ -34,6 +35,8 @@ export { FcmStore } from './fcmStore' export { ScratchlistStore } from './scratchlistStore' export { SessionStore } from './sessionStore' export { UserStore } from './userStore' +export { TurnStore } from './turnStore' +export type { SessionTurn, SessionTurnEndReason, SessionTurnTrigger } from './turns' export { UsageStore } from './usageStore' export { WorkGraphStore } from './workGraphStore' export { @@ -42,7 +45,7 @@ export { WorkGraphValidationError } from './workGraph' -const SCHEMA_VERSION: number = 27 +const SCHEMA_VERSION: number = 28 const REQUIRED_TABLES = [ 'sessions', 'machines', @@ -55,7 +58,8 @@ const REQUIRED_TABLES = [ 'usage_events', 'usage_scan_state', 'events', - 'event_links' + 'event_links', + 'session_turns' ] as const export class Store { @@ -71,6 +75,7 @@ export class Store { readonly fcm: FcmStore readonly scratchlist: ScratchlistStore readonly usage: UsageStore + readonly turns: TurnStore readonly workGraph: WorkGraphStore /** @@ -125,6 +130,7 @@ export class Store { this.fcm = new FcmStore(this.db) this.scratchlist = new ScratchlistStore(this.db) this.usage = new UsageStore(this.db) + this.turns = new TurnStore(this.db) this.workGraph = new WorkGraphStore(this.db) } @@ -353,6 +359,7 @@ export class Store { 24: () => this.migrateFromV24ToV25(), 25: () => this.migrateFromV25ToV26(), 26: () => this.migrateFromV26ToV27(), + 27: () => this.migrateFromV27ToV28(), }) if (currentVersion === 0) { @@ -609,6 +616,25 @@ export class Store { ON event_links(namespace, from_event_id); CREATE INDEX IF NOT EXISTS idx_event_links_namespace_to ON event_links(namespace, to_event_id); + + CREATE TABLE IF NOT EXISTS session_turns ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + session_id TEXT NOT NULL, + namespace TEXT NOT NULL, + started_at INTEGER NOT NULL, + ended_at INTEGER, + end_reason TEXT, + started_by TEXT NOT NULL, + recover_until INTEGER, + FOREIGN KEY (session_id) REFERENCES sessions(id) ON DELETE CASCADE + ); + CREATE INDEX IF NOT EXISTS idx_session_turns_namespace_started + ON session_turns(namespace, started_at); + CREATE INDEX IF NOT EXISTS idx_session_turns_session_started + ON session_turns(session_id, started_at); + CREATE UNIQUE INDEX IF NOT EXISTS idx_session_turns_one_open + ON session_turns(session_id) + WHERE ended_at IS NULL; `) } @@ -1030,6 +1056,40 @@ export class Store { `) } + /** + * v27→v28: persist when agents were working. `thinking` used to live only + * in the session cache, so a hub restart forgot it and nothing could say + * afterwards how long an agent worked. Each false→true→false stretch is + * now a row; at most one per session is open (ended_at IS NULL). + * recover_until marks a turn a restart interrupted, which the agent can + * still continue until then. + */ + private migrateFromV27ToV28(): void { + // In the legacy branch sessions may not exist yet; createSchema adds + // the table once it does. + if (this.getSessionColumnNames().size === 0) return + this.db.exec(` + CREATE TABLE IF NOT EXISTS session_turns ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + session_id TEXT NOT NULL, + namespace TEXT NOT NULL, + started_at INTEGER NOT NULL, + ended_at INTEGER, + end_reason TEXT, + started_by TEXT NOT NULL, + recover_until INTEGER, + FOREIGN KEY (session_id) REFERENCES sessions(id) ON DELETE CASCADE + ); + CREATE INDEX IF NOT EXISTS idx_session_turns_namespace_started + ON session_turns(namespace, started_at); + CREATE INDEX IF NOT EXISTS idx_session_turns_session_started + ON session_turns(session_id, started_at); + CREATE UNIQUE INDEX IF NOT EXISTS idx_session_turns_one_open + ON session_turns(session_id) + WHERE ended_at IS NULL; + `) + } + /** * A2A Layer 1 / P1 (#1374) + P3 substrate: hub work-graph ledger tables. * Namespace + principal_json required on every events row. diff --git a/hub/src/store/migration-v13.test.ts b/hub/src/store/migration-v13.test.ts index 4101f540..84dd38e4 100644 --- a/hub/src/store/migration-v13.test.ts +++ b/hub/src/store/migration-v13.test.ts @@ -10,7 +10,7 @@ describe('Store V12/V13→V14 schema reconciliation', () => { const store = new Store(':memory:') expect(tableExists(store, 'message_epochs')).toBe(true) expect(tableExists(store, 'session_scratchlist')).toBe(true) - expect(getUserVersion(store)).toBe(27) + expect(getUserVersion(store)).toBe(28) store.close() }) @@ -34,7 +34,7 @@ describe('Store V12/V13→V14 schema reconciliation', () => { store = new Store(dbPath) expect(tableExists(store, 'message_epochs')).toBe(true) expect(tableExists(store, 'session_scratchlist')).toBe(true) - expect(getUserVersion(store)).toBe(27) + expect(getUserVersion(store)).toBe(28) expect(store.messages.getMessageEpoch('session-1')).toBe(0) expect(store.messages.getMessages('session-1')).toHaveLength(1) } finally { @@ -72,7 +72,7 @@ describe('Store V12/V13→V14 schema reconciliation', () => { store = new Store(dbPath) expect(tableExists(store, 'message_epochs')).toBe(true) expect(tableExists(store, 'session_scratchlist')).toBe(true) - expect(getUserVersion(store)).toBe(27) + expect(getUserVersion(store)).toBe(28) expect(store.messages.getMessages('session-1')).toHaveLength(1) } finally { store?.close() diff --git a/hub/src/store/migration-v15.test.ts b/hub/src/store/migration-v15.test.ts index 02b5c556..8353dba2 100644 --- a/hub/src/store/migration-v15.test.ts +++ b/hub/src/store/migration-v15.test.ts @@ -19,7 +19,7 @@ describe('Store V14→V15 migration: scratchlist attachments column', () => { expect(cols).toContain('attachments') expect(getColumns(store, 'usage_events')).toContain('last_input_tokens') expect(getColumns(store, 'usage_scan_state')).toContain('last_seq') - expect(getUserVersion(store)).toBe(27) + expect(getUserVersion(store)).toBe(28) store.close() }) @@ -40,7 +40,7 @@ describe('Store V14→V15 migration: scratchlist attachments column', () => { expect(cols).toContain('attachments') expect(getColumns(store, 'usage_events')).toContain('last_input_tokens') expect(getColumns(store, 'usage_scan_state')).toContain('last_seq') - expect(getUserVersion(store)).toBe(27) + expect(getUserVersion(store)).toBe(28) } finally { store?.close() rmSync(dir, { recursive: true, force: true }) @@ -60,7 +60,7 @@ describe('Store V14→V15 migration: scratchlist attachments column', () => { store2 = new Store(dbPath) const cols2 = getColumns(store2, 'session_scratchlist') expect(cols2).toEqual(cols1) - expect(getUserVersion(store2)).toBe(27) + expect(getUserVersion(store2)).toBe(28) } finally { store2?.close() store1?.close() diff --git a/hub/src/store/migration-v18.test.ts b/hub/src/store/migration-v18.test.ts index 5c23e841..faee78c8 100644 --- a/hub/src/store/migration-v18.test.ts +++ b/hub/src/store/migration-v18.test.ts @@ -38,7 +38,7 @@ describe('Store V18->V19 migration: usage scan state', () => { const usageRows = internalDb.prepare('SELECT COUNT(*) AS count FROM usage_events').get() as { count: number } expect(table?.name).toBe('usage_scan_state') - expect(version.user_version).toBe(27) + expect(version.user_version).toBe(28) expect(usageRows.count).toBe(0) } finally { store?.close() diff --git a/hub/src/store/migration-v19.test.ts b/hub/src/store/migration-v19.test.ts index 7a9b3d7a..8f40f99c 100644 --- a/hub/src/store/migration-v19.test.ts +++ b/hub/src/store/migration-v19.test.ts @@ -37,7 +37,7 @@ describe('Store V19->current migration: usage re-index', () => { const scanRows = internalDb.prepare('SELECT COUNT(*) AS count FROM usage_scan_state').get() as { count: number } const usageRows = internalDb.prepare('SELECT COUNT(*) AS count FROM usage_events').get() as { count: number } - expect(version.user_version).toBe(27) + expect(version.user_version).toBe(28) expect(scanRows.count).toBe(0) expect(usageRows.count).toBe(0) } finally { diff --git a/hub/src/store/migration-v20.test.ts b/hub/src/store/migration-v20.test.ts index 5f60d446..400852fc 100644 --- a/hub/src/store/migration-v20.test.ts +++ b/hub/src/store/migration-v20.test.ts @@ -43,7 +43,7 @@ describe('Store V20->V21 migration: usage semantics re-index', () => { store = new Store(dbPath) const internalDb = (store as unknown as { db: Database }).db const count = (table: string): number => (internalDb.prepare(`SELECT COUNT(*) AS count FROM ${table}`).get() as { count: number }).count - expect((internalDb.prepare('PRAGMA user_version').get() as { user_version: number }).user_version).toBe(27) + expect((internalDb.prepare('PRAGMA user_version').get() as { user_version: number }).user_version).toBe(28) expect(count('usage_events')).toBe(0) expect(count('usage_scan_state')).toBe(0) expect(store.messages.getMessages(session.id)).toHaveLength(1) diff --git a/hub/src/store/migration-v23.test.ts b/hub/src/store/migration-v23.test.ts index 71d13291..f10599b3 100644 --- a/hub/src/store/migration-v23.test.ts +++ b/hub/src/store/migration-v23.test.ts @@ -42,7 +42,7 @@ describe('schema migration v22 to v26', () => { expect(links?.name).toBe('event_links') const columns = internalDb.prepare('PRAGMA table_info(messages)').all() as Array<{ name: string }> expect(columns.map((column) => column.name)).toContain('delivery_state') - expect(version.user_version).toBe(27) + expect(version.user_version).toBe(28) migrated.close() }) }) diff --git a/hub/src/store/migration-v24.test.ts b/hub/src/store/migration-v24.test.ts index 0837c1d5..939926a3 100644 --- a/hub/src/store/migration-v24.test.ts +++ b/hub/src/store/migration-v24.test.ts @@ -37,7 +37,7 @@ describe('schema migration v23 to v26', () => { expect(columns.some((col) => col.name === 'push_key')).toBe(true) const messageColumns = internalDb.prepare('PRAGMA table_info(messages)').all() as Array<{ name: string }> expect(messageColumns.some((col) => col.name === 'delivery_state')).toBe(true) - expect(version.user_version).toBe(27) + expect(version.user_version).toBe(28) // Existing Android rows survive with a NULL push key. const devices = migrated.fcm.getDevicesByNamespace('default') @@ -74,7 +74,7 @@ describe('schema migration v23 to v26', () => { const columns = internalDb.prepare('PRAGMA table_info(messages)').all() as Array<{ name: string }> const version = internalDb.prepare('PRAGMA user_version').get() as { user_version: number } expect(columns.some((col) => col.name === 'delivery_state')).toBe(true) - expect(version.user_version).toBe(27) + expect(version.user_version).toBe(28) migrated.close() }) }) diff --git a/hub/src/store/migration-v25.test.ts b/hub/src/store/migration-v25.test.ts index 6011c9c2..bc8e900a 100644 --- a/hub/src/store/migration-v25.test.ts +++ b/hub/src/store/migration-v25.test.ts @@ -41,7 +41,7 @@ describe('schema migration v25 to v26', () => { ORDER BY seq ASC `).all('session-id') as Array<{ detail: string }> - expect(version.user_version).toBe(27) + expect(version.user_version).toBe(28) expect(plan.some((row) => row.detail.includes('idx_messages_immediate_queued'))).toBe(true) migrated.close() }) diff --git a/hub/src/store/migration-v26.test.ts b/hub/src/store/migration-v26.test.ts index a5f49433..9bec1f9d 100644 --- a/hub/src/store/migration-v26.test.ts +++ b/hub/src/store/migration-v26.test.ts @@ -53,7 +53,7 @@ describe('schema migration v26 to v27', () => { GROUP BY session_id `).all('a', 'b', 123) as Array<{ detail: string }> - expect(version.user_version).toBe(27) + expect(version.user_version).toBe(28) expect(countPlan.some((row) => row.detail.includes('idx_messages_future_scheduled_queued'))).toBe(true) expect(minPlan.some((row) => row.detail.includes('idx_messages_future_scheduled_queued'))).toBe(true) migrated.close() diff --git a/hub/src/store/migration-v28.test.ts b/hub/src/store/migration-v28.test.ts new file mode 100644 index 00000000..2ca561ce --- /dev/null +++ b/hub/src/store/migration-v28.test.ts @@ -0,0 +1,42 @@ +import { afterEach, describe, expect, it } from 'bun:test' +import { Database } from 'bun:sqlite' +import { mkdtempSync, rmSync } from 'node:fs' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { Store } from './index' + +const tempDirs: string[] = [] + +afterEach(() => { + for (const dir of tempDirs.splice(0)) { + rmSync(dir, { recursive: true, force: true }) + } +}) + +describe('schema migration v27 to v28', () => { + it('adds session_turns, allowing one open turn per session', () => { + const dir = mkdtempSync(join(tmpdir(), 'hapi-migration-v27-')) + tempDirs.push(dir) + const dbPath = join(dir, 'hapi.db') + + const original = new Store(dbPath) + const session = original.sessions.getOrCreateSession('tag', { path: '/tmp', host: 'mac' }, null, 'default') + original.close() + const legacy = new Database(dbPath) + legacy.exec(` + DROP TABLE session_turns; + PRAGMA user_version = 27; + `) + legacy.close() + + const migrated = new Store(dbPath) + const internalDb = (migrated as unknown as { db: Database }).db + const version = internalDb.prepare('PRAGMA user_version').get() as { user_version: number } + expect(version.user_version).toBe(28) + + expect(migrated.turns.open(session.id, 'default', 1_000, 'agent')).toBe(true) + expect(migrated.turns.open(session.id, 'default', 2_000, 'agent')).toBe(false) + expect(migrated.turns.list('default', 0, 10_000)).toHaveLength(1) + migrated.close() + }) +}) diff --git a/hub/src/store/turnStore.ts b/hub/src/store/turnStore.ts new file mode 100644 index 00000000..789aab6d --- /dev/null +++ b/hub/src/store/turnStore.ts @@ -0,0 +1,47 @@ +import type { Database } from 'bun:sqlite' + +import { + abandonSessionTurnRecovery, + closeDanglingSessionTurns, + closeSessionTurn, + listSessionTurns, + openSessionTurn, + resumeInterruptedSessionTurn, + transferSessionTurns, + type SessionTurn, + type SessionTurnEndReason, + type SessionTurnTrigger +} from './turns' + +/** When agents were working: one row per stretch a session was thinking. */ +export class TurnStore { + constructor(private readonly db: Database) {} + + open(sessionId: string, namespace: string, startedAt: number, trigger: SessionTurnTrigger): boolean { + return openSessionTurn(this.db, sessionId, namespace, startedAt, trigger) + } + + close(sessionId: string, endedAt: number, reason: SessionTurnEndReason): boolean { + return closeSessionTurn(this.db, sessionId, endedAt, reason) + } + + closeDangling(now: number, recoveryWindowMs: number): string[] { + return closeDanglingSessionTurns(this.db, now, recoveryWindowMs) + } + + resumeInterrupted(sessionId: string, now: number): boolean { + return resumeInterruptedSessionTurn(this.db, sessionId, now) + } + + abandonRecovery(sessionId: string): void { + abandonSessionTurnRecovery(this.db, sessionId) + } + + list(namespace: string, since: number, until: number): SessionTurn[] { + return listSessionTurns(this.db, namespace, since, until) + } + + transferSession(fromSessionId: string, toSessionId: string, now: number = Date.now()): void { + transferSessionTurns(this.db, fromSessionId, toSessionId, now) + } +} diff --git a/hub/src/store/turns.test.ts b/hub/src/store/turns.test.ts new file mode 100644 index 00000000..b97ba6dc --- /dev/null +++ b/hub/src/store/turns.test.ts @@ -0,0 +1,130 @@ +import { describe, expect, it } from 'bun:test' +import { Store } from './index' + +function setup() { + const store = new Store(':memory:') + const session = store.sessions.getOrCreateSession('tag-a', { path: '/tmp/a', host: 'mac' }, null, 'default') + return { store, session } +} + +describe('TurnStore', () => { + it('opens and closes turns, never ending before the start', () => { + const { store, session } = setup() + try { + expect(store.turns.open(session.id, 'default', 1_000, 'hub_prompt')).toBe(true) + expect(store.turns.close(session.id, 500, 'idle')).toBe(true) + expect(store.turns.close(session.id, 9_000, 'idle')).toBe(false) + + const [turn] = store.turns.list('default', 0, 10_000) + expect(turn).toMatchObject({ sessionId: session.id, startedAt: 1_000, endedAt: 1_000, endReason: 'idle', trigger: 'hub_prompt' }) + } finally { + store.close() + } + }) + + it('lists turns overlapping the window, open ones included', () => { + const { store, session } = setup() + try { + store.turns.open(session.id, 'default', 1_000, 'agent') + store.turns.close(session.id, 2_000, 'idle') + store.turns.open(session.id, 'default', 5_000, 'agent') + + expect(store.turns.list('default', 1_500, 1_600).map((turn) => turn.startedAt)).toEqual([1_000]) + expect(store.turns.list('default', 2_000, 4_000)).toEqual([]) + expect(store.turns.list('default', 9_000, 10_000).map((turn) => turn.endedAt)).toEqual([null]) + expect(store.turns.list('other', 0, 10_000)).toEqual([]) + } finally { + store.close() + } + }) + + it('ends turns left open by a restart at the session\'s newest message', () => { + const { store, session } = setup() + const idle = store.sessions.getOrCreateSession('tag-b', { path: '/tmp/b', host: 'mac' }, null, 'default') + try { + store.turns.open(session.id, 'default', 1_000, 'agent') + store.messages.addMessage(session.id, { role: 'agent', content: { type: 'text', text: 'working' } }) + store.turns.open(idle.id, 'default', 1_000, 'agent') + + const now = Date.now() + expect(store.turns.closeDangling(now + 60_000, 10 * 60_000).sort()).toEqual([session.id, idle.id].sort()) + const turns = store.turns.list('default', 0, now + 120_000) + const withMessage = turns.find((turn) => turn.sessionId === session.id) + const withoutMessage = turns.find((turn) => turn.sessionId === idle.id) + expect(withMessage?.endReason).toBe('hub_restart') + expect(withMessage?.endedAt).toBeGreaterThanOrEqual(now - 5_000) + expect(withMessage?.endedAt).toBeLessThanOrEqual(now + 5_000) + // No message to go by: a zero-length turn rather than one running to now. + expect(withoutMessage?.endedAt).toBe(1_000) + } finally { + store.close() + } + }) + + it('lets an interrupted turn continue until a deadline counted from the restart', () => { + const { store, session } = setup() + const minute = 60_000 + try { + store.turns.open(session.id, 'default', 1_000, 'agent') + // No message for an hour before the restart: the turn ends long before it. + const restart = 60 * minute + store.turns.closeDangling(restart, 10 * minute) + expect(store.turns.list('default', 0, restart)[0]?.endedAt).toBe(1_000) + + expect(store.turns.resumeInterrupted(session.id, restart + 11 * minute)).toBe(false) + expect(store.turns.resumeInterrupted(session.id, restart + 5 * minute)).toBe(true) + expect(store.turns.list('default', 0, restart)[0]?.endedAt).toBeNull() + // Once resumed, it's an ordinary turn again. + store.turns.close(session.id, restart + 6 * minute, 'idle') + expect(store.turns.resumeInterrupted(session.id, restart + 7 * minute)).toBe(false) + } finally { + store.close() + } + }) + + it('never continues a turn whose recovery was given up, or a turn that ended normally', () => { + const { store, session } = setup() + try { + store.turns.open(session.id, 'default', 1_000, 'agent') + store.turns.closeDangling(2_000, 10 * 60_000) + store.turns.abandonRecovery(session.id) + expect(store.turns.resumeInterrupted(session.id, 3_000)).toBe(false) + + store.turns.open(session.id, 'default', 4_000, 'agent') + store.turns.close(session.id, 5_000, 'idle') + expect(store.turns.resumeInterrupted(session.id, 5_500)).toBe(false) + } finally { + store.close() + } + }) + + it('moves turns to the session that replaces it', () => { + const { store, session } = setup() + const replacement = store.sessions.getOrCreateSession('tag-c', { path: '/tmp/a', host: 'mac' }, null, 'default') + try { + store.turns.open(session.id, 'default', 1_000, 'agent') + store.turns.close(session.id, 2_000, 'idle') + store.turns.open(session.id, 'default', 3_000, 'agent') + store.turns.open(replacement.id, 'default', 4_000, 'agent') + + store.turns.transferSession(session.id, replacement.id, 5_000) + const turns = store.turns.list('default', 0, 10_000) + expect(turns.every((turn) => turn.sessionId === replacement.id)).toBe(true) + expect(turns.filter((turn) => turn.endedAt === null).map((turn) => turn.startedAt)).toEqual([4_000]) + expect(turns.find((turn) => turn.startedAt === 3_000)?.endedAt).toBe(5_000) + } finally { + store.close() + } + }) + + it('goes away with its session', () => { + const { store, session } = setup() + try { + store.turns.open(session.id, 'default', 1_000, 'agent') + store.sessions.deleteSession(session.id, 'default') + expect(store.turns.list('default', 0, 10_000)).toEqual([]) + } finally { + store.close() + } + }) +}) diff --git a/hub/src/store/turns.ts b/hub/src/store/turns.ts new file mode 100644 index 00000000..07f1b108 --- /dev/null +++ b/hub/src/store/turns.ts @@ -0,0 +1,147 @@ +import type { Database } from 'bun:sqlite' + +/** Why a turn stopped: the agent went idle, the session ended, its keepalives + * stopped, or the hub restarted while it was open. */ +export type SessionTurnEndReason = 'idle' | 'session_end' | 'expired' | 'hub_restart' + +/** What started a turn: a prompt sent through the hub, the agent reporting + * that it is thinking (e.g. a prompt typed in a local terminal), or a + * session patch. */ +export type SessionTurnTrigger = 'hub_prompt' | 'agent' | 'patch' + +export type SessionTurn = { + id: number + sessionId: string + namespace: string + startedAt: number + endedAt: number | null + endReason: SessionTurnEndReason | null + trigger: SessionTurnTrigger +} + +type SessionTurnRow = { + id: number + session_id: string + namespace: string + started_at: number + ended_at: number | null + end_reason: SessionTurnEndReason | null + started_by: SessionTurnTrigger +} + +function toSessionTurn(row: SessionTurnRow): SessionTurn { + return { + id: row.id, + sessionId: row.session_id, + namespace: row.namespace, + startedAt: row.started_at, + endedAt: row.ended_at, + endReason: row.end_reason, + trigger: row.started_by + } +} + +/** Opens a turn unless one is already open for the session. */ +export function openSessionTurn( + db: Database, + sessionId: string, + namespace: string, + startedAt: number, + trigger: SessionTurnTrigger +): boolean { + const result = db.prepare(` + INSERT OR IGNORE INTO session_turns (session_id, namespace, started_at, started_by) + VALUES (?, ?, ?, ?) + `).run(sessionId, namespace, startedAt, trigger) + return result.changes > 0 +} + +/** Closes the session's open turn, never before it started. */ +export function closeSessionTurn(db: Database, sessionId: string, endedAt: number, reason: SessionTurnEndReason): boolean { + const result = db.prepare(` + UPDATE session_turns + SET ended_at = MAX(started_at, ?), end_reason = ? + WHERE session_id = ? AND ended_at IS NULL + `).run(endedAt, reason, sessionId) + return result.changes > 0 +} + +/** + * Closes turns left open by a previous hub process. Nobody knows when they + * really ended, so each ends at its session's newest message, which is the + * last sign of the agent working that survived the restart. Until + * `now + recoveryWindowMs` the agent can still continue the turn (see + * `resumeInterruptedSessionTurn`); that deadline counts from the restart, not + * from the newest message, which may be long before it. + * + * @returns the ids of the sessions whose turns were closed. + */ +export function closeDanglingSessionTurns(db: Database, now: number, recoveryWindowMs: number): string[] { + const rows = db.prepare(` + UPDATE session_turns + SET ended_at = MAX(started_at, MIN(?, COALESCE(( + SELECT MAX(COALESCE(m.invoked_at, m.created_at)) + FROM messages m + WHERE m.session_id = session_turns.session_id + ), started_at))), + end_reason = 'hub_restart', + recover_until = ? + WHERE ended_at IS NULL + RETURNING session_id + `).all(now, now + recoveryWindowMs) as Array<{ session_id: string }> + return rows.map((row) => row.session_id) +} + +/** + * Reopens the session's turn a restart interrupted, so the agent carrying on + * with it doesn't split it in two. Only before its recovery deadline, and + * only while it is still the session's latest turn. + */ +export function resumeInterruptedSessionTurn(db: Database, sessionId: string, now: number): boolean { + const result = db.prepare(` + UPDATE session_turns + SET ended_at = NULL, end_reason = NULL, recover_until = NULL + WHERE id = ( + SELECT id FROM session_turns + WHERE session_id = ? + ORDER BY started_at DESC, id DESC + LIMIT 1 + ) + AND recover_until >= ? + `).run(sessionId, now) + return result.changes > 0 +} + +/** The agent was seen idle or got a new prompt: the interrupted turn is over for good. */ +export function abandonSessionTurnRecovery(db: Database, sessionId: string): void { + db.prepare('UPDATE session_turns SET recover_until = NULL WHERE session_id = ? AND recover_until IS NOT NULL').run(sessionId) +} + +/** Turns in `namespace` that overlap [since, until), oldest first. */ +export function listSessionTurns(db: Database, namespace: string, since: number, until: number): SessionTurn[] { + const rows = db.prepare(` + SELECT id, session_id, namespace, started_at, ended_at, end_reason, started_by + FROM session_turns + WHERE namespace = ? + AND started_at < ? + AND (ended_at IS NULL OR ended_at > ?) + ORDER BY started_at, id + `).all(namespace, until, since) as SessionTurnRow[] + return rows.map(toSessionTurn) +} + +/** Moves a merged session's turns to the session that replaces it. */ +export function transferSessionTurns(db: Database, fromSessionId: string, toSessionId: string, now: number): void { + if (fromSessionId === toSessionId) return + + db.transaction(() => { + // Only one turn per session may be open; the destination's wins. + db.prepare(` + UPDATE session_turns + SET ended_at = MAX(started_at, ?), end_reason = 'session_end' + WHERE session_id = ? AND ended_at IS NULL + AND EXISTS (SELECT 1 FROM session_turns WHERE session_id = ? AND ended_at IS NULL) + `).run(now, fromSessionId, toSessionId) + db.prepare('UPDATE session_turns SET session_id = ? WHERE session_id = ?').run(toSessionId, fromSessionId) + })() +} diff --git a/hub/src/sync/activityService.ts b/hub/src/sync/activityService.ts new file mode 100644 index 00000000..f7e2d796 --- /dev/null +++ b/hub/src/sync/activityService.ts @@ -0,0 +1,90 @@ +import { classifyKind, toSessionSummaryMetadata } from '@hapi/protocol' +import type { AgentActivityResponse, AgentActivitySession, AgentActivityWait } from '@hapi/protocol/apiTypes' +import { AgentStateSchema, MachineMetadataSchema, MetadataSchema } from '@hapi/protocol/schemas' +import type { SessionTurn, Store } from '../store' + +export const ACTIVITY_DEFAULT_WINDOW_MS = 24 * 60 * 60 * 1000 +export const ACTIVITY_MAX_WINDOW_MS = 31 * 24 * 60 * 60 * 1000 +const TITLE_MAX_LENGTH = 120 + +/** + * When agents in `namespace` were working between `since` and `until`: each + * session's turns, plus the permission prompts and questions it waited on + * the user for, so a client can leave those out. + */ +export function getAgentActivity( + store: Store, + namespace: string, + since: number, + until: number, + now: number = Date.now() +): AgentActivityResponse { + const turnsBySession = new Map() + for (const turn of store.turns.list(namespace, since, until)) { + const turns = turnsBySession.get(turn.sessionId) ?? [] + turns.push(turn) + turnsBySession.set(turn.sessionId, turns) + } + + const machineNames = new Map() + const machineName = (machineId: string): string | null => { + if (!machineNames.has(machineId)) { + const stored = store.machines.getMachineByNamespace(machineId, namespace) + const parsed = MachineMetadataSchema.safeParse(stored?.metadata) + machineNames.set(machineId, parsed.success ? (parsed.data.displayName ?? parsed.data.host) : null) + } + return machineNames.get(machineId) ?? null + } + + const sessions: AgentActivitySession[] = [] + for (const [sessionId, turns] of turnsBySession) { + const stored = store.sessions.getSessionByNamespace(sessionId, namespace) + if (!stored) continue + const parsedMetadata = MetadataSchema.safeParse(stored.metadata) + const metadata = parsedMetadata.success ? parsedMetadata.data : null + const summary = toSessionSummaryMetadata(metadata) + const machineId = summary?.machineId ?? stored.machineId ?? null + const title = (summary?.name ?? summary?.summary?.text)?.trim() + + sessions.push({ + id: sessionId, + flavor: summary?.flavor ?? null, + agentSessionId: summary?.agentSessionId ?? null, + machineId, + host: metadata?.host ?? null, + machineName: machineId ? machineName(machineId) : null, + path: summary?.path ?? null, + title: title ? title.slice(0, TITLE_MAX_LENGTH) : null, + turns: turns.map((turn) => ({ + start: turn.startedAt, + end: turn.endedAt, + endReason: turn.endReason, + startedBy: turn.trigger + })), + waits: collectWaits(stored.agentState, turns, now) + }) + } + + return { since, until, generatedAt: now, sessions } +} + +/** Permission prompts and questions raised during one of `turns`. */ +function collectWaits(agentState: unknown, turns: SessionTurn[], now: number): AgentActivityWait[] { + const parsed = AgentStateSchema.safeParse(agentState) + if (!parsed.success) return [] + const overlapsTurn = (start: number, end: number) => + turns.some((turn) => start < (turn.endedAt ?? now) && end > turn.startedAt) + + const waits: AgentActivityWait[] = [] + for (const request of Object.values(parsed.data.requests ?? {})) { + // Requests stored before createdAt existed can't be placed in time. + if (request.createdAt == null || !overlapsTurn(request.createdAt, now)) continue + waits.push({ start: request.createdAt, end: null, kind: classifyKind(request.tool), tool: request.tool }) + } + for (const request of Object.values(parsed.data.completedRequests ?? {})) { + if (request.createdAt == null || request.completedAt == null) continue + if (request.completedAt <= request.createdAt || !overlapsTurn(request.createdAt, request.completedAt)) continue + waits.push({ start: request.createdAt, end: request.completedAt, kind: classifyKind(request.tool), tool: request.tool }) + } + return waits.sort((a, b) => a.start - b.start) +} diff --git a/hub/src/sync/sessionCache.ts b/hub/src/sync/sessionCache.ts index d10b1cd1..63523c41 100644 --- a/hub/src/sync/sessionCache.ts +++ b/hub/src/sync/sessionCache.ts @@ -1,13 +1,16 @@ import { SESSION_LIFECYCLE_IDLE, SESSION_LIFECYCLE_RUNNING } from '@hapi/protocol' import { AgentStateSchema, MetadataSchema, SessionPatchSchema, TeamStateSchema } from '@hapi/protocol/schemas' import type { BackgroundTask, CodexCollaborationMode, CopilotAgentMode, PermissionMode, Session, SessionPatch } from '@hapi/protocol/types' -import type { Store } from '../store' +import type { SessionTurnEndReason, SessionTurnTrigger, Store } from '../store' import { clampAliveTime } from './aliveTime' import { EventPublisher } from './eventPublisher' import { extractTodoWriteTodosFromMessageContent, TodosSchema } from './todos' import { resolveSessionIdleTimeoutMs, shouldClearKeepaliveIdle, shouldMarkKeepaliveIdle } from './sessionIdle' const QUEUED_MESSAGE_THINKING_GRACE_MS = 15_000 +// A turn the hub closed when it restarted is picked up again if the agent +// carries on with it this soon after the restart, instead of starting a new one. +const TURN_RECOVERY_WINDOW_MS = 10 * 60_000 // tiann/hapi#919: metadata writers (renameSession, clearSessionArchiveMetadata, // restoreSessionArchiveMetadata) retry on version-mismatch with a fresh cache // snapshot. Cap retries so genuine concurrent contention still surfaces to the @@ -30,6 +33,12 @@ export class SessionCache { * so a hub restart re-derives it from the last human turn on disk. */ private readonly agentProgressAtBySessionId: Map = new Map() + /** + * Sessions whose turn the previous hub process left open. It continues if + * the agent reports thinking before the hub sees it idle or a new prompt + * arrives; see `recordTurn`. + */ + private readonly interruptedTurnSessionIds: Set = new Set() private readonly sessionIdleTimeoutMs: number = resolveSessionIdleTimeoutMs() constructor( @@ -45,6 +54,20 @@ export class SessionCache { ) { } + /** + * Closes turns the previous hub process left open; `thinking` starts out + * false here, so nothing else would. Call once at startup. + */ + closeTurnsLeftOpen(now: number = Date.now()): void { + try { + for (const sessionId of this.store.turns.closeDangling(now, TURN_RECOVERY_WINDOW_MS)) { + this.interruptedTurnSessionIds.add(sessionId) + } + } catch (error) { + console.warn('[SessionCache] Failed to close turns left open', { error }) + } + } + getSessions(): Session[] { return Array.from(this.sessions.values()) } @@ -357,6 +380,7 @@ export class SessionCache { } const patch = parsed.data + const wasThinking = session.thinking if (patch.active !== undefined) session.active = patch.active if (patch.thinking !== undefined) session.thinking = patch.thinking @@ -389,6 +413,8 @@ export class SessionCache { session.agentState = patch.agentState.value session.agentStateVersion = patch.agentState.version } + if (patch.thinking === false) this.forgetInterruptedTurn(session.id) + this.recordTurn(session, wasThinking, Date.now(), 'patch', 'idle') return true } @@ -441,6 +467,9 @@ export class SessionCache { if (requestedThinking || pendingThinkingUntil <= hubNow) { this.pendingThinkingUntilBySessionId.delete(session.id) } + // The agent says it's idle: whatever it does next is a new turn. + if (!requestedThinking) this.forgetInterruptedTurn(session.id) + this.recordTurn(session, wasThinking, hubNow, 'agent', 'idle') if (payload.permissionMode !== undefined && !this.isStaleRuntimeKeepAlive(session.id, 'permissionMode', t)) { session.permissionMode = payload.permissionMode this.persistPreferredPermissionMode(session, payload.permissionMode) @@ -555,6 +584,7 @@ export class SessionCache { session.thinking = true session.thinkingAt = nextTime if (!wasThinking) session.activeTurnStartedAt = activeTurnStartedAt + this.recordTurn(session, wasThinking, session.activeTurnStartedAt ?? nextTime, 'hub_prompt', 'idle') session.updatedAt = Math.max(session.updatedAt, nextTime) this.pendingThinkingUntilBySessionId.set(session.id, nextTime + QUEUED_MESSAGE_THINKING_GRACE_MS) this.recordAgentProgress(session.id, nextTime) @@ -573,6 +603,48 @@ export class SessionCache { } } + /** + * Persists a change of `thinking` as the start or end of a turn, so how + * long agents worked survives a hub restart. Never throws: losing a turn + * row must not break the keepalive path. + */ + private recordTurn( + session: Session, + wasThinking: boolean, + at: number, + trigger: SessionTurnTrigger, + reason: SessionTurnEndReason + ): void { + if (wasThinking === session.thinking) return + try { + if (session.thinking) { + // Only the agent itself carrying on continues an interrupted + // turn; a new prompt starts a new one. + const resumed = trigger === 'agent' + && this.interruptedTurnSessionIds.has(session.id) + && this.store.turns.resumeInterrupted(session.id, at) + this.forgetInterruptedTurn(session.id) + if (!resumed) { + this.store.turns.open(session.id, session.namespace, at, trigger) + } + } else { + this.store.turns.close(session.id, at, reason) + } + } catch (error) { + console.warn('[SessionCache] Failed to record turn', { sessionId: session.id, error }) + } + } + + /** The turn a restart interrupted can no longer continue. */ + private forgetInterruptedTurn(sessionId: string): void { + if (!this.interruptedTurnSessionIds.delete(sessionId)) return + try { + this.store.turns.abandonRecovery(sessionId) + } catch (error) { + console.warn('[SessionCache] Failed to drop turn recovery', { sessionId, error }) + } + } + setBackgroundTasks(sessionId: string, tasks: BackgroundTask[]): void { const session = this.sessions.get(sessionId) if (!session) return @@ -766,16 +838,19 @@ export class SessionCache { const session = this.sessions.get(payload.sid) ?? this.refreshSession(payload.sid) if (!session) return + this.forgetInterruptedTurn(session.id) if (!session.active && !session.thinking) { return } + const wasThinking = session.thinking session.active = false this.store.sessions.setSessionActive(session.id, false, t, session.namespace) session.thinking = false session.thinkingAt = t session.activeTurnStartedAt = null + this.recordTurn(session, wasThinking, t, 'agent', 'session_end') session.backgroundTaskCount = 0 session.backgroundTasks = [] this.pendingThinkingUntilBySessionId.delete(session.id) @@ -801,9 +876,12 @@ export class SessionCache { if (!session.active) continue const timeoutMs = session.thinking ? thinkingSessionTimeoutMs : sessionTimeoutMs if (now - session.activeAt <= timeoutMs) continue + const wasThinking = session.thinking session.active = false this.store.sessions.setSessionActive(session.id, false, now, session.namespace) session.thinking = false + // The last keepalive is the last time the agent was known to be working. + this.recordTurn(session, wasThinking, session.activeAt, 'agent', 'expired') this.pendingThinkingUntilBySessionId.delete(session.id) expired.push(session.id) this.publisher.emit({ @@ -1574,6 +1652,8 @@ export class SessionCache { } if (options.deleteOldSession) { + // Deleting the row cascades to its turns; they're the new row's history now. + this.store.turns.transferSession(oldSessionId, newSessionId) const deleted = this.store.sessions.deleteSession(oldSessionId, namespace) if (!deleted) { throw new Error('Failed to delete old session during merge') diff --git a/hub/src/sync/sessionCache.turns.test.ts b/hub/src/sync/sessionCache.turns.test.ts new file mode 100644 index 00000000..80dbfce5 --- /dev/null +++ b/hub/src/sync/sessionCache.turns.test.ts @@ -0,0 +1,166 @@ +import { describe, expect, it } from 'bun:test' +import type { SyncEvent } from '@hapi/protocol/types' +import { Store } from '../store' +import type { EventPublisher } from './eventPublisher' +import { SessionCache } from './sessionCache' + +function setup() { + const store = new Store(':memory:') + const events: SyncEvent[] = [] + const publisher = { emit: (event: SyncEvent) => { events.push(event) } } as unknown as EventPublisher + const cache = new SessionCache(store, publisher) + const session = cache.getOrCreateSession( + 'turns-' + Math.random().toString(36).slice(2, 8), + { path: '/tmp/project', host: 'mac', flavor: 'claude' }, + null, + 'default' + ) + return { store, cache, session } +} + +function turns(store: Store) { + return store.turns.list('default', 0, Date.now() + 60_000) +} + +describe('SessionCache turn recording', () => { + it('records a turn from thinking on to thinking off', () => { + const { store, cache, session } = setup() + try { + cache.handleSessionAlive({ sid: session.id, time: Date.now(), thinking: false }) + expect(turns(store)).toEqual([]) + + cache.handleSessionAlive({ sid: session.id, time: Date.now(), thinking: true }) + cache.handleSessionAlive({ sid: session.id, time: Date.now(), thinking: true }) + expect(turns(store)).toHaveLength(1) + expect(turns(store)[0]).toMatchObject({ trigger: 'agent', endedAt: null }) + + cache.handleSessionAlive({ sid: session.id, time: Date.now(), thinking: false }) + expect(turns(store)[0]).toMatchObject({ endReason: 'idle' }) + expect(turns(store)[0]?.endedAt).not.toBeNull() + } finally { + store.close() + } + }) + + it('starts a turn when a prompt is queued through the hub', () => { + const { store, cache, session } = setup() + try { + cache.handleSessionAlive({ sid: session.id, time: Date.now(), thinking: false }) + const sentAt = Date.now() - 2_000 + cache.markMessageQueued(session.id, Date.now(), sentAt) + + expect(turns(store)).toHaveLength(1) + expect(turns(store)[0]).toMatchObject({ trigger: 'hub_prompt', startedAt: sentAt, endedAt: null }) + // The agent picking it up doesn't start another one. + cache.handleSessionAlive({ sid: session.id, time: Date.now(), thinking: true }) + expect(turns(store)).toHaveLength(1) + } finally { + store.close() + } + }) + + it('ends the turn when the session ends', () => { + const { store, cache, session } = setup() + try { + cache.handleSessionAlive({ sid: session.id, time: Date.now(), thinking: true }) + cache.handleSessionEnd({ sid: session.id, time: Date.now() }) + expect(turns(store)[0]).toMatchObject({ endReason: 'session_end' }) + } finally { + store.close() + } + }) + + it('ends an expired turn at the last keepalive, not when it was noticed', () => { + const { store, cache, session } = setup() + try { + cache.handleSessionAlive({ sid: session.id, time: Date.now(), thinking: true }) + const lastKeepalive = cache.getSession(session.id)!.activeAt + cache.expireInactive(lastKeepalive + 301_000) + expect(turns(store)[0]).toMatchObject({ endReason: 'expired', endedAt: lastKeepalive }) + } finally { + store.close() + } + }) + + it('records thinking changes that arrive as patches', () => { + const { store, cache, session } = setup() + try { + cache.handleSessionAlive({ sid: session.id, time: Date.now(), thinking: false }) + expect(cache.applySessionPatch(session.id, { thinking: true })).toBe(true) + expect(turns(store)[0]).toMatchObject({ trigger: 'patch', endedAt: null }) + expect(cache.applySessionPatch(session.id, { thinking: false })).toBe(true) + expect(turns(store)[0]?.endedAt).not.toBeNull() + } finally { + store.close() + } + }) + + function restart(store: Store): SessionCache { + const restarted = new SessionCache(store, { emit: () => {} } as unknown as EventPublisher) + restarted.closeTurnsLeftOpen() + restarted.reloadAll() + return restarted + } + + it('continues a turn a hub restart cut short, however long the agent was quiet before it', () => { + const { store, session } = setup() + try { + // A command ran for 30 minutes without a message, then the hub restarted. + const startedAt = Date.now() - 30 * 60_000 + store.turns.open(session.id, 'default', startedAt, 'agent') + store.messages.addMessage(session.id, { role: 'agent', content: { type: 'text', text: 'running tests' } }, undefined, null, startedAt + 60_000) + + const restarted = restart(store) + restarted.handleSessionAlive({ sid: session.id, time: Date.now(), thinking: true }) + expect(turns(store)).toHaveLength(1) + expect(turns(store)[0]).toMatchObject({ startedAt, endedAt: null }) + } finally { + store.close() + } + }) + + it('starts a new turn once the agent was seen idle after a restart', () => { + const { store, session } = setup() + try { + store.turns.open(session.id, 'default', Date.now() - 60_000, 'agent') + const restarted = restart(store) + restarted.handleSessionAlive({ sid: session.id, time: Date.now(), thinking: false }) + restarted.handleSessionAlive({ sid: session.id, time: Date.now(), thinking: true }) + + expect(turns(store)).toHaveLength(2) + expect(turns(store)[0]?.endReason).toBe('hub_restart') + expect(turns(store)[1]?.endedAt).toBeNull() + } finally { + store.close() + } + }) + + it('starts a new turn for a prompt sent after a restart', () => { + const { store, session } = setup() + try { + store.turns.open(session.id, 'default', Date.now() - 60_000, 'agent') + const restarted = restart(store) + restarted.markSessionActive(session.id) + restarted.markMessageQueued(session.id) + + expect(turns(store)).toHaveLength(2) + expect(turns(store)[0]?.endReason).toBe('hub_restart') + expect(turns(store)[1]).toMatchObject({ trigger: 'hub_prompt', endedAt: null }) + } finally { + store.close() + } + }) + + it('keeps a merged-away session\'s turns', async () => { + const { store, cache, session } = setup() + const replacement = cache.getOrCreateSession('turns-new', { path: '/tmp/project', host: 'mac', flavor: 'claude' }, null, 'default') + try { + cache.handleSessionAlive({ sid: session.id, time: Date.now(), thinking: true }) + cache.handleSessionEnd({ sid: session.id, time: Date.now() }) + await cache.mergeSessions(session.id, replacement.id, 'default') + expect(turns(store).map((turn) => turn.sessionId)).toEqual([replacement.id]) + } finally { + store.close() + } + }) +}) diff --git a/hub/src/sync/syncEngine.ts b/hub/src/sync/syncEngine.ts index a60997e0..703f1052 100644 --- a/hub/src/sync/syncEngine.ts +++ b/hub/src/sync/syncEngine.ts @@ -278,6 +278,7 @@ export class SyncEngine { this.titleSuggestionService ) this.rpcGateway = new RpcGateway(io, rpcRegistry) + this.sessionCache.closeTurnsLeftOpen() this.reloadAll() this.inactivityTimer = setInterval(() => this.expireInactive(), 5_000) } diff --git a/hub/src/web/routes/activity.test.ts b/hub/src/web/routes/activity.test.ts new file mode 100644 index 00000000..5cd49487 --- /dev/null +++ b/hub/src/web/routes/activity.test.ts @@ -0,0 +1,102 @@ +import { describe, expect, it } from 'bun:test' +import { Hono } from 'hono' +import type { AgentActivityResponse } from '@hapi/protocol/apiTypes' +import { Store } from '../../store' +import type { WebAppEnv } from '../middleware/auth' +import { createActivityRoutes } from './activity' + +function createApp(store: Store, namespace: string): Hono { + const app = new Hono() + app.use('*', async (c, next) => { + c.set('namespace', namespace) + await next() + }) + app.route('/api', createActivityRoutes(store)) + return app +} + +describe('GET /api/activity', () => { + it('returns turns, waits and what identifies each session', async () => { + const store = new Store(':memory:') + try { + store.machines.getOrCreateMachine('machine-1', { host: 'studio.local', platform: 'darwin', happyCliVersion: '1', displayName: 'Studio' }, null, 'default') + const session = store.sessions.getOrCreateSession('tag', { + path: '/Users/me/project', + host: 'studio.local', + machineId: 'machine-1', + flavor: 'claude', + claudeSessionId: 'claude-123', + name: 'Fix the build' + }, null, 'default') + store.sessions.updateSessionAgentState(session.id, { + requests: { pending: { tool: 'Bash', arguments: {}, createdAt: 9_000 } }, + completedRequests: { + asked: { tool: 'AskUserQuestion', arguments: {}, createdAt: 2_000, completedAt: 4_000, status: 'approved' }, + old: { tool: 'Bash', arguments: {}, createdAt: 100, completedAt: 200, status: 'approved' }, + unplaced: { tool: 'Edit', arguments: {}, completedAt: 3_000, status: 'approved' } + } + }, session.agentStateVersion, 'default') + store.turns.open(session.id, 'default', 1_000, 'hub_prompt') + store.turns.close(session.id, 5_000, 'idle') + store.turns.open(session.id, 'default', 8_000, 'agent') + + const response = await createApp(store, 'default').request('/api/activity?since=0&until=20000') + expect(response.status).toBe(200) + expect(response.headers.get('cache-control')).toBe('no-store') + const body = await response.json() as AgentActivityResponse + expect(body.since).toBe(0) + expect(body.until).toBe(20_000) + expect(body.sessions).toHaveLength(1) + expect(body.sessions[0]).toMatchObject({ + id: session.id, + flavor: 'claude', + agentSessionId: 'claude-123', + machineId: 'machine-1', + host: 'studio.local', + machineName: 'Studio', + path: '/Users/me/project', + title: 'Fix the build', + turns: [ + { start: 1_000, end: 5_000, endReason: 'idle', startedBy: 'hub_prompt' }, + { start: 8_000, end: null, endReason: null, startedBy: 'agent' } + ], + waits: [ + { start: 2_000, end: 4_000, kind: 'input', tool: 'AskUserQuestion' }, + { start: 9_000, end: null, kind: 'permission', tool: 'Bash' } + ] + }) + } finally { + store.close() + } + }) + + it('only shows the caller\'s namespace', async () => { + const store = new Store(':memory:') + try { + const session = store.sessions.getOrCreateSession('tag', { path: '/tmp', host: 'mac' }, null, 'tenant') + store.turns.open(session.id, 'tenant', 1_000, 'agent') + + const owner = await (await createApp(store, 'default').request('/api/activity?since=0&until=20000')).json() as AgentActivityResponse + const tenant = await (await createApp(store, 'tenant').request('/api/activity?since=0&until=20000')).json() as AgentActivityResponse + expect(owner.sessions).toEqual([]) + expect(tenant.sessions.map((s) => s.id)).toEqual([session.id]) + } finally { + store.close() + } + }) + + it('defaults to the last day and rejects bad windows', async () => { + const store = new Store(':memory:') + try { + const app = createApp(store, 'default') + const body = await (await app.request('/api/activity')).json() as AgentActivityResponse + expect(body.until - body.since).toBe(24 * 60 * 60 * 1000) + + for (const query of ['since=abc', 'since=-5', 'since=10&until=10', `since=0&until=${32 * 24 * 60 * 60 * 1000}`]) { + expect((await app.request(`/api/activity?${query}`)).status).toBe(400) + } + } finally { + store.close() + } + }) +}) diff --git a/hub/src/web/routes/activity.ts b/hub/src/web/routes/activity.ts new file mode 100644 index 00000000..515b5e19 --- /dev/null +++ b/hub/src/web/routes/activity.ts @@ -0,0 +1,38 @@ +import { Hono } from 'hono' +import type { AgentActivityResponse } from '@hapi/protocol/apiTypes' +import type { WebAppEnv } from '../middleware/auth' +import type { Store } from '../../store' +import { ACTIVITY_DEFAULT_WINDOW_MS, ACTIVITY_MAX_WINDOW_MS, getAgentActivity } from '../../sync/activityService' + +/** Epoch ms from a query parameter; undefined when absent, null when invalid. */ +function parseTime(value: string | undefined): number | null | undefined { + if (value === undefined || value === '') return undefined + if (!/^\d{1,15}$/.test(value)) return null + return Number(value) +} + +export function createActivityRoutes(store: Store): Hono { + const app = new Hono() + + app.get('/activity', (c) => { + const now = Date.now() + const until = parseTime(c.req.query('until')) + const since = parseTime(c.req.query('since')) + if (until === null || since === null) { + return c.json({ error: 'since and until must be epoch milliseconds' }, 400) + } + const end = until ?? now + const start = since ?? end - ACTIVITY_DEFAULT_WINDOW_MS + if (start >= end) { + return c.json({ error: 'since must be before until' }, 400) + } + if (end - start > ACTIVITY_MAX_WINDOW_MS) { + return c.json({ error: 'The window can be at most 31 days' }, 400) + } + const response: AgentActivityResponse = getAgentActivity(store, c.get('namespace'), start, end, now) + c.header('Cache-Control', 'no-store') + return c.json(response) + }) + + return app +} diff --git a/hub/src/web/server.ts b/hub/src/web/server.ts index e179508a..9ee5a290 100644 --- a/hub/src/web/server.ts +++ b/hub/src/web/server.ts @@ -23,6 +23,7 @@ import { createPermissionsRoutes } from './routes/permissions' import { createMachinesRoutes } from './routes/machines' import { createStorageRoutes } from './routes/storage' import { createUsageRoutes } from './routes/usage' +import { createActivityRoutes } from './routes/activity' import { createGitRoutes } from './routes/git' import { createCliRoutes } from './routes/cli' import { createCodexDesktopRoutes } from './routes/codexDesktop' @@ -294,6 +295,7 @@ function createWebApp(options: { app.route('/api', createStorageRoutes(configuration.dbPath)) app.route('/api', createHubSettingsRoutes(configuration.dataDir)) app.route('/api', createUsageRoutes(options.store)) + app.route('/api', createActivityRoutes(options.store)) app.route('/api', createGitRoutes(options.getSyncEngine)) // 中文注释：这里提供两类 Codex 辅助能力：扫描本地 transcript 以导入到 Hapi，以及按需重启 Codex Desktop 客户端。 app.route('/api', createCodexDesktopRoutes({ diff --git a/shared/src/apiTypes.ts b/shared/src/apiTypes.ts index 372910f0..0b12ec8d 100644 --- a/shared/src/apiTypes.ts +++ b/shared/src/apiTypes.ts @@ -1134,3 +1134,43 @@ export type UsageSummaryResponse = { byModel: UsageSummaryBucket[] updatedAt: number } + +/** A stretch an agent was working: from `thinking` turning on to it turning + * off. `end` is null while the turn is still going. */ +export type AgentActivityTurn = { + start: number + end: number | null + endReason: 'idle' | 'session_end' | 'expired' | 'hub_restart' | null + startedBy: 'hub_prompt' | 'agent' | 'patch' +} + +/** The agent waiting on the user inside a turn: a permission prompt or a + * question. `end` is null while it is still waiting. */ +export type AgentActivityWait = { + start: number + end: number | null + kind: 'permission' | 'input' + tool: string +} + +export type AgentActivitySession = { + id: string + flavor: string | null + /** The agent's own session id (Claude session, Codex thread, ...). */ + agentSessionId: string | null + machineId: string | null + host: string | null + machineName: string | null + path: string | null + title: string | null + turns: AgentActivityTurn[] + waits: AgentActivityWait[] +} + +/** GET /api/activity: when agents in the caller's namespace were working. */ +export type AgentActivityResponse = { + since: number + until: number + generatedAt: number + sessions: AgentActivitySession[] +} diff --git a/shared/src/sessionSummary.ts b/shared/src/sessionSummary.ts index 20fed857..886db43b 100644 --- a/shared/src/sessionSummary.ts +++ b/shared/src/sessionSummary.ts @@ -28,7 +28,9 @@ export type PendingRequest = { since: number } -function classifyKind(tool: string): PendingRequestKind { +/** Whether a request asks the user to answer something (a question, a plan) + * or to allow a tool. */ +export function classifyKind(tool: string): PendingRequestKind { return INPUT_REQUEST_TOOLS.has(tool) ? 'input' : 'permission' }