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' }