diff --git a/src/memory/contract.ts b/src/memory/contract.ts index 07b0e42..0b93279 100644 --- a/src/memory/contract.ts +++ b/src/memory/contract.ts @@ -110,6 +110,13 @@ export interface Storage { appendMessage(m: Omit): Promise; sessionMessages(sessionId: string): Promise; + // partial-turn stash (no-hassle resume, 2026-10-21 plan): a budget-exhausted + // chat turn parks its transcript (tool results included, opaque JSON) here; + // the NEXT message in the same session takes it — one-shot, consumed on + // read — to seed the loop so the model continues instead of redoing hops. + stashPartial(sessionId: string, turnId: string, json: string): Promise; + takePartial(sessionId: string): Promise<{ turnId: string; json: string } | null>; + // notes (explicit-write only; supersession, never delete) insertNote(n: Omit): Promise; activeNotes(scopeKeyPrefix?: string): Promise; diff --git a/src/memory/postgres.ts b/src/memory/postgres.ts index ea9c02d..87a80b6 100644 --- a/src/memory/postgres.ts +++ b/src/memory/postgres.ts @@ -99,6 +99,13 @@ CREATE TABLE IF NOT EXISTS butler_artifacts( ); CREATE INDEX IF NOT EXISTS butler_artifacts_kind_name ON butler_artifacts(kind, name); CREATE INDEX IF NOT EXISTS butler_artifacts_run ON butler_artifacts(run_id); + +CREATE TABLE IF NOT EXISTS butler_turn_partials( + session_id TEXT PRIMARY KEY, + turn_id TEXT NOT NULL, + json TEXT NOT NULL, + created_at BIGINT NOT NULL +); `; export class PostgresStorage implements Storage { @@ -190,6 +197,19 @@ export class PostgresStorage implements Storage { return { ...m, id, created_at: created }; } + async stashPartial(sessionId: string, turnId: string, json: string): Promise { + await this.pool.query( + 'INSERT INTO butler_turn_partials(session_id, turn_id, json, created_at) VALUES ($1,$2,$3,$4) ON CONFLICT (session_id) DO UPDATE SET turn_id = $2, json = $3, created_at = $4', + [sessionId, turnId, json, Date.now()], + ); + } + + async takePartial(sessionId: string): Promise<{ turnId: string; json: string } | null> { + const r = await this.pool.query('DELETE FROM butler_turn_partials WHERE session_id = $1 RETURNING turn_id, json', [sessionId]); + const row = r.rows[0] as { turn_id: string; json: string } | undefined; + return row ? { turnId: String(row.turn_id), json: String(row.json) } : null; + } + async sessionMessages(sessionId: string): Promise { const r = await this.pool.query('SELECT * FROM butler_messages WHERE session_id = $1 ORDER BY created_at ASC, id ASC', [sessionId]); return r.rows.map((x) => ({ diff --git a/src/memory/sqlite.ts b/src/memory/sqlite.ts index bd5add2..a2ef542 100644 --- a/src/memory/sqlite.ts +++ b/src/memory/sqlite.ts @@ -102,6 +102,13 @@ CREATE TABLE IF NOT EXISTS butler_artifacts( ); CREATE INDEX IF NOT EXISTS butler_artifacts_kind_name ON butler_artifacts(kind, name); CREATE INDEX IF NOT EXISTS butler_artifacts_run ON butler_artifacts(run_id); + +CREATE TABLE IF NOT EXISTS butler_turn_partials( + session_id TEXT PRIMARY KEY, + turn_id TEXT NOT NULL, + json TEXT NOT NULL, + created_at INTEGER NOT NULL +); `; export class SqliteStorage implements Storage { @@ -171,6 +178,17 @@ export class SqliteStorage implements Storage { return { ...m, id, created_at: created }; } + async stashPartial(sessionId: string, turnId: string, json: string): Promise { + this.db.prepare('INSERT OR REPLACE INTO butler_turn_partials(session_id, turn_id, json, created_at) VALUES (?,?,?,?)').run(sessionId, turnId, json, Date.now()); + } + + async takePartial(sessionId: string): Promise<{ turnId: string; json: string } | null> { + const row = this.db.prepare('SELECT turn_id, json FROM butler_turn_partials WHERE session_id = ?').get(sessionId) as { turn_id: string; json: string } | undefined; + if (!row) return null; + this.db.prepare('DELETE FROM butler_turn_partials WHERE session_id = ?').run(sessionId); + return { turnId: row.turn_id, json: row.json }; + } + async sessionMessages(sessionId: string): Promise { return this.db.prepare('SELECT * FROM butler_messages WHERE session_id = ? ORDER BY rowid').all(sessionId) as MessageRow[]; } diff --git a/src/transport/conversation.ts b/src/transport/conversation.ts index c942562..705e248 100644 --- a/src/transport/conversation.ts +++ b/src/transport/conversation.ts @@ -4,7 +4,7 @@ * rows alone (condition 4). */ import type { FastifyInstance } from 'fastify'; import type { Storage } from '../memory/contract.ts'; -import { runLoop } from '../loop.ts'; +import { runLoop, LoopError } from '../loop.ts'; import type { EventSink } from '../orchestrate/event-sink.ts'; import type { Provider, ProviderMessage, ProviderResponse, ProviderTool } from '../provider/index.ts'; import type { Policy, ToolRegistry } from '../tools.ts'; @@ -172,6 +172,20 @@ export async function registerConversationApi(app: FastifyInstance, deps: Conver .filter((m) => m.role === 'user' || (m.role === 'assistant' && m.content)) .map((m) => ({ role: m.role as 'user' | 'assistant', content: m.content ?? '' })); + // no-hassle resume (2026-10-21 plan): if the previous turn died on the + // budget guard, its transcript (tool results included) was stashed — + // seed it back so the model CONTINUES instead of redoing (and re-dying + // on) the same hops. One-shot: takePartial consumes the stash. + let resumeNote = ''; + let seeded: ProviderMessage[] = []; + try { + const stash = await storage.takePartial(id); + if (stash) { + seeded = JSON.parse(stash.json) as ProviderMessage[]; + resumeNote = '\n\n[resume] The previous turn on this session ran out of runway mid-task. Its full transcript, including tool results, is prepended to the history below. Continue that work from where it stopped — do not repeat completed steps.'; + } + } catch { /* a corrupt stash must never kill the next turn */ } + const outcome: TurnOutcome = { sessionId: id, ok: false, tokens: { in: 0, out: 0 }, hops: 0 }; const ac = new AbortController(); activeTurns.set(id, ac); @@ -232,9 +246,11 @@ export async function registerConversationApi(app: FastifyInstance, deps: Conver try { prep = await deps.prepareTurn?.(id, b as { content: string; commands?: string[] }) ?? null; } catch { prep = null; } // decoration failure must never kill the turn const turnProvider = prep?.provider ?? (routedComplete ? { complete: (m, t, o) => routedComplete(m, t, o), model: routedModel ?? deps.provider.model } : deps.provider); - const result = await runLoop(deps.systemPrompt + cmdBlock + notesBlock + (prep?.promptBlock ?? ''), b.content, { + const result = await runLoop(deps.systemPrompt + cmdBlock + notesBlock + (prep?.promptBlock ?? '') + resumeNote, b.content, { sessionKey: id, // R2-F4: per-Lionidas-session claude --resume isolation - history: history.length > 1 ? history.slice(-40, -1) : [], // prior turns only — the current message rides as userMessage + // prior turns only — the current message rides as userMessage; a + // stashed partial (if any) rides FIRST: it is the most recent context + history: [...seeded, ...(history.length > 1 ? history.slice(-40, -1) : [])], provider: turnProvider, tools: toolSet.map((t) => toolFromRegistry(t.name, registry, policy, { sessionId: id })), ...(deps.budget ? { budget: deps.budget } : {}), @@ -294,10 +310,21 @@ export async function registerConversationApi(app: FastifyInstance, deps: Conver return { turn: outcome }; } catch (err) { const e = err as { code?: string; message?: string }; - outcome.error = e.message ?? String(err); + // no-hassle resume (2026-10-21 plan): budget exhaustion is a PAUSE, not + // a session death. Stash the partial transcript (LoopError carries the + // full loop messages — tool results included), keep the session usable, + // and tell the operator how to continue. Every other failure (ABORTED, + // provider, unknown) keeps today's honest session-poisoning. + const budgetExhausted = err instanceof LoopError && err.code === 'BUDGET_EXCEEDED'; + if (budgetExhausted) { + try { await storage.stashPartial(id, durableTurnId, JSON.stringify(err.partial ?? [])); } catch { /* stash is best-effort */ } + outcome.error = 'ran out of runway — the work so far is kept; send a message to continue from where it stopped'; + } else { + outcome.error = e.message ?? String(err); + } if (deps.sink) deps.sink.handle(id, { kind: 'turn.fail', sessionId: id, turnId: durableTurnId, error: outcome.error }, `chat:${durableTurnId}:fail`); outcome.code = e.code ?? undefined; - await storage.updateSession(id, { status: 'error' }); + await storage.updateSession(id, { status: budgetExhausted ? 'active' : 'error' }); activeTurns.delete(id); try { deps.afterTurn?.(id, { ok: false, turnId: durableTurnId, answer: '' }); } catch { /* host hook best-effort */ } return { turn: outcome }; // the turn failed; the REQUEST succeeded — outcome carries the verdict diff --git a/test/conversation-budget.test.ts b/test/conversation-budget.test.ts new file mode 100644 index 0000000..12a8bd7 --- /dev/null +++ b/test/conversation-budget.test.ts @@ -0,0 +1,136 @@ +import { describe, expect, it } from 'vitest'; +import Fastify from 'fastify'; +import type { Provider, ProviderMessage, ProviderTool } from '../src/provider/types.ts'; +import { MockProvider, type ScriptedTurn } from '../src/provider/mock.ts'; +import { SqliteStorage } from '../src/memory/sqlite.ts'; +import { ToolRegistry } from '../src/tools.ts'; +import { registerConversationApi } from '../src/transport/conversation.ts'; + +/** no-hassle long-running turns (2026-10-21 plan, Stage A): budget + * exhaustion is a PAUSE — friendly fail, session stays usable, partial + * transcript stashed — and the NEXT message seeds it back (invisible + * resume) instead of redoing the same hops. ABORTED keeps poisoning. */ + +const noopCall = (i: number): ProviderMessage['tool_calls'] => [ + { id: `c${i}`, type: 'function', function: { name: 'noop', arguments: '{}' } }, +]; + +function toolTurn(i: number): ScriptedTurn { + return { kind: 'assistant', content: null, toolCalls: noopCall(i) }; +} + +/** provider wrapper that CAPTURES the message list of every call — proves + * (or disproves) that a stashed partial was seeded into the next turn. */ +class CapturingProvider implements Provider { + readonly model = 'mock-model'; + readonly seen: ProviderMessage[][] = []; + calls = 0; + constructor(private readonly inner: MockProvider) {} + async complete(messages: ProviderMessage[], tools: ProviderTool[], opts: { remainingMs: number; signal?: AbortSignal }) { + this.calls += 1; + this.seen.push(messages.map((m) => ({ ...m }))); + return this.inner.complete(messages, tools, opts); + } +} + +async function boot(provider: Provider, budget: { maxHops?: number; wallMs?: number }, storage: SqliteStorage) { + const app = Fastify(); + const registry = new ToolRegistry().register({ + name: 'noop', description: 'n', permission: 'read', parameters: { type: 'object', properties: {} }, + execute: async () => `observation ${Date.now()}`, + }); + await registerConversationApi(app, { + storage, provider, registry, policy: { name: 'draft', allows: 'draft' }, + systemPrompt: 'sys', budget, emit: () => undefined, + }); + await app.listen({ port: 0, host: '127.0.0.1' }); + return { app, baseUrl: `http://127.0.0.1:${(app.server.address() as { port: number }).port}` }; +} + +const post = (b: string, sid: string, content: string) => + fetch(`${b}/api/agent/sessions/${sid}/messages`, { method: 'POST', headers: { 'content-type': 'application/json' }, body: JSON.stringify({ content }) }) + .then((r) => r.json() as Promise<{ turn: { ok: boolean; code?: string; error?: string; answer?: string } }>); + +describe('partial stash storage', () => { + it('stash → take roundtrip is one-shot and replaceable', async () => { + const s = new SqliteStorage(':memory:'); + await s.stashPartial('sess_x', 't_1', '[{"role":"tool"}]'); + await s.stashPartial('sess_x', 't_2', '[{"role":"tool","content":"newer"}]'); // REPLACE wins + const got = await s.takePartial('sess_x'); + expect(got).toEqual({ turnId: 't_2', json: '[{"role":"tool","content":"newer"}]' }); + expect(await s.takePartial('sess_x')).toBeNull(); // consumed + s.close(); + }); +}); + +describe('budget exhaustion pauses, resume continues', () => { + it('exhaust → friendly fail, session active, stash seeded into the NEXT turn (not redone)', async () => { + const storage = new SqliteStorage(':memory:'); + // app A: a model that tool-calls forever → dies at maxHops 3 + const exhaust = await boot(new MockProvider([toolTurn(0)]), { maxHops: 3, wallMs: 60_000 }, storage); + const created = await (await fetch(`${exhaust.baseUrl}/api/agent/sessions`, { method: 'POST', headers: { 'content-type': 'application/json' }, body: '{}' })).json() as { session: { id: string } }; + const sid = created.session.id; + + const failed = await post(exhaust.baseUrl, sid, 'do a deep multi-step thing'); + expect(failed.turn.ok).toBe(false); + expect(failed.turn.code).toBe('BUDGET_EXCEEDED'); + expect(failed.turn.error).toContain('continue from where it stopped'); + + // session stays usable — no dead-end error state + const sess = await (await fetch(`${exhaust.baseUrl}/api/agent/sessions/${sid}`)).json() as { session: { status: string } }; + expect(sess.session.status).toBe('active'); + + // app B (same storage): the follow-up turn completes and RECEIVES the + // stashed transcript — history rows alone never carry role:'tool' + const capturer = new CapturingProvider(new MockProvider([{ kind: 'assistant', content: 'resumed and finished' }])); + const resume = await boot(capturer, { maxHops: 5, wallMs: 60_000 }, storage); + const done = await post(resume.baseUrl, sid, 'continue'); + expect(done.turn.ok).toBe(true); + expect(done.turn.answer).toBe('resumed and finished'); + + const firstCall = capturer.seen[0]!; + const seededTool = firstCall.find((m) => m.role === 'tool'); + expect(seededTool).toBeDefined(); // the 3 exhausted hops' results rode along + expect(firstCall.some((m) => m.role === 'user' && m.content === 'continue')).toBe(true); + + // one-shot: the stash was consumed by the resume turn + expect(await storage.takePartial(sid)).toBeNull(); + + await exhaust.app.close(); await resume.app.close(); storage.close(); + }); + + it('a fresh session gets NO seed (stash is per-session)', async () => { + const storage = new SqliteStorage(':memory:'); + const exhaust = await boot(new MockProvider([toolTurn(0)]), { maxHops: 2, wallMs: 60_000 }, storage); + const a = await (await fetch(`${exhaust.baseUrl}/api/agent/sessions`, { method: 'POST', headers: { 'content-type': 'application/json' }, body: '{}' })).json() as { session: { id: string } }; + await post(exhaust.baseUrl, a.session.id, 'exhaust it'); // stashes for session A + + const capturer = new CapturingProvider(new MockProvider([{ kind: 'assistant', content: 'fresh answer' }])); + const fresh = await boot(capturer, { maxHops: 4, wallMs: 60_000 }, storage); + const b = await (await fetch(`${fresh.baseUrl}/api/agent/sessions`, { method: 'POST', headers: { 'content-type': 'application/json' }, body: '{}' })).json() as { session: { id: string } }; + const done = await post(fresh.baseUrl, b.session.id, 'unrelated ask'); + expect(done.turn.answer).toBe('fresh answer'); + expect(capturer.seen[0]!.some((m) => m.role === 'tool')).toBe(false); // no leakage + await exhaust.app.close(); await fresh.app.close(); storage.close(); + }); +}); + +describe('non-budget failures keep today\'s behavior', () => { + it('an ABORTED turn still poisons the session to error (regression guard)', async () => { + const storage = new SqliteStorage(':memory:'); + const hang = await boot(new MockProvider([{ kind: 'hang', ms: 30_000 }]), { maxHops: 5, wallMs: 60_000 }, storage); + const created = await (await fetch(`${hang.baseUrl}/api/agent/sessions`, { method: 'POST', headers: { 'content-type': 'application/json' }, body: '{}' })).json() as { session: { id: string } }; + const sid = created.session.id; + const turnPromise = fetch(`${hang.baseUrl}/api/agent/sessions/${sid}/messages`, { + method: 'POST', headers: { 'content-type': 'application/json' }, body: JSON.stringify({ content: 'hang please' }), + }).then((r) => r.json() as Promise<{ turn: { ok: boolean; code?: string } }>); + await new Promise((r) => setTimeout(r, 300)); + await fetch(`${hang.baseUrl}/api/agent/sessions/${sid}/abort`, { method: 'POST' }); + const out = await turnPromise; + expect(out.turn.ok).toBe(false); + expect(out.turn.code).toBe('ABORTED'); + const sess = await (await fetch(`${hang.baseUrl}/api/agent/sessions/${sid}`)).json() as { session: { status: string } }; + expect(sess.session.status).toBe('error'); // only the budget path softens + await hang.app.close(); storage.close(); + }); +});