Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 7 additions & 0 deletions src/memory/contract.ts
Original file line number Diff line number Diff line change
Expand Up @@ -110,6 +110,13 @@ export interface Storage {
appendMessage(m: Omit<MessageRow, 'id' | 'created_at'>): Promise<MessageRow>;
sessionMessages(sessionId: string): Promise<MessageRow[]>;

// 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<void>;
takePartial(sessionId: string): Promise<{ turnId: string; json: string } | null>;

// notes (explicit-write only; supersession, never delete)
insertNote(n: Omit<NoteRow, 'id' | 'created_at' | 'superseded_by' | 'superseded_at' | 'superseded_reason'>): Promise<NoteRow>;
activeNotes(scopeKeyPrefix?: string): Promise<NoteRow[]>;
Expand Down
20 changes: 20 additions & 0 deletions src/memory/postgres.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand Down Expand Up @@ -190,6 +197,19 @@ export class PostgresStorage implements Storage {
return { ...m, id, created_at: created };
}

async stashPartial(sessionId: string, turnId: string, json: string): Promise<void> {
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<MessageRow[]> {
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) => ({
Expand Down
18 changes: 18 additions & 0 deletions src/memory/sqlite.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand Down Expand Up @@ -171,6 +178,17 @@ export class SqliteStorage implements Storage {
return { ...m, id, created_at: created };
}

async stashPartial(sessionId: string, turnId: string, json: string): Promise<void> {
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<MessageRow[]> {
return this.db.prepare('SELECT * FROM butler_messages WHERE session_id = ? ORDER BY rowid').all(sessionId) as MessageRow[];
}
Expand Down
37 changes: 32 additions & 5 deletions src/transport/conversation.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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';
Expand Down Expand Up @@ -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);
Expand Down Expand Up @@ -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 } : {}),
Expand Down Expand Up @@ -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
Expand Down
136 changes: 136 additions & 0 deletions test/conversation-budget.test.ts
Original file line number Diff line number Diff line change
@@ -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();
});
});
Loading