feat: let agents read supervisor story graphs
Expose saved decisions, timelines and source evidence through read-only Story Graph tools across Model, Harness and local/remote OpenCode and Continue. Supervisor enablement and runtime assignments govern discovery and calls; reads do not invoke models or modify stored evidence. Remote use requires matching Desktop and Agent changes. Make Continue discussion show the complete editable message with a separated destination and shared action layout. Sending uses the visible draft without reattaching removed context. Validation: existing focused MCP, runtime, controlled Linux Host, discussion component and Electron layout checks plus typechecks; staged diff check passed. Cross-scope fact reuse, historical reconstruction and full package acceptance remain deferred. Release note: Agents can consult saved supervisor decisions and their sources when continuing work. Continue discussion now lets you review and edit the complete message before sending.
This commit is contained in:
@@ -8742,6 +8742,10 @@ export class AssistantDatabase {
|
||||
}
|
||||
}
|
||||
|
||||
readStoryGraph(name: import('../../shared/story-graph-tools').StoryGraphToolName, input: unknown, projectId?: string, signal?: AbortSignal): Record<string, unknown> {
|
||||
return readStoryGraph(this.requireDatabase(), name, input, projectId, signal)
|
||||
}
|
||||
|
||||
getSupervisionResult(id: string): { id: string; summary: string } | undefined {
|
||||
return this.requireDatabase().prepare('SELECT id, summary FROM supervision_results WHERE id = ?').get(id) as { id: string; summary: string } | undefined
|
||||
}
|
||||
@@ -11488,3 +11492,4 @@ export class AssistantDatabase {
|
||||
}
|
||||
import type { ImageOperation } from '../../shared/image-generation-contracts'
|
||||
import { SupervisionReviewStore, supervisionReviewMigration } from './supervision-review-store'
|
||||
import { readStoryGraph } from './story-graph-reader'
|
||||
|
||||
@@ -0,0 +1,138 @@
|
||||
// @vitest-environment node
|
||||
import { randomUUID } from 'node:crypto'
|
||||
import { mkdtemp, rm } from 'node:fs/promises'
|
||||
import { tmpdir } from 'node:os'
|
||||
import { join } from 'node:path'
|
||||
import { DatabaseSync } from 'node:sqlite'
|
||||
import { afterEach, expect, it } from 'vitest'
|
||||
import { AssistantDatabase } from './assistant-database'
|
||||
import type { StoredSupervisionResult } from './supervisor-service'
|
||||
import type { StoryGraphToolName } from '../../shared/story-graph-tools'
|
||||
import { readStoryGraph } from './story-graph-reader'
|
||||
|
||||
const cleanups: Array<() => void | Promise<void>> = []
|
||||
afterEach(async () => { for (const cleanup of cleanups.splice(0).reverse()) await cleanup() })
|
||||
type Page = { items: Array<{ object_ref: { type: string; id: string }; result_id: string | null; event_time: string | null;
|
||||
content: string; source_reference_ids: string[]; validity: string }>; page: { next_cursor: string | null; has_more: boolean; total_count: number } }
|
||||
|
||||
async function fixture() {
|
||||
const root = await mkdtemp(join(tmpdir(), 'story-graph-'))
|
||||
cleanups.push(() => rm(root, { recursive: true, force: true }))
|
||||
const path = join(root, 'assistant.sqlite')
|
||||
const db = new AssistantDatabase(path)
|
||||
db.initialize(root)
|
||||
cleanups.push(() => db.close())
|
||||
const sql = new DatabaseSync(path)
|
||||
cleanups.push(() => sql.close())
|
||||
const project = db.listProjects()[0]!
|
||||
const other = db.createProject({ name: 'Other', description: '', rootPath: root, defaultWorkMode: 'ask' })
|
||||
const messageId = randomUUID(), conversationId = randomUUID()
|
||||
const body = 'Choose A. Then explicitly revise A to B because of measured latency. Option C is only a proposal. '
|
||||
const update = (content: string, projectId = project.id) => db.saveLocalConversations([{ header: { id: conversationId, projectId, title: 'Decision', updatedAt: 1 },
|
||||
messages: [{ id: messageId, role: 'user', content, createdAt: Date.parse('2026-09-01T00:00:00Z'), state: 'complete' }] }])
|
||||
update(body)
|
||||
const request: StoredSupervisionResult['request'] = { trigger: 'manual', scope: { kind: 'projects', projectIds: [project.id] },
|
||||
timeRange: { from: '2026-09-01T00:00:00Z', to: '2026-09-30T00:00:00Z' } }
|
||||
const save = (description: string, occurredAt: string, persistedId?: string) => {
|
||||
const result: StoredSupervisionResult = { request, candidates: db.listSupervisionCandidates(request), evidence: [{ id: 'quote',
|
||||
sourceType: 'conversation', sourceId: conversationId, title: 'Decision evidence', content: body,
|
||||
occurredAt: '2026-09-01T00:00:00Z', locator: { source: `message:${messageId}`, projectId: project.id, conversationId, messageId, start: 0, end: Array.from(body).length, length: Array.from(body).length } }],
|
||||
output: { summary: description, changeDigest: '', openItems: ['Execution still pending'],
|
||||
entities: [{ id: 'choice', label: 'Decision', description, sourceReferenceIds: ['quote'], ...(persistedId ? { persistedId } : {}) }],
|
||||
events: [{ title: description, description, occurredAt, eventType: 'decision', entityIds: ['choice'], sourceReferenceIds: ['quote'] }],
|
||||
entityChanges: [{ entityId: 'choice', changeType: persistedId ? 'revised' : 'proposed', description, sourceReferenceIds: ['quote'] }], relations: [] } }
|
||||
db.saveSupervisionResult(result)
|
||||
return db.listSupervisionCandidates(request)[0]!.id
|
||||
}
|
||||
const read = (name: StoryGraphToolName, input: Record<string, unknown>) => readStoryGraph(sql, name, input, project.id)
|
||||
return { db, sql, read, save, project, other, request, update, body }
|
||||
}
|
||||
|
||||
it('retains A, explicit revision B and proposal C without latest-wins, with event-time order and complete nested pagination', async () => {
|
||||
const f = await fixture()
|
||||
const id = f.save('Decision A', '2026-09-01T00:00:00Z')
|
||||
f.save('Decision B explicitly revises A; execution pending', '2026-09-03T00:00:00Z', id)
|
||||
f.save('Decision C is an option, not a replacement', '2026-09-04T00:00:00Z', id)
|
||||
f.save('Decision late-recorded discussion', '2026-09-02T09:00:00+09:00', id)
|
||||
const input = { object_ref: { type: 'entity', id }, mode: 'timeline', page_size: 2, content_page_size: 71 }
|
||||
let cursor: string | null = null
|
||||
const fragments = new Map<string, string>()
|
||||
const eventTimes = new Map<string, string>()
|
||||
do {
|
||||
const result = f.read('story_graph_get_context', { ...input, ...(cursor ? { cursor } : {}) }) as Page
|
||||
for (const item of result.items) {
|
||||
const key = `${item.object_ref.type}:${item.object_ref.id}:${item.result_id}`
|
||||
fragments.set(key, (fragments.get(key) ?? '') + item.content)
|
||||
if (item.event_time) eventTimes.set(item.object_ref.id, item.event_time)
|
||||
expect(item.validity).toBe('unknown')
|
||||
}
|
||||
cursor = result.page.next_cursor
|
||||
} while (cursor)
|
||||
const data = [...fragments.values()].map(text => JSON.parse(text))
|
||||
expect(data.filter(item => item.canonical_label).map(item => item.description)).toEqual(expect.arrayContaining([
|
||||
'Decision A', 'Decision B explicitly revises A; execution pending', 'Decision C is an option, not a replacement'
|
||||
]))
|
||||
expect([...eventTimes.values()]).toEqual(['2026-09-01T00:00:00.000Z', '2026-09-02T00:00:00.000Z', '2026-09-03T00:00:00.000Z', '2026-09-04T00:00:00.000Z'])
|
||||
expect(data.some(item => item.content?.includes('measured latency'))).toBe(true)
|
||||
const search = f.read('story_graph_search', { query: 'Decision', object_types: ['event'], time_range: { from: '2026-09-02T09:00:00+09:00', to: '2026-09-02T00:00:00Z' } }) as Page
|
||||
expect(search.items).toHaveLength(1)
|
||||
expect(search.items[0]!.event_time).toBe('2026-09-02T00:00:00.000Z')
|
||||
})
|
||||
|
||||
it('enforces exact scope, rejects as_of and stale/changed cursors, and never writes during reads', async () => {
|
||||
const f = await fixture()
|
||||
const id = f.save('Decision A', '2026-09-01T00:00:00Z')
|
||||
const input = { query: 'Decision', page_size: 1 }
|
||||
const first = f.read('story_graph_search', input) as Page
|
||||
const before = f.sql.prepare('SELECT total_changes() AS n').get()!.n
|
||||
f.sql.exec('PRAGMA query_only = ON')
|
||||
expect(f.db.readStoryGraph('story_graph_search', input, f.other.id)).toMatchObject({ items: [], coverage: { status: 'unknown' } })
|
||||
expect(f.read('story_graph_search', { ...input, scope: { kind: 'global' } })).toMatchObject({ items: [] })
|
||||
expect(() => f.db.readStoryGraph('story_graph_get_context', { object_ref: { type: 'entity', id } }, f.other.id)).toThrow('scope_mismatch')
|
||||
expect(() => f.read('story_graph_get_context', { object_ref: { type: 'entity', id: 'missing' } })).toThrow('object_not_found')
|
||||
expect(() => f.db.readStoryGraph('story_graph_search', input)).toThrow('scope_required')
|
||||
expect(() => f.read('story_graph_search', { ...input, as_of: '2026-09-01' })).toThrow('unsupported_mode')
|
||||
expect(() => f.read('story_graph_get_context', { object_ref: { type: 'entity', id }, mode: 'as_of' })).toThrow('unsupported_mode')
|
||||
expect(() => f.read('story_graph_search', { ...input, query: 'Other', cursor: first.page.next_cursor })).toThrow('invalid_cursor')
|
||||
expect(f.sql.prepare('SELECT total_changes() AS n').get()!.n).toBe(before)
|
||||
f.sql.exec('PRAGMA query_only = OFF')
|
||||
f.db.applySupervisionEntityAction({ entityId: id, action: 'revise', description: 'Human confirmed preference' })
|
||||
expect(() => f.read('story_graph_search', { ...input, cursor: first.page.next_cursor })).toThrow('stale_cursor')
|
||||
const abort = new AbortController(); abort.abort()
|
||||
expect(() => f.db.readStoryGraph('story_graph_search', input, f.project.id, abort.signal)).toThrow()
|
||||
})
|
||||
|
||||
it('pages Unicode snapshots, reads explicit current versions, and reports deletion and moved scope without substituting history', async () => {
|
||||
const f = await fixture()
|
||||
f.save('Decision A', '2026-09-01T00:00:00Z')
|
||||
const source = f.sql.prepare('SELECT id FROM supervision_sources').get()!
|
||||
f.sql.prepare('UPDATE supervision_sources SET content = ? WHERE id = ?').run('A\u{1f600}B\u{1f601}C', source.id!)
|
||||
const input = { source_reference_id: source.id, page_size: 2 }
|
||||
let cursor: string | null = null, text = ''
|
||||
do {
|
||||
const page = f.read('story_graph_read_source', { ...input, ...(cursor ? { cursor } : {}) }) as { content: string; page: { next_cursor: string | null } }
|
||||
text += page.content; cursor = page.page.next_cursor
|
||||
} while (cursor)
|
||||
expect(text).toBe('A\u{1f600}B\u{1f601}C')
|
||||
const current = f.read('story_graph_read_source', { source_reference_id: source.id, content_kind: 'current' })
|
||||
expect(current.content).toBe(f.body)
|
||||
// Released incremental references can have messageId without the newer source key.
|
||||
f.sql.prepare("UPDATE supervision_sources SET locator_json = json_remove(locator_json, '$.source') WHERE id = ?").run(source.id!)
|
||||
expect(f.read('story_graph_read_source', { source_reference_id: source.id, content_kind: 'current' })).toMatchObject({ content: f.body })
|
||||
expect(f.read('story_graph_read_source', { source_reference_id: source.id, content_kind: 'version', source_version: current.source_version })).toMatchObject({ content: f.body, content_kind: 'version' })
|
||||
f.update('Changed original')
|
||||
const currentPage = f.read('story_graph_read_source', { ...input, content_kind: 'current' }) as { page: { next_cursor: string } }
|
||||
f.update('Changed again')
|
||||
expect(() => f.read('story_graph_read_source', { ...input, content_kind: 'current', cursor: currentPage.page.next_cursor })).toThrow('stale_cursor')
|
||||
expect(f.read('story_graph_read_source', { source_reference_id: source.id, content_kind: 'version', source_version: current.source_version })).toMatchObject({ error: 'historical_version_unavailable' })
|
||||
f.update('Moved original', f.other.id)
|
||||
expect(() => f.read('story_graph_read_source', { source_reference_id: source.id, content_kind: 'current' })).toThrow('scope_mismatch')
|
||||
expect(() => f.db.readStoryGraph('story_graph_read_source', { source_reference_id: source.id }, f.other.id)).toThrow('scope_mismatch')
|
||||
f.sql.exec('DELETE FROM messages')
|
||||
expect(f.read('story_graph_read_source', { source_reference_id: source.id, content_kind: 'current' })).toMatchObject({ error: 'current_source_unavailable' })
|
||||
expect(f.read('story_graph_read_source', { source_reference_id: source.id })).toMatchObject({ content: text, content_kind: 'snapshot' })
|
||||
f.sql.prepare('UPDATE supervision_sources SET source_type = ?, locator_json = ? WHERE id = ?').run('knowledge', JSON.stringify({ length: 999 }), source.id!)
|
||||
expect(f.read('story_graph_read_source', { source_reference_id: source.id })).toMatchObject({ original_source_length: 'unknown', reviewed_source_length: 999 })
|
||||
f.sql.prepare('UPDATE supervision_sources SET locator_json = ? WHERE id = ?').run(JSON.stringify({ detail: 'x'.repeat(100_001) }), source.id!)
|
||||
expect(() => f.read('story_graph_read_source', { source_reference_id: source.id })).toThrow('response_too_large')
|
||||
})
|
||||
@@ -0,0 +1,236 @@
|
||||
import { createHash } from 'node:crypto'
|
||||
import type { DatabaseSync } from 'node:sqlite'
|
||||
import { z } from 'zod'
|
||||
import type { HeartbeatScope } from '../../shared/assistant-contracts'
|
||||
import { storyGraphContextSchema, storyGraphSearchSchema, storyGraphSourceSchema, type StoryGraphToolName } from '../../shared/story-graph-tools'
|
||||
import type { SupervisionEvidence, SupervisionSummaryOutput } from '../../shared/supervision-contracts'
|
||||
|
||||
type Row = Record<string, unknown>
|
||||
type Fact = {
|
||||
object_ref: { type: string; id: string }
|
||||
object_revision: string
|
||||
result_id: string | null
|
||||
scope: HeartbeatScope
|
||||
event_time: string | null
|
||||
review_generated_at: string | null
|
||||
recorded_at: string | null
|
||||
confirmation_state: string
|
||||
validity: 'unknown'
|
||||
source_reference_ids: string[]
|
||||
data: Row
|
||||
}
|
||||
const digest = (value: unknown): string => createHash('sha256').update(JSON.stringify(value)).digest('hex')
|
||||
const cursorSchema = z.object({ key: z.string(), revision: z.string(), index: z.number().int().nonnegative(), offset: z.number().int().nonnegative() }).strict()
|
||||
const normalizeScope = (scope: HeartbeatScope): HeartbeatScope => scope.kind === 'global' ? scope : { kind: 'projects', projectIds: [...new Set(scope.projectIds)].sort() }
|
||||
const timestamp = (value: unknown): string | null => typeof value === 'string' && Number.isFinite(Date.parse(value)) ? new Date(value).toISOString() : null
|
||||
const bounded = (output: Row): Row => {
|
||||
if (Buffer.byteLength(JSON.stringify(output)) > 100_000) throw new Error('response_too_large: retry with a smaller page_size')
|
||||
return output
|
||||
}
|
||||
|
||||
// This is a projection of existing storage, not a second graph or a historical state store.
|
||||
export function readStoryGraph(db: DatabaseSync, name: StoryGraphToolName, input: unknown, projectId?: string, signal?: AbortSignal): Record<string, unknown> {
|
||||
signal?.throwIfAborted()
|
||||
if (input && typeof input === 'object' && ('as_of' in input || ('mode' in input && input.mode === 'as_of'))) throw new Error('unsupported_mode: as_of is not supported')
|
||||
const args = name === 'story_graph_search' ? storyGraphSearchSchema.parse(input)
|
||||
: name === 'story_graph_get_context' ? storyGraphContextSchema.parse(input) : storyGraphSourceSchema.parse(input)
|
||||
const requested = args.scope ?? (projectId ? { kind: 'projects' as const, projectIds: [projectId] } : undefined)
|
||||
if (!requested) throw new Error('scope_required: no current project; specify scope explicitly')
|
||||
const scope = normalizeScope(requested)
|
||||
if (scope.kind === 'projects') for (const id of scope.projectIds) {
|
||||
if (!db.prepare("SELECT id FROM projects WHERE id = ? AND status = 'active'").get(id)) throw new Error('scope_unavailable: project does not exist or is inactive')
|
||||
}
|
||||
const base = { requested_scope: args.scope ?? 'current_project', resolved_scope: scope, read_at: new Date().toISOString(),
|
||||
coverage: { status: 'unknown', semantic_completeness: 'unknown', unprocessed_sources: 'unknown', historical_reconstruction: false,
|
||||
note: 'Exact stored scope only. Saved results and fragments do not establish complete source or semantic coverage.' } }
|
||||
const key = digest({ name, ...args, scope, cursor: undefined })
|
||||
let cursor: z.infer<typeof cursorSchema> | undefined
|
||||
if (args.cursor) {
|
||||
try { cursor = cursorSchema.parse(JSON.parse(Buffer.from(args.cursor, 'base64url').toString('utf8'))) }
|
||||
catch { throw new Error('invalid_cursor') }
|
||||
if (cursor.key !== key) throw new Error('invalid_cursor: repeat the original filters and page sizes')
|
||||
}
|
||||
const page = (revision: string, index: number, offset: number, more: boolean, count: number, total: number) => ({
|
||||
returned_count: count, total_count: total, has_more: more, complete: !more,
|
||||
next_cursor: more ? Buffer.from(JSON.stringify({ key, revision, index, offset })).toString('base64url') : null,
|
||||
limit_reason: more ? 'page_size' : null
|
||||
})
|
||||
const checkRevision = (revision: string): void => { if (cursor && cursor.revision !== revision) throw new Error('stale_cursor: stored evidence changed; restart the query') }
|
||||
const sameScope = (value: unknown): boolean => JSON.stringify(normalizeScope(JSON.parse(String(value)))) === JSON.stringify(scope)
|
||||
if (name === 'story_graph_read_source') {
|
||||
const sourceArgs = storyGraphSourceSchema.parse(input)
|
||||
const source = db.prepare(`SELECT s.*, sr.scope_json, r.created_at AS generated_at FROM supervision_sources s
|
||||
JOIN supervision_results r ON r.id = s.result_id JOIN supervision_runs sr ON sr.id = r.run_id WHERE s.id = ?`).get(sourceArgs.source_reference_id)
|
||||
if (!source) throw new Error('source_not_found')
|
||||
if (!sameScope(source.scope_json)) throw new Error('scope_mismatch')
|
||||
const locator = source.locator_json ? JSON.parse(String(source.locator_json)) as Row : {}
|
||||
const version = typeof locator.revision === 'string' ? locator.revision : 'unknown'
|
||||
if (sourceArgs.content_kind === 'version' && !sourceArgs.source_version) throw new Error('source_version_required')
|
||||
let body = String(source.content)
|
||||
let returnedVersion = version
|
||||
let location = locator
|
||||
let originalLength: unknown = source.source_type === 'knowledge' ? 'unknown' : locator.length ?? 'unknown'
|
||||
if (sourceArgs.content_kind !== 'snapshot') {
|
||||
// Resolve only an exact saved locator. Never replace a missing historical original with a snapshot.
|
||||
const sourceKey = typeof locator.source === 'string' ? locator.source
|
||||
: source.source_type === 'conversation' && typeof locator.messageId === 'string' ? `message:${locator.messageId}`
|
||||
: source.source_type === 'task' ? `task:${source.source_id}` : ''
|
||||
const kind = sourceKey.slice(0, sourceKey.indexOf(':'))
|
||||
const key = sourceKey.slice(sourceKey.indexOf(':') + 1)
|
||||
const separator = key.lastIndexOf(':')
|
||||
const current = sourceKey ? db.prepare(`SELECT body, locator, project_id, context
|
||||
FROM supervision_review_current WHERE record_id = ? AND reference_index = ? AND type = ?`).get(
|
||||
kind === 'knowledge' ? key.slice(0, separator) : key,
|
||||
kind === 'knowledge' ? Number(key.slice(separator + 1)) : 0,
|
||||
kind === 'message' ? 'conversation' : kind) : undefined
|
||||
if (current && scope.kind === 'projects' && !scope.projectIds.includes(String(current.project_id))) throw new Error('scope_mismatch: source moved out of the requested scope')
|
||||
const currentRevision = current ? createHash('sha256').update(String(current.context)).digest('hex') : undefined
|
||||
if (!current || (sourceArgs.content_kind === 'version' && currentRevision !== sourceArgs.source_version)) {
|
||||
return { ...base, error: sourceArgs.content_kind === 'version' ? 'historical_version_unavailable' : 'current_source_unavailable',
|
||||
original_source_availability: current ? 'available' : 'unavailable', available_content_kinds: current ? ['snapshot', 'current'] : ['snapshot'] }
|
||||
}
|
||||
body = String(current.body)
|
||||
returnedVersion = currentRevision!
|
||||
location = JSON.parse(String(current.locator)) as Row
|
||||
originalLength = source.source_type === 'knowledge' ? 'unknown' : Array.from(body).length
|
||||
}
|
||||
const content = Array.from(body)
|
||||
const revision = digest({ source, body, returnedVersion, location })
|
||||
checkRevision(revision)
|
||||
const start = cursor?.offset ?? 0
|
||||
if (start > content.length) throw new Error('invalid_cursor')
|
||||
const end = Math.min(content.length, start + sourceArgs.page_size)
|
||||
signal?.throwIfAborted()
|
||||
return bounded({ ...base, source_reference_id: source.id, result_id: source.result_id, original_source_id: source.source_id,
|
||||
source_type: source.source_type, content_kind: sourceArgs.content_kind, requested_version: sourceArgs.source_version ?? null,
|
||||
content_extent: sourceArgs.content_kind === 'snapshot' ? 'saved_fragment' : source.source_type === 'knowledge' ? 'current_reference_fragment' : 'source_record',
|
||||
source_version: returnedVersion, reference_version: version, differs_from_reference: version === 'unknown' ? 'unknown' : returnedVersion !== version,
|
||||
object_revision: revision, source_location: location, availability: source.availability,
|
||||
original_source_availability: sourceArgs.content_kind === 'snapshot' ? 'unknown' : 'available', source_record_time: source.occurred_at, event_time: null,
|
||||
review_generated_at: source.generated_at, recorded_at: source.generated_at,
|
||||
content: content.slice(start, end).join(''), position: { start, end, unit: 'unicode_code_point' },
|
||||
saved_fragment: { start: locator.start ?? 'unknown', end: locator.end ?? 'unknown', length: Array.from(String(source.content)).length },
|
||||
reviewed_source_length: locator.length ?? 'unknown',
|
||||
original_source_length: originalLength,
|
||||
page: page(revision, 0, end, end < content.length, end - start, content.length) })
|
||||
}
|
||||
const facts: Fact[] = []
|
||||
const results = new Map<string, Row>()
|
||||
const stories = db.prepare('SELECT * FROM story_lines ORDER BY id').all().filter(row => sameScope(row.scope_json))
|
||||
const add = (type: string, data: Row, result?: Row, sources?: string[]): void => {
|
||||
facts.push({ object_ref: { type, id: String(data.id) }, object_revision: digest(data), result_id: result ? String(result.id) : null, scope,
|
||||
event_time: type === 'event' ? timestamp(data.occurred_at) : null,
|
||||
review_generated_at: result ? timestamp(result.created_at) : null,
|
||||
recorded_at: result ? timestamp(result.created_at) : type === 'story_line' ? timestamp(data.created_at) : null,
|
||||
confirmation_state: String(data.confirmation_state ?? 'automatic'), validity: 'unknown',
|
||||
source_reference_ids: sources ?? (data.source_reference_ids_json ? JSON.parse(String(data.source_reference_ids_json)) : []), data })
|
||||
}
|
||||
for (const story of stories) {
|
||||
signal?.throwIfAborted()
|
||||
add('story_line', story)
|
||||
for (const row of db.prepare(`SELECT r.*, sr.time_range_json FROM supervision_results r JOIN supervision_runs sr ON sr.id = r.run_id WHERE sr.scope_json = ? ORDER BY r.id`).all(String(story.scope_json))) {
|
||||
if (results.has(String(row.id))) continue
|
||||
results.set(String(row.id), row)
|
||||
const { graph_snapshot_json, ...summary } = row
|
||||
add('summary', summary, row)
|
||||
const snapshot = JSON.parse(String(graph_snapshot_json)) as { entities: Row[]; relations: Row[] }
|
||||
for (const entity of snapshot.entities) add('entity', entity, row)
|
||||
for (const relation of snapshot.relations) add('relation', relation, row)
|
||||
const sources = db.prepare('SELECT * FROM supervision_sources WHERE result_id = ? ORDER BY id').all(String(row.id))
|
||||
for (const source of sources) add('source_reference', source, row, [String(source.id)])
|
||||
for (const change of db.prepare('SELECT * FROM supervision_entity_changes WHERE result_id = ? ORDER BY id').all(String(row.id))) add('entity_change', change, row)
|
||||
// Full successful leaves retain descriptions that the navigation summary and identity projection can omit.
|
||||
for (const leaf of db.prepare('SELECT id, output_json, evidence_json FROM supervision_review_batches WHERE run_id = ? ORDER BY id').all(String(row.run_id))) {
|
||||
const output = JSON.parse(String(leaf.output_json)) as SupervisionSummaryOutput
|
||||
const evidence = JSON.parse(String(leaf.evidence_json)) as SupervisionEvidence[]
|
||||
const refs = new Map(evidence.map(item => [item.id, sources.filter(source => source.source_type === item.sourceType &&
|
||||
source.source_id === item.sourceId && source.content === item.content && source.locator_json === (item.locator ? JSON.stringify(item.locator) : null)).map(source => String(source.id))]))
|
||||
const references = (ids: string[]) => [...new Set(ids.flatMap(id => refs.get(id) ?? []))]
|
||||
add('summary', { id: `leaf:${leaf.id}`, summary: output.summary, change_digest: output.changeDigest,
|
||||
open_items: output.openItems, content_kind: 'successful_leaf' }, row, references(evidence.map(item => item.id)))
|
||||
// Persisted IDs are explicit identity evidence; local model IDs are namespaced to the saved leaf.
|
||||
for (const entity of output.entities) add('entity', { id: `leaf:${leaf.id}:${entity.id}`,
|
||||
canonical_label: entity.label, description: entity.description, persisted_entity_id: entity.persistedId ?? null,
|
||||
content_kind: 'successful_leaf', leaf_id: leaf.id }, row, references(entity.sourceReferenceIds))
|
||||
for (const relation of output.relations) add('relation', { id: `leaf:${leaf.id}:${digest(relation)}`,
|
||||
from_entity_id: `leaf:${leaf.id}:${relation.fromEntityId}`, to_entity_id: `leaf:${leaf.id}:${relation.toEntityId}`,
|
||||
relation_type: relation.relationType, reason: relation.reason, content_kind: 'successful_leaf' }, row, references(relation.sourceReferenceIds))
|
||||
}
|
||||
}
|
||||
for (const entity of db.prepare('SELECT * FROM supervision_entities WHERE story_line_id = ? ORDER BY id').all(String(story.id))) add('entity', entity)
|
||||
for (const relation of db.prepare('SELECT * FROM supervision_relations WHERE story_line_id = ? ORDER BY id').all(String(story.id))) add('relation', relation)
|
||||
for (const event of db.prepare('SELECT * FROM supervision_events WHERE story_line_id = ? ORDER BY id').all(String(story.id))) {
|
||||
const sources = db.prepare('SELECT source_id FROM supervision_event_sources WHERE event_id = ? ORDER BY source_id').all(String(event.id)).map(row => String(row.source_id))
|
||||
const entities = db.prepare('SELECT entity_id FROM supervision_event_entities WHERE event_id = ? ORDER BY entity_id').all(String(event.id)).map(row => String(row.entity_id))
|
||||
add('event', { ...event, entity_ids: entities }, results.get(String(event.result_id)), sources)
|
||||
}
|
||||
}
|
||||
const revision = digest(facts)
|
||||
checkRevision(revision)
|
||||
const timed = args as z.infer<typeof storyGraphSearchSchema>
|
||||
const matchesTime = (fact: Fact): boolean => !timed.time_range || Boolean(fact[timed.time_basis] &&
|
||||
Date.parse(fact[timed.time_basis]!) >= Date.parse(timed.time_range.from) && Date.parse(fact[timed.time_basis]!) <= Date.parse(timed.time_range.to))
|
||||
const metadata = { ...base, coverage: { ...base.coverage, published_result_count: results.size }, time_basis: timed.time_basis,
|
||||
time_range: timed.time_range ? { from: timestamp(timed.time_range.from), to: timestamp(timed.time_range.to) } : null,
|
||||
interpretation: 'Stored claims within the selected evidence scope; no inferred supersession, completion, or historical state.' }
|
||||
const query = 'query' in args ? args.query.toLowerCase() : ''
|
||||
let selected: Fact[]
|
||||
if (name === 'story_graph_search') {
|
||||
const search = storyGraphSearchSchema.parse(input)
|
||||
selected = facts.filter(fact => (!search.object_types || search.object_types.includes(fact.object_ref.type as typeof search.object_types[number])) && JSON.stringify(fact.data).toLowerCase().includes(query))
|
||||
} else {
|
||||
const context = storyGraphContextSchema.parse(input)
|
||||
const targets = facts.filter(fact => fact.object_ref.type === context.object_ref.type && fact.object_ref.id === context.object_ref.id)
|
||||
if (!targets.length) {
|
||||
const tables = { story_line: 'story_lines', event: 'supervision_events', entity: 'supervision_entities',
|
||||
entity_change: 'supervision_entity_changes', relation: 'supervision_relations', summary: 'supervision_results', source_reference: 'supervision_sources' }
|
||||
const leafId = context.object_ref.id.startsWith('leaf:') ? context.object_ref.id.split(':')[1] : undefined
|
||||
const exists = leafId ? db.prepare('SELECT id FROM supervision_review_batches WHERE id = ?').get(leafId)
|
||||
: db.prepare(`SELECT id FROM ${tables[context.object_ref.type]} WHERE id = ?`).get(context.object_ref.id)
|
||||
throw new Error(exists ? 'scope_mismatch: object is outside the requested scope' : 'object_not_found')
|
||||
}
|
||||
const related = facts.filter(fact => targets.includes(fact) || (context.object_ref.type === 'entity' &&
|
||||
(fact.data.entity_id === context.object_ref.id || fact.data.persisted_entity_id === context.object_ref.id || fact.data.from_entity_id === context.object_ref.id || fact.data.to_entity_id === context.object_ref.id || (Array.isArray(fact.data.entity_ids) && fact.data.entity_ids.includes(context.object_ref.id)))))
|
||||
const targetSources = new Set(related.flatMap(fact => fact.source_reference_ids))
|
||||
const entityIds = new Set(related.flatMap(fact => [fact.data.entity_id, fact.data.from_entity_id, fact.data.to_entity_id, ...(Array.isArray(fact.data.entity_ids) ? fact.data.entity_ids : [])]).filter((id): id is string => typeof id === 'string'))
|
||||
const targetResults = new Set(targets.flatMap(fact => fact.result_id ? [fact.result_id] : []))
|
||||
selected = facts.filter(fact => {
|
||||
if (context.object_ref.type === 'story_line') return true
|
||||
if (related.includes(fact) || (fact.object_ref.type === 'entity' && entityIds.has(fact.object_ref.id))) return true
|
||||
if (context.object_ref.type === 'summary') return fact.result_id !== null && targetResults.has(fact.result_id)
|
||||
return fact.source_reference_ids.some(id => targetSources.has(id))
|
||||
})
|
||||
}
|
||||
const unknownExcluded = timed.time_range ? selected.filter(fact => !fact[timed.time_basis]).length : 0
|
||||
selected = selected.filter(matchesTime)
|
||||
const score = (fact: Fact): number => {
|
||||
if (name !== 'story_graph_search') return 0
|
||||
return Object.entries(fact.data).reduce((sum, [field, value]) => sum + (JSON.stringify(value).toLowerCase().includes(query) ? (['title', 'canonical_label', 'summary'].includes(field) ? 3 : 1) : 0), 0)
|
||||
}
|
||||
selected.sort((a, b) => score(b) - score(a) || (name === 'story_graph_get_context' && 'mode' in args && args.mode === 'timeline'
|
||||
? (a.event_time ?? '\uffff').localeCompare(b.event_time ?? '\uffff') : 0) ||
|
||||
a.object_ref.type.localeCompare(b.object_ref.type) || a.object_ref.id.localeCompare(b.object_ref.id) || (a.result_id ?? '').localeCompare(b.result_id ?? ''))
|
||||
let index = cursor?.index ?? 0
|
||||
let offset = cursor?.offset ?? 0
|
||||
if (index > selected.length) throw new Error('invalid_cursor')
|
||||
const items: Row[] = []
|
||||
for (; index < selected.length && items.length < args.page_size; index++) {
|
||||
signal?.throwIfAborted()
|
||||
const { data, ...fact } = selected[index]!
|
||||
const text = Array.from(JSON.stringify(data))
|
||||
if (name === 'story_graph_search') {
|
||||
const fields = Object.keys(data).filter(field => JSON.stringify(data[field]).toLowerCase().includes(query))
|
||||
items.push({ ...fact, matched_fields: fields, preview: text.slice(0, 400).join(''), preview_is_excerpt: true })
|
||||
} else {
|
||||
const size = storyGraphContextSchema.parse(input).content_page_size
|
||||
const end = Math.min(text.length, offset + size)
|
||||
items.push({ ...fact, content: text.slice(offset, end).join(''), content_format: 'json_fragment', position: { start: offset, end, length: text.length, unit: 'unicode_code_point' } })
|
||||
if (end < text.length) { offset = end; break }
|
||||
offset = 0
|
||||
}
|
||||
// Fail explicitly rather than allowing a downstream runtime to truncate metadata.
|
||||
if (Buffer.byteLength(JSON.stringify(items)) > 90_000) throw new Error('response_too_large: retry with a smaller page_size')
|
||||
}
|
||||
const output = { ...metadata, unknown_time_excluded_count: unknownExcluded, ...('mode' in args ? { mode: args.mode } : {}), items, page: page(revision, index, offset, index < selected.length, items.length, selected.length) }
|
||||
return bounded(output)
|
||||
}
|
||||
@@ -86,6 +86,12 @@ it('persists publication failure and resumes with saved navigation without model
|
||||
expect(batch.output.summary).toBe(summary)
|
||||
expect(batch.output.events[0]!.description).toBe(description)
|
||||
}
|
||||
const calls = f.respond.mock.calls.length
|
||||
expect(f.db.readStoryGraph('story_graph_search', { query: 'Retained source qualification', scope: { kind: 'global' }, object_types: ['event'] }))
|
||||
.toMatchObject({ page: { total_count: 2 }, items: [expect.objectContaining({ source_reference_ids: [] }), expect.objectContaining({ source_reference_ids: [] })] })
|
||||
expect(f.db.readStoryGraph('story_graph_search', { query: 'Atlas', scope: { kind: 'global' }, object_types: ['entity'] }))
|
||||
.toMatchObject({ page: { has_more: false } })
|
||||
expect(f.respond).toHaveBeenCalledTimes(calls)
|
||||
})
|
||||
|
||||
it('stops oversized runtime responses with a readable error while preserving resumable progress', async () => {
|
||||
|
||||
Reference in New Issue
Block a user