diff --git a/apps/daemon/drizzle/0013_famous_randall.sql b/apps/daemon/drizzle/0013_famous_randall.sql new file mode 100644 index 000000000..d42e1ac66 --- /dev/null +++ b/apps/daemon/drizzle/0013_famous_randall.sql @@ -0,0 +1 @@ +ALTER TABLE `sessions` ADD `event_epoch` integer DEFAULT 0 NOT NULL; \ No newline at end of file diff --git a/apps/daemon/drizzle/meta/0013_snapshot.json b/apps/daemon/drizzle/meta/0013_snapshot.json new file mode 100644 index 000000000..a2bae5232 --- /dev/null +++ b/apps/daemon/drizzle/meta/0013_snapshot.json @@ -0,0 +1,1327 @@ +{ + "version": "6", + "dialect": "sqlite", + "id": "882ec2b6-a29e-44e2-893e-b19adff08199", + "prevId": "65a8fc35-ee6f-4e2c-a3dc-9f2419f3b5a6", + "tables": { + "conversation_operations": { + "name": "conversation_operations", + "columns": { + "operation_id": { + "name": "operation_id", + "type": "text", + "primaryKey": true, + "notNull": true, + "autoincrement": false + }, + "session_id": { + "name": "session_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "kind": { + "name": "kind", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "state": { + "name": "state", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "turn_id": { + "name": "turn_id", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "error_code": { + "name": "error_code", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "error_message": { + "name": "error_message", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "created_at": { + "name": "created_at", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "resolved_at": { + "name": "resolved_at", + "type": "integer", + "primaryKey": false, + "notNull": false, + "autoincrement": false + } + }, + "indexes": { + "conversation_operations_session_idx": { + "name": "conversation_operations_session_idx", + "columns": ["session_id"], + "isUnique": false + }, + "conversation_operations_open_session_unique": { + "name": "conversation_operations_open_session_unique", + "columns": ["session_id"], + "isUnique": true, + "where": "state = 'open'" + } + }, + "foreignKeys": { + "conversation_operations_session_id_sessions_session_id_fk": { + "name": "conversation_operations_session_id_sessions_session_id_fk", + "tableFrom": "conversation_operations", + "tableTo": "sessions", + "columnsFrom": ["session_id"], + "columnsTo": ["session_id"], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": {} + }, + "conversation_turns": { + "name": "conversation_turns", + "columns": { + "turn_id": { + "name": "turn_id", + "type": "text", + "primaryKey": true, + "notNull": true, + "autoincrement": false + }, + "session_id": { + "name": "session_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "parent_turn_id": { + "name": "parent_turn_id", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "sibling_ordinal": { + "name": "sibling_ordinal", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "input_type": { + "name": "input_type", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "prompt_id": { + "name": "prompt_id", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "command_name": { + "name": "command_name", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "command_arguments": { + "name": "command_arguments", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "shell_command": { + "name": "shell_command", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "run_id": { + "name": "run_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "state": { + "name": "state", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "created_at": { + "name": "created_at", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + } + }, + "indexes": { + "conversation_turns_session_idx": { + "name": "conversation_turns_session_idx", + "columns": ["session_id"], + "isUnique": false + }, + "conversation_turns_sibling_unique": { + "name": "conversation_turns_sibling_unique", + "columns": ["session_id", "parent_turn_id", "sibling_ordinal"], + "isUnique": true, + "where": "parent_turn_id IS NOT NULL" + }, + "conversation_turns_root_sibling_unique": { + "name": "conversation_turns_root_sibling_unique", + "columns": ["session_id", "sibling_ordinal"], + "isUnique": true, + "where": "parent_turn_id IS NULL" + } + }, + "foreignKeys": { + "conversation_turns_session_id_sessions_session_id_fk": { + "name": "conversation_turns_session_id_sessions_session_id_fk", + "tableFrom": "conversation_turns", + "tableTo": "sessions", + "columnsFrom": ["session_id"], + "columnsTo": ["session_id"], + "onDelete": "cascade", + "onUpdate": "no action" + }, + "conversation_turns_parent_turn_id_conversation_turns_turn_id_fk": { + "name": "conversation_turns_parent_turn_id_conversation_turns_turn_id_fk", + "tableFrom": "conversation_turns", + "tableTo": "conversation_turns", + "columnsFrom": ["parent_turn_id"], + "columnsTo": ["turn_id"], + "onDelete": "cascade", + "onUpdate": "no action" + }, + "conversation_turns_prompt_id_prompts_prompt_id_fk": { + "name": "conversation_turns_prompt_id_prompts_prompt_id_fk", + "tableFrom": "conversation_turns", + "tableTo": "prompts", + "columnsFrom": ["prompt_id"], + "columnsTo": ["prompt_id"], + "onDelete": "no action", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": {} + }, + "loop_iterations": { + "name": "loop_iterations", + "columns": { + "loop_id": { + "name": "loop_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "index": { + "name": "index", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "status": { + "name": "status", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "worker_session_id": { + "name": "worker_session_id", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "verifier_session_id": { + "name": "verifier_session_id", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "checks_json": { + "name": "checks_json", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "verdict_json": { + "name": "verdict_json", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "error": { + "name": "error", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "started_at": { + "name": "started_at", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "ended_at": { + "name": "ended_at", + "type": "integer", + "primaryKey": false, + "notNull": false, + "autoincrement": false + } + }, + "indexes": {}, + "foreignKeys": { + "loop_iterations_loop_id_loops_loop_id_fk": { + "name": "loop_iterations_loop_id_loops_loop_id_fk", + "tableFrom": "loop_iterations", + "tableTo": "loops", + "columnsFrom": ["loop_id"], + "columnsTo": ["loop_id"], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": { + "loop_iterations_loop_id_index_pk": { + "columns": ["loop_id", "index"], + "name": "loop_iterations_loop_id_index_pk" + } + }, + "uniqueConstraints": {}, + "checkConstraints": {} + }, + "loops": { + "name": "loops", + "columns": { + "loop_id": { + "name": "loop_id", + "type": "text", + "primaryKey": true, + "notNull": true, + "autoincrement": false + }, + "spec_json": { + "name": "spec_json", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "status": { + "name": "status", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "iteration_count": { + "name": "iteration_count", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": 0 + }, + "error": { + "name": "error", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "summary": { + "name": "summary", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "started_at": { + "name": "started_at", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "updated_at": { + "name": "updated_at", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "ended_at": { + "name": "ended_at", + "type": "integer", + "primaryKey": false, + "notNull": false, + "autoincrement": false + } + }, + "indexes": {}, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": {} + }, + "prompt_attachment_refs": { + "name": "prompt_attachment_refs", + "columns": { + "prompt_id": { + "name": "prompt_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "attachment_id": { + "name": "attachment_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + } + }, + "indexes": {}, + "foreignKeys": { + "prompt_attachment_refs_prompt_id_prompts_prompt_id_fk": { + "name": "prompt_attachment_refs_prompt_id_prompts_prompt_id_fk", + "tableFrom": "prompt_attachment_refs", + "tableTo": "prompts", + "columnsFrom": ["prompt_id"], + "columnsTo": ["prompt_id"], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": { + "prompt_attachment_refs_prompt_id_attachment_id_pk": { + "columns": ["prompt_id", "attachment_id"], + "name": "prompt_attachment_refs_prompt_id_attachment_id_pk" + } + }, + "uniqueConstraints": {}, + "checkConstraints": {} + }, + "prompts": { + "name": "prompts", + "columns": { + "prompt_id": { + "name": "prompt_id", + "type": "text", + "primaryKey": true, + "notNull": true, + "autoincrement": false + }, + "blocks_json": { + "name": "blocks_json", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "context_attachment_ids_json": { + "name": "context_attachment_ids_json", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "created_at": { + "name": "created_at", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + } + }, + "indexes": {}, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": {} + }, + "provider_turn_bindings": { + "name": "provider_turn_bindings", + "columns": { + "turn_id": { + "name": "turn_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "run_id": { + "name": "run_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "history_id": { + "name": "history_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "checkpoint": { + "name": "checkpoint", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "captured_from": { + "name": "captured_from", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + } + }, + "indexes": {}, + "foreignKeys": { + "provider_turn_bindings_turn_id_conversation_turns_turn_id_fk": { + "name": "provider_turn_bindings_turn_id_conversation_turns_turn_id_fk", + "tableFrom": "provider_turn_bindings", + "tableTo": "conversation_turns", + "columnsFrom": ["turn_id"], + "columnsTo": ["turn_id"], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": { + "provider_turn_bindings_turn_id_history_id_pk": { + "columns": ["turn_id", "history_id"], + "name": "provider_turn_bindings_turn_id_history_id_pk" + } + }, + "uniqueConstraints": {}, + "checkConstraints": {} + }, + "schedule_runs": { + "name": "schedule_runs", + "columns": { + "run_id": { + "name": "run_id", + "type": "text", + "primaryKey": true, + "notNull": true, + "autoincrement": false + }, + "schedule_id": { + "name": "schedule_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "status": { + "name": "status", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "trigger": { + "name": "trigger", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "session_id": { + "name": "session_id", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "error": { + "name": "error", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "summary": { + "name": "summary", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "started_at": { + "name": "started_at", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "ended_at": { + "name": "ended_at", + "type": "integer", + "primaryKey": false, + "notNull": false, + "autoincrement": false + } + }, + "indexes": { + "schedule_runs_schedule_started_idx": { + "name": "schedule_runs_schedule_started_idx", + "columns": ["schedule_id", "started_at"], + "isUnique": false + } + }, + "foreignKeys": { + "schedule_runs_schedule_id_schedules_schedule_id_fk": { + "name": "schedule_runs_schedule_id_schedules_schedule_id_fk", + "tableFrom": "schedule_runs", + "tableTo": "schedules", + "columnsFrom": ["schedule_id"], + "columnsTo": ["schedule_id"], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": {} + }, + "schedules": { + "name": "schedules", + "columns": { + "schedule_id": { + "name": "schedule_id", + "type": "text", + "primaryKey": true, + "notNull": true, + "autoincrement": false + }, + "name": { + "name": "name", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "prompt": { + "name": "prompt", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "cadence_type": { + "name": "cadence_type", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "cron_expression": { + "name": "cron_expression", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "cron_timezone": { + "name": "cron_timezone", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "interval_ms": { + "name": "interval_ms", + "type": "integer", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "target_type": { + "name": "target_type", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "target_session_id": { + "name": "target_session_id", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "target_config_json": { + "name": "target_config_json", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "status": { + "name": "status", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "completed_reason": { + "name": "completed_reason", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "misfire_policy": { + "name": "misfire_policy", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "next_run_at": { + "name": "next_run_at", + "type": "integer", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "last_run_at": { + "name": "last_run_at", + "type": "integer", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "run_count": { + "name": "run_count", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": 0 + }, + "max_runs": { + "name": "max_runs", + "type": "integer", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "expires_at": { + "name": "expires_at", + "type": "integer", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "created_at": { + "name": "created_at", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "updated_at": { + "name": "updated_at", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + } + }, + "indexes": { + "schedules_next_run_at_idx": { + "name": "schedules_next_run_at_idx", + "columns": ["next_run_at"], + "isUnique": false + } + }, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": {} + }, + "session_resources": { + "name": "session_resources", + "columns": { + "resource_id": { + "name": "resource_id", + "type": "text", + "primaryKey": true, + "notNull": true, + "autoincrement": false + }, + "session_id": { + "name": "session_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "direction": { + "name": "direction", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "name": { + "name": "name", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "kind": { + "name": "kind", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "status": { + "name": "status", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "locator_type": { + "name": "locator_type", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "locator": { + "name": "locator", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "normalized_locator_key": { + "name": "normalized_locator_key", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "mime_type": { + "name": "mime_type", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "size_bytes": { + "name": "size_bytes", + "type": "integer", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "error": { + "name": "error", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "created_at": { + "name": "created_at", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "updated_at": { + "name": "updated_at", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + } + }, + "indexes": { + "session_resources_session_idx": { + "name": "session_resources_session_idx", + "columns": ["session_id"], + "isUnique": false + }, + "session_resources_locator_idx": { + "name": "session_resources_locator_idx", + "columns": ["session_id", "normalized_locator_key"], + "isUnique": true + } + }, + "foreignKeys": { + "session_resources_session_id_sessions_session_id_fk": { + "name": "session_resources_session_id_sessions_session_id_fk", + "tableFrom": "session_resources", + "tableTo": "sessions", + "columnsFrom": ["session_id"], + "columnsTo": ["session_id"], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": {} + }, + "session_runs": { + "name": "session_runs", + "columns": { + "id": { + "name": "id", + "type": "integer", + "primaryKey": true, + "notNull": true, + "autoincrement": true + }, + "session_id": { + "name": "session_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "seq": { + "name": "seq", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "run_id": { + "name": "run_id", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "base_turn_id": { + "name": "base_turn_id", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "history_id": { + "name": "history_id", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "account_id": { + "name": "account_id", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "model": { + "name": "model", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "effort": { + "name": "effort", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "approval_policy_id": { + "name": "approval_policy_id", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "started_at": { + "name": "started_at", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "ended_at": { + "name": "ended_at", + "type": "integer", + "primaryKey": false, + "notNull": false, + "autoincrement": false + } + }, + "indexes": { + "session_runs_session_id_idx": { + "name": "session_runs_session_id_idx", + "columns": ["session_id"], + "isUnique": false + }, + "session_runs_run_id_unique": { + "name": "session_runs_run_id_unique", + "columns": ["run_id"], + "isUnique": true + } + }, + "foreignKeys": { + "session_runs_session_id_sessions_session_id_fk": { + "name": "session_runs_session_id_sessions_session_id_fk", + "tableFrom": "session_runs", + "tableTo": "sessions", + "columnsFrom": ["session_id"], + "columnsTo": ["session_id"], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": {} + }, + "sessions": { + "name": "sessions", + "columns": { + "session_id": { + "name": "session_id", + "type": "text", + "primaryKey": true, + "notNull": true, + "autoincrement": false + }, + "kind": { + "name": "kind", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "cwd": { + "name": "cwd", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "title": { + "name": "title", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "origin_type": { + "name": "origin_type", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "origin_history_id": { + "name": "origin_history_id", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "origin_imported_at": { + "name": "origin_imported_at", + "type": "integer", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "origin_source_session_id": { + "name": "origin_source_session_id", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "origin_source_turn_id": { + "name": "origin_source_turn_id", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "origin_forked_at": { + "name": "origin_forked_at", + "type": "integer", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "created_via": { + "name": "created_via", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "automation_kind": { + "name": "automation_kind", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "automation_id": { + "name": "automation_id", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "active_leaf_turn_id": { + "name": "active_leaf_turn_id", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "graph_revision": { + "name": "graph_revision", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": 0 + }, + "event_epoch": { + "name": "event_epoch", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": 0 + }, + "created_at": { + "name": "created_at", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "updated_at": { + "name": "updated_at", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + } + }, + "indexes": { + "sessions_updated_at_idx": { + "name": "sessions_updated_at_idx", + "columns": ["updated_at"], + "isUnique": false + } + }, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": {} + }, + "workspaces": { + "name": "workspaces", + "columns": { + "workspace_id": { + "name": "workspace_id", + "type": "text", + "primaryKey": true, + "notNull": true, + "autoincrement": false + }, + "cwd": { + "name": "cwd", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "name": { + "name": "name", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "kind": { + "name": "kind", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": "'project'" + }, + "parent_workspace_id": { + "name": "parent_workspace_id", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "created_at": { + "name": "created_at", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "last_used_at": { + "name": "last_used_at", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + } + }, + "indexes": { + "workspaces_cwd_unique": { + "name": "workspaces_cwd_unique", + "columns": ["cwd"], + "isUnique": true + }, + "workspaces_last_used_at_idx": { + "name": "workspaces_last_used_at_idx", + "columns": ["last_used_at"], + "isUnique": false + } + }, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": {} + }, + "worktrees": { + "name": "worktrees", + "columns": { + "worktree_path": { + "name": "worktree_path", + "type": "text", + "primaryKey": true, + "notNull": true, + "autoincrement": false + }, + "repo_root": { + "name": "repo_root", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "branch": { + "name": "branch", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "session_id": { + "name": "session_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "created_at": { + "name": "created_at", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "state": { + "name": "state", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + } + }, + "indexes": { + "worktrees_repo_root_branch_unique": { + "name": "worktrees_repo_root_branch_unique", + "columns": ["repo_root", "branch"], + "isUnique": true + }, + "worktrees_session_id_unique": { + "name": "worktrees_session_id_unique", + "columns": ["session_id"], + "isUnique": true + } + }, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": {} + } + }, + "views": {}, + "enums": {}, + "_meta": { + "schemas": {}, + "tables": {}, + "columns": {} + }, + "internal": { + "indexes": {} + } +} diff --git a/apps/daemon/drizzle/meta/_journal.json b/apps/daemon/drizzle/meta/_journal.json index 708ab95fb..d70f3890e 100644 --- a/apps/daemon/drizzle/meta/_journal.json +++ b/apps/daemon/drizzle/meta/_journal.json @@ -92,6 +92,13 @@ "when": 1788243073079, "tag": "0012_overjoyed_risque", "breakpoints": true + }, + { + "idx": 13, + "version": "6", + "when": 1788261880957, + "tag": "0013_famous_randall", + "breakpoints": true } ] } diff --git a/apps/daemon/src/__tests__/session-store.test.ts b/apps/daemon/src/__tests__/session-store.test.ts index 13df38cb8..299f69910 100644 --- a/apps/daemon/src/__tests__/session-store.test.ts +++ b/apps/daemon/src/__tests__/session-store.test.ts @@ -53,6 +53,7 @@ describe('SQLite session store', () => { updatedAt: 2, activeLeafTurnId: 'turn-leaf', graphRevision: 7, + eventEpoch: 3, runs: [ { runId: 'run-1', startedAt: 1, endedAt: 2, historyId: 'native-1', accountId: 'acc_first' }, { diff --git a/apps/daemon/src/db/schema.ts b/apps/daemon/src/db/schema.ts index 9211f69cb..ed90820f1 100644 --- a/apps/daemon/src/db/schema.ts +++ b/apps/daemon/src/db/schema.ts @@ -36,6 +36,7 @@ export const sessions = sqliteTable( * tables would otherwise cycle. */ activeLeafTurnId: text('active_leaf_turn_id'), graphRevision: integer('graph_revision').notNull().default(0), + eventEpoch: integer('event_epoch').notNull().default(0), createdAt: integer('created_at').notNull(), updatedAt: integer('updated_at').notNull(), }, diff --git a/apps/daemon/src/session-store.ts b/apps/daemon/src/session-store.ts index 2f8f9c89f..986ae07ad 100644 --- a/apps/daemon/src/session-store.ts +++ b/apps/daemon/src/session-store.ts @@ -93,6 +93,7 @@ function toSessionRow(record: SessionRecord): typeof sessions.$inferInsert { automationId: record.automation?.id ?? null, activeLeafTurnId: record.activeLeafTurnId ?? null, graphRevision: record.graphRevision, + eventEpoch: record.eventEpoch, createdAt: record.createdAt, updatedAt: record.updatedAt, }; @@ -113,6 +114,7 @@ function toRecord(row: SessionRow, runRows: RunRow[]): SessionRecord { : undefined, activeLeafTurnId: row.activeLeafTurnId ?? undefined, graphRevision: row.graphRevision, + eventEpoch: row.eventEpoch, createdAt: row.createdAt, updatedAt: row.updatedAt, runs: runRows.map((run) => ({ diff --git a/packages/client/workbench/src/mock/dev-mock-host.ts b/packages/client/workbench/src/mock/dev-mock-host.ts index 58f5977f6..17e6c3176 100644 --- a/packages/client/workbench/src/mock/dev-mock-host.ts +++ b/packages/client/workbench/src/mock/dev-mock-host.ts @@ -967,6 +967,7 @@ export class DevMockHost { updatedAt: now, runs: [], graphRevision: 0, + eventEpoch: 0, }, }); } diff --git a/packages/foundation/schema/src/model/conversation.ts b/packages/foundation/schema/src/model/conversation.ts index 55f5a8bbd..e3f2ed3f1 100644 --- a/packages/foundation/schema/src/model/conversation.ts +++ b/packages/foundation/schema/src/model/conversation.ts @@ -130,3 +130,12 @@ export const ConversationWatermarkSchema = z.object({ seq: z.number().int().nonnegative(), }); export type ConversationWatermark = z.infer; + +/** Lexicographic `(epoch, seq)` order: an event at or below a watermark is dropped by the merge + * rule, so any straggler from an older epoch compares below every position of a newer one. */ +export function compareConversationWatermarks( + a: ConversationWatermark, + b: ConversationWatermark, +): number { + return a.epoch === b.epoch ? a.seq - b.seq : a.epoch - b.epoch; +} diff --git a/packages/foundation/schema/src/model/session/record.ts b/packages/foundation/schema/src/model/session/record.ts index f40b16f5c..2eb738d95 100644 --- a/packages/foundation/schema/src/model/session/record.ts +++ b/packages/foundation/schema/src/model/session/record.ts @@ -95,6 +95,9 @@ export const SessionRecordSchema = z.object({ activeLeafTurnId: TurnIdSchema.optional(), /** Optimistic-concurrency counter for graph mutations. */ graphRevision: z.number().int().nonnegative().default(0), + /** Event-plane epoch: bumped on every run launch and daemon boot; `agent.event` seq is + * daemon-minted and monotone within one epoch (merge rule is lexicographic). */ + eventEpoch: z.number().int().nonnegative().default(0), }); export type SessionRecord = z.infer; diff --git a/packages/host/engine/src/__tests__/conversation-live-journal.test.ts b/packages/host/engine/src/__tests__/conversation-live-journal.test.ts new file mode 100644 index 000000000..775096363 --- /dev/null +++ b/packages/host/engine/src/__tests__/conversation-live-journal.test.ts @@ -0,0 +1,151 @@ +import type { AgentEvent, MessageId, RunId, SessionId } from '@linkcode/schema'; +import { compareConversationWatermarks } from '@linkcode/schema'; +import { describe, expect, it } from 'vitest'; +import type { JournaledEvent } from '../conversation/live-journal'; +import { ConversationLiveJournal, ConversationLiveJournals } from '../conversation/live-journal'; + +const runId = 'run-journal' as RunId; + +function chunk(text: string): AgentEvent { + return { + type: 'agent-message-chunk', + messageId: 'msg-1' as MessageId, + content: { type: 'text', text }, + }; +} + +function stamped(epoch: number, seq: number, text = `event-${epoch}-${seq}`): JournaledEvent { + return { epoch, seq, runId, ts: 1, event: chunk(text) }; +} + +describe('conversation watermark merge rule', () => { + it('discards an old-epoch straggler regardless of its seq', () => { + const watermark = { epoch: 2, seq: 0 }; + // The client merge drops anything ≤ watermark; a delayed old-adapter event can never fold + // over newer state, however far its own seq ran. + expect(compareConversationWatermarks({ epoch: 1, seq: 999_999 }, watermark)).toBeLessThan(0); + expect(compareConversationWatermarks({ epoch: 2, seq: 0 }, watermark)).toBe(0); + expect(compareConversationWatermarks({ epoch: 2, seq: 1 }, watermark)).toBeGreaterThan(0); + expect(compareConversationWatermarks({ epoch: 3, seq: 0 }, watermark)).toBeGreaterThan(0); + }); +}); + +describe('ConversationLiveJournal caps', () => { + it('holds the byte cap under a chunk storm, accounting exactly for what it retains', () => { + const maxBytes = 64 * 1024; + const journal = new ConversationLiveJournal(maxBytes, 100_000); + const encoder = new TextEncoder(); + for (let seq = 1; seq <= 5000; seq++) { + journal.append(stamped(1, seq, 'x'.repeat(1024))); + } + expect(journal.truncated).toBe(true); + expect(journal.bytes).toBeLessThanOrEqual(maxBytes); + // Boundedness of the retained data itself, not just a counter: the accounted bytes equal an + // independent re-encoding of every retained event. + const retained = journal + .snapshot() + .reduce((sum, { event }) => sum + encoder.encode(JSON.stringify(event)).byteLength, 0); + expect(journal.bytes).toBe(retained); + expect(retained).toBeLessThanOrEqual(maxBytes); + }); + + it('holds the event cap under a chunk storm', () => { + const journal = new ConversationLiveJournal(Number.MAX_SAFE_INTEGER, 100); + for (let seq = 1; seq <= 10_000; seq++) journal.append(stamped(1, seq)); + expect(journal.size).toBe(100); + expect(journal.truncated).toBe(true); + expect(journal.snapshot()[0]?.seq).toBe(9901); + expect(journal.watermark).toEqual({ epoch: 1, seq: 10_000 }); + }); + + it('never marks truncation while under both caps', () => { + const journal = new ConversationLiveJournal(); + for (let seq = 1; seq <= 50; seq++) journal.append(stamped(1, seq)); + expect(journal.truncated).toBe(false); + expect(journal.size).toBe(50); + }); +}); + +describe('ConversationLiveJournal tailAfter', () => { + it('returns a clean tail for a watermark inside the retained range', () => { + const journal = new ConversationLiveJournal(); + for (let seq = 1; seq <= 5; seq++) journal.append(stamped(3, seq)); + + const { events, gap } = journal.tailAfter({ epoch: 3, seq: 2 }); + expect(gap).toBe(false); + expect(events.map((event) => event.seq)).toEqual([3, 4, 5]); + }); + + it('reports a gap when eviction removed events past the watermark', () => { + const journal = new ConversationLiveJournal(Number.MAX_SAFE_INTEGER, 3); + for (let seq = 1; seq <= 6; seq++) journal.append(stamped(3, seq)); + + // seq 1–3 were evicted; a reader at seq 2 lost seq 3 forever — it must re-read, never splice. + const cut = journal.tailAfter({ epoch: 3, seq: 2 }); + expect(cut.gap).toBe(true); + expect(cut.events.map((event) => event.seq)).toEqual([4, 5, 6]); + + // A reader at or past everything evicted is complete. + expect(journal.tailAfter({ epoch: 3, seq: 3 })).toEqual({ + events: journal.snapshot(), + gap: false, + }); + }); + + it('reports a gap for a watermark from an older epoch: an epoch jump forces a re-read', () => { + const journal = new ConversationLiveJournal(); + for (let seq = 1; seq <= 3; seq++) journal.append(stamped(7, seq)); + + // The old epoch's tail died with its journal; completeness is unprovable below this epoch. + const jumped = journal.tailAfter({ epoch: 6, seq: 40 }); + expect(jumped.gap).toBe(true); + expect(jumped.events.map((event) => event.seq)).toEqual([1, 2, 3]); + + // The epoch's own base is complete: everything since the journal was born is retained. + expect(journal.tailAfter({ epoch: 7, seq: 0 }).gap).toBe(false); + }); + + it('flags a watermark above everything appended: the epoch-reuse signature', () => { + const journal = new ConversationLiveJournal(); + for (let seq = 1; seq <= 3; seq++) journal.append(stamped(1, seq)); + + // An honest current client compares at most equal; above means a reused epoch minted these + // positions before for different events, or the watermark belongs to another stream. + expect(journal.tailAfter({ epoch: 1, seq: 4 })).toEqual({ events: [], gap: true }); + expect(journal.tailAfter({ epoch: 2, seq: 0 })).toEqual({ events: [], gap: true }); + // The equal edge stays complete. + expect(journal.tailAfter({ epoch: 1, seq: 3 })).toEqual({ events: [], gap: false }); + }); + + it('treats every watermark as a gap while nothing was ever appended', () => { + const journal = new ConversationLiveJournal(); + expect(journal.tailAfter({ epoch: 0, seq: 0 })).toEqual({ events: [], gap: true }); + expect(journal.tailAfter({ epoch: 5, seq: 40 })).toEqual({ events: [], gap: true }); + }); + + it('keeps the watermark at the lexicographic max when an old-epoch straggler appends late', () => { + const journal = new ConversationLiveJournal(); + journal.append(stamped(2, 1)); + journal.append(stamped(2, 2)); + journal.append(stamped(1, 99)); + + expect(journal.watermark).toEqual({ epoch: 2, seq: 2 }); + // The straggler sits at or below every current-epoch watermark — provably discarded on read. + expect(journal.tailAfter({ epoch: 2, seq: 2 }).events).toEqual([]); + }); +}); + +describe('ConversationLiveJournals registry', () => { + it('creates one journal per session and drops it with the live session', () => { + const journals = new ConversationLiveJournals(); + const sessionId = 'sess-journal' as SessionId; + const journal = journals.open(sessionId); + journal.append(stamped(1, 1)); + expect(journals.open(sessionId)).toBe(journal); + expect(journals.get(sessionId)?.size).toBe(1); + + journals.drop(sessionId); + expect(journals.get(sessionId)).toBeUndefined(); + expect(journals.open(sessionId).size).toBe(0); + }); +}); diff --git a/packages/host/engine/src/__tests__/engine-session-input.test.ts b/packages/host/engine/src/__tests__/engine-session-input.test.ts index 7ec07a68f..59925594d 100644 --- a/packages/host/engine/src/__tests__/engine-session-input.test.ts +++ b/packages/host/engine/src/__tests__/engine-session-input.test.ts @@ -266,16 +266,18 @@ describe('engine session input', () => { message: 'Agent input was rejected', reportedInConversation: true, }); - expect(h.sent).toContainEqual({ - kind: 'agent.event', - sessionId, - event: { - type: 'error', - message: 'Agent input was rejected', - code: 'input_rejected', - recoverable: true, - }, - }); + expect(h.sent).toContainEqual( + expect.objectContaining({ + kind: 'agent.event', + sessionId, + event: { + type: 'error', + message: 'Agent input was rejected', + code: 'input_rejected', + recoverable: true, + }, + }), + ); }); it('does not leave the turn busy when source lookup fails', async () => { @@ -356,16 +358,18 @@ describe('engine session input', () => { message: `Session is busy: ${sessionId}`, reportedInConversation: true, }); - expect(h.sent).toContainEqual({ - kind: 'agent.event', - sessionId, - event: { - type: 'error', - message: `Session is busy: ${sessionId}`, - code: 'input_rejected', - recoverable: true, - }, - }); + expect(h.sent).toContainEqual( + expect.objectContaining({ + kind: 'agent.event', + sessionId, + event: { + type: 'error', + message: `Session is busy: ${sessionId}`, + code: 'input_rejected', + recoverable: true, + }, + }), + ); adapter.releaseSend(); }); }); diff --git a/packages/host/engine/src/__tests__/engine-session-lifecycle.test.ts b/packages/host/engine/src/__tests__/engine-session-lifecycle.test.ts index 6d018a4dd..e71ce08e8 100644 --- a/packages/host/engine/src/__tests__/engine-session-lifecycle.test.ts +++ b/packages/host/engine/src/__tests__/engine-session-lifecycle.test.ts @@ -372,6 +372,7 @@ describe('engine session lifecycle', () => { new FakeAdapter(), 'sess-interrupt' as SessionId, 'run-interrupt' as RunId, + 0, scope, closed, ); diff --git a/packages/host/engine/src/__tests__/engine-session-records.test.ts b/packages/host/engine/src/__tests__/engine-session-records.test.ts index 8693d9bd0..8998c30d1 100644 --- a/packages/host/engine/src/__tests__/engine-session-records.test.ts +++ b/packages/host/engine/src/__tests__/engine-session-records.test.ts @@ -748,6 +748,7 @@ describe('engine session records', () => { { runId: 'run-1' as RunId, historyId: asHistoryId('native-1'), startedAt: 1, endedAt: 2 }, ], graphRevision: 0, + eventEpoch: 0, }); const h = harness(store); await h.engine.start(); @@ -871,6 +872,7 @@ describe('engine session records', () => { updatedAt: 2, runs: [], graphRevision: 0, + eventEpoch: 0, }; const created: SessionRecord = { sessionId: 's-created' as SessionId, @@ -881,6 +883,7 @@ describe('engine session records', () => { updatedAt: 2, runs: [], graphRevision: 0, + eventEpoch: 0, }; await sessionStore.save(imported); await sessionStore.save(created); diff --git a/packages/host/engine/src/__tests__/engine-turn-tracking.test.ts b/packages/host/engine/src/__tests__/engine-turn-tracking.test.ts index ddff0e181..d447b71bb 100644 --- a/packages/host/engine/src/__tests__/engine-turn-tracking.test.ts +++ b/packages/host/engine/src/__tests__/engine-turn-tracking.test.ts @@ -441,6 +441,7 @@ describe('commitRunning idempotence', () => { updatedAt: 1, runs: [], graphRevision: 0, + eventEpoch: 0, }); const store = new InMemoryConversationStore(); const turns = new ConversationTurnService(store, registry, transport, (effect) => { diff --git a/packages/host/engine/src/__tests__/session-event-sequencing.test.ts b/packages/host/engine/src/__tests__/session-event-sequencing.test.ts new file mode 100644 index 000000000..0dc9b4b83 --- /dev/null +++ b/packages/host/engine/src/__tests__/session-event-sequencing.test.ts @@ -0,0 +1,360 @@ +import type { + AgentEvent, + ConversationWatermark, + MessageId, + RunId, + SessionId, + SessionRecord, + StartOptions, + ValidatedWireMessage, + WirePayload, +} from '@linkcode/schema'; +import { compareConversationWatermarks, OperationIdSchema } from '@linkcode/schema'; +import type { Transport } from '@linkcode/transport'; +import { Deferred, Effect, Scope } from 'effect'; +import { noop } from 'foxts/noop'; +import { describe, expect, it } from 'vitest'; +import { AgentRuntimeService } from '../agent/runtime-service'; +import { InMemoryConversationStore } from '../conversation/conversation-store'; +import { ConversationLiveJournals } from '../conversation/live-journal'; +import { ConversationTurnService } from '../conversation/turn-service'; +import type { OperationError } from '../failure'; +import { FileHostService } from '../preview/file-host-service'; +import { PreviewRouteRegistry } from '../preview/route-registry'; +import { InMemoryResourceStore } from '../resource/resource-store'; +import { ResourceService } from '../resource/service'; +import { LiveSession } from '../session/live-session'; +import { SessionEventProcessor } from '../session/session-event-processor'; +import { SessionRecordRegistry } from '../session/session-record-registry'; +import { InMemorySessionStore } from '../session/session-store'; +import { + FakeAdapter, + createSessionHarness as harness, + startedSessionId as startedId, +} from './fixtures/session-harness'; + +function chunk(text: string): AgentEvent { + return { + type: 'agent-message-chunk', + messageId: 'msg-seq' as MessageId, + content: { type: 'text', text }, + }; +} + +interface StampedFrame { + readonly epoch: number; + readonly seq: number; + readonly event: AgentEvent; +} + +function stampedFrames(sent: WirePayload[], sessionId: SessionId): StampedFrame[] { + return sent.flatMap((payload) => { + if (payload.kind !== 'agent.event' || payload.sessionId !== sessionId) return []; + expect(payload.runId).toBeDefined(); + expect(payload.epoch).toBeDefined(); + expect(payload.seq).toBeDefined(); + if (payload.epoch === undefined || payload.seq === undefined) return []; + return [{ epoch: payload.epoch, seq: payload.seq, event: payload.event }]; + }); +} + +describe('agent.event sequencing over the wire', () => { + it('stamps every frame with the run epoch and contiguous seqs', async () => { + const h = harness(); + await h.engine.start(); + await h.inject({ + kind: 'session.start', + clientReqId: 'r1', + opts: { kind: 'claude-code', cwd: '/repo' }, + }); + const sessionId = startedId(h.sent, 'r1'); + const adapter = h.adapters[0]; + + adapter.emit({ type: 'status', status: 'running' }); + adapter.emit(chunk('hello')); + adapter.emit({ type: 'status', status: 'idle' }); + + const frames = stampedFrames(h.sent, sessionId); + expect(frames).toHaveLength(3); + // A fresh record launches under epoch 0; seq is minted contiguously from 1. + expect(frames.map(({ epoch, seq }) => [epoch, seq])).toEqual([ + [0, 1], + [0, 2], + [0, 3], + ]); + }); + + it('bumps the epoch on relaunch so every old-run position compares below the new run', async () => { + const h = harness(); + await h.engine.start(); + await h.inject({ + kind: 'session.start', + clientReqId: 'r1', + opts: { kind: 'claude-code', cwd: '/repo' }, + }); + const sessionId = startedId(h.sent, 'r1'); + h.adapters[0].emit(chunk('first epoch')); + const before = stampedFrames(h.sent, sessionId); + const oldStamp = before.at(-1); + expect(oldStamp).toBeDefined(); + + await h.inject({ kind: 'session.stop', clientReqId: 'r-stop', sessionId }); + const mark = h.sent.length; + await h.inject({ kind: 'session.resume', clientReqId: 'r-resume', sessionId }); + h.adapters[1].emit(chunk('second epoch')); + + const after = stampedFrames(h.sent.slice(mark), sessionId); + const newStamp = after.at(-1); + expect(newStamp).toBeDefined(); + if (!oldStamp || !newStamp) return; + expect(newStamp.epoch).toBeGreaterThan(oldStamp.epoch); + expect(newStamp.seq).toBe(after.length); + // The merge rule discards anything at or below a watermark: the whole old epoch sits below + // position 0 of the new one, so a delayed old-run event can never fold over newer state. + expect(compareConversationWatermarks(oldStamp, { epoch: newStamp.epoch, seq: 0 })).toBeLessThan( + 0, + ); + }); + + it('re-stamps the interactive-request replay on attach above any prior watermark', async () => { + const h = harness(); + await h.engine.start(); + await h.inject({ + kind: 'session.start', + clientReqId: 'r1', + opts: { kind: 'claude-code', cwd: '/repo' }, + }); + const sessionId = startedId(h.sent, 'r1'); + h.adapters[0].emit({ type: 'status', status: 'running' }); + h.adapters[0].emit({ + type: 'permission-request', + requestId: 'perm-replay', + title: 'Run', + subject: { type: 'tool-call', toolCallId: 't1' }, + options: [{ optionId: 'ok', name: 'Allow', kind: 'allow_once' }], + }); + const original = stampedFrames(h.sent, sessionId).at(-1); + expect(original?.event.type).toBe('permission-request'); + + await h.inject({ kind: 'session.attach', sessionId }); + + const frames = stampedFrames(h.sent, sessionId); + const replayed = frames.filter(({ event }) => event.type === 'permission-request').at(-1); + expect(replayed).toBeDefined(); + if (!original || !replayed) return; + // The authoritative replay lands above every prior position, so a client merging by + // watermark can never drop an open interactive request as already-seen. + expect(compareConversationWatermarks(replayed, original)).toBeGreaterThan(0); + }); +}); + +/** Save log + on-demand failure: the launch path must prove the bumped epoch durable pre-mint. */ +class GatedSaveStore extends InMemorySessionStore { + failSaves = false; + + constructor(private readonly log: string[]) { + super(); + } + + override save(record: SessionRecord): Promise { + if (this.failSaves) return Promise.reject(new Error('session store save failed')); + this.log.push(`save:${record.eventEpoch}`); + return super.save(record); + } +} + +class StartLoggingAdapter extends FakeAdapter { + constructor(private readonly log: string[]) { + super(); + } + + override start(opts: StartOptions): Promise { + this.log.push('adapter-start'); + return super.start(opts); + } +} + +describe('durable epoch before minting', () => { + function launchHarness() { + const log: string[] = []; + const store = new GatedSaveStore(log); + return { log, store, h: harness(store, () => new StartLoggingAdapter(log)) }; + } + + it('persists the bumped epoch before the relaunched adapter starts', async () => { + const { log, h } = launchHarness(); + await h.engine.start(); + await h.inject({ + kind: 'session.start', + clientReqId: 'r1', + opts: { kind: 'claude-code', cwd: '/repo' }, + }); + const sessionId = startedId(h.sent, 'r1'); + await h.inject({ kind: 'session.stop', clientReqId: 'r-stop', sessionId }); + + await h.inject({ kind: 'session.resume', clientReqId: 'r-resume', sessionId }); + + const bumpedSave = log.indexOf('save:1'); + const relaunchStart = log.lastIndexOf('adapter-start'); + expect(bumpedSave).toBeGreaterThanOrEqual(0); + expect(relaunchStart).toBeGreaterThan(bumpedSave); + }); + + it('fails the launch loud when the epoch cannot be made durable, minting nothing', async () => { + const { store, h } = launchHarness(); + await h.engine.start(); + await h.inject({ + kind: 'session.start', + clientReqId: 'r1', + opts: { kind: 'claude-code', cwd: '/repo' }, + }); + const sessionId = startedId(h.sent, 'r1'); + await h.inject({ kind: 'session.stop', clientReqId: 'r-stop', sessionId }); + + store.failSaves = true; + const mark = h.sent.length; + await h.inject({ + kind: 'turn.submit', + clientReqId: 'r-sub', + sessionId, + operationId: OperationIdSchema.parse('op-epoch-flush'), + input: { type: 'prompt', blocks: [{ type: 'text', text: 'hello' }] }, + }); + + expect(h.sent.slice(mark)).toContainEqual( + expect.objectContaining({ kind: 'request.failed', replyTo: 'r-sub' }), + ); + // No LiveSession was constructed and nothing was minted under the undurable epoch. + expect(h.adapters).toHaveLength(1); + expect(stampedFrames(h.sent.slice(mark), sessionId)).toEqual([]); + }); +}); + +describe('stale-run events at saga cutover', () => { + async function makeProcessor() { + const sent: WirePayload[] = []; + const transport: Transport = { + connect: () => Promise.resolve(), + send(msg: ValidatedWireMessage) { + sent.push(msg.payload); + }, + onMessage: () => noop, + onClose: () => noop, + close: noop, + }; + const registry = new SessionRecordRegistry(new InMemorySessionStore(), noop); + await Effect.runPromise( + registry.start((effect) => { + void Effect.runPromise(effect); + }), + ); + const runtimes = await Effect.runPromise(AgentRuntimeService.make({ onChanged: noop }, noop)); + const journals = new ConversationLiveJournals(); + const turns = new ConversationTurnService( + new InMemoryConversationStore(), + registry, + transport, + (effect) => { + void Effect.runPromise(effect); + }, + ); + const processor = new SessionEventProcessor( + transport, + registry, + runtimes, + noop, + new ResourceService( + transport, + new InMemoryResourceStore(), + registry, + undefined, + new FileHostService(new PreviewRouteRegistry()), + ), + turns, + journals, + ); + return { sent, registry, journals, processor }; + } + + function record(sessionId: SessionId): SessionRecord { + return { + sessionId, + kind: 'claude-code', + cwd: '/repo', + origin: { type: 'created' }, + createdAt: 1, + updatedAt: 1, + runs: [], + graphRevision: 0, + eventEpoch: 0, + }; + } + + async function liveSession( + sessionId: SessionId, + runId: RunId, + epoch: number, + ): Promise { + return Effect.runPromise( + Effect.gen(function* () { + const scope = yield* Scope.make(); + const closed = yield* Deferred.make(); + return new LiveSession(new FakeAdapter(), sessionId, runId, epoch, scope, closed); + }), + ); + } + + it('drops a stale run session-scoped status at the source: no stopped flicker, nothing journaled', async () => { + const { sent, registry, journals, processor } = await makeProcessor(); + const sessionId = 'sess-cutover' as SessionId; + registry.register(record(sessionId)); + + const staleRun = registry.beginRun(sessionId); + const staleEpoch = registry.get(sessionId)?.eventEpoch ?? -1; + const stale = await liveSession(sessionId, staleRun, staleEpoch); + // The saga cut over: a new run is current while the old adapter still drains. + const currentRun = registry.beginRun(sessionId); + const currentEpoch = registry.get(sessionId)?.eventEpoch ?? -1; + + processor.handle(sessionId, stale, { type: 'status', status: 'stopped' }); + + expect(sent.filter((payload) => payload.kind === 'agent.event')).toEqual([]); + expect(journals.get(sessionId)).toBeUndefined(); + + // The replacement's session-scoped events pass and stamp under the new epoch. + const current = await liveSession(sessionId, currentRun, currentEpoch); + processor.handle(sessionId, current, { type: 'status', status: 'running' }); + const frames = stampedFrames(sent, sessionId); + expect(frames).toEqual([ + { epoch: currentEpoch, seq: 1, event: { type: 'status', status: 'running' } }, + ]); + }); + + it('passes a stale run turn-scoped straggler under its own epoch, provably below the new watermark', async () => { + const { sent, registry, journals, processor } = await makeProcessor(); + const sessionId = 'sess-straggler' as SessionId; + registry.register(record(sessionId)); + + const staleRun = registry.beginRun(sessionId); + const staleEpoch = registry.get(sessionId)?.eventEpoch ?? -1; + const stale = await liveSession(sessionId, staleRun, staleEpoch); + registry.beginRun(sessionId); + const currentEpoch = registry.get(sessionId)?.eventEpoch ?? -1; + + processor.handle(sessionId, stale, chunk('late but attributed')); + + const frames = stampedFrames(sent, sessionId); + expect(frames).toHaveLength(1); + const stamp: ConversationWatermark = frames[0]; + expect(stamp.epoch).toBe(staleEpoch); + // Any client watermark in the new epoch discards it — even at the epoch's very first position. + expect(compareConversationWatermarks(stamp, { epoch: currentEpoch, seq: 0 })).toBeLessThan(0); + // And the journal recorded exactly what went to the wire. + expect( + journals + .get(sessionId) + ?.snapshot() + .map(({ epoch, seq }) => ({ epoch, seq })), + ).toEqual([{ epoch: stamp.epoch, seq: stamp.seq }]); + }); +}); diff --git a/packages/host/engine/src/__tests__/session-record-registry.test.ts b/packages/host/engine/src/__tests__/session-record-registry.test.ts index 837a74ef5..a43b9fa6f 100644 --- a/packages/host/engine/src/__tests__/session-record-registry.test.ts +++ b/packages/host/engine/src/__tests__/session-record-registry.test.ts @@ -2,6 +2,7 @@ import { asHistoryId } from '@linkcode/agent-adapter'; import type { RunId, SessionId, SessionRecord } from '@linkcode/schema'; import { Effect } from 'effect'; import { noop } from 'foxts/noop'; +import { wait } from 'foxts/wait'; import { describe, expect, it } from 'vitest'; import { SessionRecordRegistry } from '../session/session-record-registry'; import { InMemorySessionStore } from '../session/session-store'; @@ -18,6 +19,7 @@ function makeRecord(): SessionRecord { updatedAt: 1, runs: [], graphRevision: 0, + eventEpoch: 0, }; } @@ -75,3 +77,34 @@ describe('session record registry run addressing', () => { expect(registry.get(sessionId)?.runs.at(-1)?.runId).toBe(minted); }); }); + +describe('session record registry event epoch', () => { + it('bumps the epoch on every run launch', async () => { + const registry = await startedRegistry(); + expect(registry.get(sessionId)?.eventEpoch).toBe(0); + + registry.beginRun(sessionId); + expect(registry.get(sessionId)?.eventEpoch).toBe(1); + registry.beginRun(sessionId); + expect(registry.get(sessionId)?.eventEpoch).toBe(2); + }); + + it('bumps every loaded record at boot and persists the bump on the next launch', async () => { + const store = new InMemorySessionStore(); + await store.save({ ...makeRecord(), eventEpoch: 5 }); + const registry = new SessionRecordRegistry(store, noop); + await Effect.runPromise( + registry.start((effect) => { + void Effect.runPromise(effect); + }), + ); + + // In memory immediately: a relaunch after a reboot can never reuse a pre-reboot epoch. + expect(registry.get(sessionId)?.eventEpoch).toBe(6); + + registry.beginRun(sessionId); + await wait(0); + const persisted = await store.load(); + expect(persisted[0]?.eventEpoch).toBe(7); + }); +}); diff --git a/packages/host/engine/src/conversation/live-journal.ts b/packages/host/engine/src/conversation/live-journal.ts new file mode 100644 index 000000000..e4aa96fbc --- /dev/null +++ b/packages/host/engine/src/conversation/live-journal.ts @@ -0,0 +1,148 @@ +import type { AgentEvent, ConversationWatermark, RunId, SessionId, TurnId } from '@linkcode/schema'; +import { compareConversationWatermarks } from '@linkcode/schema'; + +/** One stamped live event: what its `agent.event` envelope carried, positioned for the + * lexicographic `(epoch, seq)` watermark merge. */ +export interface JournaledEvent { + readonly epoch: number; + readonly seq: number; + readonly runId: RunId; + readonly turnId?: TurnId; + readonly ts: number; + readonly event: AgentEvent; +} + +interface JournalEntry { + readonly event: JournaledEvent; + readonly bytes: number; +} + +const DEFAULT_JOURNAL_BYTE_CAP = 10 * 1024 * 1024; +const DEFAULT_JOURNAL_EVENT_CAP = 10_000; + +function stampOf(event: JournaledEvent): ConversationWatermark { + return { epoch: event.epoch, seq: event.seq }; +} + +/** + * Byte/event-bounded live tail of one session's stamped `agent.event` stream (the + * `TerminalReplayJournal` pattern). Bounded is non-negotiable: an unthrottled chunk storm must + * evict from the front, never grow — the 2026-08 daemon OOM came from an unbounded event corpus. + */ +export class ConversationLiveJournal { + private readonly entries: JournalEntry[] = []; + private byteCount = 0; + /** First position ever appended: completeness is provable only for watermarks at or above its + * predecessor — anything older (an earlier epoch's tail) never reached this journal. */ + private first: ConversationWatermark | undefined; + /** Highest position evicted by the caps; a watermark below it lost events it never saw. */ + private evictedThrough: ConversationWatermark | undefined; + private last: ConversationWatermark | undefined; + + constructor( + private readonly maxBytes = DEFAULT_JOURNAL_BYTE_CAP, + private readonly maxEvents = DEFAULT_JOURNAL_EVENT_CAP, + ) {} + + get bytes(): number { + return this.byteCount; + } + + get size(): number { + return this.entries.length; + } + + get truncated(): boolean { + return this.evictedThrough !== undefined; + } + + /** Highest position appended — the merge watermark the final page of a read carries. */ + get watermark(): ConversationWatermark | undefined { + return this.last; + } + + snapshot(): JournaledEvent[] { + return this.entries.map(({ event }) => event); + } + + append(event: JournaledEvent): void { + const stamp = stampOf(event); + this.first ??= stamp; + if (this.last === undefined || compareConversationWatermarks(stamp, this.last) > 0) { + this.last = stamp; + } + const bytes = Buffer.byteLength(JSON.stringify(event.event)); + this.entries.push({ event, bytes }); + this.byteCount += bytes; + while ( + (this.byteCount > this.maxBytes || this.entries.length > this.maxEvents) && + this.entries.length > 0 + ) { + const removed = this.entries.shift(); + if (!removed) break; + this.byteCount -= removed.bytes; + const evicted = stampOf(removed.event); + if ( + this.evictedThrough === undefined || + compareConversationWatermarks(evicted, this.evictedThrough) > 0 + ) { + this.evictedThrough = evicted; + } + } + } + + /** + * Retained events above `watermark`, and whether that set is provably complete. `gap` means + * events past the watermark were evicted or never reached this journal (an older epoch's tail): + * the reader must clear the affected in-flight state and re-read — never splice a headless tail. + * A watermark ABOVE everything appended is also a gap: an honest current client compares at + * most equal, so above means epoch reuse or a foreign watermark, never provable completeness. + * Events return in append order, not stamp order (a stale straggler can sit after newer-epoch + * entries); consumers merge by stamp. + */ + tailAfter(watermark: ConversationWatermark): { events: JournaledEvent[]; gap: boolean } { + const gap = + this.last === undefined || + compareConversationWatermarks(watermark, this.last) > 0 || + (this.evictedThrough !== undefined && + compareConversationWatermarks(this.evictedThrough, watermark) > 0) || + (this.first !== undefined && + compareConversationWatermarks(watermark, { + epoch: this.first.epoch, + seq: this.first.seq - 1, + }) < 0); + const events: JournaledEvent[] = []; + for (let i = 0, len = this.entries.length; i < len; i++) { + const { event } = this.entries[i]; + if (compareConversationWatermarks(stampOf(event), watermark) > 0) events.push(event); + } + return { events, gap }; + } +} + +/** Per-session live journals, created on the first stamped broadcast and dropped with the live + * session — total memory bounds to concurrent live adapters × the per-journal caps. */ +export class ConversationLiveJournals { + private readonly journals = new Map(); + + constructor( + private readonly maxBytes?: number, + private readonly maxEvents?: number, + ) {} + + open(sessionId: SessionId): ConversationLiveJournal { + const existing = this.journals.get(sessionId); + if (existing) return existing; + const journal = new ConversationLiveJournal(this.maxBytes, this.maxEvents); + this.journals.set(sessionId, journal); + return journal; + } + + get(sessionId: SessionId): ConversationLiveJournal | undefined { + return this.journals.get(sessionId); + } + + drop(sessionId: SessionId): void { + this.journals.delete(sessionId); + } +} diff --git a/packages/host/engine/src/conversation/turn-service.ts b/packages/host/engine/src/conversation/turn-service.ts index 3afae3801..074cc68ab 100644 --- a/packages/host/engine/src/conversation/turn-service.ts +++ b/packages/host/engine/src/conversation/turn-service.ts @@ -317,6 +317,11 @@ export class ConversationTurnService { }); } + /** The running turn `runId` owns, for event attribution; undefined between dispatch and commit. */ + runningTurnId(sessionId: SessionId, runId: RunId): TurnId | undefined { + return this.runningFor(sessionId, runId)?.turn.turnId; + } + /** An adapter `error` while the run's turn is live; decides `failed` on a stop-less settle. */ noteError(sessionId: SessionId, runId: RunId): void { const entry = this.runningFor(sessionId, runId); diff --git a/packages/host/engine/src/engine.ts b/packages/host/engine/src/engine.ts index 8d8987e37..92cd1e595 100644 --- a/packages/host/engine/src/engine.ts +++ b/packages/host/engine/src/engine.ts @@ -23,6 +23,7 @@ import { BrowserBrokerService } from './browser/broker'; import { BrowserReplHost } from './browser/repl-host'; import { BrowserRequestHandler } from './browser/request-handler'; import { InMemoryConversationStore } from './conversation/conversation-store'; +import { ConversationLiveJournals } from './conversation/live-journal'; import { ConversationRequestHandler } from './conversation/request-handler'; import { ConversationTurnService } from './conversation/turn-service'; import type { EngineDeps } from './deps'; @@ -153,6 +154,7 @@ export const createEngineRuntime = Effect.fn('Engine.create')(function* ( transport, runTask, ); + const conversationJournals = new ConversationLiveJournals(); const sessions = new SessionOrchestrator( transport, factory, @@ -167,6 +169,7 @@ export const createEngineRuntime = Effect.fn('Engine.create')(function* ( }, resources, conversationTurns, + conversationJournals, deps.browserToolsEnabled ? () => new BrowserReplHost((op, args) => browserBroker.dispatch(op, args)) : undefined, diff --git a/packages/host/engine/src/session/lifecycle-service.ts b/packages/host/engine/src/session/lifecycle-service.ts index 9e60f3f26..c2a934499 100644 --- a/packages/host/engine/src/session/lifecycle-service.ts +++ b/packages/host/engine/src/session/lifecycle-service.ts @@ -163,6 +163,7 @@ export class SessionLifecycleService { updatedAt: now, runs: [{ runId, startedAt: now, ...runOf(resolved, accountId) }], graphRevision: 0, + eventEpoch: 0, }; yield* sessions.startLive( replyTo, @@ -203,6 +204,7 @@ export class SessionLifecycleService { updatedAt: now, runs: [], graphRevision: 0, + eventEpoch: 0, }; yield* records.importRecord(record); if (record.cwd) yield* workspaceTouch(workspaces, record.cwd); @@ -243,6 +245,7 @@ export class SessionLifecycleService { updatedAt: now, runs: [{ runId, historyId, startedAt: now, ...runOf(startOptions, accountId) }], graphRevision: 0, + eventEpoch: 0, }; yield* sessions.startLive( replyTo, @@ -850,14 +853,22 @@ export class SessionLifecycleService { runId, baseTurnId, }); - return this.sessions.startLive( - replyTo, - record, - launchedRunId, - startAdapter, - resolved.warnings, - startOptions, - ); + // The bumped epoch must be durable before the LiveSession exists to mint under it; a lost + // write here would re-mint the same (epoch, seq) pairs after a reboot, with no gap signal. + return this.records + .flush(record.sessionId) + .pipe( + Effect.andThen( + this.sessions.startLive( + replyTo, + record, + launchedRunId, + startAdapter, + resolved.warnings, + startOptions, + ), + ), + ); }); } @@ -900,6 +911,7 @@ export class SessionLifecycleService { updatedAt: now, runs: [{ runId, startedAt: now, ...runOf(startOptions, accountId) }], graphRevision: 0, + eventEpoch: 0, }; if (startOptions.cwd) yield* workspaceTouch(workspaces, startOptions.cwd); yield* sessions.startLive(undefined, record, runId, (adapter) => diff --git a/packages/host/engine/src/session/live-session.ts b/packages/host/engine/src/session/live-session.ts index 26f32bf37..e448f5765 100644 --- a/packages/host/engine/src/session/live-session.ts +++ b/packages/host/engine/src/session/live-session.ts @@ -55,12 +55,16 @@ export class LiveSession { private closing = false; private historyId: AgentHistoryId | undefined; private readonly livePrompts: LivePrompt[] = []; + private seq = 0; constructor( readonly adapter: AgentAdapter, sessionId: SessionId, /** The `SessionRun` this adapter serves — run bookkeeping addresses runs by this id. */ readonly runId: RunId, + /** The session's event epoch captured at launch: a replaced adapter's stragglers keep minting + * under their own epoch, so they always compare below the replacement's positions. */ + readonly epoch: number, readonly scope: Scope.Closeable, readonly closed: Deferred.Deferred, ) { @@ -68,6 +72,11 @@ export class LiveSession { this.capabilities = adapter.capabilities; } + /** Mint the next event-plane position; monotone within this adapter's epoch. */ + nextSeq(): number { + return ++this.seq; + } + run(effect: Effect.Effect): Effect.Effect { return Effect.suspend(() => this.closing diff --git a/packages/host/engine/src/session/orchestrator.ts b/packages/host/engine/src/session/orchestrator.ts index ef3659f5a..070723792 100644 --- a/packages/host/engine/src/session/orchestrator.ts +++ b/packages/host/engine/src/session/orchestrator.ts @@ -18,6 +18,7 @@ import { Cause, Deferred, Effect, Exit, Scope } from 'effect'; import type { AgentRuntimeService } from '../agent/runtime-service'; import type { TurnResult } from '../automation/turn-watcher'; import { watchTurn } from '../automation/turn-watcher'; +import type { ConversationLiveJournals } from '../conversation/live-journal'; import type { ConversationTurnService, PersistedTurnIntent } from '../conversation/turn-service'; import { mintOperationId, promptBlocksFromContent } from '../conversation/turn-service'; import type { EngineFailure } from '../failure'; @@ -44,6 +45,7 @@ export class SessionOrchestrator { private readonly onStopped: (sessionId: SessionId) => void, private readonly resources: ResourceService, private readonly turns: ConversationTurnService, + private readonly journals: ConversationLiveJournals, private readonly browserTools?: BrowserToolsetFactory, ) { this.events = new SessionEventProcessor( @@ -53,6 +55,7 @@ export class SessionOrchestrator { reportFailure, resources, turns, + journals, ); this.inputs = new SessionInputDispatcher(records, this.events, resources, turns); } @@ -104,7 +107,7 @@ export class SessionOrchestrator { replay(sessionId: SessionId): void { const session = this.sessions.get(sessionId); - if (session) this.events.broadcast(sessionId, session.replay()); + if (session) this.events.broadcast(sessionId, session, session.replay()); } sendInput( @@ -144,6 +147,7 @@ export class SessionOrchestrator { yield* resources.deleteSession(sessionId); yield* this.turns.deleteSession(sessionId); yield* this.records.delete(sessionId); + this.journals.drop(sessionId); }); } @@ -201,7 +205,7 @@ export class SessionOrchestrator { input: { type: 'prompt', blocks: promptBlocksFromContent(content) }, }); const result = yield* Effect.sync(() => { - this.events.broadcast(sessionId, [ + this.events.broadcast(sessionId, session, [ { type: 'user-message', messageId: nextMessageId(), content }, ]); records.setTitleFromContent(sessionId, content); @@ -284,7 +288,14 @@ export class SessionOrchestrator { if (browserTools) adapter.attachBrowserTools?.(browserTools); const scope = yield* Scope.fork(parentScope); const closed = yield* Deferred.make(); - const session = new LiveSession(adapter, sessionId, runId, scope, closed); + const session = new LiveSession( + adapter, + sessionId, + runId, + record.eventEpoch, + scope, + closed, + ); const startupEvents: AgentEvent[] = []; let bufferEvents = rewindMessageId !== undefined; session.listen((event) => { @@ -329,7 +340,7 @@ export class SessionOrchestrator { ), ); if (rewindMessageId !== undefined) { - events.broadcast(sessionId, [ + events.broadcast(sessionId, session, [ { type: 'conversation-rewind', messageId: rewindMessageId }, ]); bufferEvents = false; @@ -414,7 +425,7 @@ export class SessionOrchestrator { return Scope.close(session.scope, Exit.interrupt()).pipe( Effect.andThen( Effect.sync(() => { - this.events.broadcast(sessionId, session.closeInteractions()); + this.events.broadcast(sessionId, session, session.closeInteractions()); session.stopListening(); }), ), @@ -426,6 +437,9 @@ export class SessionOrchestrator { // Teardown mid-turn kills the turn without a stop frame; settle it here. this.turns.settleStatus(sessionId, session.runId, 'stopped'); this.records.sealRun(sessionId, session.runId); + // The live tail dies with the live session (readers see the epoch-jump gap), so + // journal memory stays bounded by the number of concurrent live adapters. + this.journals.drop(sessionId); return recordLiveSessions(this.sessions.size); }), ), @@ -461,6 +475,7 @@ export class SessionOrchestrator { // token minted while resolving start options. Normal teardown does this via `onStopped`; // a discarded failed start must too, or that token leaks until daemon shutdown. this.onStopped(sessionId); + this.journals.drop(sessionId); return recordLiveSessions(this.sessions.size); }), ), diff --git a/packages/host/engine/src/session/session-event-processor.ts b/packages/host/engine/src/session/session-event-processor.ts index 0d90833be..bc23996fa 100644 --- a/packages/host/engine/src/session/session-event-processor.ts +++ b/packages/host/engine/src/session/session-event-processor.ts @@ -1,9 +1,16 @@ import { AUTH_FAILED_ERROR_CODE } from '@linkcode/agent-adapter'; -import type { AgentEvent, SessionId, SessionNotificationReason, ToolKind } from '@linkcode/schema'; +import type { + AgentEvent, + SessionId, + SessionNotificationReason, + ToolKind, + TurnId, +} from '@linkcode/schema'; import type { Transport } from '@linkcode/transport'; import { createWireMessage } from '@linkcode/transport'; import { Effect } from 'effect'; import type { AgentRuntimeService } from '../agent/runtime-service'; +import type { ConversationLiveJournals } from '../conversation/live-journal'; import type { ConversationTurnService } from '../conversation/turn-service'; import type { ResourceService } from '../resource/service'; import type { LiveSession } from './live-session'; @@ -33,12 +40,28 @@ export class SessionEventProcessor { private readonly reportFailure: (effect: Effect.Effect) => void, private readonly resources: ResourceService, private readonly turns: ConversationTurnService, + private readonly journals: ConversationLiveJournals, ) {} - broadcast(sessionId: SessionId, events: Iterable): void { - for (const event of events) { - this.transport.send(createWireMessage({ kind: 'agent.event', sessionId, event })); - } + broadcast(sessionId: SessionId, session: LiveSession, events: Iterable): void { + const turnId = this.turns.runningTurnId(sessionId, session.runId); + for (const event of events) this.send(sessionId, session, event, turnId); + } + + /** The one stamped exit: every `agent.event` frame mints its `(epoch, seq)` position here and + * lands in the session's live journal, so the wire stream and the journal never diverge. */ + private send( + sessionId: SessionId, + session: LiveSession, + event: AgentEvent, + turnId: TurnId | undefined, + ): void { + const { epoch, runId } = session; + const seq = session.nextSeq(); + this.journals.open(sessionId).append({ epoch, seq, runId, turnId, ts: Date.now(), event }); + this.transport.send( + createWireMessage({ kind: 'agent.event', sessionId, runId, turnId, epoch, seq, event }), + ); } private registerResources(sessionId: SessionId, event: AgentEvent): void { @@ -96,14 +119,10 @@ export class SessionEventProcessor { ); } - rejectInput(sessionId: SessionId, message: string): void { - this.transport.send( - createWireMessage({ - kind: 'agent.event', - sessionId, - event: { type: 'error', message, code: 'input_rejected', recoverable: true }, - }), - ); + rejectInput(sessionId: SessionId, session: LiveSession, message: string): void { + this.broadcast(sessionId, session, [ + { type: 'error', message, code: 'input_rejected', recoverable: true }, + ]); } handle(sessionId: SessionId, session: LiveSession, event: AgentEvent): void { @@ -116,7 +135,12 @@ export class SessionEventProcessor { ) { return; } - this.broadcast(sessionId, session.apply(event)); + // Captured before the settle below so a turn-ending event still carries its turn. + const turnId = this.turns.runningTurnId(sessionId, session.runId); + const derived = session.apply(event); + for (let i = 0, len = derived.length; i < len; i++) { + this.send(sessionId, session, derived[i], turnId); + } this.registerResources(sessionId, event); switch (event.type) { case 'status': @@ -141,7 +165,7 @@ export class SessionEventProcessor { default: break; } - this.transport.send(createWireMessage({ kind: 'agent.event', sessionId, event })); + this.send(sessionId, session, event, turnId); this.notify(sessionId, event); } catch (error) { this.reportFailure(Effect.logError('Failed to process agent event', { sessionId }, error)); diff --git a/packages/host/engine/src/session/session-input-dispatcher.ts b/packages/host/engine/src/session/session-input-dispatcher.ts index 0096f3f3e..ff79d2295 100644 --- a/packages/host/engine/src/session/session-input-dispatcher.ts +++ b/packages/host/engine/src/session/session-input-dispatcher.ts @@ -41,7 +41,7 @@ export class SessionInputDispatcher { message: `Unknown slash command: /${input.name}`, reportedInConversation: true, }); - this.events.rejectInput(sessionId, error.message); + this.events.rejectInput(sessionId, session, error.message); return Effect.fail(error); } if (input.type === 'shell-command' && !session.capabilities.shellCommand) { @@ -50,7 +50,7 @@ export class SessionInputDispatcher { message: 'Shell commands are not supported by this session', reportedInConversation: true, }); - this.events.rejectInput(sessionId, error.message); + this.events.rejectInput(sessionId, session, error.message); return Effect.fail(error); } if (startsTurn && session.turnInputActive) { @@ -59,7 +59,7 @@ export class SessionInputDispatcher { message: `Session is busy: ${sessionId}`, reportedInConversation: true, }); - this.events.rejectInput(sessionId, error.message); + this.events.rejectInput(sessionId, session, error.message); return Effect.fail(error); } const { events, records, resources, turns } = this; @@ -75,7 +75,7 @@ export class SessionInputDispatcher { message: `Session is busy: ${sessionId}`, reportedInConversation: true, }); - events.rejectInput(sessionId, error.message); + events.rejectInput(sessionId, session, error.message); return yield* Effect.fail(error); } let adapterInput: AgentInput = input; @@ -120,14 +120,14 @@ export class SessionInputDispatcher { const dispatch = Effect.gen(function* () { // Echo before awaiting send: provider events can outrun the dispatch acknowledgement. if (promptMessageId !== undefined && input.type === 'prompt') { - events.broadcast(sessionId, session.trackPrompt(promptMessageId, input.content)); + events.broadcast(sessionId, session, session.trackPrompt(promptMessageId, input.content)); records.setTitleFromContent(sessionId, input.content); } else if (input.type === 'command' || input.type === 'shell-command') { const text = input.type === 'command' ? `/${input.name}${input.arguments ? ` ${input.arguments}` : ''}` : `$ ${input.command}`; - events.broadcast(sessionId, [ + events.broadcast(sessionId, session, [ { type: 'user-message', messageId: nextMessageId(), @@ -143,7 +143,7 @@ export class SessionInputDispatcher { ? session.interactions.beginResponse(responseInput) : undefined; if (responseInput && respondingAsk) { - events.broadcast(sessionId, [ + events.broadcast(sessionId, session, [ { type: 'prompt-response-status', requestId: responseInput.requestId, @@ -167,19 +167,20 @@ export class SessionInputDispatcher { if (responseInput && respondingAsk) { events.broadcast( sessionId, + session, session.interactions.restoreResponse(responseInput.requestId, respondingAsk), ); } if (promptMessageId !== undefined) { - events.broadcast(sessionId, session.untrackPrompt(promptMessageId)); + events.broadcast(sessionId, session, session.untrackPrompt(promptMessageId)); } - if (startsTurn) events.rejectInput(sessionId, error.publicMessage); + if (startsTurn) events.rejectInput(sessionId, session, error.publicMessage); }), ), ); if (responseInput && respondingAsk) { const resolution = session.interactions.resolveResponse(responseInput, respondingAsk); - if (resolution) events.broadcast(sessionId, [resolution]); + if (resolution) events.broadcast(sessionId, session, [resolution]); } // The provider accepted the dispatch: the turn flips to running and the default leaf moves. if (persisted !== undefined) yield* turns.commitRunning(persisted); diff --git a/packages/host/engine/src/session/session-record-registry.ts b/packages/host/engine/src/session/session-record-registry.ts index 616be6187..a6cd571f2 100644 --- a/packages/host/engine/src/session/session-record-registry.ts +++ b/packages/host/engine/src/session/session-record-registry.ts @@ -60,6 +60,9 @@ export class SessionRecordRegistry { Effect.sync(() => { for (let i = 0, len = records.length; i < len; i++) { const record = records[i]; + // Boot epoch bump, in memory only: nothing mints events before a launch, and every + // launch persists the record — writing all rows here would churn updatedAt (recency). + record.eventEpoch += 1; this.records.set(record.sessionId, record); } }), @@ -216,6 +219,8 @@ export class SessionRecordRegistry { const runId = run.runId ?? mintRunId(); const record = this.records.get(sessionId); if (!record) return runId; + // A replacement adapter must mint above everything the old one could have minted. + record.eventEpoch += 1; record.runs.push({ startedAt: Date.now(), ...definedFields(run), runId }); this.persist(record); // A new run re-points the identity `list()` projects — `accountId`, `historyId` — so clients @@ -225,6 +230,17 @@ export class SessionRecordRegistry { return runId; } + /** Awaited durable save, for the launch path only: the bumped epoch must reach the store before + * a LiveSession can mint under it, and a lost write must fail the launch loud — the general + * fire-and-forget {@link persist} cannot guarantee either. */ + flush(sessionId: SessionId): Effect.Effect { + const record = this.records.get(sessionId); + if (!record) return Effect.void; + return storeOperation('session-records.save', 'Failed to persist session record', () => + this.store.save(record), + ); + } + setTitleFromContent(sessionId: SessionId, content: ContentBlock[]): void { const record = this.records.get(sessionId); if (!record || record.title !== undefined) return;