diff --git a/hub/src/startHub.ts b/hub/src/startHub.ts index 4dd29fc1..3d5a7587 100644 --- a/hub/src/startHub.ts +++ b/hub/src/startHub.ts @@ -6,6 +6,7 @@ import type { NotificationChannel } from './notifications/notificationTypes' import { HappyBot } from './telegram/bot' import { startWebServer } from './web/server' import { getOrCreateJwtSecret } from './config/jwtSecret' +import { getOrCreateOwnerId } from './config/ownerId' import { createSocketServer } from './socket/server' import { SSEManager } from './sse/sseManager' import { getOrCreateVapidKeys } from './config/vapidKeys' @@ -199,6 +200,8 @@ export async function startHub(options: StartHubOptions = {}): Promise this.migrateFromV19ToV20(), 20: () => this.migrateFromV20ToV21(), 21: () => this.migrateFromV21ToV22(), + 22: () => this.migrateFromV22ToV23(), }) if (currentVersion === 0) { @@ -483,6 +495,55 @@ export class Store { last_seq INTEGER NOT NULL DEFAULT 0, FOREIGN KEY (session_id) REFERENCES sessions(id) ON DELETE CASCADE ); + + CREATE TABLE IF NOT EXISTS events ( + id TEXT PRIMARY KEY, + ts INTEGER NOT NULL, + source_kind TEXT NOT NULL, + source_ref TEXT NOT NULL, + sink_kind TEXT, + sink_ref TEXT, + event_type TEXT NOT NULL, + summary TEXT, + payload_json TEXT, + artifact_refs TEXT NOT NULL DEFAULT '[]', + tags TEXT NOT NULL DEFAULT '[]', + related_session_id TEXT, + related_event_id TEXT, + provenance TEXT, + idempotency_key TEXT, + dedupe_key TEXT, + confidence REAL, + severity TEXT, + expires_at INTEGER, + namespace TEXT NOT NULL, + principal_json TEXT NOT NULL + ); + CREATE INDEX IF NOT EXISTS idx_events_namespace_ts + ON events(namespace, ts DESC); + CREATE INDEX IF NOT EXISTS idx_events_namespace_related_session + ON events(namespace, related_session_id, ts DESC); + CREATE INDEX IF NOT EXISTS idx_events_namespace_expires + ON events(namespace, expires_at); + CREATE UNIQUE INDEX IF NOT EXISTS idx_events_namespace_idempotency + ON events(namespace, idempotency_key) + WHERE idempotency_key IS NOT NULL; + + CREATE TABLE IF NOT EXISTS event_links ( + id TEXT PRIMARY KEY, + from_event_id TEXT NOT NULL, + to_event_id TEXT NOT NULL, + relation_type TEXT NOT NULL, + created_at INTEGER NOT NULL, + metadata_json TEXT, + namespace TEXT NOT NULL, + FOREIGN KEY (from_event_id) REFERENCES events(id) ON DELETE CASCADE, + FOREIGN KEY (to_event_id) REFERENCES events(id) ON DELETE CASCADE + ); + CREATE INDEX IF NOT EXISTS idx_event_links_namespace_from + ON event_links(namespace, from_event_id); + CREATE INDEX IF NOT EXISTS idx_event_links_namespace_to + ON event_links(namespace, to_event_id); `) } @@ -853,6 +914,64 @@ export class Store { } } + /** + * A2A Layer 1 / P1 (#1374) + P3 substrate: hub work-graph ledger tables. + * Namespace + principal_json required on every events row. + * Bumped as v22→v23 because upstream main already ships schema 22. + */ + private migrateFromV22ToV23(): void { + this.db.exec(` + CREATE TABLE IF NOT EXISTS events ( + id TEXT PRIMARY KEY, + ts INTEGER NOT NULL, + source_kind TEXT NOT NULL, + source_ref TEXT NOT NULL, + sink_kind TEXT, + sink_ref TEXT, + event_type TEXT NOT NULL, + summary TEXT, + payload_json TEXT, + artifact_refs TEXT NOT NULL DEFAULT '[]', + tags TEXT NOT NULL DEFAULT '[]', + related_session_id TEXT, + related_event_id TEXT, + provenance TEXT, + idempotency_key TEXT, + dedupe_key TEXT, + confidence REAL, + severity TEXT, + expires_at INTEGER, + namespace TEXT NOT NULL, + principal_json TEXT NOT NULL + ); + CREATE INDEX IF NOT EXISTS idx_events_namespace_ts + ON events(namespace, ts DESC); + CREATE INDEX IF NOT EXISTS idx_events_namespace_related_session + ON events(namespace, related_session_id, ts DESC); + CREATE INDEX IF NOT EXISTS idx_events_namespace_expires + ON events(namespace, expires_at); + CREATE UNIQUE INDEX IF NOT EXISTS idx_events_namespace_idempotency + ON events(namespace, idempotency_key) + WHERE idempotency_key IS NOT NULL; + + CREATE TABLE IF NOT EXISTS event_links ( + id TEXT PRIMARY KEY, + from_event_id TEXT NOT NULL, + to_event_id TEXT NOT NULL, + relation_type TEXT NOT NULL, + created_at INTEGER NOT NULL, + metadata_json TEXT, + namespace TEXT NOT NULL, + FOREIGN KEY (from_event_id) REFERENCES events(id) ON DELETE CASCADE, + FOREIGN KEY (to_event_id) REFERENCES events(id) ON DELETE CASCADE + ); + CREATE INDEX IF NOT EXISTS idx_event_links_namespace_from + ON event_links(namespace, from_event_id); + CREATE INDEX IF NOT EXISTS idx_event_links_namespace_to + ON event_links(namespace, to_event_id); + `) + } + private getSessionColumnNames(): Set { const rows = this.db.prepare('PRAGMA table_info(sessions)').all() as Array<{ name: string }> return new Set(rows.map((row) => row.name)) diff --git a/hub/src/store/migration-v13.test.ts b/hub/src/store/migration-v13.test.ts index a099009b..1780813e 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(22) + expect(getUserVersion(store)).toBe(23) 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(22) + expect(getUserVersion(store)).toBe(23) 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(22) + expect(getUserVersion(store)).toBe(23) 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 ce8ade58..a323fdeb 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(22) + expect(getUserVersion(store)).toBe(23) 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(22) + expect(getUserVersion(store)).toBe(23) } 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(22) + expect(getUserVersion(store2)).toBe(23) } finally { store2?.close() store1?.close() diff --git a/hub/src/store/migration-v18.test.ts b/hub/src/store/migration-v18.test.ts index af662bf1..75861a17 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(22) + expect(version.user_version).toBe(23) 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 8e665934..5688e615 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(22) + expect(version.user_version).toBe(23) 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 7cfec238..529dff11 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(22) + expect((internalDb.prepare('PRAGMA user_version').get() as { user_version: number }).user_version).toBe(23) 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 new file mode 100644 index 00000000..9fc94893 --- /dev/null +++ b/hub/src/store/migration-v23.test.ts @@ -0,0 +1,46 @@ +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 v22 to v23', () => { + it('adds events and event_links tables to a V22 database', () => { + const dir = mkdtempSync(join(tmpdir(), 'hapi-migration-v23-')) + tempDirs.push(dir) + const dbPath = join(dir, 'hapi.db') + + new Store(dbPath).close() + const legacy = new Database(dbPath) + legacy.exec(` + DROP TABLE IF EXISTS event_links; + DROP TABLE IF EXISTS events; + PRAGMA user_version = 22; + `) + legacy.close() + + const migrated = new Store(dbPath) + const internalDb = (migrated as unknown as { db: Database }).db + const events = internalDb.prepare( + "SELECT name FROM sqlite_master WHERE type = 'table' AND name = 'events'" + ).get() as { name: string } | null + const links = internalDb.prepare( + "SELECT name FROM sqlite_master WHERE type = 'table' AND name = 'event_links'" + ).get() as { name: string } | null + const version = internalDb.prepare('PRAGMA user_version').get() as { user_version: number } + + expect(events?.name).toBe('events') + expect(links?.name).toBe('event_links') + expect(version.user_version).toBe(23) + migrated.close() + }) +}) diff --git a/hub/src/store/workGraph.test.ts b/hub/src/store/workGraph.test.ts new file mode 100644 index 00000000..79557218 --- /dev/null +++ b/hub/src/store/workGraph.test.ts @@ -0,0 +1,173 @@ +import { describe, expect, it } from 'bun:test' +import { WORK_GRAPH_MAX_SUMMARY } from '@hapi/protocol' +import { Store, WorkGraphPrincipalError, WorkGraphValidationError } from './index' + +const humanPrincipal = { kind: 'human' as const, id: '1' } + +describe('WorkGraphStore', () => { + it('refuses non-human principal without human owner', () => { + const store = new Store(':memory:') + expect(() => store.workGraph.insertEvent('default', { + source_kind: 'session', + source_ref: 'sess-a', + event_type: 'work_ad', + summary: 'no owner', + // Passes Zod min(1) but fails trim accountability (store choke point). + principal: { kind: 'agent', id: 'worker', on_behalf_of: ' ' } + })).toThrow(WorkGraphPrincipalError) + }) + + it('rejects summary longer than WORK_GRAPH_MAX_SUMMARY at insert (S4)', () => { + const store = new Store(':memory:') + const oversized = 'x'.repeat(WORK_GRAPH_MAX_SUMMARY + 1) + expect(() => store.workGraph.insertEvent('default', { + source_kind: 'session', + source_ref: 'sess-a', + event_type: 'work_ad', + summary: oversized, + principal: humanPrincipal + })).toThrow(WorkGraphValidationError) + expect(store.workGraph.listByRelatedSession('default', 'sess-a')).toHaveLength(0) + }) + + it('isolates writes and queries by namespace', () => { + const store = new Store(':memory:') + const alpha = store.workGraph.insertEvent('alpha', { + source_kind: 'session', + source_ref: 'sess-a', + event_type: 'work_ad', + related_session_id: 'sess-a', + summary: 'alpha ad', + principal: humanPrincipal + }) + store.workGraph.insertEvent('beta', { + source_kind: 'session', + source_ref: 'sess-b', + event_type: 'work_ad', + related_session_id: 'sess-a', + summary: 'beta ad same related id', + principal: humanPrincipal + }) + + expect(store.workGraph.getEvent(alpha.event.id, 'beta')).toBeNull() + expect(store.workGraph.getEvent(alpha.event.id, 'alpha')?.summary).toBe('alpha ad') + + const alphaRows = store.workGraph.listByRelatedSession('alpha', 'sess-a') + expect(alphaRows).toHaveLength(1) + expect(alphaRows[0]?.summary).toBe('alpha ad') + + const betaRows = store.workGraph.listByRelatedSession('beta', 'sess-a') + expect(betaRows).toHaveLength(1) + expect(betaRows[0]?.summary).toBe('beta ad same related id') + }) + + it('idempotent insert via idempotency_key does not duplicate', () => { + const store = new Store(':memory:') + const first = store.workGraph.insertEvent('default', { + source_kind: 'session', + source_ref: 'sess-a', + event_type: 'work_ad', + related_session_id: 'sess-a', + summary: 'first', + idempotency_key: 'key-1', + principal: humanPrincipal + }) + const second = store.workGraph.insertEvent('default', { + source_kind: 'session', + source_ref: 'sess-a', + event_type: 'work_ad', + related_session_id: 'sess-a', + summary: 'retry', + idempotency_key: 'key-1', + principal: humanPrincipal + }) + + expect(first.inserted).toBe(true) + expect(second.inserted).toBe(false) + expect(second.event.id).toBe(first.event.id) + expect(second.event.summary).toBe('first') + expect(store.workGraph.listByRelatedSession('default', 'sess-a')).toHaveLength(1) + }) + + it('allows same idempotency_key in different namespaces', () => { + const store = new Store(':memory:') + const alpha = store.workGraph.insertEvent('alpha', { + source_kind: 'session', + source_ref: 'a', + event_type: 'work_ad', + idempotency_key: 'shared-key', + principal: humanPrincipal + }) + const beta = store.workGraph.insertEvent('beta', { + source_kind: 'session', + source_ref: 'b', + event_type: 'work_ad', + idempotency_key: 'shared-key', + principal: humanPrincipal + }) + expect(alpha.inserted).toBe(true) + expect(beta.inserted).toBe(true) + expect(alpha.event.id).not.toBe(beta.event.id) + }) + + it('refuses cross-namespace link creation', () => { + const store = new Store(':memory:') + const alpha = store.workGraph.insertEvent('alpha', { + source_kind: 'session', + source_ref: 'a', + event_type: 'handoff', + principal: humanPrincipal + }) + const beta = store.workGraph.insertEvent('beta', { + source_kind: 'session', + source_ref: 'b', + event_type: 'handoff_receipt', + principal: humanPrincipal + }) + + expect(() => store.workGraph.insertLink('alpha', { + from_event_id: alpha.event.id, + to_event_id: beta.event.id, + relation_type: 'resolves' + })).toThrow(/to_event_id not found/) + }) + + it('creates links when both endpoints are in the caller namespace', () => { + const store = new Store(':memory:') + const handoff = store.workGraph.insertEvent('default', { + source_kind: 'session', + source_ref: 'a', + event_type: 'handoff', + principal: humanPrincipal + }) + const receipt = store.workGraph.insertEvent('default', { + source_kind: 'session', + source_ref: 'b', + event_type: 'handoff_receipt', + principal: humanPrincipal + }) + const link = store.workGraph.insertLink('default', { + from_event_id: receipt.event.id, + to_event_id: handoff.event.id, + relation_type: 'resolves' + }) + expect(link.relationType).toBe('resolves') + expect(store.workGraph.listLinksForEvent('default', handoff.event.id)).toHaveLength(1) + }) + + it('accepts agent principal with on_behalf_of human owner', () => { + const store = new Store(':memory:') + const result = store.workGraph.insertEvent('default', { + source_kind: 'session', + source_ref: 'worker-1', + event_type: 'work_ad', + principal: { kind: 'agent', id: 'worker-1', on_behalf_of: '1' } + }) + expect(result.inserted).toBe(true) + expect(result.event.principal).toEqual({ + kind: 'agent', + id: 'worker-1', + on_behalf_of: '1' + }) + }) +}) diff --git a/hub/src/store/workGraph.ts b/hub/src/store/workGraph.ts new file mode 100644 index 00000000..7089a0b8 --- /dev/null +++ b/hub/src/store/workGraph.ts @@ -0,0 +1,338 @@ +import type { Database } from 'bun:sqlite' +import { randomUUID } from 'node:crypto' +import { + type WorkGraphArtifactRef, + type WorkGraphEvent, + type WorkGraphEventCreate, + WorkGraphEventCreateSchema, + type WorkGraphEventLink, + type WorkGraphEventLinkCreate, + type WorkGraphPrincipal, + WorkGraphPrincipalSchema, + isPrincipalAccountable +} from '@hapi/protocol' + +export class WorkGraphPrincipalError extends Error { + readonly code = 'principal_refused' as const + + constructor(message: string) { + super(message) + this.name = 'WorkGraphPrincipalError' + } +} + +export class WorkGraphValidationError extends Error { + readonly code = 'invalid_event' as const + + constructor(message: string) { + super(message) + this.name = 'WorkGraphValidationError' + } +} + +export class WorkGraphNotFoundError extends Error { + readonly code = 'not_found' as const + + constructor(message: string) { + super(message) + this.name = 'WorkGraphNotFoundError' + } +} + +type EventRow = { + id: string + ts: number + source_kind: string + source_ref: string + sink_kind: string | null + sink_ref: string | null + event_type: string + summary: string | null + payload_json: string | null + artifact_refs: string + tags: string + related_session_id: string | null + related_event_id: string | null + provenance: string | null + idempotency_key: string | null + dedupe_key: string | null + confidence: number | null + severity: string | null + expires_at: number | null + namespace: string + principal_json: string +} + +type LinkRow = { + id: string + from_event_id: string + to_event_id: string + relation_type: string + created_at: number + metadata_json: string | null + namespace: string +} + +function parseJsonArray(raw: string, fallback: T[]): T[] { + try { + const parsed = JSON.parse(raw) as unknown + return Array.isArray(parsed) ? parsed as T[] : fallback + } catch { + return fallback + } +} + +function parseJsonUnknown(raw: string | null): unknown | null { + if (raw === null) return null + try { + return JSON.parse(raw) as unknown + } catch { + return null + } +} + +function toEvent(row: EventRow): WorkGraphEvent { + return { + id: row.id, + ts: row.ts, + sourceKind: row.source_kind, + sourceRef: row.source_ref, + sinkKind: row.sink_kind, + sinkRef: row.sink_ref, + eventType: row.event_type, + summary: row.summary, + payloadJson: parseJsonUnknown(row.payload_json), + artifactRefs: parseJsonArray(row.artifact_refs, []), + tags: parseJsonArray(row.tags, []), + relatedSessionId: row.related_session_id, + relatedEventId: row.related_event_id, + provenance: row.provenance, + idempotencyKey: row.idempotency_key, + dedupeKey: row.dedupe_key, + confidence: row.confidence, + severity: row.severity, + expiresAt: row.expires_at, + namespace: row.namespace, + principal: parsePrincipal(row.principal_json) + } +} + +function parsePrincipal(raw: string): WorkGraphPrincipal { + const parsed = parseJsonUnknown(raw) + const result = WorkGraphPrincipalSchema.safeParse(parsed) + if (result.success) { + return result.data + } + // Corrupt row must not 500 the whole list endpoint. + return { kind: 'human', id: 'invalid' } +} + +function toLink(row: LinkRow): WorkGraphEventLink { + return { + id: row.id, + fromEventId: row.from_event_id, + toEventId: row.to_event_id, + relationType: row.relation_type, + createdAt: row.created_at, + metadataJson: parseJsonUnknown(row.metadata_json), + namespace: row.namespace + } +} + +export type InsertWorkGraphEventResult = { + event: WorkGraphEvent + inserted: boolean +} + +export function insertWorkGraphEvent( + db: Database, + namespace: string, + input: WorkGraphEventCreate, + options?: { id?: string; ts?: number } +): InsertWorkGraphEventResult { + // All writers (HTTP + notify ingest + future callers) share one ledger + // bound: schema max lengths / payload size. Typed callers can still + // bypass TypeScript with forged objects; this is the choke point. + const parsed = WorkGraphEventCreateSchema.safeParse(input) + if (!parsed.success) { + throw new WorkGraphValidationError( + parsed.error.issues[0]?.message ?? 'Invalid work-graph event' + ) + } + const eventInput = parsed.data + + if (!isPrincipalAccountable(eventInput.principal)) { + throw new WorkGraphPrincipalError( + 'Non-human principal requires a resolvable human owner via on_behalf_of' + ) + } + + if (eventInput.idempotency_key) { + const existing = db.prepare(` + SELECT * FROM events + WHERE namespace = ? AND idempotency_key = ? + LIMIT 1 + `).get(namespace, eventInput.idempotency_key) as EventRow | undefined + if (existing) { + return { event: toEvent(existing), inserted: false } + } + } + + const id = options?.id ?? randomUUID() + const ts = options?.ts ?? Date.now() + const artifactRefs = JSON.stringify(eventInput.artifact_refs ?? []) + const tags = JSON.stringify(eventInput.tags ?? []) + const payloadJson = eventInput.payload_json === undefined + ? null + : JSON.stringify(eventInput.payload_json) + const principalJson = JSON.stringify(eventInput.principal) + + try { + db.prepare(` + INSERT INTO events ( + id, ts, + source_kind, source_ref, + sink_kind, sink_ref, + event_type, summary, payload_json, + artifact_refs, tags, + related_session_id, related_event_id, + provenance, idempotency_key, dedupe_key, + confidence, severity, expires_at, + namespace, principal_json + ) VALUES ( + @id, @ts, + @source_kind, @source_ref, + @sink_kind, @sink_ref, + @event_type, @summary, @payload_json, + @artifact_refs, @tags, + @related_session_id, @related_event_id, + @provenance, @idempotency_key, @dedupe_key, + @confidence, @severity, @expires_at, + @namespace, @principal_json + ) + `).run({ + id, + ts, + source_kind: eventInput.source_kind, + source_ref: eventInput.source_ref, + sink_kind: eventInput.sink_kind ?? null, + sink_ref: eventInput.sink_ref ?? null, + event_type: eventInput.event_type, + summary: eventInput.summary ?? null, + payload_json: payloadJson, + artifact_refs: artifactRefs, + tags, + related_session_id: eventInput.related_session_id ?? null, + related_event_id: eventInput.related_event_id ?? null, + provenance: eventInput.provenance ?? null, + idempotency_key: eventInput.idempotency_key ?? null, + dedupe_key: eventInput.dedupe_key ?? null, + confidence: eventInput.confidence ?? null, + severity: eventInput.severity ?? null, + expires_at: eventInput.expires_at ?? null, + namespace, + principal_json: principalJson + }) + } catch (error) { + // Race on idempotency unique index: return the winner's row. + if (eventInput.idempotency_key) { + const existing = db.prepare(` + SELECT * FROM events + WHERE namespace = ? AND idempotency_key = ? + LIMIT 1 + `).get(namespace, eventInput.idempotency_key) as EventRow | undefined + if (existing) { + return { event: toEvent(existing), inserted: false } + } + } + throw error + } + + const row = db.prepare('SELECT * FROM events WHERE id = ? AND namespace = ?') + .get(id, namespace) as EventRow + return { event: toEvent(row), inserted: true } +} + +export function getWorkGraphEventByNamespace( + db: Database, + eventId: string, + namespace: string +): WorkGraphEvent | null { + const row = db.prepare('SELECT * FROM events WHERE id = ? AND namespace = ?') + .get(eventId, namespace) as EventRow | undefined + return row ? toEvent(row) : null +} + +export function listWorkGraphEventsByRelatedSession( + db: Database, + namespace: string, + relatedSessionId: string, + options?: { limit?: number } +): WorkGraphEvent[] { + const limit = Math.min(Math.max(options?.limit ?? 100, 1), 500) + const rows = db.prepare(` + SELECT * FROM events + WHERE namespace = ? AND related_session_id = ? + ORDER BY ts DESC + LIMIT ? + `).all(namespace, relatedSessionId, limit) as EventRow[] + return rows.map(toEvent) +} + +export function insertWorkGraphEventLink( + db: Database, + namespace: string, + input: WorkGraphEventLinkCreate, + options?: { id?: string; createdAt?: number } +): WorkGraphEventLink { + const fromEvent = getWorkGraphEventByNamespace(db, input.from_event_id, namespace) + if (!fromEvent) { + throw new WorkGraphNotFoundError('from_event_id not found in namespace') + } + const toEvent = getWorkGraphEventByNamespace(db, input.to_event_id, namespace) + if (!toEvent) { + throw new WorkGraphNotFoundError('to_event_id not found in namespace') + } + + const id = options?.id ?? randomUUID() + const createdAt = options?.createdAt ?? Date.now() + const metadataJson = input.metadata_json === undefined + ? null + : JSON.stringify(input.metadata_json) + + db.prepare(` + INSERT INTO event_links ( + id, from_event_id, to_event_id, relation_type, + created_at, metadata_json, namespace + ) VALUES ( + @id, @from_event_id, @to_event_id, @relation_type, + @created_at, @metadata_json, @namespace + ) + `).run({ + id, + from_event_id: input.from_event_id, + to_event_id: input.to_event_id, + relation_type: input.relation_type, + created_at: createdAt, + metadata_json: metadataJson, + namespace + }) + + const row = db.prepare('SELECT * FROM event_links WHERE id = ? AND namespace = ?') + .get(id, namespace) as LinkRow + return toLink(row) +} + +export function listWorkGraphEventLinksForEvent( + db: Database, + namespace: string, + eventId: string +): WorkGraphEventLink[] { + const rows = db.prepare(` + SELECT * FROM event_links + WHERE namespace = ? + AND (from_event_id = ? OR to_event_id = ?) + ORDER BY created_at ASC + `).all(namespace, eventId, eventId) as LinkRow[] + return rows.map(toLink) +} diff --git a/hub/src/store/workGraphStore.ts b/hub/src/store/workGraphStore.ts new file mode 100644 index 00000000..9ec5541a --- /dev/null +++ b/hub/src/store/workGraphStore.ts @@ -0,0 +1,51 @@ +import type { Database } from 'bun:sqlite' +import type { + WorkGraphEvent, + WorkGraphEventCreate, + WorkGraphEventLink, + WorkGraphEventLinkCreate +} from '@hapi/protocol' +import { + getWorkGraphEventByNamespace, + insertWorkGraphEvent, + insertWorkGraphEventLink, + listWorkGraphEventLinksForEvent, + listWorkGraphEventsByRelatedSession, + type InsertWorkGraphEventResult +} from './workGraph' + +export class WorkGraphStore { + private readonly db: Database + + constructor(db: Database) { + this.db = db + } + + insertEvent( + namespace: string, + input: WorkGraphEventCreate, + options?: { id?: string; ts?: number } + ): InsertWorkGraphEventResult { + return insertWorkGraphEvent(this.db, namespace, input, options) + } + + getEvent(eventId: string, namespace: string): WorkGraphEvent | null { + return getWorkGraphEventByNamespace(this.db, eventId, namespace) + } + + listByRelatedSession( + namespace: string, + relatedSessionId: string, + options?: { limit?: number } + ): WorkGraphEvent[] { + return listWorkGraphEventsByRelatedSession(this.db, namespace, relatedSessionId, options) + } + + insertLink(namespace: string, input: WorkGraphEventLinkCreate): WorkGraphEventLink { + return insertWorkGraphEventLink(this.db, namespace, input) + } + + listLinksForEvent(namespace: string, eventId: string): WorkGraphEventLink[] { + return listWorkGraphEventLinksForEvent(this.db, namespace, eventId) + } +} diff --git a/hub/src/sync/syncEngine.ts b/hub/src/sync/syncEngine.ts index 8cdffa69..92f38fb9 100644 --- a/hub/src/sync/syncEngine.ts +++ b/hub/src/sync/syncEngine.ts @@ -51,6 +51,7 @@ import { type RpcUploadFileResponse } from './rpcGateway' import { SessionCache } from './sessionCache' +import { ingestNotifySummaryFromMessage } from './workGraphNotifyIngest' type PiResumeAttempt = NonNullable['piResumeAttempt']> type PtyResumeAttempt = NonNullable['ptyResumeAttempt']> @@ -185,6 +186,11 @@ export class SyncEngine { private readonly opencodeClearTails = new Map>() /** Serialize fork/rewind per session so concurrent native rollbacks cannot stack. */ private readonly historyActionsInFlight = new Set() + /** + * Hub owner id for accountable work-graph principals (A2A P1/P3). + * Defaults to "1" for unit tests; startHub overwrites with getOrCreateOwnerId(). + */ + private hubOwnerUserId: string = '1' constructor( private readonly store: Store, @@ -206,6 +212,10 @@ export class SyncEngine { this.inactivityTimer = setInterval(() => this.expireInactive(), 5_000) } + setHubOwnerUserId(ownerUserId: string | number): void { + this.hubOwnerUserId = String(ownerUserId) + } + stop(): void { if (this.inactivityTimer) { clearInterval(this.inactivityTimer) @@ -462,7 +472,31 @@ export class SyncEngine { } } + // Emit chat updates before ledger capture so SSE is not blocked on + // synchronous SQLite ingest (cold review m5). this.eventPublisher.emit(event) + + if (event.type === 'message-received' && event.sessionId && 'message' in event && event.message) { + // A2A P3: well-formed AGENT_NOTIFY_SUMMARY → work-graph work_ad. + // Capture is independent of chat display settings (#1462/#1464). + const session = this.getSession(event.sessionId) + if (session) { + try { + ingestNotifySummaryFromMessage({ + store: this.store, + namespace: session.namespace, + sessionId: session.id, + messageId: event.message.id, + content: event.message.content, + ts: event.message.createdAt, + ownerUserId: this.hubOwnerUserId, + flavor: session.metadata?.flavor ?? null + }) + } catch (error) { + console.error('[work-graph] notify ingest failed', error) + } + } + } } handleSessionAlive(payload: { diff --git a/hub/src/sync/syncEngineHandleRealtimeEvent.test.ts b/hub/src/sync/syncEngineHandleRealtimeEvent.test.ts index 6378807f..afbd6d8b 100644 --- a/hub/src/sync/syncEngineHandleRealtimeEvent.test.ts +++ b/hub/src/sync/syncEngineHandleRealtimeEvent.test.ts @@ -1,4 +1,5 @@ import { describe, expect, it } from 'bun:test' +import { WORK_GRAPH_MAX_SUMMARY } from '@hapi/protocol' import type { SessionPatch, SyncEvent } from '@hapi/protocol/types' import { RpcRegistry } from '../socket/rpcRegistry' import { Store } from '../store' @@ -139,3 +140,182 @@ describe('SyncEngine.handleRealtimeEvent dedup-on-metadata-change', () => { expect(dedupCalls).toEqual([created.id]) }) }) + +describe('SyncEngine.handleRealtimeEvent notify → work-graph ingest', () => { + function makeEngine(): { engine: SyncEngine; store: Store } { + const store = new Store(':memory:') + const engine = new SyncEngine( + store, + {} as never, + new RpcRegistry(), + { broadcast() {} } as never + ) + engine.setHubOwnerUserId(7) + return { engine, store } + } + + it('captures AGENT_NOTIFY_SUMMARY from message-received (display cannot gate)', () => { + const { engine, store } = makeEngine() + const session = store.sessions.getOrCreateSession( + 'notify-wiring', + { path: '/tmp', host: 'h', flavor: 'claude' }, + null, + 'default' + ) + // Ensure cache has the session for ingest namespace/flavor lookup. + engine.handleRealtimeEvent({ + type: 'session-updated', + sessionId: session.id + }) + + const messageTs = 1_660_000_000_000 + engine.handleRealtimeEvent({ + type: 'message-received', + sessionId: session.id, + message: { + id: 'msg-wire-1', + seq: 1, + localId: null, + createdAt: messageTs, + content: { + role: 'agent', + content: { + type: 'output', + data: { + type: 'assistant', + message: { + content: [{ + type: 'text', + text: 'Done.\n\nAGENT_NOTIFY_SUMMARY {"status":"done","summary":"Wired through engine","project":"hapi"}' + }] + } + } + } + } + } + }) + + const rows = store.workGraph.listByRelatedSession('default', session.id) + expect(rows).toHaveLength(1) + expect(rows[0]!.summary).toBe('Wired through engine') + expect(rows[0]!.ts).toBe(messageTs) + expect(rows[0]!.provenance).toBe('AGENT_NOTIFY_SUMMARY') + expect(rows[0]!.payloadJson).toMatchObject({ status: 'done' }) + }) + + it('does not ingest when the footer is missing', () => { + const { engine, store } = makeEngine() + const session = store.sessions.getOrCreateSession( + 'notify-wiring-none', + { path: '/tmp', host: 'h', flavor: 'claude' }, + null, + 'default' + ) + engine.handleRealtimeEvent({ type: 'session-updated', sessionId: session.id }) + engine.handleRealtimeEvent({ + type: 'message-received', + sessionId: session.id, + message: { + id: 'msg-wire-none', + seq: 1, + localId: null, + createdAt: Date.now(), + content: { + role: 'agent', + content: { + type: 'output', + data: { + type: 'assistant', + message: { content: [{ type: 'text', text: 'No footer here.' }] } + } + } + } + } + }) + expect(store.workGraph.listByRelatedSession('default', session.id)).toHaveLength(0) + }) + + it('captures notify footers from AGY agy_message envelopes', () => { + const { engine, store } = makeEngine() + const session = store.sessions.getOrCreateSession( + 'notify-wiring-agy', + { path: '/tmp', host: 'h', flavor: 'agy' }, + null, + 'default' + ) + engine.handleRealtimeEvent({ type: 'session-updated', sessionId: session.id }) + engine.handleRealtimeEvent({ + type: 'message-received', + sessionId: session.id, + message: { + id: 'msg-wire-agy', + seq: 1, + localId: null, + createdAt: 1_670_000_000_000, + content: { + role: 'agent', + content: { + type: 'output', + data: { + type: 'agy_message', + content: 'PINGOK\n\nAGENT_NOTIFY_SUMMARY {"status":"done","summary":"AGY footer","agent":"forged-peer"}' + } + } + } + } + }) + + const rows = store.workGraph.listByRelatedSession('default', session.id) + expect(rows).toHaveLength(1) + expect(rows[0]!.summary).toBe('AGY footer') + expect(rows[0]!.principal).toEqual({ + kind: 'agent', + id: `session:${session.id}`, + on_behalf_of: '7' + }) + expect(rows[0]!.payloadJson).toMatchObject({ agent: 'forged-peer' }) + }) + + it('does not persist footer summary beyond WORK_GRAPH_MAX_SUMMARY (S4)', () => { + const { engine, store } = makeEngine() + const session = store.sessions.getOrCreateSession( + 'notify-wiring-bound', + { path: '/tmp', host: 'h', flavor: 'claude' }, + null, + 'default' + ) + engine.handleRealtimeEvent({ type: 'session-updated', sessionId: session.id }) + + const oversized = 'y'.repeat(WORK_GRAPH_MAX_SUMMARY + 1) + engine.handleRealtimeEvent({ + type: 'message-received', + sessionId: session.id, + message: { + id: 'msg-wire-bound', + seq: 1, + localId: null, + createdAt: Date.now(), + content: { + role: 'agent', + content: { + type: 'output', + data: { + type: 'assistant', + message: { + content: [{ + type: 'text', + text: `Done.\n\nAGENT_NOTIFY_SUMMARY {"status":"done","summary":${JSON.stringify(oversized)}}` + }] + } + } + } + } + } + }) + + const rows = store.workGraph.listByRelatedSession('default', session.id) + expect(rows).toHaveLength(1) + expect(rows[0]!.summary?.length).toBe(WORK_GRAPH_MAX_SUMMARY) + expect(rows[0]!.summary?.length).toBeLessThan(oversized.length) + }) +}) diff --git a/hub/src/sync/workGraphNotifyIngest.test.ts b/hub/src/sync/workGraphNotifyIngest.test.ts new file mode 100644 index 00000000..eec0eb1b --- /dev/null +++ b/hub/src/sync/workGraphNotifyIngest.test.ts @@ -0,0 +1,374 @@ +import { describe, expect, it } from 'bun:test' +import { WORK_GRAPH_MAX_STRING, WORK_GRAPH_MAX_SUMMARY } from '@hapi/protocol' +import { Store } from '../store' +import { + WORK_AD_DEFAULT_TTL_MS, + buildWorkAdFromNotify, + ingestNotifySummaryFromMessage, + mapNotifyStatusToWorkAdStatus +} from './workGraphNotifyIngest' + +function assistantOutput(text: string) { + return { + role: 'agent', + content: { + type: 'output', + data: { + type: 'assistant', + message: { + content: [{ type: 'text', text }] + } + } + } + } +} + +describe('mapNotifyStatusToWorkAdStatus', () => { + it('maps notify contract statuses onto RFC WorkAd vocabulary', () => { + expect(mapNotifyStatusToWorkAdStatus('done')).toBe('done') + expect(mapNotifyStatusToWorkAdStatus('blocked')).toBe('blocked') + expect(mapNotifyStatusToWorkAdStatus('needs_decision')).toBe('needs_decision') + expect(mapNotifyStatusToWorkAdStatus('needs_review')).toBe('needs_decision') + expect(mapNotifyStatusToWorkAdStatus('failed')).toBe('failed') + expect(mapNotifyStatusToWorkAdStatus('stalled')).toBe('blocked') + expect(mapNotifyStatusToWorkAdStatus('stale')).toBe('unknown') + expect(mapNotifyStatusToWorkAdStatus('in_progress')).toBe('in_progress') + expect(mapNotifyStatusToWorkAdStatus(undefined)).toBe('unknown') + expect(mapNotifyStatusToWorkAdStatus('weird')).toBe('unknown') + }) +}) + +describe('buildWorkAdFromNotify field mapping', () => { + it('maps notify fields per RFC elevation table', () => { + const ts = 1_700_000_000_000 + const create = buildWorkAdFromNotify({ + sessionId: 'sess-1', + messageId: 'msg-1', + ownerUserId: 42, + flavor: 'claude', + ts, + notify: { + version: 1, + status: 'done', + summary: 'Opened PR', + action: 'review diff', + agent: 'peer-a', + project: 'hapi' + } + }) + + expect(create.event_type).toBe('work_ad') + expect(create.summary).toBe('Opened PR') + expect(create.related_session_id).toBe('sess-1') + expect(create.provenance).toBe('AGENT_NOTIFY_SUMMARY') + expect(create.idempotency_key).toBe('session:sess-1:message:msg-1:notify') + expect(create.expires_at).toBe(ts + WORK_AD_DEFAULT_TTL_MS) + expect(create.principal).toEqual({ + kind: 'agent', + id: 'session:sess-1', + on_behalf_of: '42' + }) + expect(create.tags).toContain('project:hapi') + expect(create.tags).toContain('agent:peer-a') + expect(create.payload_json).toMatchObject({ + status: 'done', + action: 'review diff', + project: 'hapi', + agent: 'peer-a', + messageId: 'msg-1' + }) + }) + + it('never trusts notify.agent as principal.id', () => { + const create = buildWorkAdFromNotify({ + sessionId: 'sess-xyz', + messageId: 'msg-1', + ownerUserId: 1, + ts: 1000, + notify: { status: 'done', summary: 'ok', agent: 'forged-peer' } + }) + expect(create.principal).toEqual({ + kind: 'agent', + id: 'session:sess-xyz', + on_behalf_of: '1' + }) + expect(create.payload_json).toMatchObject({ agent: 'forged-peer' }) + }) +}) + +describe('ingestNotifySummaryFromMessage', () => { + it('inserts a work_ad ledger row from a well-formed footer', () => { + const store = new Store(':memory:') + const session = store.sessions.getOrCreateSession('sess-ingest', {}, null, 'alpha') + const content = assistantOutput( + 'All good.\n\nAGENT_NOTIFY_SUMMARY {"version":1,"status":"done","action":"ship it","summary":"Tests green","agent":"worker","project":"hapi"}' + ) + + const result = ingestNotifySummaryFromMessage({ + store, + namespace: session.namespace, + sessionId: session.id, + messageId: 'msg-a', + content, + ts: Date.now(), + ownerUserId: 7, + flavor: 'claude' + }) + + expect(result?.inserted).toBe(true) + expect(result?.event.eventType).toBe('work_ad') + expect(result?.event.summary).toBe('Tests green') + expect(result?.event.namespace).toBe('alpha') + expect(result?.event.relatedSessionId).toBe(session.id) + expect(result?.event.provenance).toBe('AGENT_NOTIFY_SUMMARY') + expect(result?.event.payloadJson).toMatchObject({ + status: 'done', + action: 'ship it' + }) + + const listed = store.workGraph.listByRelatedSession('alpha', session.id) + expect(listed).toHaveLength(1) + expect(listed[0]!.id).toBe(result!.event.id) + }) + + it('is idempotent for the same message', () => { + const store = new Store(':memory:') + const session = store.sessions.getOrCreateSession('sess-idem', {}, null, 'default') + const content = assistantOutput( + 'AGENT_NOTIFY_SUMMARY {"status":"blocked","summary":"Waiting on review"}' + ) + const input = { + store, + namespace: 'default', + sessionId: session.id, + messageId: 'msg-same', + content, + ts: Date.now(), + ownerUserId: 1 + } + + const first = ingestNotifySummaryFromMessage(input) + const second = ingestNotifySummaryFromMessage(input) + + expect(first?.inserted).toBe(true) + expect(second?.inserted).toBe(false) + expect(second?.event.id).toBe(first?.event.id) + expect(store.workGraph.listByRelatedSession('default', session.id)).toHaveLength(1) + }) + + it('isolates ledger rows by namespace', () => { + const store = new Store(':memory:') + const alpha = store.sessions.getOrCreateSession('sess-ns', {}, null, 'alpha') + const content = assistantOutput( + 'AGENT_NOTIFY_SUMMARY {"status":"failed","summary":"Boom"}' + ) + + const result = ingestNotifySummaryFromMessage({ + store, + namespace: 'alpha', + sessionId: alpha.id, + messageId: 'msg-ns', + content, + ts: Date.now(), + ownerUserId: 1 + }) + + expect(result?.inserted).toBe(true) + expect(store.workGraph.listByRelatedSession('beta', alpha.id)).toHaveLength(0) + expect(store.workGraph.getEvent(result!.event.id, 'beta')).toBeNull() + expect(store.workGraph.listByRelatedSession('alpha', alpha.id)).toHaveLength(1) + }) + + it('ignores messages without a well-formed notify footer', () => { + const store = new Store(':memory:') + const session = store.sessions.getOrCreateSession('sess-none', {}, null, 'default') + const result = ingestNotifySummaryFromMessage({ + store, + namespace: 'default', + sessionId: session.id, + messageId: 'msg-none', + content: assistantOutput('Just prose, no footer.'), + ts: Date.now(), + ownerUserId: 1 + }) + expect(result).toBeNull() + expect(store.workGraph.listByRelatedSession('default', session.id)).toHaveLength(0) + }) + + it('does not require a chat display setting — capture always runs when well-formed', () => { + // Kill criterion: display-off does not block capture. This path has no + // display gate at all; presence of a footer is sufficient. + const store = new Store(':memory:') + const session = store.sessions.getOrCreateSession('sess-capture', {}, null, 'default') + const result = ingestNotifySummaryFromMessage({ + store, + namespace: 'default', + sessionId: session.id, + messageId: 'msg-capture', + content: assistantOutput( + 'AGENT_NOTIFY_SUMMARY {"status":"needs_review","summary":"Please look","action":"review PR"}' + ), + ts: Date.now(), + ownerUserId: 1 + }) + expect(result?.inserted).toBe(true) + expect(result?.event.payloadJson).toMatchObject({ status: 'needs_decision' }) + }) + + it('persists the message timestamp and default expires_at (M2/M3)', () => { + const store = new Store(':memory:') + const session = store.sessions.getOrCreateSession('sess-ts', {}, null, 'default') + const messageTs = 1_650_000_000_000 + const result = ingestNotifySummaryFromMessage({ + store, + namespace: 'default', + sessionId: session.id, + messageId: 'msg-ts', + content: assistantOutput( + 'AGENT_NOTIFY_SUMMARY {"status":"done","summary":"Historical import"}' + ), + ts: messageTs, + ownerUserId: 1 + }) + + expect(result?.inserted).toBe(true) + expect(result?.event.ts).toBe(messageTs) + expect(result?.event.expiresAt).toBe(messageTs + WORK_AD_DEFAULT_TTL_MS) + }) + + it('keeps work_ad rows after session delete (append-only audit, M1)', () => { + const store = new Store(':memory:') + const session = store.sessions.getOrCreateSession('sess-survive', {}, null, 'default') + const result = ingestNotifySummaryFromMessage({ + store, + namespace: 'default', + sessionId: session.id, + messageId: 'msg-survive', + content: assistantOutput( + 'AGENT_NOTIFY_SUMMARY {"status":"done","summary":"Survives delete"}' + ), + ts: Date.now(), + ownerUserId: 1 + }) + expect(result?.inserted).toBe(true) + + expect(store.sessions.deleteSession(session.id, 'default')).toBe(true) + expect(store.sessions.getSession(session.id)).toBeNull() + + const listed = store.workGraph.listByRelatedSession('default', session.id) + expect(listed).toHaveLength(1) + expect(listed[0]!.summary).toBe('Survives delete') + expect(listed[0]!.id).toBe(result!.event.id) + }) + + it('clamps oversized footer summary to WORK_GRAPH_MAX_SUMMARY (S4)', () => { + const store = new Store(':memory:') + const session = store.sessions.getOrCreateSession('sess-bound', {}, null, 'default') + const oversized = 'x'.repeat(WORK_GRAPH_MAX_SUMMARY + 1) + const result = ingestNotifySummaryFromMessage({ + store, + namespace: 'default', + sessionId: session.id, + messageId: 'msg-bound', + content: assistantOutput( + `AGENT_NOTIFY_SUMMARY {"status":"done","summary":${JSON.stringify(oversized)}}` + ), + ts: Date.now(), + ownerUserId: 1 + }) + + expect(result?.inserted).toBe(true) + expect(result?.event.summary?.length).toBe(WORK_GRAPH_MAX_SUMMARY) + expect(store.workGraph.listByRelatedSession('default', session.id)).toHaveLength(1) + }) + + it('elevates max-clamped footer fields without duplicating into payload (bot Minor)', () => { + const store = new Store(':memory:') + const session = store.sessions.getOrCreateSession('sess-budget', {}, null, 'default') + // Three near-max strings previously lived twice (flat + notify_summary) + // and blew the 32 KiB payload cap → silent null. Flat-only must insert. + const fat = 'a'.repeat(6_000) + const result = ingestNotifySummaryFromMessage({ + store, + namespace: 'default', + sessionId: session.id, + messageId: 'msg-budget', + content: assistantOutput( + `AGENT_NOTIFY_SUMMARY ${JSON.stringify({ + status: 'done', + summary: 'ok', + action: fat, + project: fat, + agent: fat + })}` + ), + ts: Date.now(), + ownerUserId: 1 + }) + + expect(result?.inserted).toBe(true) + const payload = result?.event.payloadJson as Record + expect(payload).not.toHaveProperty('notify_summary') + expect(payload?.action).toBe(fat) + expect(store.workGraph.listByRelatedSession('default', session.id)).toHaveLength(1) + }) + + it('clamps CJK footer fields by UTF-8 bytes so elevation still inserts', () => { + const store = new Store(':memory:') + const session = store.sessions.getOrCreateSession('sess-cjk', {}, null, 'default') + // 6k CJK × 3 ≈ 54 KiB UTF-8 if unclamped; must not silent-drop. + const fatCjk = '\u4e2d'.repeat(6_000) + const result = ingestNotifySummaryFromMessage({ + store, + namespace: 'default', + sessionId: session.id, + messageId: 'msg-cjk', + content: assistantOutput( + `AGENT_NOTIFY_SUMMARY ${JSON.stringify({ + status: 'done', + summary: 'ok', + action: fatCjk, + project: fatCjk, + agent: fatCjk + })}` + ), + ts: Date.now(), + ownerUserId: 1 + }) + + expect(result?.inserted).toBe(true) + const action = (result?.event.payloadJson as { action?: string })?.action ?? '' + expect(new TextEncoder().encode(action).byteLength).toBeLessThanOrEqual(WORK_GRAPH_MAX_STRING) + expect(store.workGraph.listByRelatedSession('default', session.id)).toHaveLength(1) + }) + + it('clamps JSON-escapable ASCII so elevation still inserts', () => { + const store = new Store(':memory:') + const session = store.sessions.getOrCreateSession('sess-esc', {}, null, 'default') + // Raw UTF-8 clamp would keep 8192 backslashes; JSON.stringify doubles them. + const fatEsc = '\\'.repeat(8_192) + const result = ingestNotifySummaryFromMessage({ + store, + namespace: 'default', + sessionId: session.id, + messageId: 'msg-esc', + content: assistantOutput( + `AGENT_NOTIFY_SUMMARY ${JSON.stringify({ + status: 'done', + summary: 'ok', + action: fatEsc, + project: fatEsc, + agent: fatEsc + })}` + ), + ts: Date.now(), + ownerUserId: 1 + }) + + expect(result?.inserted).toBe(true) + const action = (result?.event.payloadJson as { action?: string })?.action ?? '' + const escaped = JSON.stringify(action).slice(1, -1) + expect(new TextEncoder().encode(escaped).byteLength).toBeLessThanOrEqual(WORK_GRAPH_MAX_STRING) + expect(store.workGraph.listByRelatedSession('default', session.id)).toHaveLength(1) + }) +}) diff --git a/hub/src/sync/workGraphNotifyIngest.ts b/hub/src/sync/workGraphNotifyIngest.ts new file mode 100644 index 00000000..4f4959f7 --- /dev/null +++ b/hub/src/sync/workGraphNotifyIngest.ts @@ -0,0 +1,241 @@ +import { + WORK_GRAPH_MAX_STRING, + WORK_GRAPH_MAX_SUMMARY, + WORK_GRAPH_MAX_TAGS, + extractAssistantPlainText, + extractNotifySummary, + unwrapRoleWrappedRecordEnvelope, + type NotifySummary, + type WorkGraphEventCreate +} from '@hapi/protocol' +import type { Store } from '../store' +import { WorkGraphValidationError } from '../store' +import type { InsertWorkGraphEventResult } from '../store/workGraph' + +const WORK_GRAPH_MAX_TAG = 256 +const utf8Encoder = new TextEncoder() + +/** Truncate to a UTF-8 byte budget (event.summary Zod max is char-oriented). */ +function clampUtf8(value: string, maxBytes: number): string { + if (utf8Encoder.encode(value).byteLength <= maxBytes) return value + const chars: string[] = [] + let bytes = 0 + for (const char of value) { + const size = utf8Encoder.encode(char).byteLength + if (bytes + size > maxBytes) break + chars.push(char) + bytes += size + } + return chars.join('') +} + +/** + * Truncate so JSON-escaped UTF-8 bytes stay within budget. + * `payloadJsonWithinLimit` measures JSON.stringify, where `\` / `"` / controls expand. + */ +function clampJsonUtf8(value: string, maxBytes: number): string { + const chars: string[] = [] + let bytes = 0 + for (const char of value) { + const escaped = JSON.stringify(char).slice(1, -1) + const size = utf8Encoder.encode(escaped).byteLength + if (bytes + size > maxBytes) break + chars.push(char) + bytes += size + } + return chars.join('') +} + +function clampJsonUtf8Opt(value: string | undefined, maxBytes: number): string | undefined { + if (value === undefined) return undefined + return clampJsonUtf8(value, maxBytes) +} + +/** WorkAd status vocabulary from the A2A RFC (P3 notify elevation). */ +export const WORK_AD_STATUSES = [ + 'in_progress', + 'blocked', + 'needs_decision', + 'done', + 'failed', + 'stale', + 'unknown' +] as const +export type WorkAdStatus = (typeof WORK_AD_STATUSES)[number] + +/** + * Default work_ad TTL from message timestamp. + * Staleness filtering / reap is P4; this field must still be populated so + * "eventually stale via expires_at" is representable (cold review M2). + */ +export const WORK_AD_DEFAULT_TTL_MS = 24 * 60 * 60 * 1000 + +/** + * Map AGENT_NOTIFY_SUMMARY.status → WorkAd payload status. + * Notify contract uses needs_review / stalled; ledger uses RFC WorkAd vocab. + * + * Policy: RFC `stale` is derived from `expires_at`, never from a fresh + * self-report. `stalled` means the worker believes it is stuck → `blocked`. + * An agent emitting status "stale" is treated as `unknown`. + */ +export function mapNotifyStatusToWorkAdStatus(status: string | undefined): WorkAdStatus { + switch (status) { + case 'done': + return 'done' + case 'blocked': + case 'stalled': + return 'blocked' + case 'needs_decision': + case 'needs_review': + return 'needs_decision' + case 'failed': + return 'failed' + case 'in_progress': + return 'in_progress' + case 'stale': + return 'unknown' + default: + return status && WORK_AD_STATUSES.includes(status as WorkAdStatus) && status !== 'stale' + ? status as WorkAdStatus + : 'unknown' + } +} + +export function buildWorkAdSummaryFromNotify(notify: NotifySummary): string { + const summary = notify.summary?.trim() + if (summary) return summary + const action = notify.action?.trim() + if (action) return action + const status = notify.status?.trim() + if (status) return `Agent status: ${status}` + return 'Agent turn summary' +} + +export type NotifyIngestInput = { + store: Store + namespace: string + sessionId: string + messageId: string + content: unknown + ts: number + /** Authenticated hub owner id; required for accountable agent principal. */ + ownerUserId: string | number + flavor?: string | null +} + +export type NotifyIngestResult = InsertWorkGraphEventResult | null + +function isAgentMessageContent(content: unknown): boolean { + if (content === null || typeof content !== 'object' || Array.isArray(content)) { + return false + } + const record = content as Record + if (record.role === 'agent') return true + if (record.type === 'output' || record.type === 'codex') return true + return false +} + +function buildTags(notify: NotifySummary, flavor: string | null | undefined): string[] { + // Project stays in tags + payload for now. Indexed `project` column / + // project-scoped list query is deferred to #1374 / P4 (cold review M4). + // Tag strings are untrusted footer text — clamp to schema max before insert. + const tags: string[] = ['notify_summary'] + if (notify.project) tags.push(clampJsonUtf8(`project:${notify.project}`, WORK_GRAPH_MAX_TAG)) + if (notify.agent) tags.push(clampJsonUtf8(`agent:${notify.agent}`, WORK_GRAPH_MAX_TAG)) + if (flavor) tags.push(clampJsonUtf8(`flavor:${flavor}`, WORK_GRAPH_MAX_TAG)) + return tags.slice(0, WORK_GRAPH_MAX_TAGS) +} + +/** + * Build the work-graph create payload for a parsed notify footer (RFC P3). + * Pure helper for tests + ingest. + */ +export function buildWorkAdFromNotify(params: { + sessionId: string + messageId: string + notify: NotifySummary + ownerUserId: string | number + /** Message createdAt — also anchors default expires_at. */ + ts: number + flavor?: string | null + expiresAt?: number +}): WorkGraphEventCreate { + const status = mapNotifyStatusToWorkAdStatus(params.notify.status) + // Footer fields are untrusted. Clamp to ledger schema bounds so elevation + // still lands; store insert also validates WorkGraphEventCreateSchema. + const summary = clampUtf8(buildWorkAdSummaryFromNotify(params.notify), WORK_GRAPH_MAX_SUMMARY) + const action = clampJsonUtf8Opt(params.notify.action, WORK_GRAPH_MAX_STRING) ?? null + const project = clampJsonUtf8Opt(params.notify.project, WORK_GRAPH_MAX_STRING) ?? null + const agent = clampJsonUtf8Opt(params.notify.agent, WORK_GRAPH_MAX_STRING) ?? null + // Audit principal is always session-bound. notify.agent is untrusted + // self-label text and stays advisory in payload/tags only. + // Do not nest a full notify_summary copy — duplicating clamped strings + // can blow the 32 KiB payload_json cap and silently drop the ledger row. + return { + source_kind: 'session', + source_ref: params.sessionId, + event_type: 'work_ad', + summary, + payload_json: { + status, + action, + project, + agent, + messageId: params.messageId + }, + tags: buildTags(params.notify, params.flavor), + related_session_id: params.sessionId, + provenance: 'AGENT_NOTIFY_SUMMARY', + idempotency_key: `session:${params.sessionId}:message:${params.messageId}:notify`, + expires_at: params.expiresAt ?? (params.ts + WORK_AD_DEFAULT_TTL_MS), + principal: { + kind: 'agent', + id: `session:${params.sessionId}`, + on_behalf_of: String(params.ownerUserId) + } + } +} + +/** + * On assistant message ingest: well-formed trailing AGENT_NOTIFY_SUMMARY → + * idempotent work_ad row. Invalid/missing footer → no-op (null). + * + * Ledger rows are append-only audit: deleting the related session does not + * delete work_ad rows (cold review M1). + */ +export function ingestNotifySummaryFromMessage(input: NotifyIngestInput): NotifyIngestResult { + if (!isAgentMessageContent(input.content)) { + return null + } + + const agentBody = unwrapRoleWrappedRecordEnvelope(input.content) + const agentContent = agentBody?.role === 'agent' ? agentBody.content : input.content + const plainText = extractAssistantPlainText(agentContent) + if (!plainText) { + return null + } + + const notify = extractNotifySummary(plainText) + if (!notify) { + return null + } + + const create = buildWorkAdFromNotify({ + sessionId: input.sessionId, + messageId: input.messageId, + notify, + ownerUserId: input.ownerUserId, + flavor: input.flavor, + ts: input.ts + }) + + try { + return input.store.workGraph.insertEvent(input.namespace, create, { ts: input.ts }) + } catch (error) { + // Best-effort capture: never break message ingest on ledger bounds. + if (error instanceof WorkGraphValidationError) { + return null + } + throw error + } +} diff --git a/hub/src/web/routes/workGraph.test.ts b/hub/src/web/routes/workGraph.test.ts new file mode 100644 index 00000000..bda5aaa6 --- /dev/null +++ b/hub/src/web/routes/workGraph.test.ts @@ -0,0 +1,193 @@ +import { describe, expect, it } from 'bun:test' +import { Hono } from 'hono' +import { SignJWT } from 'jose' +import type { WebAppEnv } from '../middleware/auth' +import { createAuthMiddleware } from '../middleware/auth' +import { Store } from '../../store' +import { createWorkGraphRoutes } from './workGraph' +import { PROTOCOL_VERSION, WORK_GRAPH_MAX_BODY_BYTES } from '@hapi/protocol' + +const JWT_SECRET = new TextEncoder().encode('test-secret') + +async function authHeaders(namespace: string, userId = 1) { + const token = await new SignJWT({ uid: userId, ns: namespace }) + .setProtectedHeader({ alg: 'HS256' }) + .setIssuedAt() + .setExpirationTime('1h') + .sign(JWT_SECRET) + return { authorization: `Bearer ${token}` } +} + +function createApp(store: Store) { + const app = new Hono() + app.get('/health', (c) => c.json({ + status: 'ok', + protocolVersion: PROTOCOL_VERSION, + capabilities: { workGraph: true } + })) + app.use('/api/*', createAuthMiddleware(JWT_SECRET)) + app.route('/api', createWorkGraphRoutes(store)) + return app +} + +describe('work-graph routes', () => { + it('advertises workGraph capability on /health without auth', async () => { + const store = new Store(':memory:') + const app = createApp(store) + const response = await app.request('/health') + expect(response.status).toBe(200) + const body = await response.json() as { + status: string + protocolVersion: number + capabilities?: { workGraph?: boolean } + } + expect(body.status).toBe('ok') + expect(body.protocolVersion).toBe(PROTOCOL_VERSION) + expect(body.capabilities?.workGraph).toBe(true) + }) + + it('writes and queries events within the authenticated namespace', async () => { + const store = new Store(':memory:') + const app = createApp(store) + const headers = await authHeaders('default') + + const create = await app.request('/api/work-graph/events', { + method: 'POST', + headers: { ...headers, 'content-type': 'application/json' }, + body: JSON.stringify({ + source_kind: 'session', + source_ref: 'sess-1', + event_type: 'work_ad', + related_session_id: 'sess-1', + summary: 'Implementing ledger', + principal: { kind: 'human', id: '1' } + }) + }) + expect(create.status).toBe(201) + const created = await create.json() as { event: { id: string }; inserted: boolean } + expect(created.inserted).toBe(true) + + const list = await app.request('/api/work-graph/events?related_session_id=sess-1', { + headers + }) + expect(list.status).toBe(200) + const listed = await list.json() as { events: Array<{ id: string; summary: string | null }> } + expect(listed.events).toHaveLength(1) + expect(listed.events[0]?.id).toBe(created.event.id) + expect(listed.events[0]?.summary).toBe('Implementing ledger') + }) + + it('does not leak events across namespaces', async () => { + const store = new Store(':memory:') + const app = createApp(store) + const alphaHeaders = await authHeaders('alpha') + const betaHeaders = await authHeaders('beta') + + const create = await app.request('/api/work-graph/events', { + method: 'POST', + headers: { ...alphaHeaders, 'content-type': 'application/json' }, + body: JSON.stringify({ + source_kind: 'session', + source_ref: 'sess-a', + event_type: 'work_ad', + related_session_id: 'sess-a', + summary: 'alpha only', + principal: { kind: 'human', id: '1' } + }) + }) + const created = await create.json() as { event: { id: string } } + + const betaGet = await app.request(`/api/work-graph/events/${created.event.id}`, { + headers: betaHeaders + }) + expect(betaGet.status).toBe(404) + + const betaList = await app.request('/api/work-graph/events?related_session_id=sess-a', { + headers: betaHeaders + }) + const listed = await betaList.json() as { events: unknown[] } + expect(listed.events).toHaveLength(0) + }) + + it('refuses agent principal without matching authenticated owner', async () => { + const store = new Store(':memory:') + const app = createApp(store) + const headers = await authHeaders('default', 1) + + const response = await app.request('/api/work-graph/events', { + method: 'POST', + headers: { ...headers, 'content-type': 'application/json' }, + body: JSON.stringify({ + source_kind: 'session', + source_ref: 'worker', + event_type: 'work_ad', + principal: { kind: 'agent', id: 'worker', on_behalf_of: '999' } + }) + }) + expect(response.status).toBe(403) + }) + + it('idempotent POST returns existing row without duplicate', async () => { + const store = new Store(':memory:') + const app = createApp(store) + const headers = await authHeaders('default') + const body = { + source_kind: 'session', + source_ref: 'sess-1', + event_type: 'work_ad', + related_session_id: 'sess-1', + summary: 'once', + idempotency_key: 'dedupe-me', + principal: { kind: 'human', id: '1' } + } + + const first = await app.request('/api/work-graph/events', { + method: 'POST', + headers: { ...headers, 'content-type': 'application/json' }, + body: JSON.stringify(body) + }) + expect(first.status).toBe(201) + const firstJson = await first.json() as { event: { id: string }; inserted: boolean } + + const second = await app.request('/api/work-graph/events', { + method: 'POST', + headers: { ...headers, 'content-type': 'application/json' }, + body: JSON.stringify({ ...body, summary: 'retry' }) + }) + expect(second.status).toBe(200) + const secondJson = await second.json() as { event: { id: string; summary: string | null }; inserted: boolean } + expect(secondJson.inserted).toBe(false) + expect(secondJson.event.id).toBe(firstJson.event.id) + expect(secondJson.event.summary).toBe('once') + }) + + it('rejects oversized POST bodies with 413', async () => { + const store = new Store(':memory:') + const app = createApp(store) + const headers = await authHeaders('default') + const oversized = 'x'.repeat(WORK_GRAPH_MAX_BODY_BYTES + 1) + const response = await app.request('/api/work-graph/events', { + method: 'POST', + headers: { + ...headers, + 'content-type': 'application/json', + 'content-length': String(oversized.length) + }, + body: oversized + }) + expect(response.status).toBe(413) + }) + + it('rejects fractional list limit with 400 (not SQLite 500)', async () => { + const store = new Store(':memory:') + const app = createApp(store) + const headers = await authHeaders('default') + const response = await app.request( + '/api/work-graph/events?related_session_id=sess-1&limit=1.5', + { headers } + ) + expect(response.status).toBe(400) + const body = await response.json() as { error: string } + expect(body.error).toBe('Invalid limit') + }) +}) diff --git a/hub/src/web/routes/workGraph.ts b/hub/src/web/routes/workGraph.ts new file mode 100644 index 00000000..d8101053 --- /dev/null +++ b/hub/src/web/routes/workGraph.ts @@ -0,0 +1,121 @@ +import { Hono } from 'hono' +import { bodyLimit } from 'hono/body-limit' +import { + WORK_GRAPH_MAX_BODY_BYTES, + WorkGraphEventCreateSchema, + WorkGraphEventLinkCreateSchema, + principalMatchesAuthenticatedOwner +} from '@hapi/protocol' +import type { Store } from '../../store' +import { + WorkGraphNotFoundError, + WorkGraphPrincipalError, + WorkGraphValidationError +} from '../../store' +import type { WebAppEnv } from '../middleware/auth' + +const workGraphBodyLimit = bodyLimit({ + maxSize: WORK_GRAPH_MAX_BODY_BYTES, + onError: (c) => c.json({ error: 'Request body too large' }, 413) +}) + +/** + * Minimal HTTP surface for the A2A work-graph ledger (P1 / #1374). + * Path is intentionally NOT `/api/events` — that route is the SSE stream. + */ +export function createWorkGraphRoutes(store: Store): Hono { + const app = new Hono() + + app.post('/work-graph/events', workGraphBodyLimit, async (c) => { + const json = await c.req.json().catch(() => null) + const parsed = WorkGraphEventCreateSchema.safeParse(json) + if (!parsed.success) { + return c.json({ error: 'Invalid body', issues: parsed.error.flatten() }, 400) + } + + const namespace = c.get('namespace') + const userId = c.get('userId') + if (!principalMatchesAuthenticatedOwner(parsed.data.principal, userId)) { + return c.json({ + error: 'Principal must resolve to the authenticated hub owner' + }, 403) + } + + try { + const result = store.workGraph.insertEvent(namespace, parsed.data) + return c.json({ + event: result.event, + inserted: result.inserted + }, result.inserted ? 201 : 200) + } catch (error) { + if (error instanceof WorkGraphPrincipalError) { + return c.json({ error: error.message, code: error.code }, 403) + } + if (error instanceof WorkGraphValidationError) { + return c.json({ error: error.message, code: error.code }, 400) + } + throw error + } + }) + + app.get('/work-graph/events', (c) => { + const namespace = c.get('namespace') + const relatedSessionId = c.req.query('related_session_id') + if (!relatedSessionId) { + return c.json({ error: 'related_session_id query parameter is required' }, 400) + } + const limitRaw = c.req.query('limit') + const limit = limitRaw ? Number(limitRaw) : undefined + // Integer only — fractional LIMIT (e.g. 1.5) is a SQLite error → 500. + if (limit !== undefined && (!Number.isInteger(limit) || limit < 1)) { + return c.json({ error: 'Invalid limit' }, 400) + } + const events = store.workGraph.listByRelatedSession(namespace, relatedSessionId, { limit }) + c.header('Cache-Control', 'no-store') + return c.json({ events }) + }) + + app.get('/work-graph/events/:eventId', (c) => { + const namespace = c.get('namespace') + const eventId = c.req.param('eventId') + const event = store.workGraph.getEvent(eventId, namespace) + if (!event) { + return c.json({ error: 'Event not found' }, 404) + } + c.header('Cache-Control', 'no-store') + return c.json({ event }) + }) + + app.post('/work-graph/event-links', workGraphBodyLimit, async (c) => { + const json = await c.req.json().catch(() => null) + const parsed = WorkGraphEventLinkCreateSchema.safeParse(json) + if (!parsed.success) { + return c.json({ error: 'Invalid body', issues: parsed.error.flatten() }, 400) + } + + const namespace = c.get('namespace') + try { + const link = store.workGraph.insertLink(namespace, parsed.data) + return c.json({ link }, 201) + } catch (error) { + if (error instanceof WorkGraphNotFoundError) { + return c.json({ error: error.message, code: error.code }, 404) + } + throw error + } + }) + + app.get('/work-graph/events/:eventId/links', (c) => { + const namespace = c.get('namespace') + const eventId = c.req.param('eventId') + const event = store.workGraph.getEvent(eventId, namespace) + if (!event) { + return c.json({ error: 'Event not found' }, 404) + } + const links = store.workGraph.listLinksForEvent(namespace, eventId) + c.header('Cache-Control', 'no-store') + return c.json({ links }) + }) + + return app +} diff --git a/hub/src/web/server.ts b/hub/src/web/server.ts index 3af9dcb7..312dac63 100644 --- a/hub/src/web/server.ts +++ b/hub/src/web/server.ts @@ -30,6 +30,7 @@ import { createPushRoutes } from './routes/push' import { createDevicesRoutes } from './routes/devices' import { createVoiceRoutes } from './routes/voice' import { createHubSettingsRoutes } from './routes/hubSettings' +import { createWorkGraphRoutes } from './routes/workGraph' import type { SSEManager } from '../sse/sseManager' import type { VisibilityTracker } from '../visibility/visibilityTracker' import type { Server as BunServer, ServerWebSocket } from 'bun' @@ -231,8 +232,13 @@ function createWebApp(options: { app.use('*', logger()) - // Health check endpoint (no auth required) - app.get('/health', (c) => c.json({ status: 'ok', protocolVersion: PROTOCOL_VERSION })) + // Health check endpoint (no auth required). + // capabilities.workGraph is additive: old clients ignore unknown fields. + app.get('/health', (c) => c.json({ + status: 'ok', + protocolVersion: PROTOCOL_VERSION, + capabilities: { workGraph: true } + })) const configuration = getConfiguration() const corsOrigins = options.corsOrigins ?? configuration.corsOrigins @@ -295,6 +301,8 @@ function createWebApp(options: { app.route('/api', createPushRoutes(options.store, options.vapidPublicKey)) app.route('/api', createDevicesRoutes(options.store)) app.route('/api', createVoiceRoutes({ dataDir: configuration.dataDir })) + // Path is intentionally NOT `/api/events` — that route is the SSE stream. + app.route('/api', createWorkGraphRoutes(options.store)) // Skip static serving in relay mode, show helpful message on root if (options.relayMode) { diff --git a/shared/src/index.ts b/shared/src/index.ts index 7d182039..714001a0 100644 --- a/shared/src/index.ts +++ b/shared/src/index.ts @@ -1,4 +1,5 @@ export * from './scratchlistAttachments' +export * from './workGraph' export * from './apiTypes' export * from './cursorCliSku' export * from './messages' diff --git a/shared/src/messages.test.ts b/shared/src/messages.test.ts index cd04a131..bc1bf6cd 100644 --- a/shared/src/messages.test.ts +++ b/shared/src/messages.test.ts @@ -87,6 +87,21 @@ describe('extractAssistantPlainText', () => { expect(extractAssistantPlainText(content)).toBeNull() }) + test('extracts AGY agy_message prose', () => { + const content = { + type: 'output', + data: { type: 'agy_message', content: 'PINGOK\n\nAGENT_NOTIFY_SUMMARY {"status":"done","summary":"ok"}' } + } + expect(extractAssistantPlainText(content)).toContain('AGENT_NOTIFY_SUMMARY') + }) + + test('returns null for empty AGY agy_message', () => { + expect(extractAssistantPlainText({ + type: 'output', + data: { type: 'agy_message', content: ' ' } + })).toBeNull() + }) + test('returns null for unknown content shapes', () => { expect(extractAssistantPlainText({ type: 'event', data: {} })).toBeNull() expect(extractAssistantPlainText({ type: 'text' })).toBeNull() diff --git a/shared/src/messages.ts b/shared/src/messages.ts index fc78a8e7..d60fe6ef 100644 --- a/shared/src/messages.ts +++ b/shared/src/messages.ts @@ -109,7 +109,16 @@ export function extractAssistantPlainText(content: unknown): string | null { if (content.type === 'output') { const data = isObject(content.data) ? content.data : null - if (!data || data.type !== 'assistant') return null + if (!data) return null + + // AGY planner prose (cli wraps PLANNER_RESPONSE as agy_message). + if (data.type === 'agy_message') { + return typeof data.content === 'string' && data.content.trim().length > 0 + ? data.content + : null + } + + if (data.type !== 'assistant') return null const message = isObject(data.message) ? data.message : null const blocks = Array.isArray(message?.content) ? message.content : null if (!blocks) return null diff --git a/shared/src/workGraph.test.ts b/shared/src/workGraph.test.ts new file mode 100644 index 00000000..5c3ca49c --- /dev/null +++ b/shared/src/workGraph.test.ts @@ -0,0 +1,96 @@ +import { describe, expect, it } from 'bun:test' +import { + WORK_GRAPH_MAX_PAYLOAD_JSON_BYTES, + WorkGraphEventCreateSchema, + WorkGraphPrincipalSchema, + isPrincipalAccountable, + principalMatchesAuthenticatedOwner +} from './workGraph' + +describe('WorkGraphPrincipalSchema', () => { + it('accepts human principal without on_behalf_of', () => { + const parsed = WorkGraphPrincipalSchema.safeParse({ kind: 'human', id: '42' }) + expect(parsed.success).toBe(true) + }) + + it('rejects agent without on_behalf_of', () => { + const parsed = WorkGraphPrincipalSchema.safeParse({ kind: 'agent', id: 'session-1' }) + expect(parsed.success).toBe(false) + }) + + it('accepts agent with human owner', () => { + const parsed = WorkGraphPrincipalSchema.safeParse({ + kind: 'agent', + id: 'session-1', + on_behalf_of: '42' + }) + expect(parsed.success).toBe(true) + }) +}) + +describe('principal accountability helpers', () => { + it('refuses non-human with empty owner', () => { + expect(isPrincipalAccountable({ + kind: 'service', + id: 'ci', + on_behalf_of: ' ' + })).toBe(false) + }) + + it('requires human id to match authenticated owner', () => { + expect(principalMatchesAuthenticatedOwner({ kind: 'human', id: '1' }, 1)).toBe(true) + expect(principalMatchesAuthenticatedOwner({ kind: 'human', id: '2' }, 1)).toBe(false) + }) + + it('requires agent on_behalf_of to match authenticated owner', () => { + expect(principalMatchesAuthenticatedOwner({ + kind: 'agent', + id: 'worker', + on_behalf_of: '1' + }, 1)).toBe(true) + expect(principalMatchesAuthenticatedOwner({ + kind: 'agent', + id: 'worker', + on_behalf_of: '99' + }, 1)).toBe(false) + }) +}) + +describe('WorkGraphEventCreateSchema bounds', () => { + const base = { + source_kind: 'session', + source_ref: 'sess-1', + event_type: 'work_ad', + principal: { kind: 'human' as const, id: '1' } + } + + it('rejects oversized payload_json', () => { + const parsed = WorkGraphEventCreateSchema.safeParse({ + ...base, + payload_json: { blob: 'x'.repeat(WORK_GRAPH_MAX_PAYLOAD_JSON_BYTES) } + }) + expect(parsed.success).toBe(false) + }) + + it('rejects payload_json that fits UTF-16 length but exceeds UTF-8 bytes', () => { + // CJK is 1 UTF-16 code unit / 3 UTF-8 bytes. ~12k chars stays under + // string.length of 32 KiB but over UTF-8 byte budget. + const cjk = '\u4e2d'.repeat(12_000) + const json = JSON.stringify({ blob: cjk }) + expect(json.length).toBeLessThanOrEqual(WORK_GRAPH_MAX_PAYLOAD_JSON_BYTES) + expect(new TextEncoder().encode(json).byteLength).toBeGreaterThan(WORK_GRAPH_MAX_PAYLOAD_JSON_BYTES) + const parsed = WorkGraphEventCreateSchema.safeParse({ + ...base, + payload_json: { blob: cjk } + }) + expect(parsed.success).toBe(false) + }) + + it('rejects too many tags', () => { + const parsed = WorkGraphEventCreateSchema.safeParse({ + ...base, + tags: Array.from({ length: 33 }, (_, i) => `t${i}`) + }) + expect(parsed.success).toBe(false) + }) +}) diff --git a/shared/src/workGraph.ts b/shared/src/workGraph.ts new file mode 100644 index 00000000..494aee4e --- /dev/null +++ b/shared/src/workGraph.ts @@ -0,0 +1,155 @@ +import { z } from 'zod' + +/** + * Structured principal for A2A work-graph ledger writes (RFC Layer 1 / P1). + * Non-human principals must name an accountable human owner via on_behalf_of. + */ +export const WorkGraphPrincipalSchema = z.discriminatedUnion('kind', [ + z.object({ + kind: z.literal('human'), + id: z.string().min(1).max(256) + }), + z.object({ + kind: z.literal('agent'), + id: z.string().min(1).max(256), + on_behalf_of: z.string().min(1).max(256) + }), + z.object({ + kind: z.literal('service'), + id: z.string().min(1).max(256), + on_behalf_of: z.string().min(1).max(256) + }) +]) +export type WorkGraphPrincipal = z.infer + +/** Caps for HTTP write bodies (paired with route bodyLimit). */ +export const WORK_GRAPH_MAX_STRING = 8_192 +export const WORK_GRAPH_MAX_SUMMARY = 2_048 +export const WORK_GRAPH_MAX_TAGS = 32 +export const WORK_GRAPH_MAX_ARTIFACT_REFS = 32 +export const WORK_GRAPH_MAX_PAYLOAD_JSON_BYTES = 32 * 1024 +/** Hono bodyLimit for POST /work-graph/* (JSON + overhead). */ +export const WORK_GRAPH_MAX_BODY_BYTES = 64 * 1024 + +export const WorkGraphArtifactRefSchema = z.object({ + kind: z.string().min(1).max(128), + url: z.string().max(WORK_GRAPH_MAX_STRING).optional(), + title: z.string().max(512).optional(), + ref: z.string().max(WORK_GRAPH_MAX_STRING).nullable().optional(), + source: z.string().max(128).optional(), + created_at: z.number().optional() +}).passthrough() +export type WorkGraphArtifactRef = z.infer + +/** Open vocabulary; common Layer 1 types listed for docs, not enforced as enum. */ +export const WORK_GRAPH_EVENT_TYPES = ['work_ad', 'handoff', 'handoff_receipt'] as const + +const utf8Encoder = new TextEncoder() + +function payloadJsonWithinLimit(value: unknown): boolean { + if (value === undefined) return true + try { + const json = JSON.stringify(value) + // Persist/wire as UTF-8 — string.length is UTF-16 code units and + // under-counts CJK/emoji relative to stored bytes. + return utf8Encoder.encode(json).byteLength <= WORK_GRAPH_MAX_PAYLOAD_JSON_BYTES + } catch { + return false + } +} + +export const WorkGraphEventCreateSchema = z.object({ + source_kind: z.string().min(1).max(128), + source_ref: z.string().min(1).max(WORK_GRAPH_MAX_STRING), + sink_kind: z.string().min(1).max(128).optional(), + sink_ref: z.string().min(1).max(WORK_GRAPH_MAX_STRING).optional(), + event_type: z.string().min(1).max(128), + summary: z.string().max(WORK_GRAPH_MAX_SUMMARY).optional(), + payload_json: z.unknown().optional().refine(payloadJsonWithinLimit, { + message: `payload_json exceeds ${WORK_GRAPH_MAX_PAYLOAD_JSON_BYTES} bytes` + }), + artifact_refs: z.array(WorkGraphArtifactRefSchema).max(WORK_GRAPH_MAX_ARTIFACT_REFS).optional(), + tags: z.array(z.string().max(256)).max(WORK_GRAPH_MAX_TAGS).optional(), + related_session_id: z.string().min(1).max(256).optional(), + related_event_id: z.string().min(1).max(256).optional(), + provenance: z.string().max(512).optional(), + idempotency_key: z.string().min(1).max(512).optional(), + dedupe_key: z.string().min(1).max(512).optional(), + confidence: z.number().optional(), + severity: z.string().max(64).optional(), + expires_at: z.number().optional(), + principal: WorkGraphPrincipalSchema +}) +export type WorkGraphEventCreate = z.infer + +export const WorkGraphEventLinkCreateSchema = z.object({ + from_event_id: z.string().min(1).max(256), + to_event_id: z.string().min(1).max(256), + relation_type: z.string().min(1).max(128), + metadata_json: z.unknown().optional().refine(payloadJsonWithinLimit, { + message: `metadata_json exceeds ${WORK_GRAPH_MAX_PAYLOAD_JSON_BYTES} bytes` + }) +}) +export type WorkGraphEventLinkCreate = z.infer + +export type WorkGraphEvent = { + id: string + ts: number + sourceKind: string + sourceRef: string + sinkKind: string | null + sinkRef: string | null + eventType: string + summary: string | null + payloadJson: unknown | null + artifactRefs: WorkGraphArtifactRef[] + tags: string[] + relatedSessionId: string | null + relatedEventId: string | null + provenance: string | null + idempotencyKey: string | null + dedupeKey: string | null + confidence: number | null + severity: string | null + expiresAt: number | null + namespace: string + principal: WorkGraphPrincipal +} + +export type WorkGraphEventLink = { + id: string + fromEventId: string + toEventId: string + relationType: string + createdAt: number + metadataJson: unknown | null + namespace: string +} + +/** + * Kill criterion from the A2A RFC: a non-human principal with no resolvable + * human owner must be refused. Schema already requires on_behalf_of for + * agent/service; this is the shared runtime check for callers that build + * principals without going through Zod. + */ +export function isPrincipalAccountable(principal: WorkGraphPrincipal): boolean { + if (principal.kind === 'human') { + return principal.id.trim().length > 0 + } + return typeof principal.on_behalf_of === 'string' && principal.on_behalf_of.trim().length > 0 +} + +/** + * Hub HTTP writes: the accountable human must match the authenticated user. + * Humans write as themselves; agents/services write only under that owner. + */ +export function principalMatchesAuthenticatedOwner( + principal: WorkGraphPrincipal, + ownerUserId: number | string +): boolean { + const ownerId = String(ownerUserId) + if (principal.kind === 'human') { + return principal.id === ownerId + } + return principal.on_behalf_of === ownerId +}