diff --git a/services/agent-harness/drizzle.config.ts b/services/agent-harness/drizzle.config.ts new file mode 100644 index 0000000000..27cf9ffb72 --- /dev/null +++ b/services/agent-harness/drizzle.config.ts @@ -0,0 +1,8 @@ +import { defineConfig } from 'drizzle-kit'; + +export default defineConfig({ + out: './drizzle', + schema: './src/db/sqlite-schema.ts', + dialect: 'sqlite', + driver: 'durable-sqlite', +}); diff --git a/services/agent-harness/drizzle/0000_authoritative_store.sql b/services/agent-harness/drizzle/0000_authoritative_store.sql new file mode 100644 index 0000000000..5461a21c2a --- /dev/null +++ b/services/agent-harness/drizzle/0000_authoritative_store.sql @@ -0,0 +1,116 @@ +CREATE TABLE `attempts` ( + `id` text PRIMARY KEY NOT NULL, + `tool_call_id` text NOT NULL, + `generation` integer NOT NULL, + `intent` text NOT NULL, + `outcome` text, + `provider_reference` text, + FOREIGN KEY (`tool_call_id`) REFERENCES `calls`(`id`) ON UPDATE no action ON DELETE no action +); +--> statement-breakpoint +CREATE UNIQUE INDEX `attempt_generation` ON `attempts` (`tool_call_id`,`generation`);--> statement-breakpoint +CREATE TABLE `calls` ( + `id` text PRIMARY KEY NOT NULL, + `run_id` text NOT NULL, + `checkpoint_id` text NOT NULL, + `position` integer NOT NULL, + `input_digest` text NOT NULL, + `data` text NOT NULL, + `policy` text NOT NULL, + `state` text NOT NULL, + `revision` integer DEFAULT 0 NOT NULL, + FOREIGN KEY (`run_id`) REFERENCES `runs`(`id`) ON UPDATE no action ON DELETE no action, + FOREIGN KEY (`checkpoint_id`) REFERENCES `checkpoints`(`id`) ON UPDATE no action ON DELETE no action +); +--> statement-breakpoint +CREATE UNIQUE INDEX `call_order` ON `calls` (`run_id`,`position`);--> statement-breakpoint +CREATE TABLE `checkpoints` ( + `id` text PRIMARY KEY NOT NULL, + `run_id` text NOT NULL, + `step` integer NOT NULL, + `status` text NOT NULL, + `data` text NOT NULL, + `definition_versions` text NOT NULL, + FOREIGN KEY (`run_id`) REFERENCES `runs`(`id`) ON UPDATE no action ON DELETE no action +); +--> statement-breakpoint +CREATE UNIQUE INDEX `checkpoint_step` ON `checkpoints` (`run_id`,`step`);--> statement-breakpoint +CREATE TABLE `client_actions` ( + `tool_call_id` text PRIMARY KEY NOT NULL, + `sequence` integer NOT NULL, + `data` text NOT NULL +); +--> statement-breakpoint +CREATE TABLE `commands` ( + `id` text PRIMARY KEY NOT NULL, + `fingerprint` text NOT NULL, + `reply` text NOT NULL, + `sequence` integer NOT NULL +); +--> statement-breakpoint +CREATE TABLE `conversation` ( + `singleton` integer PRIMARY KEY DEFAULT 1 NOT NULL, + `id` text NOT NULL, + `owner_user_id` text NOT NULL, + `context` text NOT NULL, + `permission_mode` text DEFAULT 'ask' NOT NULL, + `permission_revision` integer DEFAULT 0 NOT NULL, + `sequence` integer DEFAULT 0 NOT NULL, + `compacted_through` integer DEFAULT 0 NOT NULL, + `active_run_id` text, + `legacy_cursor` integer DEFAULT 0 NOT NULL, + CONSTRAINT "one_conversation" CHECK("conversation"."singleton" = 1) +); +--> statement-breakpoint +CREATE UNIQUE INDEX `conversation_id_unique` ON `conversation` (`id`);--> statement-breakpoint +CREATE TABLE `events` ( + `sequence` integer PRIMARY KEY NOT NULL, + `data` text NOT NULL +); +--> statement-breakpoint +CREATE TABLE `grants` ( + `id` text PRIMARY KEY NOT NULL, + `tool_call_id` text NOT NULL, + `generation` integer NOT NULL, + `data` text NOT NULL, + FOREIGN KEY (`tool_call_id`) REFERENCES `calls`(`id`) ON UPDATE no action ON DELETE no action +); +--> statement-breakpoint +CREATE UNIQUE INDEX `grant_generation` ON `grants` (`tool_call_id`,`generation`);--> statement-breakpoint +CREATE TABLE `interactions` ( + `id` text PRIMARY KEY NOT NULL, + `sequence` integer NOT NULL, + `resolved` integer NOT NULL, + `data` text NOT NULL +); +--> statement-breakpoint +CREATE INDEX `unresolved_interactions` ON `interactions` (`resolved`,`sequence`);--> statement-breakpoint +CREATE TABLE `messages` ( + `id` text PRIMARY KEY NOT NULL, + `sequence` integer NOT NULL, + `created_at` text NOT NULL, + `data` text NOT NULL +); +--> statement-breakpoint +CREATE UNIQUE INDEX `messages_sequence_unique` ON `messages` (`sequence`);--> statement-breakpoint +CREATE INDEX `message_history` ON `messages` (`created_at`,`id`);--> statement-breakpoint +CREATE TABLE `runs` ( + `id` text PRIMARY KEY NOT NULL, + `position` integer NOT NULL, + `status` text NOT NULL, + `data` text NOT NULL, + `revision` integer DEFAULT 0 NOT NULL, + `step` integer DEFAULT 0 NOT NULL, + `active_slot` integer, + CONSTRAINT "active_run_slot" CHECK(("runs"."status" IN ('running', 'waiting', 'stopping') AND "runs"."active_slot" IS 1) OR ("runs"."status" NOT IN ('running', 'waiting', 'stopping') AND "runs"."active_slot" IS NULL)) +); +--> statement-breakpoint +CREATE UNIQUE INDEX `runs_position_unique` ON `runs` (`position`);--> statement-breakpoint +CREATE UNIQUE INDEX `runs_active_slot_unique` ON `runs` (`active_slot`);--> statement-breakpoint +CREATE INDEX `run_queue` ON `runs` (`status`,`position`);--> statement-breakpoint +CREATE TABLE `snapshots` ( + `singleton` integer PRIMARY KEY DEFAULT 1 NOT NULL, + `cursor` integer NOT NULL, + `data` text NOT NULL, + CONSTRAINT "one_snapshot" CHECK("snapshots"."singleton" = 1) +); diff --git a/services/agent-harness/drizzle/0001_projection_work.sql b/services/agent-harness/drizzle/0001_projection_work.sql new file mode 100644 index 0000000000..116c59d6a2 --- /dev/null +++ b/services/agent-harness/drizzle/0001_projection_work.sql @@ -0,0 +1,10 @@ +CREATE TABLE `projection_work` ( + `id` text PRIMARY KEY NOT NULL, + `message_id` text NOT NULL, + `data` text NOT NULL, + `revision` integer DEFAULT 0 NOT NULL, + `due_at` integer NOT NULL, + `acknowledged_at` text +); +--> statement-breakpoint +CREATE INDEX `due_projections` ON `projection_work` (`acknowledged_at`,`due_at`,`id`); \ No newline at end of file diff --git a/services/agent-harness/drizzle/meta/0000_snapshot.json b/services/agent-harness/drizzle/meta/0000_snapshot.json new file mode 100644 index 0000000000..f6d9371e30 --- /dev/null +++ b/services/agent-harness/drizzle/meta/0000_snapshot.json @@ -0,0 +1,758 @@ +{ + "version": "6", + "dialect": "sqlite", + "id": "0abb8b8c-affd-463d-8b01-dcbb2158c797", + "prevId": "00000000-0000-0000-0000-000000000000", + "tables": { + "attempts": { + "name": "attempts", + "columns": { + "id": { + "name": "id", + "type": "text", + "primaryKey": true, + "notNull": true, + "autoincrement": false + }, + "tool_call_id": { + "name": "tool_call_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "generation": { + "name": "generation", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "intent": { + "name": "intent", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "outcome": { + "name": "outcome", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "provider_reference": { + "name": "provider_reference", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + } + }, + "indexes": { + "attempt_generation": { + "name": "attempt_generation", + "columns": [ + "tool_call_id", + "generation" + ], + "isUnique": true + } + }, + "foreignKeys": { + "attempts_tool_call_id_calls_id_fk": { + "name": "attempts_tool_call_id_calls_id_fk", + "tableFrom": "attempts", + "tableTo": "calls", + "columnsFrom": [ + "tool_call_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "no action", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": {} + }, + "calls": { + "name": "calls", + "columns": { + "id": { + "name": "id", + "type": "text", + "primaryKey": true, + "notNull": true, + "autoincrement": false + }, + "run_id": { + "name": "run_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "checkpoint_id": { + "name": "checkpoint_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "position": { + "name": "position", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "input_digest": { + "name": "input_digest", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "data": { + "name": "data", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "policy": { + "name": "policy", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "state": { + "name": "state", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "revision": { + "name": "revision", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": 0 + } + }, + "indexes": { + "call_order": { + "name": "call_order", + "columns": [ + "run_id", + "position" + ], + "isUnique": true + } + }, + "foreignKeys": { + "calls_run_id_runs_id_fk": { + "name": "calls_run_id_runs_id_fk", + "tableFrom": "calls", + "tableTo": "runs", + "columnsFrom": [ + "run_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "no action", + "onUpdate": "no action" + }, + "calls_checkpoint_id_checkpoints_id_fk": { + "name": "calls_checkpoint_id_checkpoints_id_fk", + "tableFrom": "calls", + "tableTo": "checkpoints", + "columnsFrom": [ + "checkpoint_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "no action", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": {} + }, + "checkpoints": { + "name": "checkpoints", + "columns": { + "id": { + "name": "id", + "type": "text", + "primaryKey": true, + "notNull": true, + "autoincrement": false + }, + "run_id": { + "name": "run_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "step": { + "name": "step", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "status": { + "name": "status", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "data": { + "name": "data", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "definition_versions": { + "name": "definition_versions", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + } + }, + "indexes": { + "checkpoint_step": { + "name": "checkpoint_step", + "columns": [ + "run_id", + "step" + ], + "isUnique": true + } + }, + "foreignKeys": { + "checkpoints_run_id_runs_id_fk": { + "name": "checkpoints_run_id_runs_id_fk", + "tableFrom": "checkpoints", + "tableTo": "runs", + "columnsFrom": [ + "run_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "no action", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": {} + }, + "client_actions": { + "name": "client_actions", + "columns": { + "tool_call_id": { + "name": "tool_call_id", + "type": "text", + "primaryKey": true, + "notNull": true, + "autoincrement": false + }, + "sequence": { + "name": "sequence", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "data": { + "name": "data", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + } + }, + "indexes": {}, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": {} + }, + "commands": { + "name": "commands", + "columns": { + "id": { + "name": "id", + "type": "text", + "primaryKey": true, + "notNull": true, + "autoincrement": false + }, + "fingerprint": { + "name": "fingerprint", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "reply": { + "name": "reply", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "sequence": { + "name": "sequence", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + } + }, + "indexes": {}, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": {} + }, + "conversation": { + "name": "conversation", + "columns": { + "singleton": { + "name": "singleton", + "type": "integer", + "primaryKey": true, + "notNull": true, + "autoincrement": false, + "default": 1 + }, + "id": { + "name": "id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "owner_user_id": { + "name": "owner_user_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "context": { + "name": "context", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "permission_mode": { + "name": "permission_mode", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": "'ask'" + }, + "permission_revision": { + "name": "permission_revision", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": 0 + }, + "sequence": { + "name": "sequence", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": 0 + }, + "compacted_through": { + "name": "compacted_through", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": 0 + }, + "active_run_id": { + "name": "active_run_id", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "legacy_cursor": { + "name": "legacy_cursor", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": 0 + } + }, + "indexes": { + "conversation_id_unique": { + "name": "conversation_id_unique", + "columns": [ + "id" + ], + "isUnique": true + } + }, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": { + "one_conversation": { + "name": "one_conversation", + "value": "\"conversation\".\"singleton\" = 1" + } + } + }, + "events": { + "name": "events", + "columns": { + "sequence": { + "name": "sequence", + "type": "integer", + "primaryKey": true, + "notNull": true, + "autoincrement": false + }, + "data": { + "name": "data", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + } + }, + "indexes": {}, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": {} + }, + "grants": { + "name": "grants", + "columns": { + "id": { + "name": "id", + "type": "text", + "primaryKey": true, + "notNull": true, + "autoincrement": false + }, + "tool_call_id": { + "name": "tool_call_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "generation": { + "name": "generation", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "data": { + "name": "data", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + } + }, + "indexes": { + "grant_generation": { + "name": "grant_generation", + "columns": [ + "tool_call_id", + "generation" + ], + "isUnique": true + } + }, + "foreignKeys": { + "grants_tool_call_id_calls_id_fk": { + "name": "grants_tool_call_id_calls_id_fk", + "tableFrom": "grants", + "tableTo": "calls", + "columnsFrom": [ + "tool_call_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "no action", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": {} + }, + "interactions": { + "name": "interactions", + "columns": { + "id": { + "name": "id", + "type": "text", + "primaryKey": true, + "notNull": true, + "autoincrement": false + }, + "sequence": { + "name": "sequence", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "resolved": { + "name": "resolved", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "data": { + "name": "data", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + } + }, + "indexes": { + "unresolved_interactions": { + "name": "unresolved_interactions", + "columns": [ + "resolved", + "sequence" + ], + "isUnique": false + } + }, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": {} + }, + "messages": { + "name": "messages", + "columns": { + "id": { + "name": "id", + "type": "text", + "primaryKey": true, + "notNull": true, + "autoincrement": false + }, + "sequence": { + "name": "sequence", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "created_at": { + "name": "created_at", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "data": { + "name": "data", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + } + }, + "indexes": { + "messages_sequence_unique": { + "name": "messages_sequence_unique", + "columns": [ + "sequence" + ], + "isUnique": true + }, + "message_history": { + "name": "message_history", + "columns": [ + "created_at", + "id" + ], + "isUnique": false + } + }, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": {} + }, + "runs": { + "name": "runs", + "columns": { + "id": { + "name": "id", + "type": "text", + "primaryKey": true, + "notNull": true, + "autoincrement": false + }, + "position": { + "name": "position", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "status": { + "name": "status", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "data": { + "name": "data", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "revision": { + "name": "revision", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": 0 + }, + "step": { + "name": "step", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": 0 + }, + "active_slot": { + "name": "active_slot", + "type": "integer", + "primaryKey": false, + "notNull": false, + "autoincrement": false + } + }, + "indexes": { + "runs_position_unique": { + "name": "runs_position_unique", + "columns": [ + "position" + ], + "isUnique": true + }, + "runs_active_slot_unique": { + "name": "runs_active_slot_unique", + "columns": [ + "active_slot" + ], + "isUnique": true + }, + "run_queue": { + "name": "run_queue", + "columns": [ + "status", + "position" + ], + "isUnique": false + } + }, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": { + "active_run_slot": { + "name": "active_run_slot", + "value": "(\"runs\".\"status\" IN ('running', 'waiting', 'stopping') AND \"runs\".\"active_slot\" IS 1) OR (\"runs\".\"status\" NOT IN ('running', 'waiting', 'stopping') AND \"runs\".\"active_slot\" IS NULL)" + } + } + }, + "snapshots": { + "name": "snapshots", + "columns": { + "singleton": { + "name": "singleton", + "type": "integer", + "primaryKey": true, + "notNull": true, + "autoincrement": false, + "default": 1 + }, + "cursor": { + "name": "cursor", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "data": { + "name": "data", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + } + }, + "indexes": {}, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": { + "one_snapshot": { + "name": "one_snapshot", + "value": "\"snapshots\".\"singleton\" = 1" + } + } + } + }, + "views": {}, + "enums": {}, + "_meta": { + "schemas": {}, + "tables": {}, + "columns": {} + }, + "internal": { + "indexes": {} + } +} \ No newline at end of file diff --git a/services/agent-harness/drizzle/meta/0001_snapshot.json b/services/agent-harness/drizzle/meta/0001_snapshot.json new file mode 100644 index 0000000000..47d36c9a56 --- /dev/null +++ b/services/agent-harness/drizzle/meta/0001_snapshot.json @@ -0,0 +1,821 @@ +{ + "version": "6", + "dialect": "sqlite", + "id": "d134f757-b35b-4c49-8c32-656480944bee", + "prevId": "0abb8b8c-affd-463d-8b01-dcbb2158c797", + "tables": { + "attempts": { + "name": "attempts", + "columns": { + "id": { + "name": "id", + "type": "text", + "primaryKey": true, + "notNull": true, + "autoincrement": false + }, + "tool_call_id": { + "name": "tool_call_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "generation": { + "name": "generation", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "intent": { + "name": "intent", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "outcome": { + "name": "outcome", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "provider_reference": { + "name": "provider_reference", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + } + }, + "indexes": { + "attempt_generation": { + "name": "attempt_generation", + "columns": [ + "tool_call_id", + "generation" + ], + "isUnique": true + } + }, + "foreignKeys": { + "attempts_tool_call_id_calls_id_fk": { + "name": "attempts_tool_call_id_calls_id_fk", + "tableFrom": "attempts", + "tableTo": "calls", + "columnsFrom": [ + "tool_call_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "no action", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": {} + }, + "calls": { + "name": "calls", + "columns": { + "id": { + "name": "id", + "type": "text", + "primaryKey": true, + "notNull": true, + "autoincrement": false + }, + "run_id": { + "name": "run_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "checkpoint_id": { + "name": "checkpoint_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "position": { + "name": "position", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "input_digest": { + "name": "input_digest", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "data": { + "name": "data", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "policy": { + "name": "policy", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "state": { + "name": "state", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "revision": { + "name": "revision", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": 0 + } + }, + "indexes": { + "call_order": { + "name": "call_order", + "columns": [ + "run_id", + "position" + ], + "isUnique": true + } + }, + "foreignKeys": { + "calls_run_id_runs_id_fk": { + "name": "calls_run_id_runs_id_fk", + "tableFrom": "calls", + "tableTo": "runs", + "columnsFrom": [ + "run_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "no action", + "onUpdate": "no action" + }, + "calls_checkpoint_id_checkpoints_id_fk": { + "name": "calls_checkpoint_id_checkpoints_id_fk", + "tableFrom": "calls", + "tableTo": "checkpoints", + "columnsFrom": [ + "checkpoint_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "no action", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": {} + }, + "checkpoints": { + "name": "checkpoints", + "columns": { + "id": { + "name": "id", + "type": "text", + "primaryKey": true, + "notNull": true, + "autoincrement": false + }, + "run_id": { + "name": "run_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "step": { + "name": "step", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "status": { + "name": "status", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "data": { + "name": "data", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "definition_versions": { + "name": "definition_versions", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + } + }, + "indexes": { + "checkpoint_step": { + "name": "checkpoint_step", + "columns": [ + "run_id", + "step" + ], + "isUnique": true + } + }, + "foreignKeys": { + "checkpoints_run_id_runs_id_fk": { + "name": "checkpoints_run_id_runs_id_fk", + "tableFrom": "checkpoints", + "tableTo": "runs", + "columnsFrom": [ + "run_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "no action", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": {} + }, + "client_actions": { + "name": "client_actions", + "columns": { + "tool_call_id": { + "name": "tool_call_id", + "type": "text", + "primaryKey": true, + "notNull": true, + "autoincrement": false + }, + "sequence": { + "name": "sequence", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "data": { + "name": "data", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + } + }, + "indexes": {}, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": {} + }, + "commands": { + "name": "commands", + "columns": { + "id": { + "name": "id", + "type": "text", + "primaryKey": true, + "notNull": true, + "autoincrement": false + }, + "fingerprint": { + "name": "fingerprint", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "reply": { + "name": "reply", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "sequence": { + "name": "sequence", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + } + }, + "indexes": {}, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": {} + }, + "conversation": { + "name": "conversation", + "columns": { + "singleton": { + "name": "singleton", + "type": "integer", + "primaryKey": true, + "notNull": true, + "autoincrement": false, + "default": 1 + }, + "id": { + "name": "id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "owner_user_id": { + "name": "owner_user_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "context": { + "name": "context", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "permission_mode": { + "name": "permission_mode", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": "'ask'" + }, + "permission_revision": { + "name": "permission_revision", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": 0 + }, + "sequence": { + "name": "sequence", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": 0 + }, + "compacted_through": { + "name": "compacted_through", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": 0 + }, + "active_run_id": { + "name": "active_run_id", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "legacy_cursor": { + "name": "legacy_cursor", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": 0 + } + }, + "indexes": { + "conversation_id_unique": { + "name": "conversation_id_unique", + "columns": [ + "id" + ], + "isUnique": true + } + }, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": { + "one_conversation": { + "name": "one_conversation", + "value": "\"conversation\".\"singleton\" = 1" + } + } + }, + "events": { + "name": "events", + "columns": { + "sequence": { + "name": "sequence", + "type": "integer", + "primaryKey": true, + "notNull": true, + "autoincrement": false + }, + "data": { + "name": "data", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + } + }, + "indexes": {}, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": {} + }, + "grants": { + "name": "grants", + "columns": { + "id": { + "name": "id", + "type": "text", + "primaryKey": true, + "notNull": true, + "autoincrement": false + }, + "tool_call_id": { + "name": "tool_call_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "generation": { + "name": "generation", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "data": { + "name": "data", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + } + }, + "indexes": { + "grant_generation": { + "name": "grant_generation", + "columns": [ + "tool_call_id", + "generation" + ], + "isUnique": true + } + }, + "foreignKeys": { + "grants_tool_call_id_calls_id_fk": { + "name": "grants_tool_call_id_calls_id_fk", + "tableFrom": "grants", + "tableTo": "calls", + "columnsFrom": [ + "tool_call_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "no action", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": {} + }, + "interactions": { + "name": "interactions", + "columns": { + "id": { + "name": "id", + "type": "text", + "primaryKey": true, + "notNull": true, + "autoincrement": false + }, + "sequence": { + "name": "sequence", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "resolved": { + "name": "resolved", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "data": { + "name": "data", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + } + }, + "indexes": { + "unresolved_interactions": { + "name": "unresolved_interactions", + "columns": [ + "resolved", + "sequence" + ], + "isUnique": false + } + }, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": {} + }, + "messages": { + "name": "messages", + "columns": { + "id": { + "name": "id", + "type": "text", + "primaryKey": true, + "notNull": true, + "autoincrement": false + }, + "sequence": { + "name": "sequence", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "created_at": { + "name": "created_at", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "data": { + "name": "data", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + } + }, + "indexes": { + "messages_sequence_unique": { + "name": "messages_sequence_unique", + "columns": [ + "sequence" + ], + "isUnique": true + }, + "message_history": { + "name": "message_history", + "columns": [ + "created_at", + "id" + ], + "isUnique": false + } + }, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": {} + }, + "projection_work": { + "name": "projection_work", + "columns": { + "id": { + "name": "id", + "type": "text", + "primaryKey": true, + "notNull": true, + "autoincrement": false + }, + "message_id": { + "name": "message_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "data": { + "name": "data", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "revision": { + "name": "revision", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": 0 + }, + "due_at": { + "name": "due_at", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "acknowledged_at": { + "name": "acknowledged_at", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + } + }, + "indexes": { + "due_projections": { + "name": "due_projections", + "columns": [ + "acknowledged_at", + "due_at", + "id" + ], + "isUnique": false + } + }, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": {} + }, + "runs": { + "name": "runs", + "columns": { + "id": { + "name": "id", + "type": "text", + "primaryKey": true, + "notNull": true, + "autoincrement": false + }, + "position": { + "name": "position", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "status": { + "name": "status", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "data": { + "name": "data", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "revision": { + "name": "revision", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": 0 + }, + "step": { + "name": "step", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": 0 + }, + "active_slot": { + "name": "active_slot", + "type": "integer", + "primaryKey": false, + "notNull": false, + "autoincrement": false + } + }, + "indexes": { + "runs_position_unique": { + "name": "runs_position_unique", + "columns": [ + "position" + ], + "isUnique": true + }, + "runs_active_slot_unique": { + "name": "runs_active_slot_unique", + "columns": [ + "active_slot" + ], + "isUnique": true + }, + "run_queue": { + "name": "run_queue", + "columns": [ + "status", + "position" + ], + "isUnique": false + } + }, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": { + "active_run_slot": { + "name": "active_run_slot", + "value": "(\"runs\".\"status\" IN ('running', 'waiting', 'stopping') AND \"runs\".\"active_slot\" IS 1) OR (\"runs\".\"status\" NOT IN ('running', 'waiting', 'stopping') AND \"runs\".\"active_slot\" IS NULL)" + } + } + }, + "snapshots": { + "name": "snapshots", + "columns": { + "singleton": { + "name": "singleton", + "type": "integer", + "primaryKey": true, + "notNull": true, + "autoincrement": false, + "default": 1 + }, + "cursor": { + "name": "cursor", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "data": { + "name": "data", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + } + }, + "indexes": {}, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": { + "one_snapshot": { + "name": "one_snapshot", + "value": "\"snapshots\".\"singleton\" = 1" + } + } + } + }, + "views": {}, + "enums": {}, + "_meta": { + "schemas": {}, + "tables": {}, + "columns": {} + }, + "internal": { + "indexes": {} + } +} \ No newline at end of file diff --git a/services/agent-harness/drizzle/meta/_journal.json b/services/agent-harness/drizzle/meta/_journal.json new file mode 100644 index 0000000000..3f3187f9b2 --- /dev/null +++ b/services/agent-harness/drizzle/meta/_journal.json @@ -0,0 +1,20 @@ +{ + "version": "7", + "dialect": "sqlite", + "entries": [ + { + "idx": 0, + "version": "6", + "when": 1787948871545, + "tag": "0000_authoritative_store", + "breakpoints": true + }, + { + "idx": 1, + "version": "6", + "when": 1787949180289, + "tag": "0001_projection_work", + "breakpoints": true + } + ] +} \ No newline at end of file diff --git a/services/agent-harness/drizzle/migrations.d.ts b/services/agent-harness/drizzle/migrations.d.ts new file mode 100644 index 0000000000..27c23b98e7 --- /dev/null +++ b/services/agent-harness/drizzle/migrations.d.ts @@ -0,0 +1,4 @@ +import type { migrate } from 'drizzle-orm/durable-sqlite/migrator'; + +declare const migrations: Parameters[1]; +export default migrations; diff --git a/services/agent-harness/drizzle/migrations.js b/services/agent-harness/drizzle/migrations.js new file mode 100644 index 0000000000..661d6c42c5 --- /dev/null +++ b/services/agent-harness/drizzle/migrations.js @@ -0,0 +1,11 @@ +import journal from './meta/_journal.json'; +import m0000 from './0000_authoritative_store.sql'; +import m0001 from './0001_projection_work.sql'; + +export default { + journal, + migrations: { + m0000, + m0001, + }, +}; diff --git a/services/agent-harness/src/db/records.ts b/services/agent-harness/src/db/records.ts new file mode 100644 index 0000000000..70a9f1e65d --- /dev/null +++ b/services/agent-harness/src/db/records.ts @@ -0,0 +1,318 @@ +import { and, asc, eq, isNull, lt, sql } from 'drizzle-orm'; +import type { DrizzleSqliteDODatabase } from 'drizzle-orm/durable-sqlite'; +import { z } from 'zod'; +import { canonicalizeValidatedInput } from '@kilocode/agent-harness/commands'; +import { + ConversationSchema, + ExecutionGrantSchema, + InteractionSchema, + MessageSchema, + RunSchema, + ToolCallSchema, + type EventEnvelope, + type ToolCall, +} from '@kilocode/agent-harness/contracts'; +import * as s from './sqlite-schema'; +import { StoreError } from './wake'; + +export type StoreDatabase = DrizzleSqliteDODatabase; +export const pageLimit = (limit: number) => z.int().min(1).max(200).parse(limit); +const same = (left: unknown, right: unknown) => + canonicalizeValidatedInput(left) === canonicalizeValidatedInput(right); +const active = (status: string) => ['running', 'waiting', 'stopping'].includes(status); + +export function conversationRow(db: StoreDatabase) { + const row = db.select().from(s.conversation).where(eq(s.conversation.singleton, 1)).get(); + if (!row) throw new StoreError('invalid_input'); + return row; +} +export function readConversation(db: StoreDatabase) { + const { id, ownerUserId, context, permissionMode, permissionRevision } = conversationRow(db); + return ConversationSchema.parse({ id, ownerUserId, context, permissionMode, permissionRevision }); +} +export function compareAndSetActiveRun( + db: StoreDatabase, + expected: string | null, + next: string | null +) { + return Boolean( + db + .update(s.conversation) + .set({ activeRunId: next }) + .where( + and( + eq(s.conversation.singleton, 1), + expected === null + ? isNull(s.conversation.activeRunId) + : eq(s.conversation.activeRunId, expected) + ) + ) + .returning({ id: s.conversation.id }) + .get() + ); +} + +// Materialized state and replay deltas always share the caller's synchronous transaction. +export function applyEvent(db: StoreDatabase, event: EventEnvelope['event'], sequence: number) { + switch (event.type) { + case 'conversation': { + const current = readConversation(db), + next = event.conversation; + if ( + current.id !== next.id || + current.ownerUserId !== next.ownerUserId || + !same(current.context, next.context) + ) + throw new StoreError('invalid_input'); + if (!same(current, next) && next.permissionRevision !== current.permissionRevision + 1) + throw new StoreError('command_conflict'); + db.update(s.conversation) + .set({ permissionMode: next.permissionMode, permissionRevision: next.permissionRevision }) + .where(eq(s.conversation.singleton, 1)) + .run(); + break; + } + case 'message': { + const message = event.message; + const old = db.select().from(s.messages).where(eq(s.messages.id, message.id)).get(); + if (old) { + const prior = MessageSchema.parse(old.data); + if ( + prior.createdAt !== message.createdAt || + prior.role !== message.role || + prior.provenance !== message.provenance || + ('runId' in prior && (!('runId' in message) || prior.runId !== message.runId)) + ) + throw new StoreError('invalid_input'); + } + db.insert(s.messages) + .values({ id: message.id, createdAt: message.createdAt, sequence, data: message }) + .onConflictDoUpdate({ target: s.messages.id, set: { data: message } }) + .run(); + break; + } + case 'run': { + const run = event.run, + meta = conversationRow(db); + if (run.conversationId !== meta.id) throw new StoreError('invalid_input'); + const old = db.select().from(s.runs).where(eq(s.runs.id, run.id)).get(); + if (old) { + const prior = RunSchema.parse(old.data); + if (!same({ ...prior, state: null }, { ...run, state: null })) + throw new StoreError('invalid_input'); + if ( + ['completed', 'cancelled', 'failed'].includes(prior.state.status) && + !same(prior.state, run.state) + ) + throw new StoreError('command_conflict'); + } else if (run.state.status !== 'queued') throw new StoreError('invalid_input'); + const position = old?.position ?? sequence; + if (active(run.state.status)) { + if (meta.activeRunId !== null && meta.activeRunId !== run.id) + throw new StoreError('command_conflict'); + const earlier = db + .select({ id: s.runs.id }) + .from(s.runs) + .where(and(eq(s.runs.status, 'queued'), lt(s.runs.position, position))) + .orderBy(asc(s.runs.position)) + .limit(1) + .get(); + if (earlier) throw new StoreError('command_conflict'); + if (!compareAndSetActiveRun(db, meta.activeRunId, run.id)) + throw new StoreError('command_conflict'); + } else if (meta.activeRunId === run.id) { + if (!compareAndSetActiveRun(db, run.id, null)) throw new StoreError('command_conflict'); + } + const values = { + status: run.state.status, + data: run, + activeSlot: active(run.state.status) ? 1 : null, + revision: old ? old.revision + 1 : 0, + }; + db.insert(s.runs) + .values({ id: run.id, position, ...values }) + .onConflictDoUpdate({ target: s.runs.id, set: values }) + .run(); + break; + } + case 'interaction': { + const interaction = event.interaction; + const old = db + .select() + .from(s.interactions) + .where(eq(s.interactions.id, interaction.id)) + .get(); + if (old) { + const prior = InteractionSchema.parse(old.data); + if ( + !same( + { ...prior, resolution: null, toolCall: null }, + { ...interaction, resolution: null, toolCall: null } + ) || + !same(callIdentity(prior.toolCall), callIdentity(interaction.toolCall)) + ) + throw new StoreError('invalid_input'); + if (prior.resolution !== null && !same(prior.resolution, interaction.resolution)) + throw new StoreError('command_conflict'); + } + const values = { data: interaction, resolved: interaction.resolution !== null }; + db.insert(s.interactions) + .values({ id: interaction.id, sequence, ...values }) + .onConflictDoUpdate({ target: s.interactions.id, set: values }) + .run(); + break; + } + case 'client_action': + if (event.action) { + if (event.toolCallId !== event.action.toolCall.id) throw new StoreError('invalid_input'); + db.insert(s.clientActions) + .values({ toolCallId: event.toolCallId, sequence, data: event.action }) + .onConflictDoUpdate({ target: s.clientActions.toolCallId, set: { data: event.action } }) + .run(); + } else + db.delete(s.clientActions).where(eq(s.clientActions.toolCallId, event.toolCallId)).run(); + } +} + +function callIdentity(call: ToolCall) { + const { state: _state, result: _result, approval: _approval, ...identity } = call; + return identity; +} +const CheckpointSchema = z.strictObject({ + id: z.uuid(), + runId: z.uuid(), + step: z.int().nonnegative(), + status: z.enum(['partial', 'complete', 'failed']), + data: z.json(), + definitionVersions: z.record(z.string(), z.string().min(1)), +}); +export function insertCheckpoint(db: StoreDatabase, input: z.input) { + db.insert(s.checkpoints).values(CheckpointSchema.parse(input)).run(); +} +export function executableCheckpoint(db: StoreDatabase, id: string) { + const row = db + .select() + .from(s.checkpoints) + .where(and(eq(s.checkpoints.id, id), eq(s.checkpoints.status, 'complete'))) + .get(); + return row ? CheckpointSchema.parse(row) : null; +} +export function insertCall( + db: StoreDatabase, + input: ToolCall, + details: { checkpointId: string; inputDigest: string; position: number; policy: unknown } +) { + const call = ToolCallSchema.parse(input), + checkpoint = executableCheckpoint(db, details.checkpointId); + if ( + !checkpoint || + checkpoint.runId !== call.runId || + !same(call.context, readConversation(db).context) || + checkpoint.definitionVersions[call.name] !== call.definitionVersion || + !['pending', 'waiting'].includes(call.state) + ) + throw new StoreError('invalid_input'); + db.insert(s.calls) + .values({ + id: call.id, + runId: call.runId, + checkpointId: checkpoint.id, + inputDigest: z.string().min(1).parse(details.inputDigest), + position: z.int().nonnegative().parse(details.position), + policy: z.json().parse(details.policy), + data: call, + state: call.state, + }) + .run(); +} +export function insertGrant(db: StoreDatabase, input: unknown) { + const grant = ExecutionGrantSchema.parse(input); + const stored = db.select().from(s.calls).where(eq(s.calls.id, grant.toolCallId)).get(); + const scope = readConversation(db); + if (!stored) throw new StoreError('invalid_input'); + const call = ToolCallSchema.parse(stored.data); + if ( + grant.conversationId !== scope.id || + grant.ownerUserId !== scope.ownerUserId || + !same(grant.context, call.context) || + grant.definitionVersion !== call.definitionVersion || + grant.inputDigest !== stored.inputDigest || + call.executionTarget.kind !== 'client' || + call.executionTarget.clientId !== grant.clientId + ) + throw new StoreError('invalid_input'); + db.insert(s.grants) + .values({ + id: grant.id, + toolCallId: grant.toolCallId, + generation: grant.generation, + data: grant, + }) + .run(); +} +export function insertAttempt( + db: StoreDatabase, + input: { id: string; toolCallId: string; generation: number; grantId?: string } +) { + const row = db.select().from(s.calls).where(eq(s.calls.id, input.toolCallId)).get(); + if (!row || !executableCheckpoint(db, row.checkpointId)) throw new StoreError('invalid_input'); + const call = ToolCallSchema.parse(row.data); + const storedGrant = input.grantId + ? db.select().from(s.grants).where(eq(s.grants.id, input.grantId)).get() + : undefined; + const grant = storedGrant ? ExecutionGrantSchema.parse(storedGrant.data) : null; + if ( + (input.grantId && !grant) || + (grant && (grant.toolCallId !== call.id || grant.generation !== input.generation)) + ) + throw new StoreError('invalid_input'); + if (call.executionTarget.kind === 'client' && !grant) throw new StoreError('invalid_input'); + db.insert(s.attempts) + .values({ + id: z.uuid().parse(input.id), + toolCallId: call.id, + generation: z.int().nonnegative().parse(input.generation), + intent: { + toolCall: call, + inputDigest: row.inputDigest, + policy: z.json().parse(row.policy), + grant, + }, + }) + .run(); +} +export function compareAndSetCall( + db: StoreDatabase, + id: string, + revision: number, + changes: Pick +) { + const row = db + .select() + .from(s.calls) + .where(and(eq(s.calls.id, id), eq(s.calls.revision, revision))) + .get(); + if (!row) return false; + const old = ToolCallSchema.parse(row.data), + call = ToolCallSchema.parse({ ...old, ...changes }); + if (!same(callIdentity(old), callIdentity(call)) || old.state === 'settled') + throw new StoreError('invalid_input'); + if ( + call.state === 'executing' && + !db + .select({ id: s.attempts.id }) + .from(s.attempts) + .where(eq(s.attempts.toolCallId, id)) + .limit(1) + .get() + ) + throw new StoreError('invalid_input'); + return Boolean( + db + .update(s.calls) + .set({ data: call, state: call.state, revision: sql`${s.calls.revision} + 1` }) + .where(and(eq(s.calls.id, id), eq(s.calls.revision, revision))) + .returning({ id: s.calls.id }) + .get() + ); +} diff --git a/services/agent-harness/src/db/sqlite-schema.ts b/services/agent-harness/src/db/sqlite-schema.ts new file mode 100644 index 0000000000..3ede0d023b --- /dev/null +++ b/services/agent-harness/src/db/sqlite-schema.ts @@ -0,0 +1,171 @@ +import { sql } from 'drizzle-orm'; +import { check, index, integer, sqliteTable, text, uniqueIndex } from 'drizzle-orm/sqlite-core'; +import type { Run, ToolCall } from '@kilocode/agent-harness/contracts'; + +// One database belongs to an existing PostgreSQL thread. This row never creates thread authority. +export const conversation = sqliteTable( + 'conversation', + { + singleton: integer('singleton').primaryKey().default(1), + id: text('id').notNull().unique(), + ownerUserId: text('owner_user_id').notNull(), + context: text('context', { mode: 'json' }).$type().notNull(), + permissionMode: text('permission_mode', { enum: ['ask', 'yolo'] }) + .notNull() + .default('ask'), + permissionRevision: integer('permission_revision').notNull().default(0), + sequence: integer('sequence').notNull().default(0), + compactedThrough: integer('compacted_through').notNull().default(0), + activeRunId: text('active_run_id'), + legacyCursor: integer('legacy_cursor').notNull().default(0), + }, + table => [check('one_conversation', sql`${table.singleton} = 1`)] +); + +export const messages = sqliteTable( + 'messages', + { + id: text('id').primaryKey(), + sequence: integer('sequence').notNull().unique(), + createdAt: text('created_at').notNull(), + // Full normalized parts remain recoverable after their display events are compacted. + data: text('data', { mode: 'json' }).$type().notNull(), + }, + table => [index('message_history').on(table.createdAt, table.id)] +); + +export const commands = sqliteTable('commands', { + id: text('id').primaryKey(), + fingerprint: text('fingerprint').notNull(), + reply: text('reply', { mode: 'json' }).$type().notNull(), + sequence: integer('sequence').notNull(), +}); + +export const runs = sqliteTable( + 'runs', + { + id: text('id').primaryKey(), + position: integer('position').notNull().unique(), + status: text('status').$type().notNull(), + data: text('data', { mode: 'json' }).$type().notNull(), + revision: integer('revision').notNull().default(0), + step: integer('step').notNull().default(0), + activeSlot: integer('active_slot').unique(), + }, + table => [ + index('run_queue').on(table.status, table.position), + check( + 'active_run_slot', + sql`(${table.status} IN ('running', 'waiting', 'stopping') AND ${table.activeSlot} IS 1) OR (${table.status} NOT IN ('running', 'waiting', 'stopping') AND ${table.activeSlot} IS NULL)` + ), + ] +); + +export const checkpoints = sqliteTable( + 'checkpoints', + { + id: text('id').primaryKey(), + runId: text('run_id') + .notNull() + .references(() => runs.id), + step: integer('step').notNull(), + status: text('status', { enum: ['partial', 'complete', 'failed'] }).notNull(), + data: text('data', { mode: 'json' }).$type().notNull(), + definitionVersions: text('definition_versions', { mode: 'json' }).$type().notNull(), + }, + table => [uniqueIndex('checkpoint_step').on(table.runId, table.step)] +); + +export const calls = sqliteTable( + 'calls', + { + id: text('id').primaryKey(), + runId: text('run_id') + .notNull() + .references(() => runs.id), + checkpointId: text('checkpoint_id') + .notNull() + .references(() => checkpoints.id), + position: integer('position').notNull(), + inputDigest: text('input_digest').notNull(), + data: text('data', { mode: 'json' }).$type().notNull(), + policy: text('policy', { mode: 'json' }).$type().notNull(), + state: text('state').$type().notNull(), + revision: integer('revision').notNull().default(0), + }, + table => [uniqueIndex('call_order').on(table.runId, table.position)] +); + +export const interactions = sqliteTable( + 'interactions', + { + id: text('id').primaryKey(), + sequence: integer('sequence').notNull(), + resolved: integer('resolved', { mode: 'boolean' }).notNull(), + data: text('data', { mode: 'json' }).$type().notNull(), + }, + table => [index('unresolved_interactions').on(table.resolved, table.sequence)] +); + +export const grants = sqliteTable( + 'grants', + { + id: text('id').primaryKey(), + toolCallId: text('tool_call_id') + .notNull() + .references(() => calls.id), + generation: integer('generation').notNull(), + data: text('data', { mode: 'json' }).$type().notNull(), + }, + table => [uniqueIndex('grant_generation').on(table.toolCallId, table.generation)] +); + +export const attempts = sqliteTable( + 'attempts', + { + id: text('id').primaryKey(), + toolCallId: text('tool_call_id') + .notNull() + .references(() => calls.id), + generation: integer('generation').notNull(), + // Intent includes the immutable call, digest, policy, and grant before an external effect. + intent: text('intent', { mode: 'json' }).$type().notNull(), + outcome: text('outcome', { mode: 'json' }).$type(), + providerReference: text('provider_reference'), + }, + table => [uniqueIndex('attempt_generation').on(table.toolCallId, table.generation)] +); + +export const clientActions = sqliteTable('client_actions', { + toolCallId: text('tool_call_id').primaryKey(), + sequence: integer('sequence').notNull(), + data: text('data', { mode: 'json' }).$type().notNull(), +}); + +export const events = sqliteTable('events', { + sequence: integer('sequence').primaryKey(), + data: text('data', { mode: 'json' }).$type().notNull(), +}); + +export const snapshots = sqliteTable( + 'snapshots', + { + singleton: integer('singleton').primaryKey().default(1), + cursor: integer('cursor').notNull(), + data: text('data', { mode: 'json' }).$type().notNull(), + }, + table => [check('one_snapshot', sql`${table.singleton} = 1`)] +); + +export const projectionWork = sqliteTable( + 'projection_work', + { + id: text('id').primaryKey(), + messageId: text('message_id').notNull(), + data: text('data', { mode: 'json' }).$type().notNull(), + revision: integer('revision').notNull().default(0), + dueAt: integer('due_at').notNull(), + acknowledgedAt: text('acknowledged_at'), + }, + table => [index('due_projections').on(table.acknowledgedAt, table.dueAt, table.id)] +); diff --git a/services/agent-harness/src/db/store.test.ts b/services/agent-harness/src/db/store.test.ts new file mode 100644 index 0000000000..41f3f36be7 --- /dev/null +++ b/services/agent-harness/src/db/store.test.ts @@ -0,0 +1,1370 @@ +import { env } from 'cloudflare:workers'; +import { abortAllDurableObjects, runDurableObjectAlarm, runInDurableObject } from 'cloudflare:test'; +import { drizzle } from 'drizzle-orm/durable-sqlite'; +import { eq } from 'drizzle-orm'; +import { describe, expect, it } from 'vitest'; +import { + ConversationSchema, + MessageSchema, + ToolCallSchema, + type EventEnvelope, + type Run, + type ToolCall, +} from '@kilocode/agent-harness/contracts'; +import type { CommandReply } from '@kilocode/agent-harness/journal'; +import { createHarnessStore } from '@kilocode/agent-harness/resume'; +import { harnessReducer, initialHarnessState } from '@kilocode/agent-harness/state'; +import { + getOldTestStoreStub, + getTestStoreStub, + type OldTestStore, + type TestStore, +} from './test-worker'; +import { openStore, type ConversationStore } from './store'; +import { + compareAndSetActiveRun, + compareAndSetCall, + executableCheckpoint, + insertAttempt, + insertCall, + insertCheckpoint, + insertGrant, +} from './records'; +import { transitionWithWake } from './wake'; +import * as s from './sqlite-schema'; + +const bindings = env as { + STORE: DurableObjectNamespace; + OLD_STORE: DurableObjectNamespace; +}; +const id = (value: number) => `00000000-0000-4000-8000-${value.toString(16).padStart(12, '0')}`; +const time = '2026-08-28T12:00:00.000Z'; +const future = () => Date.now() + 3_600_000; +const conversation = ConversationSchema.parse({ + id: id(1), + ownerUserId: 'auth0|owner', + context: { type: 'personal' }, +}); +const run = (value = 10, state: Run['state'] = { status: 'queued' }): Run => ({ + id: id(value), + conversationId: conversation.id, + inputMessageId: id(value + 1000), + originClientId: id(2), + modelId: 'test/model', + variant: 'fixed', + state, +}); +const message = (value: number, content = 'hello', runId = id(10)) => + MessageSchema.parse({ + id: id(value), + role: 'user', + content, + clientId: null, + createdAt: time, + provenance: 'harness', + protocolVersion: 1, + runId, + }); +const acceptedEvents = (value = 10): EventEnvelope['event'][] => [ + { type: 'message', message: message(value + 1000, 'hello', id(value)) }, + { type: 'run', run: run(value) }, +]; +const reply: CommandReply = { + status: 'accepted', + commandId: id(3), + result: { messageId: id(1010), runId: id(10) }, +}; +const command = { id: id(3), fingerprint: 'authenticated-command-digest' }; +const call = (value = 30): ToolCall => + ToolCallSchema.parse({ + id: id(value), + runId: id(10), + name: 'app.notifications', + definitionVersion: '1', + arguments: { enabled: true }, + context: conversation.context, + effect: 'side_effect', + executionTarget: { kind: 'client', clientId: id(2) }, + approval: null, + state: 'pending', + result: null, + }); +const checkpoint = { + id: id(40), + runId: id(10), + step: 0, + status: 'complete' as const, + data: { responseMessages: [{ role: 'assistant', content: 'validated step' }] }, + definitionVersions: { 'app.notifications': '1' }, +}; +const callDetails = { + checkpointId: checkpoint.id, + inputDigest: 'immutable-input', + position: 0, + policy: { decision: 'dispatch', permissionRevision: 0 }, +}; +const grant = { + id: id(50), + conversationId: conversation.id, + ownerUserId: conversation.ownerUserId, + clientId: id(2), + toolCallId: id(30), + context: conversation.context, + definitionVersion: '1', + inputDigest: callDetails.inputDigest, + generation: 1, + expiresAt: '2030-01-01T00:00:00.000Z', +}; + +async function fresh() { + const stub = getTestStoreStub(bindings.STORE, crypto.randomUUID()); + await runInDurableObject(stub, instance => { + instance.store.bindExistingConversation(conversation); + }); + return stub; +} +function eventList(store: ConversationStore, after = 0, limit = 200) { + const page = store.eventsAfter(after, limit); + if (page.status !== 'events') throw new Error('Unexpected expired cursor'); + return page.events; +} +async function seedCall(store: ConversationStore) { + await store.transition({ wakeAt: future() }, db => { + insertCheckpoint(db, checkpoint); + insertCall(db, call(), callDetails); + return { events: [] }; + }); +} + +describe('real Durable Object SQLite storage', () => { + it('migrates empty storage without inventing a conversation or active work', async () => { + const stub = getTestStoreStub(bindings.STORE, crypto.randomUUID()); + await runInDurableObject(stub, async (instance, state) => { + expect(instance.store.snapshot()).toBeNull(); + instance.store.bindExistingConversation({ + id: conversation.id, + ownerUserId: conversation.ownerUserId, + context: conversation.context, + }); + expect(instance.store.snapshot()).toEqual({ + protocolVersion: 1, + conversation, + recentMessages: [], + historyCursor: null, + activeRun: null, + queuedRuns: [], + unresolvedInteractions: [], + pendingClientActions: [], + eventCursor: 0, + }); + expect(instance.store.pendingProjections(Date.now())).toEqual([]); + expect(instance.store.getCommand(command.id)).toBeNull(); + expect(await state.storage.getAlarm()).toBeNull(); + expect((await openStore(state)).snapshot()).toEqual(instance.store.snapshot()); + }); + }); + + it('commits state, ordered events, and the original reply once under racing commands', async () => { + const stub = await fresh(), + wakeAt = future(); + expect( + await Promise.all([ + stub.commit({ command, wakeAt }, acceptedEvents(), reply), + stub.commit({ command, wakeAt }, acceptedEvents(), reply), + ]) + ).toEqual([reply, reply]); + await runInDurableObject(stub, (instance, state) => { + const snapshot = instance.store.snapshot(); + expect(snapshot?.recentMessages.map(row => row.id)).toEqual([id(1010)]); + expect(snapshot?.queuedRuns.map(row => row.id)).toEqual([id(10)]); + expect(eventList(instance.store).map(event => event.sequence)).toEqual([1, 2]); + expect(instance.store.getCommand(command.id)?.reply).toEqual(reply); + expect(drizzle(state.storage).select().from(s.commands).all()).toHaveLength(1); + }); + }); + + it('rolls back state, events, projection work, and results after a late uniqueness failure', async () => { + const stub = await fresh(); + await runInDurableObject(stub, async instance => { + await expect( + instance.store.transition({ command, wakeAt: future() }, db => { + db.insert(s.projectionWork) + .values({ id: 'projection-1', messageId: id(1010), data: { text: 'hello' }, dueAt: 1 }) + .run(); + // The final result insert fails after the materialized state and events have been written. + db.insert(s.commands) + .values({ id: command.id, fingerprint: command.fingerprint, reply, sequence: 0 }) + .run(); + return { events: acceptedEvents(), reply }; + }) + ).rejects.toThrow(); + expect(instance.store.snapshot()).toMatchObject({ + recentMessages: [], + queuedRuns: [], + eventCursor: 0, + }); + expect(instance.store.getCommand(command.id)).toBeNull(); + expect(instance.store.pendingProjections(Date.now())).toEqual([]); + expect(eventList(instance.store)).toEqual([]); + await instance.store.transition({ command, wakeAt: future() }, () => ({ + events: acceptedEvents(), + reply, + })); + expect(eventList(instance.store).map(event => event.sequence)).toEqual([1, 2]); + }); + }); + + it('replays a stored rejection after restart instead of admitting the command again', async () => { + const stub = await fresh(); + const rejected: CommandReply = { + status: 'rejected', + commandId: command.id, + error: { + code: 'limit_exceeded', + message: 'The stored command exceeded its limit.', + retryable: false, + }, + }; + expect(await stub.commit({ command, wakeAt: null }, [], rejected)).toEqual(rejected); + await abortAllDurableObjects(); + const restarted = getTestStoreStub(bindings.STORE, stub.id); + expect(await restarted.commit({ command, wakeAt: future() }, acceptedEvents(), reply)).toEqual( + rejected + ); + await runInDurableObject(restarted, async (instance, state) => { + expect(instance.store.getCommand(command.id)?.reply).toEqual(rejected); + expect(instance.store.snapshot()).toMatchObject({ + recentMessages: [], + queuedRuns: [], + eventCursor: 0, + }); + expect(await state.storage.getAlarm()).toBeNull(); + }); + }); + + it('rejects conflicting replay without replacing the reply or admitting another run', async () => { + const stub = await fresh(); + await stub.commit({ command, wakeAt: future() }, acceptedEvents(), reply); + const conflict = await stub.commit( + { command: { ...command, fingerprint: 'changed-input' }, wakeAt: future() }, + acceptedEvents(11), + reply + ); + expect(conflict).toMatchObject({ + status: 'rejected', + commandId: command.id, + error: { code: 'command_conflict', retryable: false }, + }); + await runInDurableObject(stub, instance => { + expect(instance.store.snapshot()?.queuedRuns.map(row => row.id)).toEqual([id(10)]); + expect(instance.store.getCommand(command.id)?.reply).toEqual(reply); + expect(eventList(instance.store)).toHaveLength(2); + }); + }); + + it('keeps a waiting active run ahead of the durable queue across restart', async () => { + const stub = await fresh(); + await stub.commit({ wakeAt: future() }, [ + ...acceptedEvents(), + ...acceptedEvents(11), + ...acceptedEvents(12), + ]); + await stub.commit({ wakeAt: future() }, [{ type: 'run', run: run(10, { status: 'running' }) }]); + const waiting = run(10, { + status: 'waiting', + waiting: { reason: 'approval', toolCallId: id(30) }, + }); + await stub.commit({ wakeAt: null }, [{ type: 'run', run: waiting }]); + await abortAllDurableObjects(); + const restarted = getTestStoreStub(bindings.STORE, stub.id); + await runInDurableObject(restarted, async instance => { + await expect( + instance.store.transition({ wakeAt: future() }, () => ({ + events: [{ type: 'run', run: run(11, { status: 'running' }) }], + })) + ).rejects.toThrow('command_conflict'); + expect(instance.store.snapshot()?.activeRun).toEqual(waiting); + expect(instance.store.queuedRuns().map(row => row.data.id)).toEqual([id(11), id(12)]); + const first = instance.store.queuedRuns(0, 1)[0]; + expect(instance.store.queuedRuns(first.position, 1).map(row => row.data.id)).toEqual([ + id(12), + ]); + }); + await restarted.commit({ wakeAt: future() }, [ + { type: 'run', run: run(10, { status: 'completed' }) }, + ]); + await restarted.commit({ wakeAt: future() }, [ + { type: 'run', run: run(11, { status: 'running' }) }, + ]); + await runInDurableObject(restarted, instance => { + expect(instance.store.snapshot()?.activeRun?.id).toBe(id(11)); + }); + }); + + it('enforces one active SQLite slot and ordered activation without changing run inputs', async () => { + const stub = await fresh(); + await stub.commit({ wakeAt: future() }, [...acceptedEvents(), ...acceptedEvents(11)]); + await runInDurableObject(stub, async (instance, state) => { + await expect( + instance.store.transition({ wakeAt: future() }, () => ({ + events: [{ type: 'run', run: run(11, { status: 'running' }) }], + })) + ).rejects.toThrow('command_conflict'); + await instance.store.transition({ wakeAt: future() }, () => ({ + events: [{ type: 'run', run: run(10, { status: 'running' }) }], + })); + await expect( + instance.store.transition({ wakeAt: future() }, () => ({ + events: [ + { type: 'run', run: { ...run(10, { status: 'running' }), modelId: 'changed/model' } }, + ], + })) + ).rejects.toThrow('invalid_input'); + const db = drizzle(state.storage); + expect(() => + db + .update(s.runs) + .set({ status: 'running', activeSlot: 1 }) + .where(eq(s.runs.id, id(11))) + .run() + ).toThrow(); + expect(() => + db + .update(s.runs) + .set({ status: 'running', activeSlot: null }) + .where(eq(s.runs.id, id(11))) + .run() + ).toThrow(); + expect(instance.store.snapshot()?.activeRun).toEqual(run(10, { status: 'running' })); + expect(instance.store.snapshot()?.queuedRuns.map(row => row.id)).toEqual([id(11)]); + }); + }); + + it('retains original results and complete tool parts after compaction and restart', async () => { + const stub = await fresh(); + await stub.commit({ command, wakeAt: future() }, acceptedEvents(), reply); + const settledCall = ToolCallSchema.parse({ + ...call(), + state: 'settled', + result: { + status: 'outcome_unknown', + reason: 'Lost provider reply', + providerReference: 'provider-operation-7', + }, + }); + const output = MessageSchema.parse({ + ...message(1020), + role: 'assistant', + parts: [{ type: 'tool_call', toolCall: settledCall }], + }); + await stub.commit({ wakeAt: null }, [ + { type: 'message', message: output }, + { type: 'run', run: run(10, { status: 'completed' }) }, + ]); + await runInDurableObject(stub, instance => { + expect(instance.store.compactEvents(2)).toBe(2); + expect(instance.store.eventsAfter(0)).toEqual({ status: 'cursor_expired' }); + expect(eventList(instance.store, 2).map(event => event.sequence)).toEqual([3, 4]); + expect(instance.store.compactEvents()).toBe(2); + }); + await abortAllDurableObjects(); + const restarted = getTestStoreStub(bindings.STORE, stub.id); + expect( + await restarted.commit({ command, wakeAt: future() }, acceptedEvents(11), reply) + ).toEqual(reply); + await runInDurableObject(restarted, (instance, state) => { + expect(instance.store.snapshot()).toMatchObject({ + activeRun: null, + queuedRuns: [], + eventCursor: 4, + }); + expect(instance.store.history().messages.find(row => row.id === output.id)).toEqual(output); + expect(instance.store.getCommand(command.id)?.reply).toEqual(reply); + expect(drizzle(state.storage).select().from(s.events).all()).toEqual([]); + expect(drizzle(state.storage).select().from(s.snapshots).get()?.data).toEqual( + instance.store.snapshot() + ); + }); + }); + + it('upgrades older SQLite storage without losing legacy text, defaults, or command results', async () => { + const stub = getOldTestStoreStub(bindings.OLD_STORE, crypto.randomUUID()); + await runInDurableObject(stub, async (_instance, state) => { + const db = drizzle(state.storage); + const legacy = { id: id(1010), role: 'assistant', content: 'old text', createdAt: time }; + db.insert(s.conversation) + .values({ + id: conversation.id, + ownerUserId: conversation.ownerUserId, + context: conversation.context, + }) + .run(); + db.insert(s.messages) + .values({ id: legacy.id, sequence: 1, createdAt: time, data: legacy }) + .run(); + db.insert(s.commands) + .values({ id: command.id, fingerprint: command.fingerprint, reply, sequence: 0 }) + .run(); + expect(() => db.select().from(s.projectionWork).all()).toThrow(); + const store = await openStore(state); + expect(store.pendingProjections(Date.now())).toEqual([]); + expect(store.snapshot()?.conversation).toEqual(conversation); + expect(store.history().messages).toEqual([ + { + ...legacy, + clientId: null, + provenance: 'legacy', + parts: [{ type: 'text', text: legacy.content }], + }, + ]); + expect(store.getCommand(command.id)?.reply).toEqual(reply); + expect((await openStore(state)).getCommand(command.id)?.reply).toEqual(reply); + }); + }); + + it('bounds ordered replay to 200 events and resumes strictly after the cursor', async () => { + const stub = await fresh(); + const inputs: EventEnvelope['event'][] = Array.from({ length: 205 }, (_, n) => ({ + type: 'message', + message: message(10_000 + n), + })); + await stub.commit({ wakeAt: null }, inputs); + await runInDurableObject(stub, instance => { + const first = eventList(instance.store), + second = eventList(instance.store, first.at(-1)!.sequence); + expect(first.map(event => event.sequence)).toEqual( + Array.from({ length: 200 }, (_, n) => n + 1) + ); + expect(second.map(event => event.sequence)).toEqual([201, 202, 203, 204, 205]); + expect(eventList(instance.store, 205)).toEqual([]); + for (const limit of [0, -1, 1.5, 201]) + expect(() => instance.store.eventsAfter(0, limit)).toThrow(); + expect(instance.store.eventsAfter(206)).toEqual({ status: 'cursor_expired' }); + }); + }); + + it('bounds replay by encoded bytes and rejects an event that cannot fit a page', async () => { + const stub = await fresh(), + text = 'é'.repeat(24 * 1024); + await stub.commit( + { wakeAt: null }, + [1, 2, 3].map(n => ({ type: 'message', message: message(10_000 + n, text) })) + ); + await runInDurableObject(stub, async instance => { + const first = instance.store.eventsAfter(0); + expect(new TextEncoder().encode(JSON.stringify(first)).byteLength).toBeLessThanOrEqual( + 256 * 1024 + ); + expect(eventList(instance.store).map(event => event.sequence)).toEqual([1, 2]); + expect(eventList(instance.store, 2).map(event => event.sequence)).toEqual([3]); + await expect( + instance.store.transition({ command, wakeAt: null }, () => ({ + events: [{ type: 'message', message: message(20_000, 'é'.repeat(128 * 1024)) }], + reply, + })) + ).rejects.toThrow('limit_exceeded'); + expect(instance.store.snapshot()?.eventCursor).toBe(3); + expect(instance.store.getCommand(command.id)).toBeNull(); + expect(instance.store.history().messages).toHaveLength(3); + }); + }); + + it('keeps the snapshot cursor consistent and history keysets complete at timestamp ties', async () => { + const stub = await fresh(); + await stub.commit( + { wakeAt: null }, + Array.from({ length: 60 }, (_, n) => ({ type: 'message', message: message(10_000 + n) })) + ); + await runInDurableObject(stub, async instance => { + const snapshot = instance.store.snapshot()!; + expect(snapshot.recentMessages.map(row => row.id)).toEqual( + Array.from({ length: 50 }, (_, n) => id(10_010 + n)) + ); + const earlier = instance.store.history(snapshot.historyCursor); + expect(earlier.messages.map(row => row.id)).toEqual( + Array.from({ length: 10 }, (_, n) => id(10_000 + n)) + ); + expect(earlier.historyCursor).toBeNull(); + const nextMessage = message(11_000, 'after snapshot'); + await instance.store.transition({ wakeAt: null }, () => ({ + events: [ + { + type: 'conversation', + conversation: { ...conversation, permissionMode: 'yolo', permissionRevision: 1 }, + }, + { type: 'message', message: nextMessage }, + ], + })); + const resumed = eventList(instance.store, snapshot.eventCursor).reduce( + (state, envelope) => harnessReducer(state, { type: 'event', envelope }), + harnessReducer(initialHarnessState(), { type: 'snapshot', snapshot }) + ); + expect(resumed.messages[nextMessage.id]).toEqual(nextMessage); + expect(resumed.conversation?.permissionMode).toBe('yolo'); + expect(resumed.eventCursor).toBe(62); + expect(() => instance.store.history('not-a-cursor')).toThrow('invalid_input'); + }); + }); + + it('retains old unresolved interactions and client waits outside display history', async () => { + const stub = await fresh(); + const interaction = { + id: id(70), + kind: 'approval' as const, + toolCall: call(), + resolution: null, + }; + const action = { toolCall: call(), grant: null, reason: 'locked' as const }; + await stub.commit({ wakeAt: future() }, [ + ...acceptedEvents(), + { type: 'interaction', interaction }, + { type: 'client_action', toolCallId: call().id, action }, + ]); + await stub.commit({ wakeAt: null }, [ + { + type: 'run', + run: run(10, { status: 'waiting', waiting: { toolCallId: call().id, reason: 'client' } }), + }, + ...Array.from({ length: 60 }, (_, n): EventEnvelope['event'] => ({ + type: 'message', + message: message(10_000 + n), + })), + ]); + await runInDurableObject(stub, instance => { + instance.store.compactEvents(); + }); + await abortAllDurableObjects(); + await runInDurableObject(getTestStoreStub(bindings.STORE, stub.id), instance => { + const snapshot = instance.store.snapshot(); + expect(snapshot?.unresolvedInteractions).toEqual([interaction]); + expect(snapshot?.pendingClientActions).toEqual([action]); + expect(snapshot?.activeRun?.state).toEqual({ + status: 'waiting', + waiting: { toolCallId: call().id, reason: 'client' }, + }); + expect(snapshot?.recentMessages.some(row => row.id === id(1010))).toBe(false); + }); + }); + + it('rejects a conflicting interaction resolution without replacing the first decision', async () => { + const stub = await fresh(); + const interaction = { + id: id(70), + kind: 'approval' as const, + toolCall: call(), + resolution: null, + }; + await stub.commit({ wakeAt: null }, [{ type: 'interaction', interaction }]); + const approved = { + ...interaction, + resolution: { + interactionId: interaction.id, + commandId: command.id, + decision: 'approve' as const, + }, + }; + await stub.commit( + { command, wakeAt: null }, + [{ type: 'interaction', interaction: approved }], + reply + ); + await runInDurableObject(stub, async (instance, state) => { + await expect( + instance.store.transition({ wakeAt: null }, () => ({ + events: [ + { + type: 'interaction', + interaction: { + ...approved, + resolution: { ...approved.resolution, decision: 'deny' }, + }, + }, + ], + })) + ).rejects.toThrow('command_conflict'); + expect(drizzle(state.storage).select().from(s.interactions).get()?.data).toEqual(approved); + expect(instance.store.snapshot()?.unresolvedInteractions).toEqual([]); + expect(instance.store.getCommand(command.id)?.reply).toEqual(reply); + }); + }); + + it('fails a snapshot instead of silently truncating unresolved work', async () => { + const stub = await fresh(); + await stub.commit( + { wakeAt: null }, + Array.from({ length: 201 }, (_, n) => ({ + type: 'interaction', + interaction: { id: id(1000 + n), kind: 'approval', toolCall: call(), resolution: null }, + })) + ); + await runInDurableObject(stub, (instance, state) => { + expect(() => instance.store.snapshot()).toThrow('limit_exceeded'); + expect(() => instance.store.compactEvents()).toThrow('limit_exceeded'); + expect(drizzle(state.storage).select().from(s.interactions).all()).toHaveLength(201); + expect(eventList(instance.store)).toHaveLength(200); + }); + }); + + it('deduplicates legacy UUIDs atomically and never upgrades supplied text to run authority', async () => { + const stub = await fresh(); + await runInDurableObject(stub, async (instance, state) => { + const forged = { + ...message(1010), + role: 'assistant', + parts: [{ type: 'tool_call', toolCall: call() }], + }; + expect(await instance.store.importLegacy(forged, 1)).toBe(true); + expect( + await instance.store.importLegacy({ ...forged, content: 'conflicting redelivery' }, 2) + ).toBe(false); + expect(instance.store.snapshot()).toMatchObject({ + activeRun: null, + queuedRuns: [], + eventCursor: 1, + }); + expect(instance.store.history().messages).toEqual([ + { + id: id(1010), + role: 'assistant', + content: 'hello', + clientId: null, + createdAt: time, + provenance: 'legacy', + parts: [{ type: 'text', text: 'hello' }], + }, + ]); + expect(drizzle(state.storage).select().from(s.conversation).get()?.legacyCursor).toBe(2); + expect(instance.store.history().messages).toHaveLength(1); + }); + }); + + it('recovers large legacy text through bounded replay, snapshots, and history after restart', async () => { + const stub = await fresh(); + const legacy = { + id: id(1011), + role: 'assistant', + content: 'é'.repeat(128 * 1024) + '\n終わり', + createdAt: time, + }; + const expected = { + ...legacy, + clientId: null, + provenance: 'legacy', + parts: [{ type: 'text', text: legacy.content }], + }; + await runInDurableObject(stub, async instance => { + await instance.store.importLegacy({ ...legacy, id: id(1010), content: 'before' }, 1); + }); + const client = createHarnessStore({ + conversationId: conversation.id, + clientId: id(2), + clock: { now: () => Date.now(), schedule: () => () => {} }, + transport: { + read: input => + runInDurableObject(getTestStoreStub(bindings.STORE, stub.id), instance => { + switch (input.type) { + case 'getSnapshot': + return instance.store.snapshot(); + case 'getHistory': + return instance.store.history(input.before, input.limit); + case 'getEvents': { + const page = instance.store.eventsAfter(input.after, input.limit); + expect( + new TextEncoder().encode(JSON.stringify(page)).byteLength + ).toBeLessThanOrEqual(256 * 1024); + return page; + } + default: + throw new Error('Unexpected storage read'); + } + }), + }, + }); + try { + await client.refresh(); + expect(client.getSnapshot().eventCursor).toBe(1); + await runInDurableObject(stub, async (instance, state) => { + expect(await instance.store.importLegacy(legacy, 2)).toBe(true); + expect( + await instance.store.importLegacy({ ...legacy, content: 'conflicting redelivery' }, 3) + ).toBe(false); + expect( + await instance.store.importLegacy({ ...legacy, id: id(1012), content: 'after' }, 4) + ).toBe(true); + expect(drizzle(state.storage).select().from(s.conversation).get()?.legacyCursor).toBe(4); + expect(eventList(instance.store).map(event => event.sequence)).toEqual([1]); + expect(instance.store.eventsAfter(1)).toEqual({ status: 'cursor_expired' }); + expect(eventList(instance.store, 2).map(event => event.sequence)).toEqual([3]); + }); + await abortAllDurableObjects(); + await client.refresh(); + expect(client.getSnapshot().connection.status).toBe('connected'); + expect(client.getSnapshot().eventCursor).toBe(3); + expect(client.getSnapshot().messages[legacy.id]).toEqual(expected); + expect(client.getSnapshot().messages[id(1012)]?.content).toBe('after'); + await runInDurableObject( + getTestStoreStub(bindings.STORE, stub.id), + async (instance, state) => { + expect(await instance.store.importLegacy(legacy, 5)).toBe(false); + expect(instance.store.history().messages.find(row => row.id === legacy.id)).toEqual( + expected + ); + expect( + instance.store.snapshot()?.recentMessages.find(row => row.id === legacy.id) + ).toEqual(expected); + for (let n = 0; n < 55; n++) + expect( + await instance.store.importLegacy( + { ...legacy, id: id(2000 + n), content: `later ${n}` }, + 6 + n + ) + ).toBe(true); + expect(drizzle(state.storage).select().from(s.conversation).get()?.legacyCursor).toBe(60); + instance.store.compactEvents(); + } + ); + await abortAllDurableObjects(); + await client.refresh(); + expect(client.getSnapshot().connection.status).toBe('connected'); + expect(client.getSnapshot().messages[legacy.id]).toBeUndefined(); + expect(client.getSnapshot().historyCursor).not.toBeNull(); + await client.loadHistory(); + expect(client.getSnapshot().messages[legacy.id]).toEqual(expected); + expect(Object.keys(client.getSnapshot().messages)).toHaveLength(58); + expect(client.getSnapshot().historyCursor).toBeNull(); + expect(client.getSnapshot().eventCursor).toBe(58); + expect(client.getSnapshot().activeRunId).toBeNull(); + expect(client.getSnapshot().queuedRunIds).toEqual([]); + } finally { + client.dispose(); + } + }); + + it('rolls back the legacy cursor with a failed large import and permits subsequent imports', async () => { + const stub = await fresh(); + const legacy = { + id: id(1011), + role: 'assistant', + content: 'é'.repeat(128 * 1024), + createdAt: time, + }; + await runInDurableObject(stub, async (instance, state) => { + await instance.store.importLegacy({ ...legacy, id: id(1010), content: 'before' }, 1); + const before = instance.store.snapshot(); + // Fail event insertion after the message write, inside the same SQLite transaction. + drizzle(state.storage) + .insert(s.events) + .values({ + sequence: 2, + data: { + protocolVersion: 1, + conversationId: conversation.id, + sequence: 2, + event: { type: 'message', message: message(9999) }, + }, + }) + .run(); + await expect(instance.store.importLegacy(legacy, 2)).rejects.toThrow(); + expect(instance.store.snapshot()).toEqual(before); + expect(instance.store.history().messages.map(row => row.id)).toEqual([id(1010)]); + expect(drizzle(state.storage).select().from(s.conversation).get()?.legacyCursor).toBe(1); + }); + await abortAllDurableObjects(); + await runInDurableObject(getTestStoreStub(bindings.STORE, stub.id), async (instance, state) => { + const db = drizzle(state.storage); + expect(db.select().from(s.conversation).get()?.legacyCursor).toBe(1); + expect(instance.store.history().messages.map(row => row.id)).toEqual([id(1010)]); + db.delete(s.events).where(eq(s.events.sequence, 2)).run(); + expect(await instance.store.importLegacy(legacy, 2)).toBe(true); + expect( + await instance.store.importLegacy({ ...legacy, id: id(1012), content: 'after retry' }, 3) + ).toBe(true); + expect(db.select().from(s.conversation).get()?.legacyCursor).toBe(3); + expect(instance.store.history().messages.map(row => row.content)).toEqual([ + 'before', + legacy.content, + 'after retry', + ]); + expect(instance.store.snapshot()?.eventCursor).toBe(3); + }); + }); + + it('persists projection work and fences stale or duplicate acknowledgments', async () => { + const stub = await fresh(); + await runInDurableObject(stub, async instance => { + await instance.store.transition({ command, wakeAt: future() }, db => { + db.insert(s.projectionWork) + .values([ + { id: 'b', messageId: id(1010), data: { text: 'hello' }, dueAt: 1 }, + { id: 'a', messageId: id(1011), data: { text: 'second' }, dueAt: 1 }, + { id: 'later', messageId: id(1012), data: {}, dueAt: 100 }, + ]) + .run(); + return { events: acceptedEvents(), reply }; + }); + }); + await abortAllDurableObjects(); + await runInDurableObject(getTestStoreStub(bindings.STORE, stub.id), instance => { + expect(instance.store.pendingProjections(1, 1).map(row => row.id)).toEqual(['a']); + expect(instance.store.acknowledgeProjection('a', 1, time)).toBe(false); + expect(instance.store.pendingProjections(1).map(row => row.id)).toEqual(['a', 'b']); + expect(instance.store.acknowledgeProjection('a', 0, time)).toBe(true); + expect(instance.store.acknowledgeProjection('a', 0, time)).toBe(false); + expect(instance.store.pendingProjections(1).map(row => row.id)).toEqual(['b']); + expect(instance.store.getCommand(command.id)?.reply).toEqual(reply); + }); + }); + + it.each(['partial', 'failed'] as const)( + 'never authorizes calls from a %s checkpoint', + async status => { + const stub = await fresh(); + await stub.commit({ wakeAt: future() }, acceptedEvents()); + await runInDurableObject(stub, async (instance, state) => { + await instance.store.transition({ wakeAt: null }, db => { + insertCheckpoint(db, { ...checkpoint, status }); + return { events: [] }; + }); + await expect( + instance.store.transition({ wakeAt: future() }, db => { + insertCall(db, call(), callDetails); + return { events: [] }; + }) + ).rejects.toThrow('invalid_input'); + const db = drizzle(state.storage); + expect(executableCheckpoint(db, checkpoint.id)).toBeNull(); + expect(db.select().from(s.calls).all()).toEqual([]); + expect(db.select().from(s.attempts).all()).toEqual([]); + expect(db.select().from(s.checkpoints).get()?.status).toBe(status); + }); + } + ); + + it('retains immutable dispatch inputs, grants, intent, and CAS outcomes through restart', async () => { + const stub = await fresh(); + await stub.commit({ wakeAt: future() }, acceptedEvents()); + await runInDurableObject(stub, async instance => { + await seedCall(instance.store); + await expect( + instance.store.transition({ wakeAt: future() }, db => { + compareAndSetCall(db, id(30), 0, { state: 'executing', approval: null, result: null }); + return { events: [] }; + }) + ).rejects.toThrow('invalid_input'); + await instance.store.transition({ wakeAt: future() }, db => { + insertGrant(db, grant); + insertAttempt(db, { id: id(60), toolCallId: id(30), generation: 1, grantId: grant.id }); + expect( + compareAndSetCall(db, id(30), 0, { state: 'executing', approval: null, result: null }) + ).toBe(true); + return { events: [] }; + }); + }); + await abortAllDurableObjects(); + await runInDurableObject(getTestStoreStub(bindings.STORE, stub.id), async (instance, state) => { + const db = drizzle(state.storage); + expect(executableCheckpoint(db, checkpoint.id)).toEqual(checkpoint); + expect(db.select().from(s.grants).get()?.data).toEqual(grant); + expect(db.select().from(s.attempts).get()?.intent).toEqual({ + toolCall: call(), + inputDigest: callDetails.inputDigest, + policy: callDetails.policy, + grant, + }); + expect(instance.store.callsForRun(id(10))[0]).toMatchObject({ + inputDigest: callDetails.inputDigest, + data: { ...call(), state: 'executing' }, + revision: 1, + }); + await instance.store.transition({ wakeAt: null }, transaction => { + expect( + compareAndSetCall(transaction, id(30), 0, { + state: 'settled', + approval: null, + result: { status: 'cancelled' }, + }) + ).toBe(false); + return { events: [] }; + }); + expect(instance.store.callsForRun(id(10))[0].data.state).toBe('executing'); + await expect( + instance.store.transition({ wakeAt: null }, transaction => { + const changed = { + state: 'waiting' as const, + approval: null, + result: null, + arguments: { enabled: false }, + }; + compareAndSetCall(transaction, id(30), 1, changed); + return { events: [] }; + }) + ).rejects.toThrow('invalid_input'); + const result = { + status: 'outcome_unknown' as const, + reason: 'Lost receipt', + providerReference: 'operation-1', + }; + await instance.store.transition({ wakeAt: null }, transaction => { + expect( + compareAndSetCall(transaction, id(30), 1, { state: 'settled', approval: null, result }) + ).toBe(true); + return { events: [] }; + }); + expect(instance.store.callsForRun(id(10))[0].data).toEqual({ + ...call(), + state: 'settled', + result, + }); + }); + }); + + it('orders calls and rejects duplicate call positions and mismatched definition versions', async () => { + const stub = await fresh(); + await stub.commit({ wakeAt: future() }, acceptedEvents()); + await runInDurableObject(stub, async instance => { + await seedCall(instance.store); + await instance.store.transition({ wakeAt: future() }, db => { + insertCall(db, call(32), { ...callDetails, position: 2 }); + insertCall(db, call(31), { ...callDetails, position: 1 }); + return { events: [] }; + }); + expect(instance.store.callsForRun(id(10), -1, 2).map(row => row.id)).toEqual([ + id(30), + id(31), + ]); + expect(instance.store.callsForRun(id(10), 1).map(row => row.id)).toEqual([id(32)]); + await expect( + instance.store.transition({ wakeAt: future() }, db => { + insertCall(db, call(33), callDetails); + return { events: [] }; + }) + ).rejects.toThrow(); + await expect( + instance.store.transition({ wakeAt: future() }, db => { + insertCall(db, { ...call(33), definitionVersion: '2' }, { ...callDetails, position: 3 }); + return { events: [] }; + }) + ).rejects.toThrow('invalid_input'); + expect(instance.store.callsForRun(id(10))).toHaveLength(3); + }); + }); +}); + +describe('stored identity and dispatch constraints', () => { + it('keeps an existing thread identity and scope immutable', async () => { + const stub = await fresh(); + await runInDurableObject(stub, async instance => { + for (const changes of [ + { id: id(99) }, + { ownerUserId: 'another-user' }, + { context: { type: 'organization' as const, organizationId: id(99) } }, + ]) { + expect(() => + instance.store.bindExistingConversation({ ...conversation, ...changes }) + ).toThrow('invalid_input'); + } + await instance.store.transition({ wakeAt: null }, () => ({ + events: [ + { + type: 'conversation', + conversation: { ...conversation, permissionMode: 'yolo', permissionRevision: 1 }, + }, + ], + })); + instance.store.bindExistingConversation(conversation); + expect(instance.store.snapshot()?.conversation).toEqual({ + ...conversation, + permissionMode: 'yolo', + permissionRevision: 1, + }); + }); + }); + + it('rejects calls and grants that do not match the stored conversation and target', async () => { + const stub = await fresh(); + await stub.commit({ wakeAt: future() }, acceptedEvents()); + await runInDurableObject(stub, async (instance, state) => { + await instance.store.transition({ wakeAt: null }, db => { + insertCheckpoint(db, checkpoint); + return { events: [] }; + }); + await expect( + instance.store.transition({ wakeAt: future() }, db => { + insertCall( + db, + { ...call(), context: { type: 'organization', organizationId: id(99) } }, + callDetails + ); + return { events: [] }; + }) + ).rejects.toThrow('invalid_input'); + expect(instance.store.callsForRun(id(10))).toEqual([]); + await instance.store.transition({ wakeAt: future() }, db => { + insertCall(db, call(), callDetails); + return { events: [] }; + }); + for (const changes of [ + { conversationId: id(99) }, + { ownerUserId: 'another-user' }, + { clientId: id(99) }, + { inputDigest: 'changed-input' }, + { definitionVersion: '2' }, + { context: { type: 'organization', organizationId: id(99) } }, + ]) { + await expect( + instance.store.transition({ wakeAt: future() }, db => { + insertGrant(db, { ...grant, ...changes }); + return { events: [] }; + }) + ).rejects.toThrow('invalid_input'); + } + expect(drizzle(state.storage).select().from(s.grants).all()).toEqual([]); + expect(instance.store.callsForRun(id(10))[0].data).toEqual(call()); + }); + }); + + it('retains one grant and execution intent per call generation', async () => { + const stub = await fresh(); + await stub.commit({ wakeAt: future() }, acceptedEvents()); + await runInDurableObject(stub, async (instance, state) => { + await seedCall(instance.store); + await instance.store.transition({ wakeAt: future() }, db => { + insertGrant(db, grant); + insertAttempt(db, { id: id(60), toolCallId: id(30), generation: 1, grantId: grant.id }); + return { events: [] }; + }); + await expect( + instance.store.transition({ wakeAt: future() }, db => { + insertGrant(db, { ...grant, id: id(51) }); + return { events: [] }; + }) + ).rejects.toThrow(); + await expect( + instance.store.transition({ wakeAt: future() }, db => { + insertAttempt(db, { id: id(61), toolCallId: id(30), generation: 1, grantId: grant.id }); + return { events: [] }; + }) + ).rejects.toThrow(); + const db = drizzle(state.storage); + expect( + db + .select() + .from(s.grants) + .all() + .map(row => row.data) + ).toEqual([grant]); + expect( + db + .select() + .from(s.attempts) + .all() + .map(row => row.id) + ).toEqual([id(60)]); + }); + }); +}); + +describe('durable prearm and transition gate', () => { + const terminalStates = [ + { status: 'completed' }, + { status: 'cancelled' }, + { + status: 'failed', + error: { code: 'invalid_output', message: 'Invalid model output.', retryable: false }, + }, + ] satisfies Run['state'][]; + + async function waitingWithQueue() { + const stub = await fresh(); + await stub.commit({ wakeAt: future() }, [ + ...acceptedEvents(), + ...acceptedEvents(11), + { + type: 'run', + run: run(10, { status: 'waiting', waiting: { reason: 'question', toolCallId: id(30) } }), + }, + ]); + // Consume the admission alarm while the active wait still blocks the queue. + expect(await runDurableObjectAlarm(stub)).toBe(true); + return stub; + } + + it.each(terminalStates)( + 'rolls back a waiting run becoming $status when the queue has no prearmed wake', + async terminal => { + const stub = await waitingWithQueue(); + await runInDurableObject(stub, async (instance, state) => { + const before = instance.store.snapshot(); + expect(await state.storage.getAlarm()).toBeNull(); + for (const releaseBeforeEvent of [false, true]) { + await expect( + instance.store.transition({ command, wakeAt: null }, db => { + if (releaseBeforeEvent) compareAndSetActiveRun(db, id(10), null); + return { events: [{ type: 'run', run: run(10, terminal) }], reply }; + }) + ).rejects.toThrow('invalid_input'); + expect(instance.store.snapshot()).toEqual(before); + expect(instance.store.getCommand(command.id)).toBeNull(); + expect(eventList(instance.store).map(event => event.sequence)).toEqual([1, 2, 3, 4, 5]); + } + const failing = await openStore(state, { + getAlarm: () => state.storage.getAlarm(), + setAlarm: async () => { + throw new Error('injected alarm failure'); + }, + }); + await expect( + failing.transition({ command, wakeAt: future() }, () => ({ + events: [{ type: 'run', run: run(10, terminal) }], + reply, + })) + ).rejects.toMatchObject({ code: 'storage_unavailable', retryable: true }); + expect(instance.store.snapshot()).toEqual(before); + expect(instance.store.getCommand(command.id)).toBeNull(); + expect(eventList(instance.store).map(event => event.sequence)).toEqual([1, 2, 3, 4, 5]); + expect(await state.storage.getAlarm()).toBeNull(); + const wakeAt = future(); + expect( + await instance.store.transition({ command, wakeAt }, () => ({ + events: [{ type: 'run', run: run(10, terminal) }], + reply, + })) + ).toEqual(reply); + expect(instance.store.snapshot()).toMatchObject({ activeRun: null, eventCursor: 6 }); + expect(instance.store.queuedRuns().map(row => row.id)).toEqual([id(11)]); + expect(await state.storage.getAlarm()).toBe(wakeAt); + }); + } + ); + + it.each(terminalStates)( + 'wakes queued work after a waiting run becomes $status and restarts without another submission', + async terminal => { + const stub = await waitingWithQueue(), + earliest = future(); + const results = await Promise.all([ + stub.commit( + { command, wakeAt: earliest + 1000 }, + [{ type: 'run', run: run(10, terminal) }], + reply + ), + stub.commit({ wakeAt: earliest }, []), + stub.commit({ wakeAt: earliest + 2000 }, []), + ]); + expect(results[0]).toEqual(reply); + await runInDurableObject(stub, async (instance, state) => { + expect(await state.storage.getAlarm()).toBe(earliest); + expect(instance.store.snapshot()).toMatchObject({ activeRun: null, eventCursor: 6 }); + expect(instance.store.queuedRuns().map(row => row.id)).toEqual([id(11)]); + expect(instance.store.getCommand(command.id)?.reply).toEqual(reply); + }); + await abortAllDurableObjects(); + const restarted = getTestStoreStub(bindings.STORE, stub.id); + expect(await runDurableObjectAlarm(restarted)).toBe(true); + await runInDurableObject(restarted, (instance, state) => { + expect(instance.store.snapshot()).toMatchObject({ + activeRun: null, + queuedRuns: [], + eventCursor: 7, + }); + const db = drizzle(state.storage); + expect( + db + .select() + .from(s.runs) + .where(eq(s.runs.id, id(10))) + .get()?.data + ).toEqual(run(10, terminal)); + expect( + db + .select() + .from(s.runs) + .where(eq(s.runs.id, id(11))) + .get()?.data + ).toEqual(run(11, { status: 'completed' })); + expect(instance.store.getCommand(command.id)?.reply).toEqual(reply); + }); + } + ); + + it('completes a waiting run without polling when no queued work remains', async () => { + const stub = await fresh(); + await stub.commit({ wakeAt: future() }, [ + ...acceptedEvents(), + { + type: 'run', + run: run(10, { status: 'waiting', waiting: { reason: 'question', toolCallId: id(30) } }), + }, + ]); + expect(await runDurableObjectAlarm(stub)).toBe(true); + expect( + await stub.commit( + { command, wakeAt: null }, + [{ type: 'run', run: run(10, { status: 'completed' }) }], + reply + ) + ).toEqual(reply); + await runInDurableObject(stub, async (instance, state) => { + expect(await state.storage.getAlarm()).toBeNull(); + expect(instance.store.snapshot()).toMatchObject({ + activeRun: null, + queuedRuns: [], + eventCursor: 4, + }); + expect(instance.store.getCommand(command.id)?.reply).toEqual(reply); + }); + }); + + it('leaves no accepted work after failed prearming and permits retry with the same command', async () => { + const stub = await fresh(); + await runInDurableObject(stub, async (instance, state) => { + const failing = await openStore(state, { + getAlarm: () => state.storage.getAlarm(), + setAlarm: async () => { + throw new Error('injected alarm failure'); + }, + }); + await expect( + failing.transition({ command, wakeAt: future() }, () => ({ + events: acceptedEvents(), + reply, + })) + ).rejects.toMatchObject({ code: 'storage_unavailable', retryable: true }); + expect(instance.store.snapshot()).toMatchObject({ + queuedRuns: [], + recentMessages: [], + eventCursor: 0, + }); + expect(instance.store.getCommand(command.id)).toBeNull(); + expect(await state.storage.getAlarm()).toBeNull(); + const wakeAt = future(); + expect( + await instance.store.transition({ command, wakeAt }, () => ({ + events: acceptedEvents(), + reply, + })) + ).toEqual(reply); + expect(await state.storage.getAlarm()).toBe(wakeAt); + expect(instance.store.snapshot()?.queuedRuns.map(row => row.id)).toEqual([id(10)]); + }); + }); + + it('arms before writing and preserves the earliest deadline across concurrent handlers', async () => { + const stub = await fresh(), + earliest = future(); + await runInDurableObject(stub, async (instance, state) => { + const observed = await openStore(state, { + getAlarm: () => state.storage.getAlarm(), + async setAlarm(deadline) { + expect(instance.store.getCommand(command.id)).toBeNull(); + expect(instance.store.snapshot()?.queuedRuns).toEqual([]); + await state.storage.setAlarm(deadline); + }, + }); + await observed.transition({ command, wakeAt: earliest + 1000 }, () => ({ + events: acceptedEvents(), + reply, + })); + }); + await Promise.all([ + stub.commit({ wakeAt: earliest + 2000 }, []), + stub.commit({ wakeAt: earliest }, []), + stub.commit({ wakeAt: earliest + 3000 }, []), + ]); + await runInDurableObject(stub, async (instance, state) => { + await instance.store.transition({ wakeAt: null }, () => ({ events: [] })); + expect(await state.storage.getAlarm()).toBe(earliest); + expect(instance.store.getCommand(command.id)?.reply).toEqual(reply); + }); + }); + + it('leaves a harmless durable alarm when the transaction fails after prearming', async () => { + const stub = await fresh(); + await runInDurableObject(stub, async (instance, state) => { + await expect( + instance.store.transition({ command, wakeAt: future() }, () => { + throw new Error('crash after arm'); + }) + ).rejects.toThrow('crash after arm'); + expect(await state.storage.getAlarm()).not.toBeNull(); + expect(instance.store.getCommand(command.id)).toBeNull(); + }); + await abortAllDurableObjects(); + const restarted = getTestStoreStub(bindings.STORE, stub.id); + expect(await runDurableObjectAlarm(restarted)).toBe(true); + await runInDurableObject(restarted, instance => { + expect(instance.store.snapshot()).toMatchObject({ + recentMessages: [], + queuedRuns: [], + activeRun: null, + eventCursor: 0, + }); + }); + }); + + it('wakes committed work after restart without another client submission', async () => { + const stub = await fresh(); + await stub.commit({ command, wakeAt: future() }, acceptedEvents(), reply); + await abortAllDurableObjects(); + const restarted = getTestStoreStub(bindings.STORE, stub.id); + expect(await runDurableObjectAlarm(restarted)).toBe(true); + await runInDurableObject(restarted, (instance, state) => { + expect(instance.store.snapshot()).toMatchObject({ + queuedRuns: [], + activeRun: null, + eventCursor: 3, + }); + expect(drizzle(state.storage).select().from(s.runs).get()?.data).toEqual( + run(10, { status: 'completed' }) + ); + expect(instance.store.getCommand(command.id)?.reply).toEqual(reply); + }); + }); + + it('does not poll wait-only state and requires a prearm when work leaves waiting', async () => { + const stub = await fresh(); + await stub.commit({ wakeAt: future() }, acceptedEvents()); + await runInDurableObject(stub, async (instance, state) => { + await state.storage.deleteAlarm(); + const waiting = run(10, { + status: 'waiting', + waiting: { toolCallId: id(30), reason: 'question' }, + }); + await instance.store.transition({ wakeAt: null }, () => ({ + events: [{ type: 'run', run: waiting }], + })); + expect(await state.storage.getAlarm()).toBeNull(); + await expect( + instance.store.transition({ wakeAt: null }, () => ({ + events: [{ type: 'run', run: run(10, { status: 'running' }) }], + })) + ).rejects.toThrow('invalid_input'); + expect(instance.store.snapshot()?.activeRun).toEqual(waiting); + const wakeAt = future(); + await instance.store.transition({ wakeAt }, () => ({ + events: [{ type: 'run', run: run(10, { status: 'running' }) }], + })); + expect(await state.storage.getAlarm()).toBe(wakeAt); + expect(instance.store.snapshot()?.activeRun?.state.status).toBe('running'); + }); + }); + + it('rejects an asynchronous transaction result and rolls back its synchronous writes', async () => { + const stub = await fresh(); + await runInDurableObject(stub, async (instance, state) => { + await expect( + transitionWithWake(state, () => ({ + wakeAt: future(), + commit: async () => { + drizzle(state.storage) + .insert(s.projectionWork) + .values({ id: 'bad-async', messageId: id(1010), data: {}, dueAt: 0 }) + .run(); + return 'not synchronous'; + }, + })) + ).rejects.toThrow('invalid_input'); + expect(instance.store.pendingProjections(Date.now())).toEqual([]); + }); + }); +}); diff --git a/services/agent-harness/src/db/store.ts b/services/agent-harness/src/db/store.ts new file mode 100644 index 0000000000..c1206bf45f --- /dev/null +++ b/services/agent-harness/src/db/store.ts @@ -0,0 +1,401 @@ +import { and, asc, desc, eq, gt, inArray, isNull, lt, lte, or, sql } from 'drizzle-orm'; +import { drizzle } from 'drizzle-orm/durable-sqlite'; +import { migrate } from 'drizzle-orm/durable-sqlite/migrator'; +import { z } from 'zod'; +import { + canonicalizeValidatedInput, + commandReplayDecision, +} from '@kilocode/agent-harness/commands'; +import { + ConversationSchema, + EventCursorSchema, + EventEnvelopeSchema, + InteractionSchema, + LegacyMessageSchema, + MessageSchema, + PendingClientActionSchema, + RunSchema, + SnapshotSchema, + ToolCallSchema, + type ConversationProducer, + type EventEnvelope, + type Snapshot, +} from '@kilocode/agent-harness/contracts'; +import { CommandReplySchema, type CommandReply } from '@kilocode/agent-harness/journal'; +import migrations from '../../drizzle/migrations.js'; +import * as s from './sqlite-schema'; +import { + applyEvent, + conversationRow, + pageLimit, + readConversation, + type StoreDatabase, +} from './records'; +import { StoreError, transitionWithWake, type AlarmStorage } from './wake'; + +const maxPageBytes = 256 * 1024; +const byteLength = (value: unknown) => new TextEncoder().encode(JSON.stringify(value)).byteLength; +const HistoryCursorSchema = z.strictObject({ createdAt: z.iso.datetime(), id: z.uuid() }); +type Changes = { events: EventEnvelope['event'][]; reply?: CommandReply }; +type Transition = { command?: { id: string; fingerprint: string }; wakeAt: number | null }; + +// Call from the Durable Object constructor's blockConcurrencyWhile, before serving any operation. +export async function openStore(state: DurableObjectState, alarms: AlarmStorage = state.storage) { + const db = drizzle(state.storage); + await migrate(db, migrations); + + function appendEvents(inputs: EventEnvelope['event'][], wakeAt: number | null) { + const meta = conversationRow(db); + let sequence = meta.sequence; + for (const input of inputs) { + const envelope = EventEnvelopeSchema.parse({ + protocolVersion: 1, + conversationId: meta.id, + sequence: ++sequence, + event: input, + }); + if ( + envelope.event.type === 'run' && + ['queued', 'running', 'stopping'].includes(envelope.event.run.state.status) && + wakeAt === null + ) + throw new StoreError('invalid_input'); + // Deployed legacy text has no size cap. Oversized legacy deltas recover through snapshots. + if ( + byteLength({ status: 'events', events: [envelope] }) > maxPageBytes && + (envelope.event.type !== 'message' || envelope.event.message.provenance !== 'legacy') + ) + throw new StoreError('limit_exceeded'); + applyEvent(db, envelope.event, sequence); + db.insert(s.events).values({ sequence, data: envelope }).run(); + } + db.update(s.conversation).set({ sequence }).where(eq(s.conversation.singleton, 1)).run(); + return sequence; + } + function history(before: string | null = null, limit = 50) { + pageLimit(limit); + let cursor: z.infer | null = null; + if (before !== null) { + try { + cursor = HistoryCursorSchema.parse( + JSON.parse(atob(before.replace(/-/g, '+').replace(/_/g, '/'))) + ); + } catch { + throw new StoreError('invalid_input'); + } + } + const rows = db + .select() + .from(s.messages) + .where( + cursor + ? or( + lt(s.messages.createdAt, cursor.createdAt), + and(eq(s.messages.createdAt, cursor.createdAt), lt(s.messages.id, cursor.id)) + ) + : undefined + ) + .orderBy(desc(s.messages.createdAt), desc(s.messages.id)) + .limit(limit + 1) + .all(); + const page = rows.slice(0, limit), + oldest = page.at(-1); + return { + messages: page.reverse().map(row => MessageSchema.parse(row.data)), + historyCursor: + rows.length > limit && oldest + ? btoa(JSON.stringify({ createdAt: oldest.createdAt, id: oldest.id })) + .replace(/\+/g, '-') + .replace(/\//g, '_') + .replace(/=+$/, '') + : null, + }; + } + function buildSnapshot(): Snapshot | null { + const meta = db.select().from(s.conversation).where(eq(s.conversation.singleton, 1)).get(); + if (!meta) return null; + const recent = history(); + const queued = db + .select() + .from(s.runs) + .where(eq(s.runs.status, 'queued')) + .orderBy(asc(s.runs.position)) + .limit(201) + .all(); + const unresolved = db + .select() + .from(s.interactions) + .where(eq(s.interactions.resolved, false)) + .orderBy(asc(s.interactions.sequence)) + .limit(201) + .all(); + const actions = db + .select() + .from(s.clientActions) + .orderBy(asc(s.clientActions.sequence)) + .limit(201) + .all(); + // A bounded snapshot must fail, not silently hide unresolved work. + if ([queued, unresolved, actions].some(rows => rows.length > 200)) + throw new StoreError('limit_exceeded'); + const current = meta.activeRunId + ? db.select().from(s.runs).where(eq(s.runs.id, meta.activeRunId)).get() + : null; + if (meta.activeRunId && !current) throw new StoreError('invalid_input'); + return SnapshotSchema.parse({ + protocolVersion: 1, + conversation: readConversation(db), + recentMessages: recent.messages, + historyCursor: recent.historyCursor, + activeRun: current ? RunSchema.parse(current.data) : null, + queuedRuns: queued.map(row => RunSchema.parse(row.data)), + unresolvedInteractions: unresolved.map(row => InteractionSchema.parse(row.data)), + pendingClientActions: actions.map(row => PendingClientActionSchema.parse(row.data)), + eventCursor: meta.sequence, + }); + } + function getCommand(id: string) { + const row = db + .select() + .from(s.commands) + .where(eq(s.commands.id, z.uuid().parse(id))) + .get(); + return row + ? { + fingerprint: row.fingerprint, + reply: CommandReplySchema.parse(row.reply), + sequence: row.sequence, + } + : null; + } + return { + // Authority must come from the authenticated PostgreSQL lookup, never a client-selected new ID. + bindExistingConversation(input: ConversationProducer) { + const value = ConversationSchema.parse(input); + return db.transaction(() => { + const old = db.select().from(s.conversation).where(eq(s.conversation.singleton, 1)).get(); + if (old) { + if ( + old.id !== value.id || + old.ownerUserId !== value.ownerUserId || + canonicalizeValidatedInput(old.context) !== canonicalizeValidatedInput(value.context) + ) + throw new StoreError('invalid_input'); + } else db.insert(s.conversation).values(value).run(); + return readConversation(db); + }); + }, + getCommand, + transition(options: Transition, write: (transaction: StoreDatabase) => Changes) { + return transitionWithWake( + state, + () => { + const command = options.command; + if (command) { + z.string().min(1).parse(command.fingerprint); + const prior = getCommand(command.id); + const decision = commandReplayDecision(prior?.fingerprint, command.fingerprint); + if (decision === 'replay' && prior) return { value: prior.reply }; + if (decision === 'command_conflict') + return { + value: { + status: 'rejected', + commandId: command.id, + error: { + code: 'command_conflict', + message: 'This command has different stored input.', + retryable: false, + }, + }, + }; + } + return { + wakeAt: options.wakeAt, + commit: () => { + const activeRunId = conversationRow(db).activeRunId; + const changes = write(db); + const reply = changes.reply ? CommandReplySchema.parse(changes.reply) : undefined; + if (command && (!reply || reply.commandId !== command.id)) + throw new StoreError('invalid_input'); + const sequence = appendEvents(changes.events, options.wakeAt); + // Include callback writes: releasing the active run can make the existing queue runnable. + if ( + options.wakeAt === null && + activeRunId !== null && + conversationRow(db).activeRunId === null && + db + .select({ id: s.runs.id }) + .from(s.runs) + .where(eq(s.runs.status, 'queued')) + .limit(1) + .get() + ) + throw new StoreError('invalid_input'); + if (command) + db.insert(s.commands) + .values({ id: command.id, fingerprint: command.fingerprint, reply, sequence }) + .run(); + return reply; + }, + }; + }, + alarms + ); + }, + snapshot: () => db.transaction(buildSnapshot), + history, + eventsAfter(after: number, limit = 200) { + EventCursorSchema.parse(after); + pageLimit(limit); + return db.transaction(() => { + const meta = conversationRow(db); + if (after < meta.compactedThrough || after > meta.sequence) + return { status: 'cursor_expired' } as const; + const page: EventEnvelope[] = []; + let bytes = byteLength({ status: 'events', events: page }), + cursor = after; + // Read at most one extra record, rather than loading 200 maximum-size records into memory. + for (let count = 0; count < limit; count++) { + const row = db + .select() + .from(s.events) + .where(gt(s.events.sequence, cursor)) + .orderBy(asc(s.events.sequence)) + .limit(1) + .get(); + if (!row) break; + const event = EventEnvelopeSchema.parse(row.data); + const size = byteLength(event) + (page.length ? 1 : 0); + if (bytes + size > maxPageBytes) { + if (!page.length) return { status: 'cursor_expired' } as const; + break; + } + page.push(event); + bytes += size; + cursor = event.sequence; + } + return { status: 'events', events: page } as const; + }); + }, + queuedRuns(afterPosition = 0, limit = 200) { + EventCursorSchema.parse(afterPosition); + return db + .select() + .from(s.runs) + .where(and(eq(s.runs.status, 'queued'), gt(s.runs.position, afterPosition))) + .orderBy(asc(s.runs.position)) + .limit(pageLimit(limit)) + .all() + .map(row => ({ ...row, data: RunSchema.parse(row.data) })); + }, + callsForRun(runId: string, afterPosition = -1, limit = 200) { + z.int().min(-1).parse(afterPosition); + return db + .select() + .from(s.calls) + .where(and(eq(s.calls.runId, z.uuid().parse(runId)), gt(s.calls.position, afterPosition))) + .orderBy(asc(s.calls.position)) + .limit(pageLimit(limit)) + .all() + .map(row => ({ + ...row, + data: ToolCallSchema.parse(row.data), + policy: z.json().parse(row.policy), + })); + }, + compactEvents(limit = 200) { + pageLimit(limit); + return db.transaction(() => { + const snapshot = buildSnapshot(); + if (!snapshot) return 0; + const rows = db + .select({ sequence: s.events.sequence }) + .from(s.events) + .where(lte(s.events.sequence, snapshot.eventCursor)) + .orderBy(asc(s.events.sequence)) + .limit(limit) + .all(); + const last = rows.at(-1); + if (!last) return 0; + db.insert(s.snapshots) + .values({ cursor: snapshot.eventCursor, data: snapshot }) + .onConflictDoUpdate({ + target: s.snapshots.singleton, + set: { cursor: snapshot.eventCursor, data: snapshot }, + }) + .run(); + db.delete(s.events) + .where( + inArray( + s.events.sequence, + rows.map(row => row.sequence) + ) + ) + .run(); + db.update(s.conversation) + .set({ compactedThrough: last.sequence }) + .where(eq(s.conversation.singleton, 1)) + .run(); + // Keep original commands, canonical messages/parts, checkpoints, and unresolved records. + return rows.length; + }); + }, + importLegacy(input: unknown, appendSequence: number) { + const message = LegacyMessageSchema.parse(input); + EventCursorSchema.positive().parse(appendSequence); + return transitionWithWake( + state, + () => ({ + wakeAt: null, + commit: () => { + const meta = conversationRow(db); + const exists = db + .select({ id: s.messages.id }) + .from(s.messages) + .where(eq(s.messages.id, message.id)) + .get(); + if (!exists) appendEvents([{ type: 'message', message }], null); + db.update(s.conversation) + .set({ legacyCursor: Math.max(meta.legacyCursor, appendSequence) }) + .where(eq(s.conversation.singleton, 1)) + .run(); + return !exists; + }, + }), + alarms + ); + }, + pendingProjections(now: number, limit = 200) { + z.int().nonnegative().parse(now); + return db + .select() + .from(s.projectionWork) + .where(and(isNull(s.projectionWork.acknowledgedAt), lte(s.projectionWork.dueAt, now))) + .orderBy(asc(s.projectionWork.dueAt), asc(s.projectionWork.id)) + .limit(pageLimit(limit)) + .all() + .map(row => ({ ...row, data: z.json().parse(row.data) })); + }, + acknowledgeProjection(id: string, revision: number, acknowledgedAt: string) { + z.int().nonnegative().parse(revision); + z.iso.datetime().parse(acknowledgedAt); + return db.transaction(() => + Boolean( + db + .update(s.projectionWork) + .set({ acknowledgedAt, revision: sql`${s.projectionWork.revision} + 1` }) + .where( + and( + eq(s.projectionWork.id, id), + eq(s.projectionWork.revision, revision), + isNull(s.projectionWork.acknowledgedAt) + ) + ) + .returning({ id: s.projectionWork.id }) + .get() + ) + ); + }, + }; +} +export type ConversationStore = Awaited>; diff --git a/services/agent-harness/src/db/test-env.d.ts b/services/agent-harness/src/db/test-env.d.ts new file mode 100644 index 0000000000..9b4ad83333 --- /dev/null +++ b/services/agent-harness/src/db/test-env.d.ts @@ -0,0 +1 @@ +import '@cloudflare/vitest-pool-workers/types'; diff --git a/services/agent-harness/src/db/test-worker.ts b/services/agent-harness/src/db/test-worker.ts new file mode 100644 index 0000000000..0c647fd8b6 --- /dev/null +++ b/services/agent-harness/src/db/test-worker.ts @@ -0,0 +1,64 @@ +import { DurableObject } from 'cloudflare:workers'; +import { drizzle } from 'drizzle-orm/durable-sqlite'; +import { migrate } from 'drizzle-orm/durable-sqlite/migrator'; +import type { EventEnvelope } from '@kilocode/agent-harness/contracts'; +import type { CommandReply } from '@kilocode/agent-harness/journal'; +import migrations from '../../drizzle/migrations.js'; +import { openStore, type ConversationStore } from './store'; + +export class TestStore extends DurableObject { + private initialized: ConversationStore | undefined; + + constructor(ctx: DurableObjectState, env: unknown) { + super(ctx, env); + void ctx.blockConcurrencyWhile(async () => { + this.initialized = await openStore(ctx); + }); + } + + get store() { + if (!this.initialized) throw new Error('Store initialization has not finished'); + return this.initialized; + } + + commit( + options: Parameters[0], + events: EventEnvelope['event'][], + reply?: CommandReply + ) { + return this.store.transition(options, () => ({ events, reply })); + } + + // Test-only progress marker. This is not the production scheduler or alarm handler. + async alarm() { + if (this.store.snapshot()?.activeRun) return; + const queued = this.store.queuedRuns(0, 1)[0]; + if (queued) + await this.store.transition({ wakeAt: null }, () => ({ + events: [{ type: 'run', run: { ...queued.data, state: { status: 'completed' } } }], + })); + } +} + +export class OldTestStore extends DurableObject { + constructor(ctx: DurableObjectState, env: unknown) { + super(ctx, env); + void ctx.blockConcurrencyWhile(() => + migrate(drizzle(ctx.storage), { + ...migrations, + journal: { ...migrations.journal, entries: migrations.journal.entries.slice(0, 1) }, + }) + ); + } +} + +export const getTestStoreStub = ( + namespace: DurableObjectNamespace, + name: string | DurableObjectId +) => (typeof name === 'string' ? namespace.getByName(name) : namespace.get(name)); +export const getOldTestStoreStub = ( + namespace: DurableObjectNamespace, + name: string +) => namespace.getByName(name); + +export default {} satisfies ExportedHandler; diff --git a/services/agent-harness/src/db/wake.ts b/services/agent-harness/src/db/wake.ts new file mode 100644 index 0000000000..fd1537aced --- /dev/null +++ b/services/agent-harness/src/db/wake.ts @@ -0,0 +1,54 @@ +export class StoreError extends Error { + constructor( + readonly code: 'storage_unavailable' | 'invalid_input' | 'limit_exceeded' | 'command_conflict', + readonly retryable = false + ) { + super(code); + } +} + +export type AlarmStorage = Pick; +// wakeAt is the earliest deadline required by the change; null is reserved for wait-only changes. +type Prepared = { value: T } | { wakeAt: number | null; commit: () => T }; + +// Both command handlers and alarm handlers use this gate. No network I/O belongs in prepare/commit. +export async function transitionWithWake( + state: DurableObjectState, + prepare: () => Prepared, + alarms: AlarmStorage = state.storage +): Promise { + const outcome = await state.blockConcurrencyWhile(async () => { + try { + const prepared = prepare(); + if ('value' in prepared) return { ok: true, value: prepared.value } as const; + if (prepared.wakeAt !== null) { + if (!Number.isSafeInteger(prepared.wakeAt) || prepared.wakeAt < 0) + throw new StoreError('invalid_input'); + try { + const existing = await alarms.getAlarm(); + await alarms.setAlarm(Math.min(existing ?? prepared.wakeAt, prepared.wakeAt)); + } catch { + throw new StoreError('storage_unavailable', true); + } + } + // An armed but uncommitted wake is harmless. A committed transition always retains its wake. + const value = state.storage.transactionSync(() => { + const result = prepared.commit(); + if ( + result !== null && + (typeof result === 'object' || typeof result === 'function') && + 'then' in result + ) { + throw new StoreError('invalid_input'); + } + return result; + }); + return { ok: true, value } as const; + } catch (error) { + // Keep expected command failures from resetting the object through blockConcurrencyWhile. + return { ok: false, error } as const; + } + }); + if (!outcome.ok) throw outcome.error; + return outcome.value; +} diff --git a/services/agent-harness/vitest.config.ts b/services/agent-harness/vitest.config.ts index 3941cf0e5b..a5748491d5 100644 --- a/services/agent-harness/vitest.config.ts +++ b/services/agent-harness/vitest.config.ts @@ -1,17 +1,21 @@ -import { existsSync } from 'node:fs'; -import { URL } from 'node:url'; import { cloudflareTest } from '@cloudflare/vitest-pool-workers'; import { defineConfig } from 'vitest/config'; export default defineConfig({ + resolve: { alias: [{ find: /^(.*\.sql)$/, replacement: '$1?raw' }] }, plugins: [ cloudflareTest({ - // The deployment config arrives with the Worker entrypoint, not before it. - wrangler: existsSync(new URL('./wrangler.jsonc', import.meta.url)) - ? { configPath: './wrangler.jsonc' } - : undefined, + // Storage tests must not load production bindings or environment files. + main: './src/db/test-worker.ts', remoteBindings: false, - miniflare: { compatibilityDate: '2026-06-05', compatibilityFlags: ['nodejs_compat'] }, + miniflare: { + compatibilityDate: '2026-06-05', + compatibilityFlags: ['nodejs_compat'], + durableObjects: { + STORE: { className: 'TestStore', useSQLite: true }, + OLD_STORE: { className: 'OldTestStore', useSQLite: true }, + }, + }, }), ], test: { include: ['src/**/*.test.ts'], passWithNoTests: true },