diff --git a/.gitignore b/.gitignore index 2a2a9f1..f78db31 100644 --- a/.gitignore +++ b/.gitignore @@ -26,6 +26,9 @@ .DS_Store *.pem .repos/effect +.tmp/ +.cursor/ +.agent/ # debug npm-debug.log* diff --git a/drizzle/0013_first_magus.sql b/drizzle/0013_first_magus.sql new file mode 100644 index 0000000..7aa2e5c --- /dev/null +++ b/drizzle/0013_first_magus.sql @@ -0,0 +1,30 @@ +CREATE TABLE "fluid_memory_items" ( + "id" uuid PRIMARY KEY DEFAULT gen_random_uuid() NOT NULL, + "workspace_id" uuid NOT NULL, + "kind" text NOT NULL, + "payload" jsonb NOT NULL, + "abstract_l0" text NOT NULL, + "overview_l1" text NOT NULL, + "source_message_id" uuid, + "confidence" double precision NOT NULL, + "status" text NOT NULL, + "version" integer DEFAULT 1 NOT NULL, + "created_at" timestamp with time zone DEFAULT now() NOT NULL, + "updated_at" timestamp with time zone DEFAULT now() NOT NULL +); +--> statement-breakpoint +CREATE TABLE "memory_diffs" ( + "id" uuid PRIMARY KEY DEFAULT gen_random_uuid() NOT NULL, + "workspace_id" uuid NOT NULL, + "source_message_id" uuid, + "operations" jsonb NOT NULL, + "created_at" timestamp with time zone DEFAULT now() NOT NULL +); +--> statement-breakpoint +ALTER TABLE "fluid_memory_items" ADD CONSTRAINT "fluid_memory_items_workspace_id_workspaces_id_fk" FOREIGN KEY ("workspace_id") REFERENCES "public"."workspaces"("id") ON DELETE cascade ON UPDATE no action;--> statement-breakpoint +ALTER TABLE "fluid_memory_items" ADD CONSTRAINT "fluid_memory_items_source_message_id_chat_messages_id_fk" FOREIGN KEY ("source_message_id") REFERENCES "public"."chat_messages"("id") ON DELETE set null ON UPDATE no action;--> statement-breakpoint +ALTER TABLE "memory_diffs" ADD CONSTRAINT "memory_diffs_workspace_id_workspaces_id_fk" FOREIGN KEY ("workspace_id") REFERENCES "public"."workspaces"("id") ON DELETE cascade ON UPDATE no action;--> statement-breakpoint +ALTER TABLE "memory_diffs" ADD CONSTRAINT "memory_diffs_source_message_id_chat_messages_id_fk" FOREIGN KEY ("source_message_id") REFERENCES "public"."chat_messages"("id") ON DELETE set null ON UPDATE no action;--> statement-breakpoint +CREATE INDEX "fluid_memory_items_workspace_status_idx" ON "fluid_memory_items" USING btree ("workspace_id","status");--> statement-breakpoint +CREATE INDEX "fluid_memory_items_workspace_kind_idx" ON "fluid_memory_items" USING btree ("workspace_id","kind");--> statement-breakpoint +CREATE INDEX "memory_diffs_workspace_created_idx" ON "memory_diffs" USING btree ("workspace_id","created_at"); \ No newline at end of file diff --git a/drizzle/0014_messy_apocalypse.sql b/drizzle/0014_messy_apocalypse.sql new file mode 100644 index 0000000..9233999 --- /dev/null +++ b/drizzle/0014_messy_apocalypse.sql @@ -0,0 +1,13 @@ +CREATE TABLE "fluid_memory_tokens" ( + "id" uuid PRIMARY KEY DEFAULT gen_random_uuid() NOT NULL, + "workspace_id" uuid NOT NULL, + "item_id" uuid NOT NULL, + "kind" text NOT NULL, + "token" text NOT NULL, + "frequency" integer DEFAULT 1 NOT NULL +); +--> statement-breakpoint +ALTER TABLE "fluid_memory_tokens" ADD CONSTRAINT "fluid_memory_tokens_workspace_id_workspaces_id_fk" FOREIGN KEY ("workspace_id") REFERENCES "public"."workspaces"("id") ON DELETE cascade ON UPDATE no action;--> statement-breakpoint +ALTER TABLE "fluid_memory_tokens" ADD CONSTRAINT "fluid_memory_tokens_item_id_fluid_memory_items_id_fk" FOREIGN KEY ("item_id") REFERENCES "public"."fluid_memory_items"("id") ON DELETE cascade ON UPDATE no action;--> statement-breakpoint +CREATE INDEX "fluid_memory_tokens_lookup_idx" ON "fluid_memory_tokens" USING btree ("workspace_id","kind","token");--> statement-breakpoint +CREATE INDEX "fluid_memory_tokens_item_idx" ON "fluid_memory_tokens" USING btree ("item_id"); \ No newline at end of file diff --git a/drizzle/meta/0013_snapshot.json b/drizzle/meta/0013_snapshot.json new file mode 100644 index 0000000..f46aa78 --- /dev/null +++ b/drizzle/meta/0013_snapshot.json @@ -0,0 +1,1185 @@ +{ + "id": "22b44d7b-bd80-4378-9789-756b55c75b0f", + "prevId": "a2a2f9a9-d567-4413-89d8-fa714b994351", + "version": "7", + "dialect": "postgresql", + "tables": { + "public.chat_messages": { + "name": "chat_messages", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "uuid", + "primaryKey": true, + "notNull": true, + "default": "gen_random_uuid()" + }, + "thread_id": { + "name": "thread_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "role": { + "name": "role", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "content": { + "name": "content", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "citations": { + "name": "citations", + "type": "jsonb", + "primaryKey": false, + "notNull": false + }, + "artifacts": { + "name": "artifacts", + "type": "jsonb", + "primaryKey": false, + "notNull": false + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": { + "chat_messages_thread_created_idx": { + "name": "chat_messages_thread_created_idx", + "columns": [ + { + "expression": "thread_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "created_at", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": { + "chat_messages_thread_id_chat_threads_id_fk": { + "name": "chat_messages_thread_id_chat_threads_id_fk", + "tableFrom": "chat_messages", + "tableTo": "chat_threads", + "columnsFrom": [ + "thread_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.chat_threads": { + "name": "chat_threads", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "uuid", + "primaryKey": true, + "notNull": true, + "default": "gen_random_uuid()" + }, + "workspace_id": { + "name": "workspace_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "title": { + "name": "title", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "demo_key": { + "name": "demo_key", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "deleted_at": { + "name": "deleted_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false + } + }, + "indexes": { + "chat_threads_workspace_updated_idx": { + "name": "chat_threads_workspace_updated_idx", + "columns": [ + { + "expression": "workspace_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "updated_at", + "isExpression": false, + "asc": false, + "nulls": "last" + } + ], + "isUnique": false, + "where": "deleted_at IS NULL", + "concurrently": false, + "method": "btree", + "with": {} + }, + "chat_threads_workspace_demo_key_idx": { + "name": "chat_threads_workspace_demo_key_idx", + "columns": [ + { + "expression": "workspace_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "demo_key", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": { + "chat_threads_workspace_id_workspaces_id_fk": { + "name": "chat_threads_workspace_id_workspaces_id_fk", + "tableFrom": "chat_threads", + "tableTo": "workspaces", + "columnsFrom": [ + "workspace_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.demo_source_visibilities": { + "name": "demo_source_visibilities", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "uuid", + "primaryKey": true, + "notNull": true, + "default": "gen_random_uuid()" + }, + "workspace_id": { + "name": "workspace_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "demo_source_id": { + "name": "demo_source_id", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "hidden_at": { + "name": "hidden_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false + }, + "deleted_at": { + "name": "deleted_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": { + "demo_source_visibilities_workspace_source_idx": { + "name": "demo_source_visibilities_workspace_source_idx", + "columns": [ + { + "expression": "workspace_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "demo_source_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "concurrently": false, + "method": "btree", + "with": {} + }, + "demo_source_visibilities_workspace_idx": { + "name": "demo_source_visibilities_workspace_idx", + "columns": [ + { + "expression": "workspace_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": { + "demo_source_visibilities_workspace_id_workspaces_id_fk": { + "name": "demo_source_visibilities_workspace_id_workspaces_id_fk", + "tableFrom": "demo_source_visibilities", + "tableTo": "workspaces", + "columnsFrom": [ + "workspace_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.fluid_memory_items": { + "name": "fluid_memory_items", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "uuid", + "primaryKey": true, + "notNull": true, + "default": "gen_random_uuid()" + }, + "workspace_id": { + "name": "workspace_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "kind": { + "name": "kind", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "payload": { + "name": "payload", + "type": "jsonb", + "primaryKey": false, + "notNull": true + }, + "abstract_l0": { + "name": "abstract_l0", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "overview_l1": { + "name": "overview_l1", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "source_message_id": { + "name": "source_message_id", + "type": "uuid", + "primaryKey": false, + "notNull": false + }, + "confidence": { + "name": "confidence", + "type": "double precision", + "primaryKey": false, + "notNull": true + }, + "status": { + "name": "status", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "version": { + "name": "version", + "type": "integer", + "primaryKey": false, + "notNull": true, + "default": 1 + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": { + "fluid_memory_items_workspace_status_idx": { + "name": "fluid_memory_items_workspace_status_idx", + "columns": [ + { + "expression": "workspace_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "status", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + }, + "fluid_memory_items_workspace_kind_idx": { + "name": "fluid_memory_items_workspace_kind_idx", + "columns": [ + { + "expression": "workspace_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "kind", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": { + "fluid_memory_items_workspace_id_workspaces_id_fk": { + "name": "fluid_memory_items_workspace_id_workspaces_id_fk", + "tableFrom": "fluid_memory_items", + "tableTo": "workspaces", + "columnsFrom": [ + "workspace_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + }, + "fluid_memory_items_source_message_id_chat_messages_id_fk": { + "name": "fluid_memory_items_source_message_id_chat_messages_id_fk", + "tableFrom": "fluid_memory_items", + "tableTo": "chat_messages", + "columnsFrom": [ + "source_message_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "set null", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.memory_diffs": { + "name": "memory_diffs", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "uuid", + "primaryKey": true, + "notNull": true, + "default": "gen_random_uuid()" + }, + "workspace_id": { + "name": "workspace_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "source_message_id": { + "name": "source_message_id", + "type": "uuid", + "primaryKey": false, + "notNull": false + }, + "operations": { + "name": "operations", + "type": "jsonb", + "primaryKey": false, + "notNull": true + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": { + "memory_diffs_workspace_created_idx": { + "name": "memory_diffs_workspace_created_idx", + "columns": [ + { + "expression": "workspace_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "created_at", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": { + "memory_diffs_workspace_id_workspaces_id_fk": { + "name": "memory_diffs_workspace_id_workspaces_id_fk", + "tableFrom": "memory_diffs", + "tableTo": "workspaces", + "columnsFrom": [ + "workspace_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + }, + "memory_diffs_source_message_id_chat_messages_id_fk": { + "name": "memory_diffs_source_message_id_chat_messages_id_fk", + "tableFrom": "memory_diffs", + "tableTo": "chat_messages", + "columnsFrom": [ + "source_message_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "set null", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.parsed_document_sync_leases": { + "name": "parsed_document_sync_leases", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "uuid", + "primaryKey": true, + "notNull": true, + "default": "gen_random_uuid()" + }, + "workspace_id": { + "name": "workspace_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "source_id": { + "name": "source_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "document_id": { + "name": "document_id", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "revision_key": { + "name": "revision_key", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "lease_token": { + "name": "lease_token", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "acquired_at": { + "name": "acquired_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "expires_at": { + "name": "expires_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true + }, + "released_at": { + "name": "released_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false + }, + "release_reason": { + "name": "release_reason", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": { + "parsed_document_sync_leases_token_idx": { + "name": "parsed_document_sync_leases_token_idx", + "columns": [ + { + "expression": "lease_token", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "concurrently": false, + "method": "btree", + "with": {} + }, + "parsed_document_sync_leases_active_idx": { + "name": "parsed_document_sync_leases_active_idx", + "columns": [ + { + "expression": "expires_at", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "released_at IS NULL", + "concurrently": false, + "method": "btree", + "with": {} + }, + "parsed_document_sync_leases_workspace_active_idx": { + "name": "parsed_document_sync_leases_workspace_active_idx", + "columns": [ + { + "expression": "workspace_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "released_at IS NULL", + "concurrently": false, + "method": "btree", + "with": {} + }, + "parsed_document_sync_leases_document_active_idx": { + "name": "parsed_document_sync_leases_document_active_idx", + "columns": [ + { + "expression": "document_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "released_at IS NULL", + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": { + "parsed_document_sync_leases_workspace_id_workspaces_id_fk": { + "name": "parsed_document_sync_leases_workspace_id_workspaces_id_fk", + "tableFrom": "parsed_document_sync_leases", + "tableTo": "workspaces", + "columnsFrom": [ + "workspace_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + }, + "parsed_document_sync_leases_source_id_sources_id_fk": { + "name": "parsed_document_sync_leases_source_id_sources_id_fk", + "tableFrom": "parsed_document_sync_leases", + "tableTo": "sources", + "columnsFrom": [ + "source_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.source_parse_results": { + "name": "source_parse_results", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "uuid", + "primaryKey": true, + "notNull": true, + "default": "gen_random_uuid()" + }, + "source_id": { + "name": "source_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "result_blob_url": { + "name": "result_blob_url", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "snapshot_manifest_url": { + "name": "snapshot_manifest_url", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "snapshot_manifest_key": { + "name": "snapshot_manifest_key", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "revision_key": { + "name": "revision_key", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "sync_status": { + "name": "sync_status", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "sync_error": { + "name": "sync_error", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "asset_urls": { + "name": "asset_urls", + "type": "jsonb", + "primaryKey": false, + "notNull": true + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": { + "source_parse_results_source_id_idx": { + "name": "source_parse_results_source_id_idx", + "columns": [ + { + "expression": "source_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": { + "source_parse_results_source_id_sources_id_fk": { + "name": "source_parse_results_source_id_sources_id_fk", + "tableFrom": "source_parse_results", + "tableTo": "sources", + "columnsFrom": [ + "source_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": { + "source_parse_results_source_id_unique": { + "name": "source_parse_results_source_id_unique", + "nullsNotDistinct": false, + "columns": [ + "source_id" + ] + } + }, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.sources": { + "name": "sources", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "uuid", + "primaryKey": true, + "notNull": true, + "default": "gen_random_uuid()" + }, + "workspace_id": { + "name": "workspace_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "title": { + "name": "title", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "mime_type": { + "name": "mime_type", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "size_bytes": { + "name": "size_bytes", + "type": "bigint", + "primaryKey": false, + "notNull": true + }, + "status": { + "name": "status", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "failure_reason": { + "name": "failure_reason", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "failure_stage": { + "name": "failure_stage", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "knowhere_job_id": { + "name": "knowhere_job_id", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "knowhere_document_id": { + "name": "knowhere_document_id", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "staged_blob_pathname": { + "name": "staged_blob_pathname", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "staged_blob_url": { + "name": "staged_blob_url", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "original_blob_pathname": { + "name": "original_blob_pathname", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "original_blob_url": { + "name": "original_blob_url", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "demo_key": { + "name": "demo_key", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "deleted_at": { + "name": "deleted_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false + } + }, + "indexes": { + "sources_workspace_created_idx": { + "name": "sources_workspace_created_idx", + "columns": [ + { + "expression": "workspace_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "created_at", + "isExpression": false, + "asc": false, + "nulls": "last" + } + ], + "isUnique": false, + "where": "deleted_at IS NULL", + "concurrently": false, + "method": "btree", + "with": {} + }, + "sources_workspace_status_idx": { + "name": "sources_workspace_status_idx", + "columns": [ + { + "expression": "workspace_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "status", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + }, + "sources_workspace_demo_key_idx": { + "name": "sources_workspace_demo_key_idx", + "columns": [ + { + "expression": "workspace_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "demo_key", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "concurrently": false, + "method": "btree", + "with": {} + }, + "sources_workspace_document_idx": { + "name": "sources_workspace_document_idx", + "columns": [ + { + "expression": "workspace_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "knowhere_document_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "where": "knowhere_document_id IS NOT NULL AND deleted_at IS NULL", + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": { + "sources_workspace_id_workspaces_id_fk": { + "name": "sources_workspace_id_workspaces_id_fk", + "tableFrom": "sources", + "tableTo": "workspaces", + "columnsFrom": [ + "workspace_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.workspaces": { + "name": "workspaces", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "uuid", + "primaryKey": true, + "notNull": true, + "default": "gen_random_uuid()" + }, + "user_id": { + "name": "user_id", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "namespace": { + "name": "namespace", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": { + "workspaces_user_id_idx": { + "name": "workspaces_user_id_idx", + "columns": [ + { + "expression": "user_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": { + "workspaces_user_id_unique": { + "name": "workspaces_user_id_unique", + "nullsNotDistinct": false, + "columns": [ + "user_id" + ] + }, + "workspaces_namespace_unique": { + "name": "workspaces_namespace_unique", + "nullsNotDistinct": false, + "columns": [ + "namespace" + ] + } + }, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + } + }, + "enums": {}, + "schemas": {}, + "sequences": {}, + "roles": {}, + "policies": {}, + "views": {}, + "_meta": { + "columns": {}, + "schemas": {}, + "tables": {} + } +} \ No newline at end of file diff --git a/drizzle/meta/0014_snapshot.json b/drizzle/meta/0014_snapshot.json new file mode 100644 index 0000000..4fe0e5f --- /dev/null +++ b/drizzle/meta/0014_snapshot.json @@ -0,0 +1,1306 @@ +{ + "id": "72bfb4fe-4048-4dce-a5fa-ce8eea880aa3", + "prevId": "22b44d7b-bd80-4378-9789-756b55c75b0f", + "version": "7", + "dialect": "postgresql", + "tables": { + "public.chat_messages": { + "name": "chat_messages", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "uuid", + "primaryKey": true, + "notNull": true, + "default": "gen_random_uuid()" + }, + "thread_id": { + "name": "thread_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "role": { + "name": "role", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "content": { + "name": "content", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "citations": { + "name": "citations", + "type": "jsonb", + "primaryKey": false, + "notNull": false + }, + "artifacts": { + "name": "artifacts", + "type": "jsonb", + "primaryKey": false, + "notNull": false + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": { + "chat_messages_thread_created_idx": { + "name": "chat_messages_thread_created_idx", + "columns": [ + { + "expression": "thread_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "created_at", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": { + "chat_messages_thread_id_chat_threads_id_fk": { + "name": "chat_messages_thread_id_chat_threads_id_fk", + "tableFrom": "chat_messages", + "tableTo": "chat_threads", + "columnsFrom": [ + "thread_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.chat_threads": { + "name": "chat_threads", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "uuid", + "primaryKey": true, + "notNull": true, + "default": "gen_random_uuid()" + }, + "workspace_id": { + "name": "workspace_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "title": { + "name": "title", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "demo_key": { + "name": "demo_key", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "deleted_at": { + "name": "deleted_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false + } + }, + "indexes": { + "chat_threads_workspace_updated_idx": { + "name": "chat_threads_workspace_updated_idx", + "columns": [ + { + "expression": "workspace_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "updated_at", + "isExpression": false, + "asc": false, + "nulls": "last" + } + ], + "isUnique": false, + "where": "deleted_at IS NULL", + "concurrently": false, + "method": "btree", + "with": {} + }, + "chat_threads_workspace_demo_key_idx": { + "name": "chat_threads_workspace_demo_key_idx", + "columns": [ + { + "expression": "workspace_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "demo_key", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": { + "chat_threads_workspace_id_workspaces_id_fk": { + "name": "chat_threads_workspace_id_workspaces_id_fk", + "tableFrom": "chat_threads", + "tableTo": "workspaces", + "columnsFrom": [ + "workspace_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.demo_source_visibilities": { + "name": "demo_source_visibilities", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "uuid", + "primaryKey": true, + "notNull": true, + "default": "gen_random_uuid()" + }, + "workspace_id": { + "name": "workspace_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "demo_source_id": { + "name": "demo_source_id", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "hidden_at": { + "name": "hidden_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false + }, + "deleted_at": { + "name": "deleted_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": { + "demo_source_visibilities_workspace_source_idx": { + "name": "demo_source_visibilities_workspace_source_idx", + "columns": [ + { + "expression": "workspace_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "demo_source_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "concurrently": false, + "method": "btree", + "with": {} + }, + "demo_source_visibilities_workspace_idx": { + "name": "demo_source_visibilities_workspace_idx", + "columns": [ + { + "expression": "workspace_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": { + "demo_source_visibilities_workspace_id_workspaces_id_fk": { + "name": "demo_source_visibilities_workspace_id_workspaces_id_fk", + "tableFrom": "demo_source_visibilities", + "tableTo": "workspaces", + "columnsFrom": [ + "workspace_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.fluid_memory_items": { + "name": "fluid_memory_items", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "uuid", + "primaryKey": true, + "notNull": true, + "default": "gen_random_uuid()" + }, + "workspace_id": { + "name": "workspace_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "kind": { + "name": "kind", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "payload": { + "name": "payload", + "type": "jsonb", + "primaryKey": false, + "notNull": true + }, + "abstract_l0": { + "name": "abstract_l0", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "overview_l1": { + "name": "overview_l1", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "source_message_id": { + "name": "source_message_id", + "type": "uuid", + "primaryKey": false, + "notNull": false + }, + "confidence": { + "name": "confidence", + "type": "double precision", + "primaryKey": false, + "notNull": true + }, + "status": { + "name": "status", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "version": { + "name": "version", + "type": "integer", + "primaryKey": false, + "notNull": true, + "default": 1 + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": { + "fluid_memory_items_workspace_status_idx": { + "name": "fluid_memory_items_workspace_status_idx", + "columns": [ + { + "expression": "workspace_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "status", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + }, + "fluid_memory_items_workspace_kind_idx": { + "name": "fluid_memory_items_workspace_kind_idx", + "columns": [ + { + "expression": "workspace_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "kind", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": { + "fluid_memory_items_workspace_id_workspaces_id_fk": { + "name": "fluid_memory_items_workspace_id_workspaces_id_fk", + "tableFrom": "fluid_memory_items", + "tableTo": "workspaces", + "columnsFrom": [ + "workspace_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + }, + "fluid_memory_items_source_message_id_chat_messages_id_fk": { + "name": "fluid_memory_items_source_message_id_chat_messages_id_fk", + "tableFrom": "fluid_memory_items", + "tableTo": "chat_messages", + "columnsFrom": [ + "source_message_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "set null", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.fluid_memory_tokens": { + "name": "fluid_memory_tokens", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "uuid", + "primaryKey": true, + "notNull": true, + "default": "gen_random_uuid()" + }, + "workspace_id": { + "name": "workspace_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "item_id": { + "name": "item_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "kind": { + "name": "kind", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "token": { + "name": "token", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "frequency": { + "name": "frequency", + "type": "integer", + "primaryKey": false, + "notNull": true, + "default": 1 + } + }, + "indexes": { + "fluid_memory_tokens_lookup_idx": { + "name": "fluid_memory_tokens_lookup_idx", + "columns": [ + { + "expression": "workspace_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "kind", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "token", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + }, + "fluid_memory_tokens_item_idx": { + "name": "fluid_memory_tokens_item_idx", + "columns": [ + { + "expression": "item_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": { + "fluid_memory_tokens_workspace_id_workspaces_id_fk": { + "name": "fluid_memory_tokens_workspace_id_workspaces_id_fk", + "tableFrom": "fluid_memory_tokens", + "tableTo": "workspaces", + "columnsFrom": [ + "workspace_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + }, + "fluid_memory_tokens_item_id_fluid_memory_items_id_fk": { + "name": "fluid_memory_tokens_item_id_fluid_memory_items_id_fk", + "tableFrom": "fluid_memory_tokens", + "tableTo": "fluid_memory_items", + "columnsFrom": [ + "item_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.memory_diffs": { + "name": "memory_diffs", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "uuid", + "primaryKey": true, + "notNull": true, + "default": "gen_random_uuid()" + }, + "workspace_id": { + "name": "workspace_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "source_message_id": { + "name": "source_message_id", + "type": "uuid", + "primaryKey": false, + "notNull": false + }, + "operations": { + "name": "operations", + "type": "jsonb", + "primaryKey": false, + "notNull": true + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": { + "memory_diffs_workspace_created_idx": { + "name": "memory_diffs_workspace_created_idx", + "columns": [ + { + "expression": "workspace_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "created_at", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": { + "memory_diffs_workspace_id_workspaces_id_fk": { + "name": "memory_diffs_workspace_id_workspaces_id_fk", + "tableFrom": "memory_diffs", + "tableTo": "workspaces", + "columnsFrom": [ + "workspace_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + }, + "memory_diffs_source_message_id_chat_messages_id_fk": { + "name": "memory_diffs_source_message_id_chat_messages_id_fk", + "tableFrom": "memory_diffs", + "tableTo": "chat_messages", + "columnsFrom": [ + "source_message_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "set null", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.parsed_document_sync_leases": { + "name": "parsed_document_sync_leases", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "uuid", + "primaryKey": true, + "notNull": true, + "default": "gen_random_uuid()" + }, + "workspace_id": { + "name": "workspace_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "source_id": { + "name": "source_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "document_id": { + "name": "document_id", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "revision_key": { + "name": "revision_key", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "lease_token": { + "name": "lease_token", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "acquired_at": { + "name": "acquired_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "expires_at": { + "name": "expires_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true + }, + "released_at": { + "name": "released_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false + }, + "release_reason": { + "name": "release_reason", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": { + "parsed_document_sync_leases_token_idx": { + "name": "parsed_document_sync_leases_token_idx", + "columns": [ + { + "expression": "lease_token", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "concurrently": false, + "method": "btree", + "with": {} + }, + "parsed_document_sync_leases_active_idx": { + "name": "parsed_document_sync_leases_active_idx", + "columns": [ + { + "expression": "expires_at", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "released_at IS NULL", + "concurrently": false, + "method": "btree", + "with": {} + }, + "parsed_document_sync_leases_workspace_active_idx": { + "name": "parsed_document_sync_leases_workspace_active_idx", + "columns": [ + { + "expression": "workspace_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "released_at IS NULL", + "concurrently": false, + "method": "btree", + "with": {} + }, + "parsed_document_sync_leases_document_active_idx": { + "name": "parsed_document_sync_leases_document_active_idx", + "columns": [ + { + "expression": "document_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "released_at IS NULL", + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": { + "parsed_document_sync_leases_workspace_id_workspaces_id_fk": { + "name": "parsed_document_sync_leases_workspace_id_workspaces_id_fk", + "tableFrom": "parsed_document_sync_leases", + "tableTo": "workspaces", + "columnsFrom": [ + "workspace_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + }, + "parsed_document_sync_leases_source_id_sources_id_fk": { + "name": "parsed_document_sync_leases_source_id_sources_id_fk", + "tableFrom": "parsed_document_sync_leases", + "tableTo": "sources", + "columnsFrom": [ + "source_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.source_parse_results": { + "name": "source_parse_results", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "uuid", + "primaryKey": true, + "notNull": true, + "default": "gen_random_uuid()" + }, + "source_id": { + "name": "source_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "result_blob_url": { + "name": "result_blob_url", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "snapshot_manifest_url": { + "name": "snapshot_manifest_url", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "snapshot_manifest_key": { + "name": "snapshot_manifest_key", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "revision_key": { + "name": "revision_key", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "sync_status": { + "name": "sync_status", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "sync_error": { + "name": "sync_error", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "asset_urls": { + "name": "asset_urls", + "type": "jsonb", + "primaryKey": false, + "notNull": true + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": { + "source_parse_results_source_id_idx": { + "name": "source_parse_results_source_id_idx", + "columns": [ + { + "expression": "source_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": { + "source_parse_results_source_id_sources_id_fk": { + "name": "source_parse_results_source_id_sources_id_fk", + "tableFrom": "source_parse_results", + "tableTo": "sources", + "columnsFrom": [ + "source_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": { + "source_parse_results_source_id_unique": { + "name": "source_parse_results_source_id_unique", + "nullsNotDistinct": false, + "columns": [ + "source_id" + ] + } + }, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.sources": { + "name": "sources", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "uuid", + "primaryKey": true, + "notNull": true, + "default": "gen_random_uuid()" + }, + "workspace_id": { + "name": "workspace_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "title": { + "name": "title", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "mime_type": { + "name": "mime_type", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "size_bytes": { + "name": "size_bytes", + "type": "bigint", + "primaryKey": false, + "notNull": true + }, + "status": { + "name": "status", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "failure_reason": { + "name": "failure_reason", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "failure_stage": { + "name": "failure_stage", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "knowhere_job_id": { + "name": "knowhere_job_id", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "knowhere_document_id": { + "name": "knowhere_document_id", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "staged_blob_pathname": { + "name": "staged_blob_pathname", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "staged_blob_url": { + "name": "staged_blob_url", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "original_blob_pathname": { + "name": "original_blob_pathname", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "original_blob_url": { + "name": "original_blob_url", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "demo_key": { + "name": "demo_key", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "deleted_at": { + "name": "deleted_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false + } + }, + "indexes": { + "sources_workspace_created_idx": { + "name": "sources_workspace_created_idx", + "columns": [ + { + "expression": "workspace_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "created_at", + "isExpression": false, + "asc": false, + "nulls": "last" + } + ], + "isUnique": false, + "where": "deleted_at IS NULL", + "concurrently": false, + "method": "btree", + "with": {} + }, + "sources_workspace_status_idx": { + "name": "sources_workspace_status_idx", + "columns": [ + { + "expression": "workspace_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "status", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + }, + "sources_workspace_demo_key_idx": { + "name": "sources_workspace_demo_key_idx", + "columns": [ + { + "expression": "workspace_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "demo_key", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "concurrently": false, + "method": "btree", + "with": {} + }, + "sources_workspace_document_idx": { + "name": "sources_workspace_document_idx", + "columns": [ + { + "expression": "workspace_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "knowhere_document_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "where": "knowhere_document_id IS NOT NULL AND deleted_at IS NULL", + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": { + "sources_workspace_id_workspaces_id_fk": { + "name": "sources_workspace_id_workspaces_id_fk", + "tableFrom": "sources", + "tableTo": "workspaces", + "columnsFrom": [ + "workspace_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.workspaces": { + "name": "workspaces", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "uuid", + "primaryKey": true, + "notNull": true, + "default": "gen_random_uuid()" + }, + "user_id": { + "name": "user_id", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "namespace": { + "name": "namespace", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": { + "workspaces_user_id_idx": { + "name": "workspaces_user_id_idx", + "columns": [ + { + "expression": "user_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": { + "workspaces_user_id_unique": { + "name": "workspaces_user_id_unique", + "nullsNotDistinct": false, + "columns": [ + "user_id" + ] + }, + "workspaces_namespace_unique": { + "name": "workspaces_namespace_unique", + "nullsNotDistinct": false, + "columns": [ + "namespace" + ] + } + }, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + } + }, + "enums": {}, + "schemas": {}, + "sequences": {}, + "roles": {}, + "policies": {}, + "views": {}, + "_meta": { + "columns": {}, + "schemas": {}, + "tables": {} + } +} \ No newline at end of file diff --git a/drizzle/meta/_journal.json b/drizzle/meta/_journal.json index 93f0ced..b9be4f5 100644 --- a/drizzle/meta/_journal.json +++ b/drizzle/meta/_journal.json @@ -92,6 +92,20 @@ "when": 1783231408481, "tag": "0012_lazy_wendell_rand", "breakpoints": true + }, + { + "idx": 13, + "version": "7", + "when": 1788247035130, + "tag": "0013_first_magus", + "breakpoints": true + }, + { + "idx": 14, + "version": "7", + "when": 1788252035435, + "tag": "0014_messy_apocalypse", + "breakpoints": true } ] } \ No newline at end of file diff --git a/src/app/api/memory/extract/route.ts b/src/app/api/memory/extract/route.ts new file mode 100644 index 0000000..ee25dbe --- /dev/null +++ b/src/app/api/memory/extract/route.ts @@ -0,0 +1,27 @@ +import { serve } from "@upstash/workflow/nextjs" + +import { + normalizeMemoryExtractPayload, + runMemoryExtractWorkflow, + type MemoryExtractPayload, +} from "@/domains/memory/extract-workflow" +import { logger } from "@/lib/logger" + +export const { POST } = serve( + async (context) => { + const payload = normalizeMemoryExtractPayload(context.requestPayload) + if (!payload) { + logger.warn("memory: extract workflow received invalid payload") + return + } + await runMemoryExtractWorkflow({ context, payload }) + }, + { + failureFunction: async ({ context, failResponse }) => { + logger.error("memory: extract workflow failed", { + payload: context.requestPayload, + failResponse, + }) + }, + }, +) diff --git a/src/domains/chat/route-answer.ts b/src/domains/chat/route-answer.ts index 73cb38f..cc579b0 100644 --- a/src/domains/chat/route-answer.ts +++ b/src/domains/chat/route-answer.ts @@ -20,6 +20,7 @@ import { type ChatTurnValue, } from "@/domains/chat/service" import { chatTurnPersistence } from "@/domains/chat/chat-turn-persistence" +import { triggerMemoryExtraction } from "@/domains/memory/extract-trigger" import { startBackgroundReconciliation } from "@/domains/sources/background-reconcile" import { BlobParsedDocumentStorage } from "@/domains/sources/parsed-document-blob-storage" import { sourceWorkflowRuntime } from "@/domains/sources/workflow-runtime" @@ -196,7 +197,17 @@ const answerChatEffect = (input: AnswerChatInput) => return Either.match(result, { onLeft: (error): RouteResponse => routeResult.error(error.status, error.message), - onRight: (value): RouteResponse => routeResult.ok(value), + onRight: (value): RouteResponse => { + // Fire-and-forget: extract fluid memory from this turn without + // blocking the chat response. + void triggerMemoryExtraction({ + workspaceId: workspace.id, + threadId: value.threadId, + userMessageId: value.messages[0].id, + assistantMessageId: value.messages[1].id, + }) + return routeResult.ok(value) + }, }) }) diff --git a/src/domains/chat/route-service.test.ts b/src/domains/chat/route-service.test.ts index e71d38f..02eceb9 100644 --- a/src/domains/chat/route-service.test.ts +++ b/src/domains/chat/route-service.test.ts @@ -25,6 +25,7 @@ const mocks = vi.hoisted(() => ({ parsedStorageWriteAsset: vi.fn(), softDeleteChatThread: vi.fn(), startBackgroundReconciliation: vi.fn(), + triggerMemoryExtraction: vi.fn(), })) vi.mock("ai", async (importOriginal) => { @@ -62,6 +63,10 @@ vi.mock("@/domains/sources/background-reconcile", () => ({ startBackgroundReconciliation: mocks.startBackgroundReconciliation, })) +vi.mock("@/domains/memory/extract-trigger", () => ({ + triggerMemoryExtraction: mocks.triggerMemoryExtraction, +})) + vi.mock("@/domains/sources/workflow-runtime", () => ({ sourceWorkflowRuntime: { listForWorkspace: mocks.listSourcesForWorkspace, @@ -836,7 +841,10 @@ describe("chat route services", () => { mocks.handleChatTurn.mockResolvedValue( Either.right({ threadId: "thread_1", - messages: [], + messages: [ + { id: "message_user", role: "user", content: "Summarize it" }, + { id: "message_assistant", role: "assistant", content: "Summary" }, + ], }), ) @@ -845,6 +853,12 @@ describe("chat route services", () => { }) expect(result.status).toBe(200) + expect(mocks.triggerMemoryExtraction).toHaveBeenCalledWith({ + workspaceId: workspace.id, + threadId: "thread_1", + userMessageId: "message_user", + assistantMessageId: "message_assistant", + }) expect(mocks.startBackgroundReconciliation).toHaveBeenCalledWith( workspace.id, parsingSource.id, diff --git a/src/domains/memory/extract-trigger.ts b/src/domains/memory/extract-trigger.ts new file mode 100644 index 0000000..35b6467 --- /dev/null +++ b/src/domains/memory/extract-trigger.ts @@ -0,0 +1,43 @@ +import "server-only" + +import { Client } from "@upstash/workflow" + +import type { MemoryExtractPayload } from "./extract-workflow" +import { logger } from "@/lib/logger" + +function resolveBaseURL(): string { + return process.env.NOTEBOOK_PUBLIC_URL ?? "http://localhost:3000" +} + +/** + * Fire-and-forget trigger for the post-turn fluid-memory extraction + * workflow. Each turn is uniquely keyed by its message ids, so unlike the + * source reconcile trigger no cooldown/dedup guard is needed; QStash + * retries cover transient delivery failures. + */ +export async function triggerMemoryExtraction( + payload: MemoryExtractPayload, +): Promise { + const token = process.env.QSTASH_TOKEN + if (!token) { + logger.warn("memory: skipping extraction — QSTASH_TOKEN not set", { + workspaceId: payload.workspaceId, + assistantMessageId: payload.assistantMessageId, + }) + return + } + + try { + await new Client({ token }).trigger({ + url: `${resolveBaseURL()}/api/memory/extract`, + body: payload, + retries: 3, + }) + } catch (error) { + logger.error("memory: failed to trigger extraction workflow", { + workspaceId: payload.workspaceId, + assistantMessageId: payload.assistantMessageId, + message: error instanceof Error ? error.message : String(error), + }) + } +} diff --git a/src/domains/memory/extract-workflow.ts b/src/domains/memory/extract-workflow.ts new file mode 100644 index 0000000..aa6b5ee --- /dev/null +++ b/src/domains/memory/extract-workflow.ts @@ -0,0 +1,169 @@ +import "server-only" + +import type { WorkflowContext } from "@upstash/workflow" + +import { extractMemoryOperations } from "./extraction-model" +import { + summarizePayloadForContext, + type ExistingMemoryContextItem, +} from "./prompts" +import { resolveMemoryOperations } from "./resolve-operations" +import { tokenizeMemoryText } from "./search-index" +import { memoryService } from "./service" +import { fluidMemoryKinds, isFluidMemoryKind } from "./types" +import { chatThreadService } from "@/domains/chat/thread-service" +import { logger } from "@/lib/logger" + +export type MemoryExtractPayload = { + readonly workspaceId: string + readonly threadId: string + readonly userMessageId: string + readonly assistantMessageId: string +} + +type MemoryExtractWorkflowContext = Pick< + WorkflowContext, + "run" +> + +/** Prompt-context item that also carries status/payload for resolution. */ +type MemoryWorkflowItem = ExistingMemoryContextItem & { + readonly status: string + readonly payload: unknown +} + +/** Per-kind cap on lexical neighbors fed into the merge-decision prompt. */ +const DEDUP_CANDIDATES_PER_KIND = 8 + +export function normalizeMemoryExtractPayload( + raw: unknown, +): MemoryExtractPayload | null { + if (!raw || typeof raw !== "object") return null + const record = raw as Record + const workspaceId = getNonEmptyString(record.workspaceId) + const threadId = getNonEmptyString(record.threadId) + const userMessageId = getNonEmptyString(record.userMessageId) + const assistantMessageId = getNonEmptyString(record.assistantMessageId) + if (!workspaceId || !threadId || !userMessageId || !assistantMessageId) { + return null + } + return { workspaceId, threadId, userMessageId, assistantMessageId } +} + +export async function runMemoryExtractWorkflow(input: { + readonly context: MemoryExtractWorkflowContext + readonly payload: MemoryExtractPayload +}): Promise { + const { context, payload } = input + + const turn = await context.run("load-turn", async () => { + const messages = await chatThreadService.listMessages( + payload.workspaceId, + payload.threadId, + ) + const userMessage = messages?.find( + (message) => message.id === payload.userMessageId, + ) + const assistantMessage = messages?.find( + (message) => message.id === payload.assistantMessageId, + ) + if (!userMessage || !assistantMessage) return null + return { + userText: userMessage.content, + assistantText: assistantMessage.content, + referencedDocumentIds: collectCitationDocumentIds( + assistantMessage.citations, + ), + } + }) + if (!turn) { + logger.warn("memory: extract skipped — turn messages not found", { + workspaceId: payload.workspaceId, + threadId: payload.threadId, + }) + return + } + + const existingItems = await context.run("retrieve-candidates", async () => { + const queryTokens = tokenizeMemoryText(turn.userText).map( + (entry) => entry.token, + ) + if (queryTokens.length === 0) return [] + + const byId = new Map() + for (const kind of fluidMemoryKinds) { + const items = await memoryService.findDedupCandidates( + payload.workspaceId, + kind, + queryTokens, + DEDUP_CANDIDATES_PER_KIND, + ) + for (const item of items) { + if (!isFluidMemoryKind(item.kind) || byId.has(item.id)) continue + byId.set(item.id, { + id: item.id, + kind: item.kind, + status: item.status, + payload: item.payload, + abstractL0: item.abstractL0, + payloadSummary: summarizePayloadForContext(item.kind, item.payload), + }) + } + } + return [...byId.values()] + }) + + const operations = await context.run("extract-operations", () => + extractMemoryOperations({ + workspaceId: payload.workspaceId, + userText: turn.userText, + assistantText: turn.assistantText, + referencedDocumentIds: turn.referencedDocumentIds, + existingItems, + }), + ) + if (!operations) return + + const applied = await context.run("apply-operations", async () => { + const resolved = resolveMemoryOperations({ + operations, + existingItems, + referencedDocumentIds: turn.referencedDocumentIds, + }) + if (resolved.length === 0) return null + return memoryService.applyOperations( + payload.workspaceId, + payload.assistantMessageId, + resolved, + ) + }) + + logger.info("memory: extract workflow finished", { + workspaceId: payload.workspaceId, + threadId: payload.threadId, + assistantMessageId: payload.assistantMessageId, + candidateCount: existingItems.length, + appliedOperations: applied?.map((operation) => operation.op) ?? [], + }) +} + +function collectCitationDocumentIds(citations: unknown): string[] { + if (!Array.isArray(citations)) return [] + const ids = new Set() + for (const citation of citations) { + if (!citation || typeof citation !== "object") continue + const source = (citation as Record).source + if (!source || typeof source !== "object") continue + const documentId = (source as Record).documentId + if (typeof documentId === "string" && documentId.length > 0) { + ids.add(documentId) + } + } + return [...ids] +} + +function getNonEmptyString(value: unknown): string | null { + return typeof value === "string" && value.trim().length > 0 + ? value + : null +} diff --git a/src/domains/memory/extraction-model.ts b/src/domains/memory/extraction-model.ts new file mode 100644 index 0000000..fa49011 --- /dev/null +++ b/src/domains/memory/extraction-model.ts @@ -0,0 +1,51 @@ +import "server-only" + +import { generateObject } from "ai" + +import { + buildMemoryExtractionPrompt, + memoryOperationsSchema, + type ExistingMemoryContextItem, + type MemoryOperations, +} from "./prompts" +import { CHAT_MODEL } from "@/lib/ai" +import { summarizeUnknownError } from "@/lib/format-log-value" +import { logger } from "@/lib/logger" + +const MEMORY_EXTRACTION_MODEL = process.env.MEMORY_EXTRACTION_MODEL ?? CHAT_MODEL + +/** + * One structured-output call: turn + existing active memories in, typed + * operations out. Best-effort by design — this runs as a background job, + * so a model failure skips the turn (logged) instead of degrading through + * fallbacks; the insight typically resurfaces in a later turn. + */ +export async function extractMemoryOperations(input: { + readonly workspaceId: string + readonly userText: string + readonly assistantText: string + readonly referencedDocumentIds: readonly string[] + readonly existingItems: readonly ExistingMemoryContextItem[] +}): Promise { + try { + const response = await generateObject({ + model: MEMORY_EXTRACTION_MODEL, + schema: memoryOperationsSchema, + messages: [ + { + role: "user", + content: buildMemoryExtractionPrompt(input), + }, + ], + }) + return response.object + } catch (error) { + logger.warn("memory: extraction model call failed; skipping turn", { + workspaceId: input.workspaceId, + model: MEMORY_EXTRACTION_MODEL, + existingItemCount: input.existingItems.length, + error: summarizeUnknownError(error), + }) + return null + } +} diff --git a/src/domains/memory/prompts.test.ts b/src/domains/memory/prompts.test.ts new file mode 100644 index 0000000..2ae15ff --- /dev/null +++ b/src/domains/memory/prompts.test.ts @@ -0,0 +1,35 @@ +import { describe, expect, it } from "vitest" + +import { buildMemoryExtractionPrompt } from "./prompts" + +describe("buildMemoryExtractionPrompt", () => { + const prompt = buildMemoryExtractionPrompt({ + userText: "毛利率是核心。", + assistantText: "明白。", + referencedDocumentIds: ["doc-1"], + existingItems: [], + }) + + it("keeps main instructions domain-agnostic", () => { + const main = prompt.slice(0, prompt.indexOf("## Illustrative examples")) + expect(main).not.toMatch(/gross margin|市盈率|\bPE\b|Fed|美联储|NVIDIA|英伟达/i) + expect(main).toContain("Write every free-text value") + expect(main).toMatch(/same language the\s+USER wrote/) + }) + + it("keeps illustrative examples in a separate section", () => { + expect(prompt).toContain("## Illustrative examples (finance vertical") + expect(prompt).toContain("not exhaustive, not required vocabulary") + const examples = prompt.slice( + prompt.indexOf("## Illustrative examples"), + prompt.indexOf("## Output JSON schema"), + ) + expect(examples).toContain("finance vertical") + expect(examples).toContain("do not force the conversation into this domain") + }) + + it("still injects turn context after the fixed blocks", () => { + expect(prompt).toContain("[user]\n毛利率是核心。") + expect(prompt).toContain("doc-1") + }) +}) diff --git a/src/domains/memory/prompts.ts b/src/domains/memory/prompts.ts new file mode 100644 index 0000000..e2fbeb0 --- /dev/null +++ b/src/domains/memory/prompts.ts @@ -0,0 +1,285 @@ +import { z } from "zod" + +import { + decisionRulePayloadSchema, + entityOfInterestPayloadSchema, + indicatorPreferencePayloadSchema, + stancePayloadSchema, + type FluidMemoryKind, +} from "./types" + +/** + * LLM contract for fluid-memory extraction. + * + * Design borrowed from OpenViking's session-commit extraction (schema-driven + * typed operations, prefetch-then-decide), reimplemented as a single + * structured-output call: the model sees the turn plus lexically retrieved + * dedup candidates and directly outputs per-kind operations + * (create / skip / merge / deprecate), mirroring OpenViking's generated + * operations model without its ReAct tool loop. + */ + +const decisionSchema = z.object({ + op: z.enum(["create", "skip", "merge", "deprecate"]), + targetItemId: z.preprocess( + (value) => (value === null ? undefined : value), + z + .string() + .optional() + .describe( + "Required for merge/deprecate: the id of the existing memory item this operation targets. Omit for create/skip.", + ), + ), + reason: z.preprocess( + (value) => (value === null ? undefined : value), + z + .string() + .optional() + .describe("Short justification, especially for skip/merge/deprecate."), + ), +}) + +const memorySidecarFields = { + abstractL0: z + .string() + .min(1) + .describe("One line, <= 30 words: the essence of this insight."), + overviewL1: z + .string() + .min(1) + .describe("2-3 sentences: what it means and when it applies."), + confidence: z + .number() + .min(0) + .max(1) + .describe("How explicitly the user stated this (1 = explicit)."), + decision: decisionSchema, +} + +const stanceEntrySchema = z.preprocess((value) => { + if (!value || typeof value !== "object" || Array.isArray(value)) return value + const record = value as Record + // Models sometimes emit "name" for a stance; the contract field is "statement". + if ( + (typeof record.statement !== "string" || record.statement.length === 0) && + typeof record.name === "string" && + record.name.length > 0 + ) { + const { name, ...rest } = record + return { ...rest, statement: name } + } + return value +}, stancePayloadSchema.extend(memorySidecarFields)) + +const entityEntrySchema = z.preprocess((value) => { + if (!value || typeof value !== "object" || Array.isArray(value)) return value + const record = value as Record + // Keep provenance reason required; if the model omitted it, fall back to L0. + if ( + (typeof record.reason !== "string" || record.reason.length === 0) && + typeof record.abstractL0 === "string" && + record.abstractL0.length > 0 + ) { + return { ...record, reason: record.abstractL0 } + } + return value +}, entityOfInterestPayloadSchema.extend(memorySidecarFields)) + +export const memoryOperationsSchema = z.object({ + indicatorPrefs: z + .array(indicatorPreferencePayloadSchema.extend(memorySidecarFields)) + .default([]), + stances: z.array(stanceEntrySchema).default([]), + decisionRules: z + .array(decisionRulePayloadSchema.extend(memorySidecarFields)) + .default([]), + entities: z.array(entityEntrySchema).default([]), +}) + +export type MemoryOperations = z.infer + +export type ExistingMemoryContextItem = { + readonly id: string + readonly kind: FluidMemoryKind + readonly abstractL0: string + readonly payloadSummary: string +} + +/** Structural output shape only — no domain content. */ +const OUTPUT_SCHEMA_BLOCK = `{ + "indicatorPrefs": [{ + "name": "string", + "aliases": ["string"], + "definition": "string", + "polarity": "higher_better|lower_better|context", + "importance": "core|secondary", + "formulaHint": "string (optional — omit if none)", + "abstractL0": "string", + "overviewL1": "string", + "confidence": 0.0, + "decision": { "op": "create|skip|merge|deprecate", "targetItemId": "id for merge/deprecate only", "reason": "string optional" } + }], + "stances": [{ + "statement": "string (the stance text; do not use a name field)", + "scope": "string", + "rationale": "string", + "abstractL0": "string", + "overviewL1": "string", + "confidence": 0.0, + "decision": { "op": "create|skip|merge|deprecate", "targetItemId": "id for merge/deprecate only", "reason": "string optional" } + }], + "decisionRules": [{ + "when": "string", + "then": "string", + "priority": "high|medium|low", + "rationale": "string", + "abstractL0": "string", + "overviewL1": "string", + "confidence": 0.0, + "decision": { "op": "create|skip|merge|deprecate", "targetItemId": "id for merge/deprecate only", "reason": "string optional" } + }], + "entities": [{ + "name": "string", + "ticker": "string optional", + "aliases": ["string"], + "knowhereDocumentIds": ["only ids listed under REFERENCED DOCUMENT IDS"], + "reason": "string", + "abstractL0": "string", + "overviewL1": "string", + "confidence": 0.0, + "decision": { "op": "create|skip|merge|deprecate", "targetItemId": "id for merge/deprecate only", "reason": "string optional" } + }] +}` + +/** + * Illustrative only — kept separate from the main instructions so the model + * does not treat these domain phrases as required vocabulary. + * Finance is the first vertical; add other industry blocks here later if needed. + */ +const ILLUSTRATIVE_EXAMPLES_BLOCK = `## Illustrative examples (finance vertical — not exhaustive, not required vocabulary) + +These show shape and judgement only. Extract whatever the user actually said; +do not force the conversation into this domain or these metric names. + +- indicatorPref: user says a named metric they repeatedly use to judge quality + (shape: name + short definition + polarity + importance). Same idea applies + outside finance (any recurring evaluation metric). +- stance: user states a durable judgement frame that changes how evidence is + weighted (e.g. long-horizon vs short-horizon). +- decisionRule: user states a reusable when → then discipline over their metrics. +- entity: user says they actively track a named company/issuer and why. +- skip: a one-off factual question about a page/number in a document, small talk, + or an assistant suggestion the user did not endorse.` + +/** Domain-agnostic extraction instructions. */ +const MAIN_INSTRUCTIONS_BLOCK = `You maintain a user's FLUID MEMORY: durable insights about how this user thinks, extracted from their conversation with an AI analyst. + +Document facts live elsewhere (crystal memory). Never extract document facts, retrieved numbers, or page content as fluid memory. + +## What to extract + +Extract ONLY these four kinds, and ONLY when the turn gives real evidence from the USER: + +- indicatorPrefs — recurring metrics or criteria the user uses to evaluate things. + Fields: name, aliases, definition, polarity (higher_better | lower_better | context), + importance (core | secondary), optional formulaHint. +- stances — durable positions that shape how the user weighs evidence. + Fields: statement (required; do not invent a "name" field), scope, rationale. +- decisionRules — reusable when → then disciplines the user stated or clearly endorsed. + Fields: when, then, priority (high | medium | low), rationale. +- entities — named subjects the user is actively tracking. + Fields: name, optional ticker, aliases, reason (required), knowhereDocumentIds + (only from REFERENCED DOCUMENT IDS below; never invent ids). + +## Language + +- Keep this instruction set and enum/field names in English. +- Write every free-text value (name, definition, statement, when/then, reason, + abstractL0, overviewL1, aliases the user used, etc.) in the same language the + USER wrote in this turn. Do not translate the user's terms into English unless + the user themselves used English. + +## Decision rules + +- Extract only durable, reusable insights about the USER. +- Skip one-off questions, document facts, small talk, and assistant claims the user did not endorse. +- For every candidate, choose exactly one op against EXISTING MEMORIES: + - create — genuinely new + - skip — already covered, or too weak/ephemeral + - merge — same insight refined; emit the full merged fields and set targetItemId + - deprecate — user explicitly reversed a stored item; set targetItemId +- Be conservative: prefer create over merge when overlap is only partial; deprecate only on clear contradiction. +- Prefer one record per insight. If a preference already encodes how a metric should be read, do not also invent a near-duplicate decisionRule unless the user stated an explicit when → then action. +- abstractL0: one line. overviewL1: 2–3 sentences. confidence=1 only when the user stated it explicitly. +- Omit optional fields instead of setting them to null. +- If nothing is worth remembering, return all four arrays empty.` + +export function buildMemoryExtractionPrompt(input: { + readonly userText: string + readonly assistantText: string + readonly referencedDocumentIds: readonly string[] + readonly existingItems: readonly ExistingMemoryContextItem[] +}): string { + const existingBlock = + input.existingItems.length === 0 + ? "(no existing memories yet)" + : input.existingItems + .map( + (item) => + `- [${item.kind}] id=${item.id} :: ${item.abstractL0} :: ${item.payloadSummary}`, + ) + .join("\n") + + const documentsBlock = + input.referencedDocumentIds.length === 0 + ? "(no documents referenced in this turn)" + : input.referencedDocumentIds.join(", ") + + return `${MAIN_INSTRUCTIONS_BLOCK} + +${ILLUSTRATIVE_EXAMPLES_BLOCK} + +## Output JSON schema (follow exactly; do not invent fields) + +${OUTPUT_SCHEMA_BLOCK} + +## EXISTING MEMORIES + +${existingBlock} + +## REFERENCED DOCUMENT IDS + +${documentsBlock} + +## CONVERSATION TURN + +[user] +${input.userText} + +[assistant] +${input.assistantText}` +} + +export function summarizePayloadForContext( + kind: FluidMemoryKind, + payload: unknown, +): string { + if (!payload || typeof payload !== "object") return "" + const record = payload as Record + switch (kind) { + case "indicator_pref": + return [record.name, record.definition] + .filter((part) => typeof part === "string" && part.length > 0) + .join(" — ") + case "stance": + return typeof record.statement === "string" ? record.statement : "" + case "decision_rule": + return [record.when, record.then] + .filter((part) => typeof part === "string" && part.length > 0) + .join(" => ") + case "entity_of_interest": + return [record.name, record.ticker] + .filter((part) => typeof part === "string" && part.length > 0) + .join(" ") + } +} diff --git a/src/domains/memory/repository.ts b/src/domains/memory/repository.ts new file mode 100644 index 0000000..f7f6b79 --- /dev/null +++ b/src/domains/memory/repository.ts @@ -0,0 +1,254 @@ +import "server-only" + +import { and, eq, inArray, sql } from "drizzle-orm" +import { Effect } from "effect" + +import { toDiffOperation, type ResolvedMemoryOperation } from "./resolve-operations" +import { buildMemoryItemTokens } from "./search-index" +import type { + FluidMemoryKind, + FluidMemoryPayload, + MemoryDiffOperation, +} from "./types" +import { DbClient } from "@/infrastructure/db" +import { + fluidMemoryItems, + fluidMemoryTokens, + memoryDiffs, + type FluidMemoryItem, + type NewFluidMemoryToken, +} from "@/infrastructure/db/schema" + +type MemoryRepository = { + readonly findDedupCandidatesEffect: ( + workspaceId: string, + kind: FluidMemoryKind, + tokens: readonly string[], + limit: number, + ) => Effect.Effect + readonly applyOperationsEffect: ( + workspaceId: string, + sourceMessageId: string | null, + operations: readonly ResolvedMemoryOperation[], + ) => Effect.Effect +} + +type RawRowsResult = readonly Row[] | { readonly rows: readonly Row[] } + +/** + * Retrieve the most lexically-similar active items of one kind, ranked by + * idf-weighted token overlap computed entirely in SQL. Common tokens (high + * document frequency within this workspace + kind) are down-weighted so a + * shared rare term outranks several shared filler characters. + * + * Token rows only exist for active items (see schema invariant), so no + * status filter is needed here. + */ +const findDedupCandidatesEffect: MemoryRepository["findDedupCandidatesEffect"] = + (workspaceId, kind, tokens, limit) => + Effect.gen(function* () { + const db = yield* DbClient + if (tokens.length === 0 || limit <= 0) return [] + + const tokenList = sql.join( + tokens.map((token) => sql`${token}`), + sql`, `, + ) + + const scored = yield* Effect.promise(() => + db.execute<{ itemId: string }>(sql` + SELECT t.item_id AS "itemId", SUM(t.frequency::float8 / df.df) AS score + FROM fluid_memory_tokens t + JOIN ( + SELECT token, COUNT(DISTINCT item_id)::float8 AS df + FROM fluid_memory_tokens + WHERE workspace_id = ${workspaceId}::uuid + AND kind = ${kind} + AND token IN (${tokenList}) + GROUP BY token + ) df ON df.token = t.token + WHERE t.workspace_id = ${workspaceId}::uuid + AND t.kind = ${kind} + AND t.token IN (${tokenList}) + GROUP BY t.item_id + ORDER BY score DESC + LIMIT ${limit} + `), + ) + + const orderedIds = getRawRows(scored).map((row) => row.itemId) + if (orderedIds.length === 0) return [] + + const items = yield* Effect.promise(() => + db + .select() + .from(fluidMemoryItems) + .where(inArray(fluidMemoryItems.id, orderedIds)), + ) + const byId = new Map(items.map((item) => [item.id, item] as const)) + return orderedIds.flatMap((id) => { + const item = byId.get(id) + return item ? [item] : [] + }) + }) + +const applyOperationsEffect: MemoryRepository["applyOperationsEffect"] = ( + workspaceId, + sourceMessageId, + operations, +) => + Effect.gen(function* () { + const db = yield* DbClient + return yield* Effect.promise(() => + db.transaction(async (tx) => { + const diffOperations: MemoryDiffOperation[] = [] + + for (const operation of operations) { + switch (operation.op) { + case "create": { + const [inserted] = await tx + .insert(fluidMemoryItems) + .values({ + workspaceId, + kind: operation.kind, + payload: operation.payload, + abstractL0: operation.abstractL0, + overviewL1: operation.overviewL1, + sourceMessageId, + confidence: operation.confidence, + status: "active", + }) + .returning() + if (inserted?.id) { + const tokenRows = tokenRowsFor( + workspaceId, + inserted.id, + operation.kind, + operation.payload, + ) + if (tokenRows.length > 0) { + await tx.insert(fluidMemoryTokens).values(tokenRows) + } + } + diffOperations.push(toDiffOperation(operation, inserted?.id)) + break + } + case "merge": { + const [updated] = await tx + .update(fluidMemoryItems) + .set({ + payload: operation.payload, + abstractL0: operation.abstractL0, + overviewL1: operation.overviewL1, + confidence: operation.confidence, + sourceMessageId, + version: sql`${fluidMemoryItems.version} + 1`, + updatedAt: sql`now()`, + }) + .where( + and( + eq(fluidMemoryItems.id, operation.targetItemId), + eq(fluidMemoryItems.status, "active"), + ), + ) + .returning({ id: fluidMemoryItems.id }) + if (updated) { + await tx + .delete(fluidMemoryTokens) + .where(eq(fluidMemoryTokens.itemId, operation.targetItemId)) + const tokenRows = tokenRowsFor( + workspaceId, + operation.targetItemId, + operation.kind, + operation.payload, + ) + if (tokenRows.length > 0) { + await tx.insert(fluidMemoryTokens).values(tokenRows) + } + } + diffOperations.push( + updated + ? toDiffOperation(operation) + : { + op: "skip", + kind: operation.kind, + summary: operation.summary, + reason: "merge target no longer active", + }, + ) + break + } + case "deprecate": { + const [updated] = await tx + .update(fluidMemoryItems) + .set({ + status: "deprecated", + updatedAt: sql`now()`, + }) + .where( + and( + eq(fluidMemoryItems.id, operation.targetItemId), + eq(fluidMemoryItems.status, "active"), + ), + ) + .returning({ id: fluidMemoryItems.id }) + if (updated) { + await tx + .delete(fluidMemoryTokens) + .where(eq(fluidMemoryTokens.itemId, operation.targetItemId)) + } + diffOperations.push( + updated + ? toDiffOperation(operation) + : { + op: "skip", + kind: operation.kind, + summary: operation.summary, + reason: "deprecate target no longer active", + }, + ) + break + } + case "skip": + diffOperations.push(toDiffOperation(operation)) + break + } + } + + if (diffOperations.length > 0) { + await tx.insert(memoryDiffs).values({ + workspaceId, + sourceMessageId, + operations: [...diffOperations], + }) + } + + return diffOperations + }), + ) + }) + +export const memoryRepository: MemoryRepository = { + findDedupCandidatesEffect, + applyOperationsEffect, +} + +function tokenRowsFor( + workspaceId: string, + itemId: string, + kind: FluidMemoryKind, + payload: FluidMemoryPayload, +): NewFluidMemoryToken[] { + return buildMemoryItemTokens(kind, payload).map((token) => ({ + workspaceId, + itemId, + kind, + token: token.token, + frequency: token.frequency, + })) +} + +function getRawRows(value: RawRowsResult): readonly Row[] { + if (Array.isArray(value)) return value + return (value as { readonly rows: readonly Row[] }).rows +} diff --git a/src/domains/memory/resolve-operations.test.ts b/src/domains/memory/resolve-operations.test.ts new file mode 100644 index 0000000..9764ccf --- /dev/null +++ b/src/domains/memory/resolve-operations.test.ts @@ -0,0 +1,378 @@ +import { describe, expect, it } from "vitest" + +import type { MemoryOperations } from "./prompts" +import { + resolveMemoryOperations, + toDiffOperation, +} from "./resolve-operations" + +const existingItems = [ + { + id: "item-1", + kind: "indicator_pref", + status: "active", + payload: { + name: "毛利率", + aliases: ["gross margin"], + definition: "毛利占营收的比例", + polarity: "higher_better", + importance: "core", + }, + }, + { id: "item-2", kind: "stance", status: "active" }, + { id: "item-3", kind: "stance", status: "deprecated" }, + { + id: "item-4", + kind: "entity_of_interest", + status: "active", + payload: { + name: "英伟达", + ticker: "NVDA", + aliases: ["NVIDIA"], + knowhereDocumentIds: ["doc-earlier"], + reason: "用户持续跟踪", + }, + }, +] as const + +function makeOperations( + overrides: Partial = {}, +): MemoryOperations { + return { + indicatorPrefs: [], + stances: [], + decisionRules: [], + entities: [], + ...overrides, + } +} + +function makeIndicatorEntry(decision: { + op: "create" | "skip" | "merge" | "deprecate" + targetItemId?: string + reason?: string +}) { + return { + name: "毛利率", + aliases: ["gross margin"], + definition: "毛利占营收的比例", + polarity: "higher_better" as const, + importance: "core" as const, + abstractL0: "用户看重毛利率", + overviewL1: "用户在分析公司时首先看毛利率。", + confidence: 0.9, + decision, + } +} + +describe("resolveMemoryOperations", () => { + it("passes create through and ignores a stray targetItemId", () => { + const resolved = resolveMemoryOperations({ + operations: makeOperations({ + indicatorPrefs: [ + makeIndicatorEntry({ op: "create", targetItemId: "item-1" }), + ], + }), + existingItems, + referencedDocumentIds: [], + }) + + expect(resolved).toHaveLength(1) + expect(resolved[0]).toMatchObject({ + op: "create", + kind: "indicator_pref", + payload: { + name: "毛利率", + aliases: ["gross margin"], + polarity: "higher_better", + }, + }) + expect(resolved[0]).not.toHaveProperty("targetItemId") + }) + + it("merges into an existing active item of the same kind", () => { + const resolved = resolveMemoryOperations({ + operations: makeOperations({ + indicatorPrefs: [ + makeIndicatorEntry({ op: "merge", targetItemId: "item-1" }), + ], + }), + existingItems, + referencedDocumentIds: [], + }) + + expect(resolved[0]).toMatchObject({ + op: "merge", + kind: "indicator_pref", + targetItemId: "item-1", + }) + }) + + it("downgrades merge to skip when the target is missing", () => { + const resolved = resolveMemoryOperations({ + operations: makeOperations({ + indicatorPrefs: [ + makeIndicatorEntry({ op: "merge", targetItemId: "item-999" }), + ], + }), + existingItems, + referencedDocumentIds: [], + }) + + expect(resolved[0]).toMatchObject({ + op: "skip", + kind: "indicator_pref", + }) + expect(resolved[0]).not.toHaveProperty("targetItemId") + }) + + it("downgrades merge to skip on kind mismatch", () => { + const resolved = resolveMemoryOperations({ + operations: makeOperations({ + indicatorPrefs: [ + makeIndicatorEntry({ op: "merge", targetItemId: "item-2" }), + ], + }), + existingItems, + referencedDocumentIds: [], + }) + + expect(resolved[0]?.op).toBe("skip") + }) + + it("downgrades merge to skip when the target is already deprecated", () => { + const resolved = resolveMemoryOperations({ + operations: makeOperations({ + stances: [ + { + statement: "长期持有，忽略短期波动", + scope: "投资", + rationale: "用户做长期投资", + abstractL0: "长期投资立场", + overviewL1: "用户强调长期持有。", + confidence: 1, + decision: { op: "merge", targetItemId: "item-3" }, + }, + ], + }), + existingItems, + referencedDocumentIds: [], + }) + + expect(resolved[0]?.op).toBe("skip") + }) + + it("keeps deprecate for an active target", () => { + const resolved = resolveMemoryOperations({ + operations: makeOperations({ + stances: [ + { + statement: "美联储短期观点不重要", + scope: "宏观", + rationale: "长期投资", + abstractL0: "不看美联储短期观点", + overviewL1: "用户认为美联储短期观点权重低。", + confidence: 1, + decision: { + op: "deprecate", + targetItemId: "item-2", + reason: "用户改口", + }, + }, + ], + }), + existingItems, + referencedDocumentIds: [], + }) + + expect(resolved[0]).toMatchObject({ + op: "deprecate", + targetItemId: "item-2", + reason: "用户改口", + }) + }) + + it("filters entity document ids to the turn's referenced documents", () => { + const resolved = resolveMemoryOperations({ + operations: makeOperations({ + entities: [ + { + name: "英伟达", + ticker: "NVDA", + aliases: ["NVIDIA"], + knowhereDocumentIds: ["doc-real", "doc-hallucinated"], + reason: "用户持续跟踪", + abstractL0: "用户关注英伟达", + overviewL1: "用户多次询问英伟达财报。", + confidence: 0.8, + decision: { op: "create" }, + }, + ], + }), + existingItems, + referencedDocumentIds: ["doc-real"], + }) + + expect(resolved[0]).toMatchObject({ + op: "create", + kind: "entity_of_interest", + payload: { + name: "英伟达", + knowhereDocumentIds: ["doc-real"], + }, + }) + }) + + it("unions stored document ids when merging an entity", () => { + const resolved = resolveMemoryOperations({ + operations: makeOperations({ + entities: [ + { + name: "英伟达", + ticker: "NVDA", + aliases: ["NVDA Corp"], + knowhereDocumentIds: ["doc-this-turn"], + reason: "用户持续跟踪", + abstractL0: "用户关注英伟达", + overviewL1: "用户多次询问英伟达财报。", + confidence: 0.9, + decision: { op: "merge", targetItemId: "item-4" }, + }, + ], + }), + existingItems, + referencedDocumentIds: ["doc-this-turn"], + }) + + expect(resolved[0]).toMatchObject({ + op: "merge", + targetItemId: "item-4", + payload: { + aliases: ["NVIDIA", "NVDA Corp"], + knowhereDocumentIds: ["doc-earlier", "doc-this-turn"], + }, + }) + }) + + it("unions aliases when merging an indicator preference", () => { + const resolved = resolveMemoryOperations({ + operations: makeOperations({ + indicatorPrefs: [ + { + name: "毛利率", + aliases: ["同比毛利率"], + definition: "毛利占营收的比例，也看同比", + polarity: "higher_better" as const, + importance: "core" as const, + abstractL0: "毛利率也看同比", + overviewL1: "用户补充了同比视角。", + confidence: 1, + decision: { op: "merge", targetItemId: "item-1" }, + }, + ], + }), + existingItems, + referencedDocumentIds: [], + }) + + expect(resolved[0]).toMatchObject({ + op: "merge", + targetItemId: "item-1", + payload: { + aliases: ["gross margin", "同比毛利率"], + }, + }) + }) + + it("skips create when the payload has no searchable tokens", () => { + const resolved = resolveMemoryOperations({ + operations: makeOperations({ + indicatorPrefs: [ + { + name: "!!!", + aliases: [], + definition: "???", + polarity: "context" as const, + importance: "secondary" as const, + abstractL0: "无效符号", + overviewL1: "无法检索。", + confidence: 0.1, + decision: { op: "create" }, + }, + ], + }), + existingItems, + referencedDocumentIds: [], + }) + + expect(resolved[0]).toMatchObject({ + op: "skip", + kind: "indicator_pref", + reason: "payload has no searchable tokens", + }) + }) + + it("skips merge when the merged payload has no searchable tokens", () => { + const resolved = resolveMemoryOperations({ + operations: makeOperations({ + stances: [ + { + statement: "!!!", + scope: "???", + rationale: "...", + abstractL0: "无效符号", + overviewL1: "无法检索。", + confidence: 0.1, + decision: { op: "merge", targetItemId: "item-2" }, + }, + ], + }), + existingItems, + referencedDocumentIds: [], + }) + + expect(resolved[0]).toMatchObject({ + op: "skip", + kind: "stance", + reason: "merged payload has no searchable tokens", + }) + }) +}) + +describe("toDiffOperation", () => { + it("records create with the inserted item id", () => { + const resolved = resolveMemoryOperations({ + operations: makeOperations({ + indicatorPrefs: [makeIndicatorEntry({ op: "create" })], + }), + existingItems, + referencedDocumentIds: [], + }) + + expect(toDiffOperation(resolved[0]!, "new-id")).toEqual({ + op: "create", + kind: "indicator_pref", + summary: "用户看重毛利率", + itemId: "new-id", + }) + }) + + it("records merge/deprecate with their target item id", () => { + const resolved = resolveMemoryOperations({ + operations: makeOperations({ + indicatorPrefs: [ + makeIndicatorEntry({ op: "merge", targetItemId: "item-1" }), + ], + }), + existingItems, + referencedDocumentIds: [], + }) + + expect(toDiffOperation(resolved[0]!)).toEqual({ + op: "merge", + kind: "indicator_pref", + summary: "用户看重毛利率", + itemId: "item-1", + }) + }) +}) diff --git a/src/domains/memory/resolve-operations.ts b/src/domains/memory/resolve-operations.ts new file mode 100644 index 0000000..ee1e6a4 --- /dev/null +++ b/src/domains/memory/resolve-operations.ts @@ -0,0 +1,314 @@ +import { buildMemoryItemTokens } from "./search-index" +import type { MemoryOperations } from "./prompts" +import { + parseFluidMemoryPayload, + type FluidMemoryKind, + type FluidMemoryPayload, + type MemoryDiffOperation, +} from "./types" + +/** + * Pure normalization from raw LLM operations to repository-ready + * operations. The LLM output already passed zod validation; this layer + * enforces the invariants the schema cannot express: + * - merge/deprecate must target an existing active item of the same kind + * (otherwise downgraded to skip — conservative, never fabricates) + * - entity knowhereDocumentIds are intersected with the document ids + * actually referenced in the turn (the model cannot invent provenance) + * - create ignores any targetItemId the model may have emitted + * - create/merge payloads must yield at least one lexical token, otherwise + * the item could never be retrieved for later dedup + * - merge unions aliases (and entity document ids) with the target so + * prior search terms / provenance are not wiped by a partial rewrite + */ + +export type ResolvedMemoryOperation = + | { + readonly op: "create" + readonly kind: FluidMemoryKind + readonly payload: FluidMemoryPayload + readonly abstractL0: string + readonly overviewL1: string + readonly confidence: number + readonly summary: string + readonly reason?: string + } + | { + readonly op: "skip" + readonly kind: FluidMemoryKind + readonly summary: string + readonly reason?: string + } + | { + readonly op: "merge" + readonly kind: FluidMemoryKind + readonly targetItemId: string + readonly payload: FluidMemoryPayload + readonly abstractL0: string + readonly overviewL1: string + readonly confidence: number + readonly summary: string + readonly reason?: string + } + | { + readonly op: "deprecate" + readonly kind: FluidMemoryKind + readonly targetItemId: string + readonly summary: string + readonly reason?: string + } + +export type ExistingMemoryItemRef = { + readonly id: string + readonly kind: string + readonly status: string + readonly payload?: unknown +} + +type CandidateEntry = { + readonly abstractL0: string + readonly overviewL1: string + readonly confidence: number + readonly decision: { + readonly op: "create" | "skip" | "merge" | "deprecate" + readonly targetItemId?: string + readonly reason?: string + } +} + +const kindToArrayKey = { + indicator_pref: "indicatorPrefs", + stance: "stances", + decision_rule: "decisionRules", + entity_of_interest: "entities", +} as const + +export function resolveMemoryOperations(input: { + readonly operations: MemoryOperations + readonly existingItems: readonly ExistingMemoryItemRef[] + readonly referencedDocumentIds: readonly string[] +}): ResolvedMemoryOperation[] { + const activeById = new Map( + input.existingItems + .filter((item) => item.status === "active") + .map((item) => [item.id, item] as const), + ) + const allowedDocumentIds = new Set(input.referencedDocumentIds) + + const resolved: ResolvedMemoryOperation[] = [] + + for (const kind of Object.keys(kindToArrayKey) as FluidMemoryKind[]) { + const entries = input.operations[kindToArrayKey[kind]] as readonly (CandidateEntry & + Record)[] + + for (const entry of entries) { + const summary = entry.abstractL0 + const reason = entry.decision.reason + + if (entry.decision.op === "skip") { + resolved.push({ op: "skip", kind, summary, ...(reason ? { reason } : {}) }) + continue + } + + if (entry.decision.op === "merge" || entry.decision.op === "deprecate") { + const targetId = entry.decision.targetItemId + const target = targetId ? activeById.get(targetId) : undefined + if (!target || target.kind !== kind) { + resolved.push({ + op: "skip", + kind, + summary, + reason: `${entry.decision.op} target missing, inactive, or kind mismatch`, + }) + continue + } + if (entry.decision.op === "deprecate") { + resolved.push({ + op: "deprecate", + kind, + targetItemId: target.id, + summary, + ...(reason ? { reason } : {}), + }) + continue + } + const mergePayload = toPayload(kind, entry, allowedDocumentIds) + if (!mergePayload) { + resolved.push({ + op: "skip", + kind, + summary, + reason: "merged payload failed validation", + }) + continue + } + const preserved = preserveFieldsOnMerge( + kind, + mergePayload, + target.payload, + ) + if (!isIndexable(kind, preserved)) { + resolved.push({ + op: "skip", + kind, + summary, + reason: "merged payload has no searchable tokens", + }) + continue + } + resolved.push({ + op: "merge", + kind, + targetItemId: target.id, + payload: preserved, + abstractL0: entry.abstractL0, + overviewL1: entry.overviewL1, + confidence: entry.confidence, + summary, + ...(reason ? { reason } : {}), + }) + continue + } + + const createPayload = toPayload(kind, entry, allowedDocumentIds) + if (!createPayload) { + resolved.push({ + op: "skip", + kind, + summary, + reason: "payload failed validation", + }) + continue + } + if (!isIndexable(kind, createPayload)) { + resolved.push({ + op: "skip", + kind, + summary, + reason: "payload has no searchable tokens", + }) + continue + } + resolved.push({ + op: "create", + kind, + payload: createPayload, + abstractL0: entry.abstractL0, + overviewL1: entry.overviewL1, + confidence: entry.confidence, + summary, + ...(reason ? { reason } : {}), + }) + } + } + + return resolved +} + +function toPayload( + kind: FluidMemoryKind, + entry: Record, + allowedDocumentIds: ReadonlySet, +): FluidMemoryPayload | null { + const candidate: Record = { + name: entry.name, + aliases: entry.aliases, + definition: entry.definition, + polarity: entry.polarity, + importance: entry.importance, + formulaHint: entry.formulaHint, + statement: entry.statement, + scope: entry.scope, + rationale: entry.rationale, + when: entry.when, + then: entry.then, + priority: entry.priority, + ticker: entry.ticker, + reason: entry.reason, + knowhereDocumentIds: Array.isArray(entry.knowhereDocumentIds) + ? entry.knowhereDocumentIds.filter( + (id): id is string => + typeof id === "string" && allowedDocumentIds.has(id), + ) + : [], + } + return parseFluidMemoryPayload(kind, candidate) +} + +/** Reject payloads that could never be found again by the token index. */ +function isIndexable(kind: FluidMemoryKind, payload: FluidMemoryPayload): boolean { + return buildMemoryItemTokens(kind, payload).length > 0 +} + +/** + * Merge replaces the stored payload, but the model only sees this turn. + * Union aliases (and entity document ids) with the target so earlier search + * terms / provenance survive a partial rewrite. + */ +function preserveFieldsOnMerge( + kind: FluidMemoryKind, + mergedPayload: FluidMemoryPayload, + existingPayload: unknown, +): FluidMemoryPayload { + const existing = parseFluidMemoryPayload(kind, existingPayload) + if (!existing) return mergedPayload + + if ( + kind === "indicator_pref" && + "aliases" in mergedPayload && + "aliases" in existing + ) { + return { + ...mergedPayload, + aliases: unionStrings(existing.aliases, mergedPayload.aliases), + } + } + + if ( + kind === "entity_of_interest" && + "aliases" in mergedPayload && + "aliases" in existing && + "knowhereDocumentIds" in mergedPayload && + "knowhereDocumentIds" in existing + ) { + return { + ...mergedPayload, + aliases: unionStrings(existing.aliases, mergedPayload.aliases), + knowhereDocumentIds: unionStrings( + existing.knowhereDocumentIds, + mergedPayload.knowhereDocumentIds, + ), + } + } + + return mergedPayload +} + +function unionStrings( + left: readonly string[], + right: readonly string[], +): string[] { + return [...new Set([...left, ...right])] +} + +/** Diff-audit view of a resolved operation (itemId filled after write). */ +export function toDiffOperation( + operation: ResolvedMemoryOperation, + itemId?: string, +): MemoryDiffOperation { + const base = { + kind: operation.kind, + summary: operation.summary, + ...(operation.reason ? { reason: operation.reason } : {}), + } + switch (operation.op) { + case "create": + return { op: "create", ...base, ...(itemId ? { itemId } : {}) } + case "merge": + return { op: "merge", ...base, itemId: operation.targetItemId } + case "deprecate": + return { op: "deprecate", ...base, itemId: operation.targetItemId } + case "skip": + return { op: "skip", ...base } + } +} diff --git a/src/domains/memory/search-index.test.ts b/src/domains/memory/search-index.test.ts new file mode 100644 index 0000000..cbf5d0b --- /dev/null +++ b/src/domains/memory/search-index.test.ts @@ -0,0 +1,94 @@ +import { describe, expect, it } from "vitest" + +import { + buildMemoryItemTokens, + buildMemorySearchText, + tokenizeMemoryText, +} from "./search-index" + +describe("tokenizeMemoryText", () => { + it("splits CJK into single characters and latin into words", () => { + expect(tokenizeMemoryText("毛利率 PE gross_margin")).toEqual([ + { token: "毛", frequency: 1 }, + { token: "利", frequency: 1 }, + { token: "率", frequency: 1 }, + { token: "pe", frequency: 1 }, + { token: "gross_margin", frequency: 1 }, + ]) + }) + + it("counts repeated tokens", () => { + expect(tokenizeMemoryText("PE pe Pe")).toEqual([ + { token: "pe", frequency: 3 }, + ]) + }) + + it("returns empty for whitespace-only input", () => { + expect(tokenizeMemoryText(" ")).toEqual([]) + }) +}) + +describe("buildMemorySearchText", () => { + it("indexes indicator name, aliases, and definition", () => { + expect( + buildMemorySearchText("indicator_pref", { + name: "毛利率", + aliases: ["gross margin"], + definition: "毛利除以营收", + polarity: "higher_better", + importance: "core", + }), + ).toBe("毛利率 gross margin 毛利除以营收") + }) + + it("indexes entity name, aliases, and ticker without reason", () => { + expect( + buildMemorySearchText("entity_of_interest", { + name: "英伟达", + aliases: ["NVIDIA"], + ticker: "NVDA", + knowhereDocumentIds: ["doc-1"], + reason: "一直在跟踪", + }), + ).toBe("英伟达 NVIDIA NVDA") + }) + + it("indexes stance statement and scope", () => { + expect( + buildMemorySearchText("stance", { + statement: "做长期投资", + scope: "宏观短期观点", + rationale: "美联储短期说法不重要", + }), + ).toBe("做长期投资 宏观短期观点") + }) + + it("indexes decision rule when and then", () => { + expect( + buildMemorySearchText("decision_rule", { + when: "毛利率连续两季下滑", + then: "减仓观望", + priority: "high", + rationale: "用户明确说过", + }), + ).toBe("毛利率连续两季下滑 减仓观望") + }) +}) + +describe("buildMemoryItemTokens", () => { + it("tokenizes the search text of an item", () => { + const tokens = buildMemoryItemTokens("indicator_pref", { + name: "PE", + aliases: [], + definition: "市盈率", + polarity: "context", + importance: "secondary", + }) + expect(tokens.map((token) => token.token)).toEqual([ + "pe", + "市", + "盈", + "率", + ]) + }) +}) diff --git a/src/domains/memory/search-index.ts b/src/domains/memory/search-index.ts new file mode 100644 index 0000000..84a53e9 --- /dev/null +++ b/src/domains/memory/search-index.ts @@ -0,0 +1,84 @@ +import type { + DecisionRulePayload, + EntityOfInterestPayload, + FluidMemoryKind, + FluidMemoryPayload, + IndicatorPreferencePayload, + StancePayload, +} from "./types" + +/** + * Lexical search-index helpers for fluid memory dedup retrieval. + * + * Pure and dependency-free so both the write path (indexing an item) and the + * read path (turning a turn into a query) share one tokenizer. Tokenization + * mirrors Knowhere map-nav: lowercase, then emit single CJK characters and + * `[a-z0-9_]+` runs. This handles Chinese (no whitespace segmentation) and + * Latin/alphanumeric terms without any Postgres extension. + */ + +export type MemoryToken = { + readonly token: string + readonly frequency: number +} + +// Single CJK char OR a run of latin letters / digits / underscore. +const TOKEN_PATTERN = + /[a-z0-9_]+|[\u3400-\u4dbf\u4e00-\u9fff\uf900-\ufaff]/g + +/** + * Build the text that represents an item for lexical matching. Only the + * fields a user would phrase a query against are included (names, aliases, + * short definitions), not provenance or bookkeeping fields. + */ +export function buildMemorySearchText( + kind: FluidMemoryKind, + payload: FluidMemoryPayload, +): string { + return collectSearchParts(kind, payload) + .filter((part) => part.length > 0) + .join(" ") +} + +/** Tokenize free text into deduped tokens with occurrence counts. */ +export function tokenizeMemoryText(text: string): MemoryToken[] { + const counts = new Map() + const matches = text.toLowerCase().match(TOKEN_PATTERN) + if (!matches) return [] + for (const token of matches) { + counts.set(token, (counts.get(token) ?? 0) + 1) + } + return [...counts].map(([token, frequency]) => ({ token, frequency })) +} + +/** Tokens that index one memory item (search text of its payload). */ +export function buildMemoryItemTokens( + kind: FluidMemoryKind, + payload: FluidMemoryPayload, +): MemoryToken[] { + return tokenizeMemoryText(buildMemorySearchText(kind, payload)) +} + +function collectSearchParts( + kind: FluidMemoryKind, + payload: FluidMemoryPayload, +): readonly string[] { + switch (kind) { + case "indicator_pref": { + const p = payload as IndicatorPreferencePayload + return [p.name, ...p.aliases, p.definition] + } + case "stance": { + const p = payload as StancePayload + return [p.statement, p.scope] + } + case "decision_rule": { + const p = payload as DecisionRulePayload + return [p.when, p.then] + } + case "entity_of_interest": { + const p = payload as EntityOfInterestPayload + return [p.name, ...p.aliases, ...(p.ticker ? [p.ticker] : [])] + } + } +} diff --git a/src/domains/memory/service.ts b/src/domains/memory/service.ts new file mode 100644 index 0000000..da772f6 --- /dev/null +++ b/src/domains/memory/service.ts @@ -0,0 +1,54 @@ +import "server-only" + +import { databaseRuntime } from "@/domains/workspace/database-runtime" +import { memoryRepository } from "./repository" +import type { ResolvedMemoryOperation } from "./resolve-operations" +import type { FluidMemoryKind, MemoryDiffOperation } from "./types" +import type { FluidMemoryItem } from "@/infrastructure/db/schema" + +type MemoryService = { + readonly findDedupCandidates: ( + workspaceId: string, + kind: FluidMemoryKind, + tokens: readonly string[], + limit: number, + ) => Promise + readonly applyOperations: ( + workspaceId: string, + sourceMessageId: string | null, + operations: readonly ResolvedMemoryOperation[], + ) => Promise +} + +const findDedupCandidates: MemoryService["findDedupCandidates"] = ( + workspaceId, + kind, + tokens, + limit, +) => + databaseRuntime.runPromise( + memoryRepository.findDedupCandidatesEffect( + workspaceId, + kind, + tokens, + limit, + ), + ) + +const applyOperations: MemoryService["applyOperations"] = ( + workspaceId, + sourceMessageId, + operations, +) => + databaseRuntime.runPromise( + memoryRepository.applyOperationsEffect( + workspaceId, + sourceMessageId, + operations, + ), + ) + +export const memoryService: MemoryService = { + findDedupCandidates, + applyOperations, +} diff --git a/src/domains/memory/types.ts b/src/domains/memory/types.ts new file mode 100644 index 0000000..dac7602 --- /dev/null +++ b/src/domains/memory/types.ts @@ -0,0 +1,101 @@ +import { z } from "zod" + +/** + * Fluid memory type contract. + * + * Four typed payload kinds, extracted from conversation turns. The DB + * stores `payload` as jsonb; these schemas are the validation boundary on + * both write (LLM output) and read (repository decode) paths. + */ + +export const fluidMemoryKinds = [ + "indicator_pref", + "stance", + "decision_rule", + "entity_of_interest", +] as const + +export type FluidMemoryKind = (typeof fluidMemoryKinds)[number] + +export const indicatorPreferencePayloadSchema = z.object({ + name: z.string().min(1), + aliases: z.array(z.string()).default([]), + definition: z.string().min(1), + polarity: z.enum(["higher_better", "lower_better", "context"]), + importance: z.enum(["core", "secondary"]), + formulaHint: z.preprocess( + (value) => (value === null ? undefined : value), + z.string().optional(), + ), +}) + +export const stancePayloadSchema = z.object({ + statement: z.string().min(1), + scope: z.string().min(1), + rationale: z.string().min(1), +}) + +export const decisionRulePayloadSchema = z.object({ + when: z.string().min(1), + then: z.string().min(1), + priority: z.enum(["high", "medium", "low"]), + rationale: z.string().min(1), +}) + +export const entityOfInterestPayloadSchema = z.object({ + name: z.string().min(1), + ticker: z.preprocess( + (value) => (value === null ? undefined : value), + z.string().optional(), + ), + aliases: z.array(z.string()).default([]), + knowhereDocumentIds: z.array(z.string()).default([]), + reason: z.string().min(1), +}) + +export type IndicatorPreferencePayload = z.infer< + typeof indicatorPreferencePayloadSchema +> +export type StancePayload = z.infer +export type DecisionRulePayload = z.infer +export type EntityOfInterestPayload = z.infer< + typeof entityOfInterestPayloadSchema +> + +export type FluidMemoryPayload = + | IndicatorPreferencePayload + | StancePayload + | DecisionRulePayload + | EntityOfInterestPayload + +const payloadSchemas: Record> = { + indicator_pref: indicatorPreferencePayloadSchema, + stance: stancePayloadSchema, + decision_rule: decisionRulePayloadSchema, + entity_of_interest: entityOfInterestPayloadSchema, +} + +export function isFluidMemoryKind(value: unknown): value is FluidMemoryKind { + return ( + typeof value === "string" && + (fluidMemoryKinds as readonly string[]).includes(value) + ) +} + +/** Decode a persisted jsonb payload; returns null when the row is malformed. */ +export function parseFluidMemoryPayload( + kind: FluidMemoryKind, + value: unknown, +): FluidMemoryPayload | null { + const result = payloadSchemas[kind].safeParse(value) + return result.success ? result.data : null +} + +/** One decided operation over the memory set; persisted into memory_diffs. */ +export type MemoryDiffOperation = { + readonly op: "create" | "skip" | "merge" | "deprecate" + readonly kind: FluidMemoryKind + readonly itemId?: string + readonly summary: string + readonly reason?: string +} diff --git a/src/domains/sources/parsed-sync-route-workflow.test.ts b/src/domains/sources/parsed-sync-route-workflow.test.ts index 670ee67..9a2bcff 100644 --- a/src/domains/sources/parsed-sync-route-workflow.test.ts +++ b/src/domains/sources/parsed-sync-route-workflow.test.ts @@ -7,12 +7,24 @@ const mocks = vi.hoisted(() => ({ updateSyncStatus: vi.fn(), loggerInfo: vi.fn(), loggerError: vi.fn(), + ensureFreshKnowhereApiKey: vi.fn(async (apiKey: string) => apiKey), + withFreshKnowhereApiKey: vi.fn( + async (apiKey: string, run: (apiKey: string) => Promise) => ({ + result: await run(apiKey), + apiKey, + }), + ), })) vi.mock("@/integrations/knowhere", () => ({ makeKnowhereClientWithParsedStorage: mocks.makeKnowhereClientWithParsedStorage, })) +vi.mock("@/integrations/dashboard/api-key-service", () => ({ + ensureFreshKnowhereApiKey: mocks.ensureFreshKnowhereApiKey, + withFreshKnowhereApiKey: mocks.withFreshKnowhereApiKey, +})) + vi.mock("./workflow-runtime", () => ({ sourceWorkflowRuntime: { updateSyncStatus: mocks.updateSyncStatus, @@ -62,6 +74,13 @@ describe("parsedSyncRouteWorkflow.runParsedSyncWorkflow", () => { }, }) mocks.releaseSyncCapacity.mockResolvedValue(undefined) + mocks.ensureFreshKnowhereApiKey.mockImplementation(async (apiKey: string) => apiKey) + mocks.withFreshKnowhereApiKey.mockImplementation( + async (apiKey: string, run: (apiKey: string) => Promise) => ({ + result: await run(apiKey), + apiKey, + }), + ) }) afterEach(() => { @@ -107,12 +126,17 @@ describe("parsedSyncRouteWorkflow.runParsedSyncWorkflow", () => { client: {}, knowledge: { syncParsedDocument }, }) - const triggered: Array<{ workflowRunId: string; segmentIndex?: number }> = [] + const triggered: Array<{ + workflowRunId: string + segmentIndex?: number + apiKey?: string + }> = [] const restore = parsedSyncRouteWorkflow.setContinuationTriggerForTesting( async (input) => { triggered.push({ workflowRunId: input.workflowRunId, segmentIndex: input.payload.segmentIndex, + apiKey: input.payload.apiKey, }) }, ) @@ -129,6 +153,7 @@ describe("parsedSyncRouteWorkflow.runParsedSyncWorkflow", () => { expect(triggered).toHaveLength(1) expect(triggered[0]?.segmentIndex).toBe(1) expect(triggered[0]?.workflowRunId).toBe("doc_1-sync-rev_1-1") + expect(triggered[0]?.apiKey).toBe("key_1") expect(mocks.updateSyncStatus).toHaveBeenLastCalledWith( "workspace_1", "source_1", @@ -186,6 +211,7 @@ describe("parsedSyncRouteWorkflow.runParsedSyncWorkflow", () => { readonly workflowRunId: string readonly segmentIndex?: number readonly delaySeconds?: number + readonly apiKey?: string }> = [] const restore = parsedSyncRouteWorkflow.setContinuationTriggerForTesting( async (input) => { @@ -193,6 +219,7 @@ describe("parsedSyncRouteWorkflow.runParsedSyncWorkflow", () => { workflowRunId: input.workflowRunId, segmentIndex: input.payload.segmentIndex, delaySeconds: input.delaySeconds, + apiKey: input.payload.apiKey, }) }, ) @@ -221,11 +248,50 @@ describe("parsedSyncRouteWorkflow.runParsedSyncWorkflow", () => { workflowRunId: "doc_1-sync-rev_1-1", segmentIndex: 1, delaySeconds: 60, + apiKey: "key_1", }, ]) expect(mocks.releaseSyncCapacity).not.toHaveBeenCalled() }) + it("forwards a refreshed Knowhere JWT on sync continuation and capacity retry", async () => { + mocks.ensureFreshKnowhereApiKey.mockResolvedValue("jwt_refreshed") + mocks.withFreshKnowhereApiKey.mockImplementation( + async (_apiKey: string, run: (apiKey: string) => Promise) => ({ + result: await run("jwt_refreshed"), + apiKey: "jwt_refreshed", + }), + ) + const syncParsedDocument = vi.fn(async () => ({ + documentId: "doc_1", + revisionKey: "rev_1", + completed: false, + })) + mocks.makeKnowhereClientWithParsedStorage.mockReturnValue({ + client: {}, + knowledge: { syncParsedDocument }, + }) + const triggered: Array<{ apiKey?: string }> = [] + const restore = parsedSyncRouteWorkflow.setContinuationTriggerForTesting( + async (input) => { + triggered.push({ apiKey: input.payload.apiKey }) + }, + ) + + try { + await parsedSyncRouteWorkflow.runParsedSyncWorkflow({ + context: createContext(), + payload: basePayload, + }) + } finally { + restore() + } + + expect(mocks.ensureFreshKnowhereApiKey).toHaveBeenCalledWith("key_1") + expect(syncParsedDocument).toHaveBeenCalled() + expect(triggered[0]?.apiKey).toBe("jwt_refreshed") + }) + it("does not release capacity when Upstash aborts during a planned step", async () => { const workflowAbort = new Error("planned workflow step") workflowAbort.name = "WorkflowAbort" diff --git a/src/domains/sources/parsed-sync-route-workflow.ts b/src/domains/sources/parsed-sync-route-workflow.ts index e832b01..6634fb8 100644 --- a/src/domains/sources/parsed-sync-route-workflow.ts +++ b/src/domains/sources/parsed-sync-route-workflow.ts @@ -4,6 +4,10 @@ import { Client, WorkflowAbort, type WorkflowContext } from "@upstash/workflow" import type { KnowledgeSyncParsedDocumentResponse } from "@ontos-ai/knowhere-sdk" import { makeKnowhereClientWithParsedStorage } from "@/integrations/knowhere" +import { + ensureFreshKnowhereApiKey, + withFreshKnowhereApiKey, +} from "@/integrations/dashboard/api-key-service" import { logger } from "@/lib/logger" import { getParsedSyncWorkflowRunId, @@ -66,10 +70,11 @@ async function runParsedSyncWorkflow(input: { readonly payload: NormalizedParsedSyncPayload }): Promise { const { context, payload } = input - const { workspaceId, sourceId, documentId, apiKey } = payload - const { knowledge } = makeKnowhereClientWithParsedStorage(apiKey, { - workspaceId, - }) + const { workspaceId, sourceId, documentId } = payload + let apiKey = await context.run( + `refresh-knowhere-jwt-${payload.segmentIndex}`, + async () => ensureFreshKnowhereApiKey(payload.apiKey), + ) let revisionKey = payload.revisionKey let completed = false @@ -131,14 +136,22 @@ async function runParsedSyncWorkflow(input: { try { for (let step = 0; step < maxSyncStepsPerSegment; step++) { - const result: KnowledgeSyncParsedDocumentResponse = await context.run( + const stepResult = await context.run( `sync-${payload.segmentIndex}-${step}`, async () => - knowledge.syncParsedDocument({ - documentId, - ...(revisionKey ? { revisionKey } : {}), + withFreshKnowhereApiKey(apiKey, async (freshKey) => { + const { knowledge } = makeKnowhereClientWithParsedStorage( + freshKey, + { workspaceId }, + ) + return knowledge.syncParsedDocument({ + documentId, + ...(revisionKey ? { revisionKey } : {}), + }) }), ) + apiKey = stepResult.apiKey + const result: KnowledgeSyncParsedDocumentResponse = stepResult.result revisionKey = result.revisionKey await context.run(`record-progress-${payload.segmentIndex}-${step}`, () => diff --git a/src/domains/sources/source-reconcile-route-workflow.test.ts b/src/domains/sources/source-reconcile-route-workflow.test.ts index 730fcf0..1498fed 100644 --- a/src/domains/sources/source-reconcile-route-workflow.test.ts +++ b/src/domains/sources/source-reconcile-route-workflow.test.ts @@ -13,6 +13,12 @@ const mocks = vi.hoisted(() => ({ makeKnowhereClientWithParsedStorage: vi.fn(), markSourceReadyAfterReconciliation: vi.fn(), pollSourceReconciliation: vi.fn(), + withFreshKnowhereApiKey: vi.fn( + async (apiKey: string, run: (apiKey: string) => Promise) => ({ + result: await run(apiKey), + apiKey, + }), + ), })) vi.mock("@/domains/sources/source-reconcile-workflow", () => ({ @@ -44,6 +50,10 @@ vi.mock("@/integrations/knowhere", () => ({ mocks.makeKnowhereClientWithParsedStorage, })) +vi.mock("@/integrations/dashboard/api-key-service", () => ({ + withFreshKnowhereApiKey: mocks.withFreshKnowhereApiKey, +})) + vi.mock("@/lib/logger", () => ({ logger: { error: mocks.loggerError, @@ -97,6 +107,12 @@ describe("sourceReconcileRouteWorkflow", () => { status: "ready", }) mocks.updateRevisionKey.mockResolvedValue({ id: "source_1" }) + mocks.withFreshKnowhereApiKey.mockImplementation( + async (apiKey: string, run: (apiKey: string) => Promise) => ({ + result: await run(apiKey), + apiKey, + }), + ) }) afterEach(() => { @@ -273,6 +289,89 @@ describe("sourceReconcileRouteWorkflow", () => { ]) }) + it("forwards a refreshed Knowhere JWT on poll continuation and parsed-sync enqueue", async () => { + const context = createWorkflowContext() + const continuations: ContinuationTriggerInput[] = [] + const restore = + sourceReconcileRouteWorkflow.setContinuationTriggerForTesting( + async (input) => { + continuations.push(input) + }, + ) + mocks.withFreshKnowhereApiKey.mockImplementation( + async (_apiKey: string, run: (apiKey: string) => Promise) => ({ + result: await run("jwt_refreshed"), + apiKey: "jwt_refreshed", + }), + ) + mocks.makeKnowhereClientWithParsedStorage.mockReturnValue({ + client: { jobs: {}, documents: { listChunks: vi.fn() } }, + knowledge: { syncParsedDocument: vi.fn() }, + }) + mocks.pollSourceReconciliation.mockResolvedValue({ + kind: "waiting", + jobId: "job_1", + jobStatus: "running", + }) + + try { + await sourceReconcileRouteWorkflow.runPollAndMirrorWorkflow({ + context, + payload: sourceReconcileRouteWorkflow.normalizeReconcilePayload({ + workspaceId: "workspace_1", + sourceId: "source_1", + apiKey: "jwt_expired", + segmentIndex: 0, + }), + }) + } finally { + restore() + } + + expect(mocks.makeKnowhereClientWithParsedStorage).toHaveBeenCalledWith( + "jwt_refreshed", + { workspaceId: "workspace_1" }, + ) + expect(continuations[0]?.payload.apiKey).toBe("jwt_refreshed") + }) + + it("enqueues parsed-sync with a refreshed Knowhere JWT", async () => { + const context = createWorkflowContext() + mocks.withFreshKnowhereApiKey.mockImplementation( + async (_apiKey: string, run: (apiKey: string) => Promise) => ({ + result: await run("jwt_refreshed"), + apiKey: "jwt_refreshed", + }), + ) + const wired = createClient({}) + mocks.makeKnowhereClientWithParsedStorage.mockReturnValue({ + client: wired.client, + knowledge: wired.knowledge, + }) + mocks.pollSourceReconciliation.mockResolvedValue({ + kind: "ready-to-prepare", + jobId: "job_1", + documentId: "doc_1", + }) + + await sourceReconcileRouteWorkflow.runPollAndMirrorWorkflow({ + context, + payload: sourceReconcileRouteWorkflow.normalizeReconcilePayload({ + workspaceId: "workspace_1", + sourceId: "source_1", + apiKey: "jwt_expired", + }), + }) + + expect(mocks.enqueueParsedDocumentSync).toHaveBeenCalledWith({ + workspaceId: "workspace_1", + sourceId: "source_1", + documentId: "doc_1", + apiKey: "jwt_refreshed", + revisionKey: "rev_1", + }) + }) + it("marks a parsing source failed after workflow retry exhaustion", async () => { mocks.markFailed.mockResolvedValue({ id: "source_1" }) diff --git a/src/domains/sources/source-reconcile-route-workflow.ts b/src/domains/sources/source-reconcile-route-workflow.ts index 5d2cb2c..a1a316b 100644 --- a/src/domains/sources/source-reconcile-route-workflow.ts +++ b/src/domains/sources/source-reconcile-route-workflow.ts @@ -7,6 +7,7 @@ import { pollSourceReconciliation, } from "@/domains/sources/source-reconcile-workflow" import { makeKnowhereClientWithParsedStorage } from "@/integrations/knowhere" +import { withFreshKnowhereApiKey } from "@/integrations/dashboard/api-key-service" import { logger } from "@/lib/logger" import { enqueueParsedDocumentSync } from "./parsed-document-sync-scheduler" import { sourceWorkflowRuntime } from "./workflow-runtime" @@ -68,10 +69,8 @@ async function runPollAndMirrorWorkflow(input: { readonly payload: NormalizedReconcilePayload }): Promise { const { context, payload } = input - const { workspaceId, sourceId, apiKey } = payload - const { client } = makeKnowhereClientWithParsedStorage(apiKey, { - workspaceId, - }) + const { workspaceId, sourceId } = payload + let apiKey = payload.apiKey let delay = initialDelaySeconds let completedJob: { readonly jobId: string @@ -79,13 +78,20 @@ async function runPollAndMirrorWorkflow(input: { } | null = null for (let attempt = 0; attempt < maxPollAttempts; attempt++) { - const poll = await context.run(`poll-${attempt}`, async () => { - return pollSourceReconciliation({ - workspaceId, - sourceId, - client, - }) - }) + const step = await context.run(`poll-${attempt}`, async () => + withFreshKnowhereApiKey(apiKey, async (freshKey) => { + const { client } = makeKnowhereClientWithParsedStorage(freshKey, { + workspaceId, + }) + return pollSourceReconciliation({ + workspaceId, + sourceId, + client, + }) + }), + ) + apiKey = step.apiKey + const poll = step.result if (poll.kind === "ready-to-prepare") { completedJob = { @@ -146,14 +152,21 @@ async function runPollAndMirrorWorkflow(input: { ) if (ready.status === "gone") return - const revisionKey = await context.run("resolve-revision-key", async () => - resolveParsedRevisionKey({ - client, - sourceId, - documentId: jobToPrepare.documentId, - fallbackRevisionKey: jobToPrepare.jobId, + const revision = await context.run("resolve-revision-key", async () => + withFreshKnowhereApiKey(apiKey, async (freshKey) => { + const { client } = makeKnowhereClientWithParsedStorage(freshKey, { + workspaceId, + }) + return resolveParsedRevisionKey({ + client, + sourceId, + documentId: jobToPrepare.documentId, + fallbackRevisionKey: jobToPrepare.jobId, + }) }), ) + apiKey = revision.apiKey + const revisionKey = revision.result await context.run("record-source-revision-key", async () => sourceWorkflowRuntime.updateRevisionKey(workspaceId, sourceId, revisionKey), diff --git a/src/infrastructure/db/schema.ts b/src/infrastructure/db/schema.ts index 4e9470f..a3c9805 100644 --- a/src/infrastructure/db/schema.ts +++ b/src/infrastructure/db/schema.ts @@ -1,7 +1,9 @@ import { sql } from "drizzle-orm"; import { bigint, + doublePrecision, index, + integer, jsonb, pgTable, text, @@ -338,3 +340,139 @@ export const chatMessages = pgTable( export type ChatMessage = typeof chatMessages.$inferSelect; export type NewChatMessage = typeof chatMessages.$inferInsert; + +/** + * Fluid memory: typed insights extracted from human-AI conversation turns + * (as opposed to "crystal memory", which is the parsed document knowledge + * that stays upstream in Knowhere). + * + * One row per extracted insight. `kind` discriminates the typed `payload` + * (see src/domains/memory/types.ts for the payload contract per kind): + * - indicator_pref — a metric the user cares about (name, aliases, + * polarity, importance) + * - stance — a stated position that shapes judgement + * - decision_rule — a when/then rule over indicators + * - entity_of_interest — a company/topic the user tracks + * + * `abstract_l0` / `overview_l1` are the tiered sidecar summaries (L0 = + * one line for pre-filter/dedup context, L1 = short paragraph for later + * cognition injection). L2 is the payload itself. + * + * Lifecycle: rows start `active`; user revisions deprecate rather than + * delete (conservative merge policy), with `version` bumped on merge. + * + * `source_message_id` points at the assistant message of the turn the + * insight was extracted from; it is set-null on message deletion because + * the insight outlives any single turn. + */ +export const fluidMemoryItems = pgTable( + "fluid_memory_items", + { + id: uuid("id").primaryKey().defaultRandom(), + workspaceId: uuid("workspace_id") + .notNull() + .references(() => workspaces.id, { onDelete: "cascade" }), + kind: text("kind").notNull(), + payload: jsonb("payload").notNull(), + abstractL0: text("abstract_l0").notNull(), + overviewL1: text("overview_l1").notNull(), + sourceMessageId: uuid("source_message_id").references( + () => chatMessages.id, + { onDelete: "set null" }, + ), + confidence: doublePrecision("confidence").notNull(), + status: text("status").notNull(), + version: integer("version").notNull().default(1), + createdAt: timestamp("created_at", { withTimezone: true }) + .notNull() + .defaultNow(), + updatedAt: timestamp("updated_at", { withTimezone: true }) + .notNull() + .defaultNow(), + }, + (t) => [ + // Workspace lifecycle scans (active vs deprecated). + index("fluid_memory_items_workspace_status_idx").on( + t.workspaceId, + t.status, + ), + index("fluid_memory_items_workspace_kind_idx").on(t.workspaceId, t.kind), + ], +); + +export type FluidMemoryItem = typeof fluidMemoryItems.$inferSelect; +export type NewFluidMemoryItem = typeof fluidMemoryItems.$inferInsert; + +/** + * Lexical inverted index over active fluid memory items, used to retrieve + * dedup candidates at extraction time instead of loading the whole memory + * set into the prompt. One row per (item, token); `frequency` counts token + * occurrences in the item's search text. + * + * Invariant: token rows exist iff the owning item is `active`. Writers keep + * this in sync — create inserts rows, merge replaces them, deprecate deletes + * them — so lookups scan tokens alone (no status join) and never surface a + * deprecated item. + * + * Tokenization mirrors Knowhere map-nav: single CJK characters plus + * `[a-z0-9_]+` runs. Scoring is idf-weighted token overlap computed in SQL, + * keeping the mechanism on portable Postgres (no pg_trgm/pgvector). + */ +export const fluidMemoryTokens = pgTable( + "fluid_memory_tokens", + { + id: uuid("id").primaryKey().defaultRandom(), + workspaceId: uuid("workspace_id") + .notNull() + .references(() => workspaces.id, { onDelete: "cascade" }), + itemId: uuid("item_id") + .notNull() + .references(() => fluidMemoryItems.id, { onDelete: "cascade" }), + kind: text("kind").notNull(), + token: text("token").notNull(), + frequency: integer("frequency").notNull().default(1), + }, + (t) => [ + // Lookup: candidate tokens within a workspace + kind scope. + index("fluid_memory_tokens_lookup_idx").on( + t.workspaceId, + t.kind, + t.token, + ), + // Rebuild/delete a single item's rows on merge/deprecate. + index("fluid_memory_tokens_item_idx").on(t.itemId), + ], +); + +export type FluidMemoryToken = typeof fluidMemoryTokens.$inferSelect; +export type NewFluidMemoryToken = typeof fluidMemoryTokens.$inferInsert; + +/** + * Append-only audit of extraction decisions, one row per processed turn. + * `operations` is a JSONB array of { op, kind, itemId?, summary, reason? } + * records (op = create | skip | merge | deprecate), mirroring OpenViking's + * memory_diff.json so memory growth stays observable and reversible. + */ +export const memoryDiffs = pgTable( + "memory_diffs", + { + id: uuid("id").primaryKey().defaultRandom(), + workspaceId: uuid("workspace_id") + .notNull() + .references(() => workspaces.id, { onDelete: "cascade" }), + sourceMessageId: uuid("source_message_id").references( + () => chatMessages.id, + { onDelete: "set null" }, + ), + operations: jsonb("operations").notNull(), + createdAt: timestamp("created_at", { withTimezone: true }) + .notNull() + .defaultNow(), + }, + (t) => [ + index("memory_diffs_workspace_created_idx").on(t.workspaceId, t.createdAt), + ], +); + +export type MemoryDiff = typeof memoryDiffs.$inferSelect; +export type NewMemoryDiff = typeof memoryDiffs.$inferInsert; diff --git a/src/integrations/dashboard/api-key-service.test.ts b/src/integrations/dashboard/api-key-service.test.ts index 88347e0..74e9dae 100644 --- a/src/integrations/dashboard/api-key-service.test.ts +++ b/src/integrations/dashboard/api-key-service.test.ts @@ -9,8 +9,13 @@ vi.mock("next/cache", () => nextCacheMocks) import { ensureApiKeyForWorkspace, + ensureFreshKnowhereApiKey, fetchKnowhereJwt, isAuthError, + readJwtExpirySeconds, + refreshKnowhereJwt, + shouldRefreshKnowhereJwt, + withFreshKnowhereApiKey, } from "./api-key-service" function getHeaderValue(headers: HeadersInit | undefined, name: string): string | null { @@ -70,6 +75,12 @@ describe("isAuthError", () => { expect(isAuthError({ message: "Invalid API key" })).toBe(true) expect(isAuthError({ message: "Auth error occurred" })).toBe(true) expect(isAuthError({ message: "unauthenticated" })).toBe(true) + expect(isAuthError({ message: "Authentication required" })).toBe(true) + }) + + it("detects auth-related error names", () => { + expect(isAuthError({ name: "AuthenticationError" })).toBe(true) + expect(isAuthError({ name: "UnauthorizedError" })).toBe(true) }) it("detects 401/403 in error message string", () => { @@ -240,3 +251,183 @@ describe("ensureApiKeyForWorkspace", () => { expect(fetchSpy).not.toHaveBeenCalled() }) }) + +const REFRESH_JWT_PATH = "/api/orpc/knowhereServiceJwt/refresh" + +function makeJwt(payload: Record): string { + const header = Buffer.from( + JSON.stringify({ alg: "none", typ: "JWT" }), + ).toString("base64url") + const body = Buffer.from(JSON.stringify(payload)).toString("base64url") + return `${header}.${body}.sig` +} + +describe("refreshKnowhereJwt", () => { + const originalFetch = globalThis.fetch + const originalOrigin = process.env.DASHBOARD_ORIGIN + + afterEach(() => { + globalThis.fetch = originalFetch + if (originalOrigin === undefined) + delete process.env.DASHBOARD_ORIGIN + else process.env.DASHBOARD_ORIGIN = originalOrigin + }) + + it("POSTs the snapshot token to the Dashboard refresh endpoint", async () => { + process.env.DASHBOARD_ORIGIN = "https://dashboard.example" + const expectedUrl = `https://dashboard.example${REFRESH_JWT_PATH}` + const fetchSpy = vi.fn().mockResolvedValue( + new Response( + JSON.stringify({ + json: { token: "refreshed.jwt.token", expiresInSeconds: 3600 }, + }), + { status: 200, headers: { "content-type": "application/json" } }, + ), + ) + globalThis.fetch = fetchSpy + + const token = await refreshKnowhereJwt("expired.jwt.snapshot") + + expect(token).toBe("refreshed.jwt.token") + expect(fetchSpy).toHaveBeenCalledOnce() + const [url, init] = fetchSpy.mock.calls[0]! + expect(url instanceof URL ? url.href : url).toBe(expectedUrl) + expect((init as RequestInit)?.method).toBe("POST") + expect(getHeaderValue((init as RequestInit)?.headers, "content-type")).toContain( + "application/json", + ) + expect(await readBodyText((init as RequestInit)?.body)).toBe( + JSON.stringify({ json: { token: "expired.jwt.snapshot" } }), + ) + }) + + it("throws on non-2xx from Dashboard", async () => { + process.env.DASHBOARD_ORIGIN = "https://dashboard.example" + globalThis.fetch = vi + .fn() + .mockResolvedValue(new Response("oops", { status: 401 })) + await expect(refreshKnowhereJwt("expired.jwt.snapshot")).rejects.toThrow( + /Dashboard JWT refresh: non-2xx/, + ) + }) +}) + +describe("ensureFreshKnowhereApiKey", () => { + const originalFetch = globalThis.fetch + const originalApiKey = process.env.KNOWHERE_API_KEY + const originalOrigin = process.env.DASHBOARD_ORIGIN + + afterEach(() => { + globalThis.fetch = originalFetch + if (originalApiKey === undefined) delete process.env.KNOWHERE_API_KEY + else process.env.KNOWHERE_API_KEY = originalApiKey + if (originalOrigin === undefined) + delete process.env.DASHBOARD_ORIGIN + else process.env.DASHBOARD_ORIGIN = originalOrigin + }) + + it("does not refresh persistent API keys or the KNOWHERE_API_KEY override", async () => { + process.env.KNOWHERE_API_KEY = "sk_dev_key" + const fetchSpy = vi.fn() + globalThis.fetch = fetchSpy + + await expect(ensureFreshKnowhereApiKey("sk_dev_key", { force: true })).resolves.toBe( + "sk_dev_key", + ) + await expect(ensureFreshKnowhereApiKey("sk_other")).resolves.toBe("sk_other") + expect(fetchSpy).not.toHaveBeenCalled() + }) + + it("refreshes a JWT that is within the safety window", async () => { + process.env.DASHBOARD_ORIGIN = "https://dashboard.example" + const nearExpiry = makeJwt({ exp: Math.floor(Date.now() / 1000) + 5 }) + globalThis.fetch = vi.fn().mockResolvedValue( + new Response( + JSON.stringify({ + json: { token: "fresh.jwt.token", expiresInSeconds: 3600 }, + }), + { status: 200, headers: { "content-type": "application/json" } }, + ), + ) + + await expect(ensureFreshKnowhereApiKey(nearExpiry)).resolves.toBe( + "fresh.jwt.token", + ) + }) + + it("returns a still-valid JWT without calling Dashboard", async () => { + const fresh = makeJwt({ exp: Math.floor(Date.now() / 1000) + 3_600 }) + const fetchSpy = vi.fn() + globalThis.fetch = fetchSpy + + await expect(ensureFreshKnowhereApiKey(fresh)).resolves.toBe(fresh) + expect(fetchSpy).not.toHaveBeenCalled() + }) +}) + +describe("shouldRefreshKnowhereJwt", () => { + it("is true when exp is inside the safety window or already past", () => { + const now = 1_700_000_000 + expect(shouldRefreshKnowhereJwt(makeJwt({ exp: now + 15 }), now)).toBe(true) + expect(shouldRefreshKnowhereJwt(makeJwt({ exp: now - 1 }), now)).toBe(true) + expect(shouldRefreshKnowhereJwt(makeJwt({ exp: now + 16 }), now)).toBe(false) + expect(shouldRefreshKnowhereJwt("sk_persistent", now)).toBe(false) + }) +}) + +describe("readJwtExpirySeconds", () => { + it("reads exp from a three-part JWT and ignores other secrets", () => { + expect(readJwtExpirySeconds(makeJwt({ exp: 42 }))).toBe(42) + expect(readJwtExpirySeconds("sk_key")).toBeNull() + expect(readJwtExpirySeconds(makeJwt({ sub: "user" }))).toBeNull() + }) +}) + +describe("withFreshKnowhereApiKey", () => { + const originalFetch = globalThis.fetch + const originalOrigin = process.env.DASHBOARD_ORIGIN + + afterEach(() => { + globalThis.fetch = originalFetch + if (originalOrigin === undefined) + delete process.env.DASHBOARD_ORIGIN + else process.env.DASHBOARD_ORIGIN = originalOrigin + }) + + it("retries once after Authentication required, then uses the refreshed token", async () => { + process.env.DASHBOARD_ORIGIN = "https://dashboard.example" + const stillValid = makeJwt({ exp: Math.floor(Date.now() / 1000) + 3_600 }) + globalThis.fetch = vi.fn().mockResolvedValue( + new Response( + JSON.stringify({ + json: { token: "retry.jwt.token", expiresInSeconds: 3600 }, + }), + { status: 200, headers: { "content-type": "application/json" } }, + ), + ) + const run = vi + .fn<(apiKey: string) => Promise>() + .mockRejectedValueOnce(new Error("Authentication required")) + .mockResolvedValueOnce("ok") + + const result = await withFreshKnowhereApiKey(stillValid, run) + + expect(result).toEqual({ result: "ok", apiKey: "retry.jwt.token" }) + expect(run).toHaveBeenCalledTimes(2) + expect(run).toHaveBeenNthCalledWith(1, stillValid) + expect(run).toHaveBeenNthCalledWith(2, "retry.jwt.token") + }) + + it("does not retry non-auth errors", async () => { + const fresh = makeJwt({ exp: Math.floor(Date.now() / 1000) + 3_600 }) + const fetchSpy = vi.fn() + globalThis.fetch = fetchSpy + const run = vi.fn().mockRejectedValue(new Error("Parser rejected this document.")) + + await expect(withFreshKnowhereApiKey(fresh, run)).rejects.toThrow( + /Parser rejected/, + ) + expect(run).toHaveBeenCalledOnce() + expect(fetchSpy).not.toHaveBeenCalled() + }) +}) diff --git a/src/integrations/dashboard/api-key-service.ts b/src/integrations/dashboard/api-key-service.ts index 743c99d..225b90f 100644 --- a/src/integrations/dashboard/api-key-service.ts +++ b/src/integrations/dashboard/api-key-service.ts @@ -9,7 +9,7 @@ import { } from "@effect/platform" import { logger } from "@/lib/logger" import { knowhereApiKeyOverride } from "@/integrations/knowhere-api-key" -import { setEmptyJsonBody } from "./orpc-request" +import { setEmptyJsonBody, setJsonBody } from "./orpc-request" import { formatUnknownForLog } from "@/lib/format-log-value" /** @@ -23,6 +23,9 @@ const JwtResponse = Schema.Struct({ }), }) +const ISSUE_JWT_PATH = "/api/orpc/users/issueServiceJwt" +const REFRESH_JWT_PATH = "/api/orpc/knowhereServiceJwt/refresh" + /** * Request a short-lived Knowhere JWT from Dashboard's generic issuance * endpoint. The returned token is passed directly to the Knowhere SDK @@ -44,7 +47,7 @@ export const fetchKnowhereJwtEffect = (cookieHeader: string) => } const http = yield* HttpClient.HttpClient - const url = `${origin}/api/orpc/users/issueServiceJwt` + const url = `${origin}${ISSUE_JWT_PATH}` const body = yield* HttpClientRequest.post(url).pipe( HttpClientRequest.setHeader("cookie", cookieHeader), setEmptyJsonBody, @@ -138,6 +141,151 @@ export async function fetchKnowhereJwt( } } +export const refreshKnowhereJwtEffect = (token: string) => + Effect.gen(function* () { + const origin = process.env.DASHBOARD_ORIGIN + if (!origin) { + return yield* Effect.die( + new Error( + "DASHBOARD_ORIGIN must be set. " + + "It should point to the Dashboard origin (see .env.local.example).", + ), + ) + } + + const http = yield* HttpClient.HttpClient + const url = `${origin}${REFRESH_JWT_PATH}` + const body = yield* setJsonBody(HttpClientRequest.post(url), { + json: { token }, + }).pipe( + http.execute, + Effect.flatMap((response) => + Effect.gen(function* () { + const status = response.status + + if (status < 200 || status >= 300) { + const rawText = yield* Effect.either(response.text) + return yield* Effect.die( + new Error( + `Dashboard JWT refresh: non-2xx (status=${status}) body=${Either.getOrElse(rawText, () => "").slice(0, 1000)}`, + ), + ) + } + + const parsed = yield* Effect.either(response.json) + if (Either.isLeft(parsed)) { + return yield* Effect.die( + new Error( + `Dashboard JWT refresh: invalid JSON (status=${status}) error=${String(parsed.left)}`, + ), + ) + } + + const result = Schema.decodeUnknownEither(JwtResponse)(parsed.right) + if (Either.isLeft(result)) { + return yield* Effect.die( + new Error( + `Dashboard JWT refresh: schema mismatch (status=${status}) body=${formatUnknownForLog(parsed.right).slice(0, 1000)}`, + ), + ) + } + + return result.right.json.token + }), + ), + ) + + return body + }) + +export async function refreshKnowhereJwt(token: string): Promise { + const start = Date.now() + try { + const refreshed = await Effect.runPromise( + refreshKnowhereJwtEffect(token).pipe( + Effect.provide(FetchHttpClient.layer), + ), + ) + logger.info("dashboard: POST /api/orpc/knowhereServiceJwt/refresh ok", { + durationMs: Date.now() - start, + }) + return refreshed + } catch (error) { + logger.error("dashboard: POST /api/orpc/knowhereServiceJwt/refresh failed", { + durationMs: Date.now() - start, + error: error instanceof Error ? error.message : String(error), + }) + throw error + } +} + +export type EnsureFreshKnowhereApiKeyOptions = { + readonly force?: boolean +} + +/** + * Keep a QStash-stored Knowhere credential usable across long workflows. + * Dashboard JWTs last one hour; this refreshes when the snapshot is within + * the safety window, already expired, or `force` is set. Persistent API keys + * are returned unchanged. + */ +export async function ensureFreshKnowhereApiKey( + apiKey: string, + options: EnsureFreshKnowhereApiKeyOptions = {}, +): Promise { + const override = knowhereApiKeyOverride.getApiKey() + if (override && apiKey === override) return apiKey + if (!looksLikeJwt(apiKey)) return apiKey + if (!options.force && !shouldRefreshKnowhereJwt(apiKey)) return apiKey + return refreshKnowhereJwt(apiKey) +} + +export async function withFreshKnowhereApiKey( + apiKey: string, + run: (apiKey: string) => Promise, +): Promise<{ readonly result: T; readonly apiKey: string }> { + let current = await ensureFreshKnowhereApiKey(apiKey) + try { + return { result: await run(current), apiKey: current } + } catch (error) { + if (!isAuthError(error)) throw error + current = await ensureFreshKnowhereApiKey(current, { force: true }) + return { result: await run(current), apiKey: current } + } +} + +export function shouldRefreshKnowhereJwt( + apiKey: string, + nowSeconds: number = Math.floor(Date.now() / 1000), +): boolean { + const expiresAt = readJwtExpirySeconds(apiKey) + if (expiresAt === null) return false + return expiresAt - nowSeconds <= jwtExpirationSafetySeconds +} + +export function looksLikeJwt(apiKey: string): boolean { + return apiKey.split(".").length === 3 +} + +const JwtExpiryPayload = Schema.Struct({ + exp: Schema.Number, +}) + +export function readJwtExpirySeconds(apiKey: string): number | null { + const parts = apiKey.split(".") + const payloadSegment = parts[1] + if (parts.length !== 3 || payloadSegment === undefined) return null + try { + const parsed: unknown = JSON.parse( + Buffer.from(payloadSegment, "base64url").toString("utf8"), + ) + const decoded = Schema.decodeUnknownEither(JwtExpiryPayload)(parsed) + return Either.isRight(decoded) ? decoded.right.exp : null + } catch { + return null + } +} + /** * Resolve the credential used for Knowhere SDK calls. Development can * short-circuit Dashboard JWT issuance by setting KNOWHERE_API_KEY. @@ -175,12 +323,19 @@ export function isAuthError(error: unknown): boolean { if ( msg.includes("unauthorized") || msg.includes("unauthenticated") || + msg.includes("authentication required") || msg.includes("forbidden") || msg.includes("invalid api key") || msg.includes("auth error") ) return true } + if (typeof err.name === "string") { + const name = err.name.toLowerCase() + if (name.includes("authentication") || name.includes("unauthorized")) { + return true + } + } return false } diff --git a/src/integrations/dashboard/orpc-request.ts b/src/integrations/dashboard/orpc-request.ts index ea73d57..a6fd5ab 100644 --- a/src/integrations/dashboard/orpc-request.ts +++ b/src/integrations/dashboard/orpc-request.ts @@ -17,3 +17,14 @@ export function setEmptyJsonBody( JSON_CONTENT_TYPE, ) } + +export function setJsonBody( + request: HttpClientRequest.HttpClientRequest, + body: unknown, +): HttpClientRequest.HttpClientRequest { + return HttpClientRequest.bodyText( + request, + JSON.stringify(body), + JSON_CONTENT_TYPE, + ) +}