mirror of
https://github.com/wu736139669/hapi.git
synced 2026-11-08 01:43:07 +00:00
feat(opencode): import local session history (#1677)
* feat(cli): list local OpenCode sessions over machine RPC * feat(hub): import OpenCode sessions into HAPI * feat(web): import OpenCode history from new-session picker * fix(cli): derive OpenCode imported message ids from part ids Random ids made every scan differ, so delta classification flagged unchanged transcripts as diverged on re-import. * fix(cli): surface unreadable OpenCode db as rpc error A locked or unopenable opencode.db now fails the listing instead of reporting an empty success; summaries normalize timestamps and the web import toast counts only created or actually-appended sessions. * fix(hub): fail closed on opencode re-import past live HAPI continuation Imported entries must remain a strict prefix of the native transcript. Insertions ahead of the imported boundary or new native messages after the session continued in HAPI now fail closed instead of appending duplicated or reordered history. * fix(web,hub): apply selected launch config when importing OpenCode history Pass the model, reasoning effort, and permission mode chosen in the new-session picker through the import request and persist them on the imported HAPI session via applySessionConfig, so reopen launches with the visible selection instead of stale or default values. * fix(hub,cli): preserve explicit launch resets and skip streaming parts on import - Launch config keys are applied by presence, so explicit null/default selections from the picker must reach applySessionConfig to clear stale model, effort, or permission values on re-import. Requests that omit the fields entirely still leave persisted config untouched. - Assistant text parts are mutable while a response streams. Mirror the live scanner gate and import only finished (time.end set), synthetic, or user text so an in-progress row can never be stored and later finalized as transcript_diverged. * test(hub): fix launch-config reset assertions to satisfy strict types * fix(cli): fail opencode transcript reads on query errors instead of partial snapshots Swallowed SQLite errors could return a truncated message list that the hub later rejects as transcript_diverged with no repair path. Full transcript reads now propagate query failures; summary lastUserMessage extraction stays resilient per message. * test(cli): cover unreadable-part-table transcript read stays non-partial * fix(hub): scope launch-config failures to the per-session import result A thrown applySessionConfig call used to escape the bulk handler as an unstructured 500 even though transcripts were already persisted. Catch it per session and report it as a config_failed result so remaining sessions in a bulk request still import. Also cover the mid-transcript part-query failure with an injected opener that rejects instead of returning a holey transcript. * fix(cli): read opencode transcripts inside a deferred snapshot transaction Messages and their parts were read in separate statements, so a concurrent OpenCode write could leave a hole in the imported sequence that positional re-import checks then reject forever. Wrap each full transcript read in BEGIN DEFERRED/COMMIT with rollback on failure. * fix(cli): follow machine path policy for OpenCode imports * fix(web): use unified permission mode for OpenCode imports * fix(opencode): harden session import consistency * fix(hub): preserve metadata on rejected OpenCode imports * fix(hub): refresh OpenCode import state under lock * fix(hub): resolve OpenCode imports after locking * fix(opencode): bind imported resume to source machine
This commit is contained in:
Junmo Kim
@@ -11,10 +11,12 @@ import type { ClientToServerEvents, ServerToClientEvents, Update, UpdateMachineB
|
||||
import {
|
||||
ArchiveCodexSessionRpcRequestSchema,
|
||||
ListCodexSessionsRpcRequestSchema,
|
||||
ListOpencodeSessionsRpcRequestSchema,
|
||||
ListPiSessionsRpcRequestSchema,
|
||||
type ArchiveCodexSessionRpcResponse,
|
||||
type AgentAvailabilityResponse,
|
||||
type ListCodexSessionsRpcResponse,
|
||||
type ListOpencodeSessionsRpcResponse,
|
||||
type ListPiSessionsRpcResponse,
|
||||
type MachineDirectoryEntry,
|
||||
type MachineListDirectoryResponse,
|
||||
@@ -36,6 +38,10 @@ import {
|
||||
type ListOpencodeModelsForCwdRequest,
|
||||
type ListOpencodeModelsForCwdResponse
|
||||
} from '../modules/common/opencodeModels'
|
||||
import {
|
||||
listLocalOpencodeSessionSummaries,
|
||||
listLocalOpencodeSessionsWithMessagesByIds
|
||||
} from '../modules/common/opencodeSessions'
|
||||
import {
|
||||
listOpencodeModelVariants,
|
||||
type ListOpencodeModelVariantsResponse
|
||||
@@ -410,6 +416,35 @@ export class ApiMachineClient {
|
||||
return { success: true, sessions }
|
||||
}
|
||||
)
|
||||
|
||||
this.rpcHandlerManager.registerHandler<unknown, ListOpencodeSessionsRpcResponse>(
|
||||
RPC_METHODS.ListOpencodeSessions,
|
||||
async (params) => {
|
||||
const parsed = ListOpencodeSessionsRpcRequestSchema.safeParse(params)
|
||||
if (!parsed.success) return { success: false, error: 'Invalid Opencode sessions request' }
|
||||
const rawCwd = typeof parsed.data.cwd === 'string' ? parsed.data.cwd.trim() : ''
|
||||
if (rawCwd) {
|
||||
const resolvedCwd = await this.pathPolicy.resolveForCheck(rawCwd)
|
||||
if (!this.pathPolicy.isWithinSpawnRoots(resolvedCwd)) {
|
||||
return { success: false, error: 'Path is outside workspace roots' }
|
||||
}
|
||||
}
|
||||
const requestedIds = parsed.data.sessionIds ? new Set(parsed.data.sessionIds) : null
|
||||
let allSessions
|
||||
try {
|
||||
allSessions = requestedIds
|
||||
? await listLocalOpencodeSessionsWithMessagesByIds(requestedIds)
|
||||
: await listLocalOpencodeSessionSummaries()
|
||||
} catch (error) {
|
||||
return { success: false, error: error instanceof Error ? error.message : 'Failed to read local OpenCode sessions' }
|
||||
}
|
||||
const sessions = []
|
||||
for (const session of allSessions) {
|
||||
if (await this.isLocalSessionWithinWorkspaceRoots(session)) sessions.push(session)
|
||||
}
|
||||
return { success: true, sessions }
|
||||
}
|
||||
)
|
||||
}
|
||||
|
||||
private async isLocalSessionWithinWorkspaceRoots(session: { cwd?: string | null }): Promise<boolean> {
|
||||
|
||||
@@ -0,0 +1,284 @@
|
||||
import { describe, expect, it } from 'vitest'
|
||||
import { spawnSync } from 'node:child_process'
|
||||
import { join } from 'node:path'
|
||||
|
||||
// bun:sqlite is only resolvable under the Bun runtime, while the unit suite
|
||||
// runs on node (vitest). Mirror the agy modules' approach: the production
|
||||
// module lazy-imports bun:sqlite, and these tests drive the real module +
|
||||
// a real sqlite db inside a `bun` subprocess, asserting on its JSON output.
|
||||
|
||||
const RUNNER = /* ts */ `
|
||||
import { mkdtempSync, writeFileSync } from 'node:fs'
|
||||
import { join } from 'node:path'
|
||||
import { tmpdir } from 'node:os'
|
||||
import { Database } from 'bun:sqlite'
|
||||
import {
|
||||
getOpencodeDbPath,
|
||||
listLocalOpencodeSessionSummaries,
|
||||
listLocalOpencodeSessionsWithMessagesByIds
|
||||
} from './opencodeSessions'
|
||||
|
||||
function createDb(root) {
|
||||
const db = new Database(join(root, 'opencode.db'))
|
||||
db.exec('CREATE TABLE session (id TEXT PRIMARY KEY, title TEXT, directory TEXT, time_updated REAL, time_archived REAL); CREATE TABLE message (id TEXT PRIMARY KEY, session_id TEXT, data TEXT, time_created REAL); CREATE TABLE part (id TEXT PRIMARY KEY, message_id TEXT, data TEXT, time_created REAL);')
|
||||
return db
|
||||
}
|
||||
|
||||
function insertMessages(dbPath, messages, parts) {
|
||||
const db = new Database(dbPath)
|
||||
try {
|
||||
for (const m of messages) {
|
||||
db.query('INSERT INTO message (id, session_id, data, time_created) VALUES (?, ?, ?, ?)')
|
||||
.run(m.id, m.sessionId, JSON.stringify({ role: m.role }), m.timeCreated)
|
||||
}
|
||||
let i = 0
|
||||
for (const p of parts) {
|
||||
i += 1
|
||||
const data = Object.assign(
|
||||
{ type: p.type },
|
||||
p.text === undefined ? {} : { text: p.text },
|
||||
p.extra ?? {}
|
||||
)
|
||||
db.query('INSERT INTO part (id, message_id, data, time_created) VALUES (?, ?, ?, ?)')
|
||||
.run('part-' + i, p.messageId, JSON.stringify(data), i)
|
||||
}
|
||||
} finally {
|
||||
db.close()
|
||||
}
|
||||
}
|
||||
|
||||
const scenario = process.env.SCENARIO
|
||||
|
||||
if (scenario === 'sessions') {
|
||||
const root = mkdtempSync(join(tmpdir(), 'opencode-sessions-'))
|
||||
process.env.OPENCODE_HOME = root
|
||||
const dbPath = join(root, 'opencode.db')
|
||||
createDb(root)
|
||||
const db = new Database(dbPath)
|
||||
db.query('INSERT INTO session (id, title, directory, time_updated) VALUES (?, ?, ?, ?)').run('session-1', 'Normal session', '/tmp/project', 2000)
|
||||
db.query('INSERT INTO session (id, title, directory, time_updated, time_archived) VALUES (?, ?, ?, ?, ?)').run('session-2', 'Archived session', '/tmp/project', 3000, 4000)
|
||||
db.query('INSERT INTO session (id, title, directory, time_updated) VALUES (?, ?, ?, ?)').run('session-3', 'Malformed', '/tmp/project', 1000)
|
||||
db.close()
|
||||
insertMessages(dbPath, [
|
||||
{ id: 'm1-u', sessionId: 'session-1', role: 'user', timeCreated: 1100 },
|
||||
{ id: 'm1-a', sessionId: 'session-1', role: 'assistant', timeCreated: 1200 },
|
||||
{ id: 'm2-u', sessionId: 'session-2', role: 'user', timeCreated: 1400 }
|
||||
], [
|
||||
{ messageId: 'm1-a', type: 'tool' },
|
||||
{ messageId: 'm1-u', type: 'text', text: 'hello world' },
|
||||
{ messageId: 'm1-a', type: 'text', text: 'hi there', extra: { time: { end: 1250 } } },
|
||||
{ messageId: 'm2-u', type: 'text', text: '<system-reminder>internal</system-reminder>' }
|
||||
])
|
||||
{
|
||||
const db = new Database(dbPath)
|
||||
try {
|
||||
// malformed message-data JSON row must be skipped silently
|
||||
db.query('INSERT INTO message (id, session_id, data, time_created) VALUES (?, ?, ?, ?)')
|
||||
.run('m3-u', 'session-3', '{invalid json', 1300)
|
||||
db.query('INSERT INTO part (id, message_id, data, time_created) VALUES (?, ?, ?, ?)')
|
||||
.run('part-m3', 'm3-u', JSON.stringify({ type: 'text', text: 'unreachable' }), 1)
|
||||
} finally {
|
||||
db.close()
|
||||
}
|
||||
}
|
||||
const summaries = await listLocalOpencodeSessionSummaries()
|
||||
const sessions = await listLocalOpencodeSessionsWithMessagesByIds(new Set(['session-1', 'session-2']))
|
||||
console.log(JSON.stringify({ summaries, sessions }))
|
||||
} else if (scenario === 'truncate') {
|
||||
const root = mkdtempSync(join(tmpdir(), 'opencode-truncate-'))
|
||||
process.env.OPENCODE_HOME = root
|
||||
const dbPath = join(root, 'opencode.db')
|
||||
createDb(root)
|
||||
const db = new Database(dbPath)
|
||||
db.query('INSERT INTO session (id, title, directory, time_updated) VALUES (?, ?, ?, ?)').run('session-long', 'Long', '/tmp/project', 5000)
|
||||
db.close()
|
||||
insertMessages(dbPath, [
|
||||
{ id: 'ml-u', sessionId: 'session-long', role: 'user', timeCreated: 5100 }
|
||||
], [
|
||||
{ messageId: 'ml-u', type: 'text', text: 'x'.repeat(200) }
|
||||
])
|
||||
const summaries = await listLocalOpencodeSessionSummaries()
|
||||
console.log(JSON.stringify({ summaries }))
|
||||
} else if (scenario === 'streaming') {
|
||||
const root = mkdtempSync(join(tmpdir(), 'opencode-streaming-'))
|
||||
process.env.OPENCODE_HOME = root
|
||||
const dbPath = join(root, 'opencode.db')
|
||||
createDb(root)
|
||||
const db = new Database(dbPath)
|
||||
db.query('INSERT INTO session (id, title, directory, time_updated) VALUES (?, ?, ?, ?)').run('session-stream', 'Streaming', '/tmp/project', 6000)
|
||||
db.close()
|
||||
insertMessages(dbPath, [
|
||||
{ id: 'ms-u', sessionId: 'session-stream', role: 'user', timeCreated: 6100 },
|
||||
{ id: 'ms-a', sessionId: 'session-stream', role: 'assistant', timeCreated: 6200 }
|
||||
], [
|
||||
{ messageId: 'ms-u', type: 'text', text: 'question' },
|
||||
{ messageId: 'ms-a', type: 'text', text: 'finished answer', extra: { time: { end: 6300 } } },
|
||||
{ messageId: 'ms-a', type: 'text', text: 'partial answer being generated' }
|
||||
])
|
||||
const sessions = await listLocalOpencodeSessionsWithMessagesByIds(new Set(['session-stream']))
|
||||
console.log(JSON.stringify({ sessions }))
|
||||
} else if (scenario === 'queryfail') {
|
||||
const root = mkdtempSync(join(tmpdir(), 'opencode-queryfail-'))
|
||||
process.env.OPENCODE_HOME = root
|
||||
const dbPath = join(root, 'opencode.db')
|
||||
createDb(root)
|
||||
const db = new Database(dbPath)
|
||||
db.query('INSERT INTO session (id, title, directory, time_updated) VALUES (?, ?, ?, ?)').run('session-qf', 'QueryFail', '/tmp/project', 7000)
|
||||
insertMessages(dbPath, [
|
||||
{ id: 'mq-u', sessionId: 'session-qf', role: 'user', timeCreated: 7100 },
|
||||
{ id: 'mq-a', sessionId: 'session-qf', role: 'assistant', timeCreated: 7200 }
|
||||
], [
|
||||
{ messageId: 'mq-u', type: 'text', text: 'question' },
|
||||
{ messageId: 'mq-a', type: 'text', text: 'answer', extra: { time: { end: 7300 } } }
|
||||
])
|
||||
// Simulate a transient SQLite failure reading parts of later messages.
|
||||
db.exec('DROP TABLE part')
|
||||
db.close()
|
||||
const out = await listLocalOpencodeSessionsWithMessagesByIds(new Set(['session-qf']))
|
||||
console.log(JSON.stringify({ sessions: out }))
|
||||
} else if (scenario === 'failclosed') {
|
||||
const missingRoot = mkdtempSync(join(tmpdir(), 'opencode-missing-'))
|
||||
process.env.OPENCODE_HOME = missingRoot
|
||||
const noDb = {
|
||||
path: getOpencodeDbPath(),
|
||||
summaries: await listLocalOpencodeSessionSummaries(),
|
||||
byIds: await listLocalOpencodeSessionsWithMessagesByIds(new Set(['any']))
|
||||
}
|
||||
|
||||
const emptyRoot = mkdtempSync(join(tmpdir(), 'opencode-empty-'))
|
||||
process.env.OPENCODE_HOME = emptyRoot
|
||||
writeFileSync(join(emptyRoot, 'opencode.db'), '')
|
||||
const emptyFile = {
|
||||
summaries: await listLocalOpencodeSessionSummaries(),
|
||||
byIds: await listLocalOpencodeSessionsWithMessagesByIds(new Set(['any']))
|
||||
}
|
||||
|
||||
const partialRoot = mkdtempSync(join(tmpdir(), 'opencode-partial-'))
|
||||
process.env.OPENCODE_HOME = partialRoot
|
||||
const partial = new Database(join(partialRoot, 'opencode.db'))
|
||||
partial.exec('CREATE TABLE session (id TEXT PRIMARY KEY);')
|
||||
partial.close()
|
||||
const missingTables = {
|
||||
summaries: await listLocalOpencodeSessionSummaries(),
|
||||
byIds: await listLocalOpencodeSessionsWithMessagesByIds(new Set(['any']))
|
||||
}
|
||||
console.log(JSON.stringify({ noDb, emptyFile, missingTables }))
|
||||
}
|
||||
`
|
||||
|
||||
function runScenario(name: string): Record<string, unknown> {
|
||||
const result = spawnSync('bun', ['-e', RUNNER], {
|
||||
cwd: import.meta.dirname,
|
||||
env: { ...process.env, SCENARIO: name },
|
||||
encoding: 'utf-8'
|
||||
})
|
||||
if (result.status !== 0) throw new Error(`bun scenario '${name}' failed:\n${result.stderr}`)
|
||||
const line = result.stdout.trim().split('\n').filter(Boolean).pop()
|
||||
return JSON.parse(line ?? '{}')
|
||||
}
|
||||
|
||||
describe('local opencode sessions', () => {
|
||||
it('lists summaries with lastUserMessage rules and converts message envelopes', () => {
|
||||
const output = runScenario('sessions') as any
|
||||
expect(output.summaries.map((summary: any) => summary.id)).toEqual(['session-1', 'session-3'])
|
||||
expect(output.summaries[0]).toMatchObject({
|
||||
id: 'session-1',
|
||||
title: 'Normal session',
|
||||
cwd: '/tmp/project',
|
||||
modifiedAt: 2_000_000,
|
||||
lastUserMessage: 'hello world'
|
||||
})
|
||||
// tool part skipped; malformed-JSON user row yields null lastUserMessage
|
||||
expect(output.summaries[1]).toMatchObject({ id: 'session-3', lastUserMessage: null })
|
||||
|
||||
expect(output.sessions.map((session: any) => session.id)).toEqual(['session-2', 'session-1'])
|
||||
const archived = output.sessions.find((session: any) => session.id === 'session-2')
|
||||
// archived sessions are loadable by ids; their '<'-prefixed user text
|
||||
// is excluded from lastUserMessage but still converts to an envelope
|
||||
expect(archived?.messages).toHaveLength(1)
|
||||
expect(archived?.messages[0]).toMatchObject({ content: { role: 'user' } })
|
||||
expect(archived?.lastUserMessage ?? null).toBe(null)
|
||||
const normal = output.sessions.find((session: any) => session.id === 'session-1')
|
||||
expect(normal?.messages).toHaveLength(2)
|
||||
expect(normal.messages[0]).toMatchObject({
|
||||
content: { role: 'user', content: { type: 'text', text: 'hello world' }, meta: { sentFrom: 'cli' } }
|
||||
})
|
||||
expect(normal.messages[1]).toMatchObject({
|
||||
content: {
|
||||
role: 'agent',
|
||||
content: { type: 'codex', data: { type: 'message', message: 'hi there' } },
|
||||
meta: { sentFrom: 'cli' }
|
||||
}
|
||||
})
|
||||
expect(typeof normal.messages[0].localId).toBe('string')
|
||||
})
|
||||
|
||||
it('truncates long last user messages to 140 chars', () => {
|
||||
const output = runScenario('truncate') as any
|
||||
const lastUserMessage = output.summaries[0]?.lastUserMessage as string | undefined
|
||||
expect(lastUserMessage).toHaveLength(140)
|
||||
expect(lastUserMessage?.endsWith('…')).toBe(true)
|
||||
})
|
||||
|
||||
it('skips unfinished assistant text parts that are still streaming', () => {
|
||||
const output = runScenario('streaming') as any
|
||||
const messages = output.sessions[0]?.messages ?? []
|
||||
expect(messages).toHaveLength(2)
|
||||
expect(messages[0]).toMatchObject({ content: { role: 'user', content: { type: 'text', text: 'question' } } })
|
||||
expect(messages[1]).toMatchObject({
|
||||
content: {
|
||||
role: 'agent',
|
||||
content: { type: 'codex', data: { type: 'message', message: 'finished answer' } },
|
||||
meta: { sentFrom: 'cli' }
|
||||
}
|
||||
})
|
||||
})
|
||||
|
||||
it('never surfaces a holey partial transcript when parts are unreadable', () => {
|
||||
const output = runScenario('queryfail') as any
|
||||
// Missing part table short-circuits the read entirely — the caller
|
||||
// sees an empty result, never a transcript with a hole in the middle.
|
||||
expect(output.sessions).toEqual([])
|
||||
})
|
||||
|
||||
it('fails closed when the db or required tables are missing', () => {
|
||||
const output = runScenario('failclosed') as any
|
||||
expect(output.noDb.path.endsWith(join('opencode.db'))).toBe(true)
|
||||
expect(output.noDb.summaries).toEqual([])
|
||||
expect(output.noDb.byIds).toEqual([])
|
||||
expect(output.emptyFile.summaries).toEqual([])
|
||||
expect(output.emptyFile.byIds).toEqual([])
|
||||
expect(output.missingTables.summaries).toEqual([])
|
||||
expect(output.missingTables.byIds).toEqual([])
|
||||
})
|
||||
|
||||
it('rejects instead of returning a partial transcript when a middle part query fails', async () => {
|
||||
const { listLocalOpencodeSessionsWithMessagesByIds } = await import('./opencodeSessions')
|
||||
let partReads = 0
|
||||
const stmt = (rows: unknown[]) => ({ all: () => rows, get: () => rows[0] })
|
||||
const fakeDb = {
|
||||
query(sql: string) {
|
||||
if (/FROM part WHERE/.test(sql)) {
|
||||
partReads += 1
|
||||
if (partReads === 2) {
|
||||
return { all: () => { throw new Error('injected part read failure') }, get: () => undefined }
|
||||
}
|
||||
}
|
||||
if (/sqlite_master/.test(sql)) return stmt([{ name: 'x' }])
|
||||
if (/FROM session WHERE id/.test(sql)) {
|
||||
return stmt([{ id: 's1', title: 'T', directory: '/tmp', time_updated: 7000 }])
|
||||
}
|
||||
if (/FROM message WHERE session_id/.test(sql)) {
|
||||
return stmt([
|
||||
{ id: 'm1', data: JSON.stringify({ role: 'user' }), time_created: 7100 },
|
||||
{ id: 'm2', data: JSON.stringify({ role: 'assistant' }), time_created: 7200 }
|
||||
])
|
||||
}
|
||||
return stmt([])
|
||||
},
|
||||
close: () => {}
|
||||
}
|
||||
await expect(listLocalOpencodeSessionsWithMessagesByIds(new Set(['s1']), async () => fakeDb))
|
||||
.rejects.toThrow('injected part read failure')
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,311 @@
|
||||
import { existsSync } from 'node:fs'
|
||||
import { homedir } from 'node:os'
|
||||
import { join } from 'node:path'
|
||||
import { AGENT_MESSAGE_PAYLOAD_TYPE } from '@hapi/protocol'
|
||||
import type {
|
||||
OpencodeImportedMessage,
|
||||
OpencodeImportedMessageContent,
|
||||
OpencodeLocalSessionSummary,
|
||||
OpencodeLocalSessionWithMessages
|
||||
} from '@hapi/protocol/apiTypes'
|
||||
|
||||
const DEFAULT_OPENCODE_SESSION_SCAN_LIMIT = 200
|
||||
const LAST_USER_MESSAGE_MAX_LENGTH = 140
|
||||
|
||||
type JsonRecord = Record<string, unknown>
|
||||
|
||||
type SqlParams = unknown[]
|
||||
|
||||
type StatementLike = {
|
||||
get: (...params: SqlParams) => unknown
|
||||
all: (...params: SqlParams) => unknown[]
|
||||
}
|
||||
|
||||
type DatabaseLike = {
|
||||
query: (sql: string) => StatementLike
|
||||
close: () => void
|
||||
exec?: (sql: string) => void
|
||||
}
|
||||
|
||||
export type OpencodeDatabaseOpener = (dbPath: string) => Promise<DatabaseLike | null>
|
||||
|
||||
type SessionRow = {
|
||||
id: string
|
||||
title: string | null
|
||||
directory: string | null
|
||||
time_updated: number | string
|
||||
}
|
||||
|
||||
let cachedOpener: OpencodeDatabaseOpener | null = null
|
||||
|
||||
async function getDefaultOpener(): Promise<OpencodeDatabaseOpener> {
|
||||
if (!cachedOpener) {
|
||||
cachedOpener = async (dbPath: string) => {
|
||||
const { Database } = await import('bun:sqlite')
|
||||
if (!existsSync(dbPath)) return null
|
||||
const db = new Database(dbPath, { readonly: true }) as unknown as DatabaseLike & { exec?: (sql: string) => void }
|
||||
try { db.exec?.('PRAGMA busy_timeout = 5000') } catch {}
|
||||
return db
|
||||
}
|
||||
}
|
||||
return cachedOpener
|
||||
}
|
||||
|
||||
function asRecord(value: unknown): JsonRecord | null {
|
||||
return value !== null && typeof value === 'object' && !Array.isArray(value)
|
||||
? value as JsonRecord
|
||||
: null
|
||||
}
|
||||
|
||||
export function getOpencodeSessionsRoot(overrideRoot?: string): string {
|
||||
const direct = overrideRoot ?? process.env.OPENCODE_HOME?.trim()
|
||||
if (direct) return direct
|
||||
const dataHome = process.env.XDG_DATA_HOME?.trim() || join(homedir(), '.local', 'share')
|
||||
return join(dataHome, 'opencode')
|
||||
}
|
||||
|
||||
export function getOpencodeDbPath(overrideRoot?: string): string {
|
||||
return join(getOpencodeSessionsRoot(overrideRoot), 'opencode.db')
|
||||
}
|
||||
|
||||
function truncateText(value: string, maxLength: number): string {
|
||||
return value.length > maxLength ? `${value.slice(0, maxLength - 1)}…` : value
|
||||
}
|
||||
|
||||
async function openDb(dbPath: string, opener?: OpencodeDatabaseOpener): Promise<DatabaseLike | null> {
|
||||
const resolveOpener = opener ?? await getDefaultOpener()
|
||||
return await resolveOpener(dbPath)
|
||||
}
|
||||
|
||||
function hasRequiredTables(db: DatabaseLike): boolean {
|
||||
for (const table of ['session', 'message', 'part']) {
|
||||
try {
|
||||
const row = db.query(
|
||||
`SELECT name FROM sqlite_master WHERE type = 'table' AND name = ?`
|
||||
).get(table) as { name: string } | null | undefined
|
||||
if (!row?.name) return false
|
||||
} catch {
|
||||
return false
|
||||
}
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
function parseJson(value: unknown): JsonRecord | null {
|
||||
if (typeof value !== 'string' || !value.trim()) return null
|
||||
try {
|
||||
return asRecord(JSON.parse(value))
|
||||
} catch {
|
||||
return null
|
||||
}
|
||||
}
|
||||
|
||||
function collectTextParts(db: DatabaseLike, messageId: string, role: string): Array<{ id: string; text: string }> {
|
||||
// Query failures must propagate: a swallowed error here would yield a
|
||||
// partial transcript that the hub later rejects as transcript_diverged.
|
||||
const partRows = db.query(
|
||||
`SELECT id, data FROM part WHERE message_id = ? ORDER BY time_created ASC`
|
||||
).all(messageId) as Array<{ id: string; data: unknown }>
|
||||
const texts: Array<{ id: string; text: string }> = []
|
||||
for (const partRow of partRows) {
|
||||
const parsed = parseJson(partRow.data)
|
||||
if (parsed?.type !== 'text' || typeof parsed.text !== 'string') continue
|
||||
if (!parsed.text.trim()) continue
|
||||
// Mirror the live scanner gate: assistant text parts are mutable while
|
||||
// the response is still streaming; only finished (or synthetic/user)
|
||||
// parts are stable enough to import.
|
||||
if (role !== 'user' && parsed.synthetic !== true) {
|
||||
const time = asRecord(parsed.time)
|
||||
if (typeof time?.end !== 'number') continue
|
||||
}
|
||||
texts.push({ id: partRow.id, text: parsed.text })
|
||||
}
|
||||
return texts
|
||||
}
|
||||
|
||||
function extractLastUserMessage(db: DatabaseLike, sessionId: string): string | null {
|
||||
let messageRows: Array<{ id: string; data: unknown }> = []
|
||||
try {
|
||||
messageRows = db.query(
|
||||
`SELECT id, data FROM message WHERE session_id = ? ORDER BY time_created DESC`
|
||||
).all(sessionId) as Array<{ id: string; data: unknown }>
|
||||
} catch {
|
||||
return null
|
||||
}
|
||||
for (const messageRow of messageRows) {
|
||||
const parsed = parseJson(messageRow.data)
|
||||
if (parsed?.role !== 'user') continue
|
||||
let text: string
|
||||
try {
|
||||
text = collectTextParts(db, messageRow.id, 'user')
|
||||
.filter((part) => !part.text.startsWith('<'))
|
||||
.map((part) => part.text)
|
||||
.join('')
|
||||
.trim()
|
||||
} catch {
|
||||
continue
|
||||
}
|
||||
if (!text) continue
|
||||
return truncateText(text, LAST_USER_MESSAGE_MAX_LENGTH)
|
||||
}
|
||||
return null
|
||||
}
|
||||
|
||||
function importedUser(text: string): OpencodeImportedMessageContent {
|
||||
return {
|
||||
role: 'user',
|
||||
content: { type: 'text', text },
|
||||
meta: { sentFrom: 'cli' }
|
||||
}
|
||||
}
|
||||
|
||||
function importedAgent(text: string, id: string): OpencodeImportedMessageContent {
|
||||
return {
|
||||
role: 'agent',
|
||||
content: { type: AGENT_MESSAGE_PAYLOAD_TYPE, data: { type: 'message', message: text, id } },
|
||||
meta: { sentFrom: 'cli' }
|
||||
}
|
||||
}
|
||||
|
||||
function normalizeTimestamp(value: number | string | null | undefined, fallback: number): number {
|
||||
const timestamp = Number(value)
|
||||
if (!Number.isFinite(timestamp)) return fallback
|
||||
return timestamp < 1_000_000_000_000 ? timestamp * 1000 : timestamp
|
||||
}
|
||||
|
||||
async function buildSessionMessages(
|
||||
db: DatabaseLike,
|
||||
summary: OpencodeLocalSessionSummary
|
||||
): Promise<OpencodeLocalSessionWithMessages> {
|
||||
const messages: OpencodeImportedMessage[] = []
|
||||
let createdAt = summary.modifiedAt
|
||||
try {
|
||||
const messageRows = db.query(
|
||||
`SELECT id, data, time_created FROM message WHERE session_id = ? ORDER BY time_created ASC`
|
||||
).all(summary.id) as Array<{ id: string; data: unknown; time_created: number | string }>
|
||||
for (const messageRow of messageRows) {
|
||||
const parsed = parseJson(messageRow.data)
|
||||
const role = typeof parsed?.role === 'string' ? parsed.role : null
|
||||
if (!role) continue
|
||||
createdAt = normalizeTimestamp(messageRow.time_created, createdAt)
|
||||
// No silent catch here either: a mid-transcript query failure must
|
||||
// fail the whole read instead of returning a partial snapshot that
|
||||
// the hub would permanently reject as transcript_diverged later.
|
||||
for (const part of collectTextParts(db, messageRow.id, role)) {
|
||||
messages.push({
|
||||
localId: `opencode:${summary.id}:${messageRow.id}:${part.id}`,
|
||||
createdAt,
|
||||
content: role === 'user' ? importedUser(part.text) : importedAgent(part.text, part.id)
|
||||
})
|
||||
}
|
||||
}
|
||||
} catch (error) {
|
||||
if (messages.length === 0 && error instanceof Error && /no such table: message/i.test(error.message)) {
|
||||
// Fresh installations without the message table yet: empty history.
|
||||
// Anything else (including a missing part table mid-transcript)
|
||||
// must propagate so callers never persist a partial snapshot.
|
||||
return { ...summary, messages }
|
||||
}
|
||||
throw error
|
||||
}
|
||||
return { ...summary, messages }
|
||||
}
|
||||
|
||||
export async function listLocalOpencodeSessionSummaries(
|
||||
limit = DEFAULT_OPENCODE_SESSION_SCAN_LIMIT,
|
||||
opener?: OpencodeDatabaseOpener
|
||||
): Promise<OpencodeLocalSessionSummary[]> {
|
||||
if (limit <= 0) return []
|
||||
const dbPath = getOpencodeDbPath()
|
||||
const db = await openDb(dbPath, opener)
|
||||
if (!db) return []
|
||||
try {
|
||||
if (!hasRequiredTables(db)) return []
|
||||
const exec = db.exec?.bind(db)
|
||||
const inTransaction = typeof exec === 'function'
|
||||
if (inTransaction) exec('BEGIN DEFERRED')
|
||||
let rows: SessionRow[] = []
|
||||
try {
|
||||
rows = db.query(
|
||||
`SELECT id, title, directory, time_updated FROM session WHERE time_archived IS NULL ORDER BY time_updated DESC LIMIT ?`
|
||||
).all(limit) as SessionRow[]
|
||||
} catch {
|
||||
if (inTransaction) {
|
||||
try { exec('ROLLBACK') } catch {}
|
||||
}
|
||||
return []
|
||||
}
|
||||
try {
|
||||
const summaries = rows.map((row) => ({
|
||||
id: row.id,
|
||||
title: row.title ?? '',
|
||||
lastUserMessage: extractLastUserMessage(db, row.id),
|
||||
cwd: row.directory ?? null,
|
||||
file: dbPath,
|
||||
modifiedAt: normalizeTimestamp(row.time_updated, Date.now())
|
||||
}))
|
||||
if (inTransaction) exec('COMMIT')
|
||||
return summaries
|
||||
} catch (error) {
|
||||
if (inTransaction) {
|
||||
try { exec('ROLLBACK') } catch {}
|
||||
}
|
||||
throw error
|
||||
}
|
||||
} finally {
|
||||
db.close()
|
||||
}
|
||||
}
|
||||
|
||||
export async function listLocalOpencodeSessionsWithMessagesByIds(
|
||||
ids: Set<string>,
|
||||
opener?: OpencodeDatabaseOpener
|
||||
): Promise<OpencodeLocalSessionWithMessages[]> {
|
||||
if (ids.size === 0) return []
|
||||
const dbPath = getOpencodeDbPath()
|
||||
const db = await openDb(dbPath, opener)
|
||||
if (!db) return []
|
||||
try {
|
||||
if (!hasRequiredTables(db)) return []
|
||||
const sessions: OpencodeLocalSessionWithMessages[] = []
|
||||
for (const id of ids) {
|
||||
let rows: SessionRow[] = []
|
||||
try {
|
||||
rows = db.query(
|
||||
`SELECT id, title, directory, time_updated FROM session WHERE id = ?`
|
||||
).all(id) as SessionRow[]
|
||||
} catch {
|
||||
continue
|
||||
}
|
||||
const row = rows[0]
|
||||
if (!row) continue
|
||||
// Read messages and their parts inside one deferred transaction:
|
||||
// OpenCode may write while we scan, and a snapshot hole here would
|
||||
// permanently fail re-imports as transcript_diverged.
|
||||
const exec = db.exec?.bind(db)
|
||||
const inTransaction = typeof exec === 'function'
|
||||
if (inTransaction) exec('BEGIN DEFERRED')
|
||||
try {
|
||||
const summary: OpencodeLocalSessionSummary = {
|
||||
id: row.id,
|
||||
title: row.title ?? '',
|
||||
lastUserMessage: extractLastUserMessage(db, row.id),
|
||||
cwd: row.directory ?? null,
|
||||
file: dbPath,
|
||||
modifiedAt: normalizeTimestamp(row.time_updated, Date.now())
|
||||
}
|
||||
sessions.push(await buildSessionMessages(db, summary))
|
||||
if (inTransaction) exec('COMMIT')
|
||||
} catch (error) {
|
||||
if (inTransaction) {
|
||||
try { exec('ROLLBACK') } catch {}
|
||||
}
|
||||
throw error
|
||||
}
|
||||
}
|
||||
// same ordering semantics as summaries: modifiedAt desc
|
||||
return sessions.sort((a, b) => b.modifiedAt - a.modifiedAt)
|
||||
} finally {
|
||||
db.close()
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user