From 5869016c429da346dca9fa1c857d382be8bec141 Mon Sep 17 00:00:00 2001 From: Jyong Date: Wed, 29 Jul 2026 13:32:01 -0400 Subject: [PATCH] Restore Research answer synthesis routing --- api/knowledge-fs-contract.lock.json | 4 +- .../api/src/answer-generation-options.test.ts | 8 +- knowledge-fs/apps/api/src/index.ts | 14 +- knowledge-fs/apps/api/src/migrate.test.ts | 8 +- knowledge-fs/docs/api-reference.md | 2 +- .../adapters/src/migration-runner.test.ts | 2 + .../src/profile-aware-query-generator.test.ts | 38 ++++ .../api/src/profile-aware-query-generator.ts | 24 +++ .../api/src/research-task-job.test.ts | 193 ++++++++++++++++++ .../packages/api/src/research-task-job.ts | 23 +++ ...partial-result-database-repository.test.ts | 15 +- ...task-partial-result-database-repository.ts | 19 +- .../research-task-response-schemas.test.ts | 5 +- .../api/src/research-task-response-schemas.ts | 3 + .../api/src/research-task-runtime.test.ts | 44 +++- .../packages/api/src/research-task-runtime.ts | 23 ++- ...3_research_task_final_answers.postgres.sql | 7 + .../0033_research_task_final_answers.tidb.sql | 7 + .../src/migration-artifacts.generated.ts | 2 + .../database/src/migration-file.test.ts | 4 + ...search-task-final-answer-migration.test.ts | 24 +++ knowledge-fs/packages/database/src/schema.ts | 1 + 22 files changed, 447 insertions(+), 23 deletions(-) create mode 100644 knowledge-fs/packages/database/migrations/0033_research_task_final_answers.postgres.sql create mode 100644 knowledge-fs/packages/database/migrations/0033_research_task_final_answers.tidb.sql create mode 100644 knowledge-fs/packages/database/src/research-task-final-answer-migration.test.ts diff --git a/api/knowledge-fs-contract.lock.json b/api/knowledge-fs-contract.lock.json index fe9f9446740..3eb174c8a99 100644 --- a/api/knowledge-fs-contract.lock.json +++ b/api/knowledge-fs-contract.lock.json @@ -1,7 +1,7 @@ { "schemaVersion": 5, - "subtreeTree": "aae9105ec7afa906fe73d9f7599118caf22dafb8", - "openapiSha256": "840c3a77214d9132f2c70220fbab655d8bc791dad8b1fead6f272d4d692081fd", + "subtreeTree": "16aebf6d7f5fccf5abd2465d8a56b685d37998ae", + "openapiSha256": "a3b085216f1f4b3db7d4f024642447b87d5fd49eb85723d4e49d5890f17fc7fb", "capabilityV2AuthManifestSha256": "fc0a47e23cce12544882f0298522b4933002e892b84ce1815df7e81d36a7a0c7", "capabilityV2AuthTestVectorSha256": "ae0de37b1ff05c40f905cf17a7b410d8971acacf64db07d5ee3d6fecfa559ce3", "productOperationManifestSha256": "77f2704c823d3db324c06f8f2d0109e7bda60799a6be74106ac047af3eef4492", diff --git a/knowledge-fs/apps/api/src/answer-generation-options.test.ts b/knowledge-fs/apps/api/src/answer-generation-options.test.ts index 7d8d1b7d79e..1f67b91253d 100644 --- a/knowledge-fs/apps/api/src/answer-generation-options.test.ts +++ b/knowledge-fs/apps/api/src/answer-generation-options.test.ts @@ -113,7 +113,7 @@ describe("createApiAnswerGenerationOptions", () => { expect(indexSource).not.toContain("model: answerGenerationOptions.model"); }); - it("keeps interactive retrieval evidence-only and reserves LLM synthesis for Research", async () => { + it("keeps Fast and Deep evidence-only while restoring final LLM synthesis for Research", async () => { const { readFile } = await import("node:fs/promises"); const indexSource = await readFile(new URL("./index.ts", import.meta.url), "utf8"); const citationOptionsStart = indexSource.indexOf("const multimodalCitationOptions ="); @@ -145,7 +145,11 @@ describe("createApiAnswerGenerationOptions", () => { expect(retrievalAssembly).toContain("...multimodalCitationOptions"); expect(retrievalAssembly).not.toContain("multimodalAnswerOptions"); expect(indexSource).toContain("generator: researchAnswerQueryGenerator"); - expect(indexSource).toContain("{ queryGenerator: retrievalEvidenceQueryGenerator }"); + expect(indexSource).toContain("createResearchAwareQueryGenerator"); + expect(indexSource).toContain("researchGenerator: researchAnswerQueryGenerator"); + expect(indexSource).toContain("retrievalGenerator: retrievalEvidenceQueryGenerator"); + expect(indexSource).toContain("{ queryGenerator: interactiveQueryGenerator }"); + expect(indexSource).not.toContain("{ queryGenerator: retrievalEvidenceQueryGenerator }"); expect(indexSource).not.toContain("{ queryGenerator: researchAnswerQueryGenerator }"); }); diff --git a/knowledge-fs/apps/api/src/index.ts b/knowledge-fs/apps/api/src/index.ts index 6ed1e37bf92..976952f1ba1 100644 --- a/knowledge-fs/apps/api/src/index.ts +++ b/knowledge-fs/apps/api/src/index.ts @@ -24,6 +24,7 @@ import { createProfileAwareKnowledgeSpaceManifestRepository, createProfileAwareQueryGenerator, createPublishedProjectionReadSnapshotResolver, + createResearchAwareQueryGenerator, createRetrievalExecutionLeaseCoordinator, createRetrievalPlanner, createRetrievalTestExecutor, @@ -577,8 +578,8 @@ const researchAnswerMultimodalOptions = { ...multimodalAnswerOptions, ...multimodalCitationOptions, }; -// Interactive query-stream is a retrieval surface: it returns bounded evidence and citations, -// never an LLM/VLM-synthesized answer. Answer synthesis remains a separate Research capability. +// Fast and Deep query-stream requests return bounded evidence and citations without answer +// synthesis. Research uses the same retrieval foundation, then performs one final LLM synthesis. const retrievalEvidenceQueryGenerator = retriever ? createHybridQueryGenerator({ limit: 5, @@ -609,6 +610,13 @@ const researchAnswerQueryGenerator = profileLlmGenerator: profileLlmAnswerQueryGenerator, }) : undefined; +const interactiveQueryGenerator = + retrievalEvidenceQueryGenerator && researchAnswerQueryGenerator + ? createResearchAwareQueryGenerator({ + researchGenerator: researchAnswerQueryGenerator, + retrievalGenerator: retrievalEvidenceQueryGenerator, + }) + : undefined; const researchProjectionSnapshotResolver = repositoryOptions.projectionSetPublications ? createPublishedProjectionReadSnapshotResolver({ publications: repositoryOptions.projectionSetPublications, @@ -773,7 +781,7 @@ const app = createKnowledgeGateway({ }, } : {}), - ...(retrievalEvidenceQueryGenerator ? { queryGenerator: retrievalEvidenceQueryGenerator } : {}), + ...(interactiveQueryGenerator ? { queryGenerator: interactiveQueryGenerator } : {}), ...(retrievalTestExecutor ? { retrievalTestExecutor } : {}), ...(publishedGraph ? { publishedGraph } : {}), ...(researchTaskRuntime diff --git a/knowledge-fs/apps/api/src/migrate.test.ts b/knowledge-fs/apps/api/src/migrate.test.ts index afe7b237cd2..a6f447b6507 100644 --- a/knowledge-fs/apps/api/src/migrate.test.ts +++ b/knowledge-fs/apps/api/src/migrate.test.ts @@ -72,8 +72,9 @@ describe("runApiDatabaseMigrations", () => { "0030_bulk_operations", "0031_source_connection_capability_provenance", "0032_capability_source_sync_policies", + "0033_research_task_final_answers", ], - pendingBeforeRun: 32, + pendingBeforeRun: 33, }); expect(operations).toEqual([ "schema", @@ -143,8 +144,10 @@ describe("runApiDatabaseMigrations", () => { "insert", "schema", "insert", + "schema", + "insert", ]); - expect(migrationSql).toHaveLength(32); + expect(migrationSql).toHaveLength(33); expect(migrationSql[2]).toContain("-- Migration id: 0003_projection_set_publications\n"); expect(migrationSql[2]).toContain("-- Dialect: postgres\n"); expect(migrationSql[2]).toContain('CREATE TABLE IF NOT EXISTS "projection_set_publications"'); @@ -188,6 +191,7 @@ describe("runApiDatabaseMigrations", () => { "-- Migration id: 0031_source_connection_capability_provenance\n", ); expect(migrationSql[31]).toContain("-- Migration id: 0032_capability_source_sync_policies\n"); + expect(migrationSql[32]).toContain("-- Migration id: 0033_research_task_final_answers\n"); expect(closed).toBe(true); }); diff --git a/knowledge-fs/docs/api-reference.md b/knowledge-fs/docs/api-reference.md index f4bbb4b7186..e62115dbce5 100644 --- a/knowledge-fs/docs/api-reference.md +++ b/knowledge-fs/docs/api-reference.md @@ -865,7 +865,7 @@ compatibility path does not admit Auto. **Auth**: Bearer; scope `knowledge-spaces:read`. **Path params**: `id` (string, min 1). **Query** (strict): `limit` (int 1–100, optional, default 25); `cursor` (optional). **Responses**: -- `200`: `{ items: [{ tenantId, knowledgeSpaceId, researchTaskJobId, sequence: int>0, evidenceBundle }], nextCursor? }`. `EvidenceBundle` `{ id, query, state: enum(answerable|partial|not-enough-evidence|conflict|permission-limited), items: [{ nodeId, text, score: 0–1, scores, freshness, citations[], conflicts[], metadata }], missingEvidence: [{ text, reason: enum(not-retrieved|permission-filtered|stale|conflict|unknown), expectedEvidenceId?, metadata }], traceId?, createdAt }`. +- `200`: `{ items: [{ tenantId, knowledgeSpaceId, researchTaskJobId, sequence: int>0, answer?: string, evidenceBundle }], nextCursor? }`. `answer` is the final Research-only LLM synthesis; legacy rows and evidence-only results omit it. `EvidenceBundle` `{ id, query, state: enum(answerable|partial|not-enough-evidence|conflict|permission-limited), items: [{ nodeId, text, score: 0–1, scores, freshness, citations[], conflicts[], metadata }], missingEvidence: [{ text, reason: enum(not-retrieved|permission-filtered|stale|conflict|unknown), expectedEvidenceId?, metadata }], traceId?, createdAt }`. - `404`; `401`/`403`. ### `GET /research-tasks/{id}/events` diff --git a/knowledge-fs/packages/adapters/src/migration-runner.test.ts b/knowledge-fs/packages/adapters/src/migration-runner.test.ts index 70d33dfec09..a8eb248947c 100644 --- a/knowledge-fs/packages/adapters/src/migration-runner.test.ts +++ b/knowledge-fs/packages/adapters/src/migration-runner.test.ts @@ -39,6 +39,7 @@ const bulkOperationsMigrationId = "0030_bulk_operations"; const sourceConnectionCapabilityProvenanceMigrationId = "0031_source_connection_capability_provenance"; const capabilitySourceSyncPoliciesMigrationId = "0032_capability_source_sync_policies"; +const researchTaskFinalAnswersMigrationId = "0033_research_task_final_answers"; const migrationsAfterDurableDeletion = [ versionedSpaceProfilesMigrationId, profilePublicationBindingsMigrationId, @@ -55,6 +56,7 @@ const migrationsAfterDurableDeletion = [ bulkOperationsMigrationId, sourceConnectionCapabilityProvenanceMigrationId, capabilitySourceSyncPoliciesMigrationId, + researchTaskFinalAnswersMigrationId, ] as const; const migrationsAfterTidbBaselineRepair = [ spaceAccessControlMigrationId, diff --git a/knowledge-fs/packages/api/src/profile-aware-query-generator.test.ts b/knowledge-fs/packages/api/src/profile-aware-query-generator.test.ts index 3dcfac482a1..0e453a405fc 100644 --- a/knowledge-fs/packages/api/src/profile-aware-query-generator.test.ts +++ b/knowledge-fs/packages/api/src/profile-aware-query-generator.test.ts @@ -8,6 +8,7 @@ import type { import { ReasoningCapabilityUnavailableError, createProfileAwareQueryGenerator, + createResearchAwareQueryGenerator, } from "./profile-aware-query-generator"; const INPUT: QueryGenerationInput = { @@ -127,6 +128,43 @@ describe("profile-aware query generator", () => { }); }); +describe("research-aware query generator", () => { + it.each(["fast", "deep"] as const)( + "keeps %s interactive queries on the evidence-only generator", + async (mode) => { + const retrieval = recordingGenerator("retrieval"); + const research = recordingGenerator("research-answer"); + const generator = createResearchAwareQueryGenerator({ + researchGenerator: research.generator, + retrievalGenerator: retrieval.generator, + }); + + await expect(drain(generator, { ...INPUT, mode })).resolves.toEqual([ + expect.objectContaining({ delta: "retrieval", type: "delta" }), + ]); + expect(retrieval.stream).toHaveBeenCalledOnce(); + expect(retrieval.stream).toHaveBeenCalledWith(expect.objectContaining({ mode })); + expect(research.stream).not.toHaveBeenCalled(); + }, + ); + + it("routes Research through retrieval plus final LLM synthesis", async () => { + const retrieval = recordingGenerator("retrieval"); + const research = recordingGenerator("research-answer"); + const generator = createResearchAwareQueryGenerator({ + researchGenerator: research.generator, + retrievalGenerator: retrieval.generator, + }); + + await expect(drain(generator, { ...INPUT, mode: "research" })).resolves.toEqual([ + expect.objectContaining({ delta: "research-answer", type: "delta" }), + ]); + expect(research.stream).toHaveBeenCalledOnce(); + expect(research.stream).toHaveBeenCalledWith(expect.objectContaining({ mode: "research" })); + expect(retrieval.stream).not.toHaveBeenCalled(); + }); +}); + function recordingGenerator(delta: string): { readonly generator: QueryGenerator; readonly stream: ReturnType; diff --git a/knowledge-fs/packages/api/src/profile-aware-query-generator.ts b/knowledge-fs/packages/api/src/profile-aware-query-generator.ts index ece10d1ac0f..222dede7e4e 100644 --- a/knowledge-fs/packages/api/src/profile-aware-query-generator.ts +++ b/knowledge-fs/packages/api/src/profile-aware-query-generator.ts @@ -13,6 +13,30 @@ export interface ProfileAwareQueryGeneratorOptions { readonly profileLlmGenerator?: QueryGenerator | undefined; } +export interface ResearchAwareQueryGeneratorOptions { + /** Evidence-only generator used by Fast and Deep interactive retrieval. */ + readonly retrievalGenerator: QueryGenerator; + /** Evidence retrieval followed by one final LLM synthesis, used only by Research. */ + readonly researchGenerator: QueryGenerator; +} + +/** + * Keeps interactive retrieval mode semantics explicit at the final generator boundary. + * + * Fast and Deep are retrieval-only product surfaces. Research shares the same retrieval + * foundations, then invokes the profile-selected reasoning model for one final synthesis. + */ +export function createResearchAwareQueryGenerator({ + researchGenerator, + retrievalGenerator, +}: ResearchAwareQueryGeneratorOptions): QueryGenerator { + return { + stream: async function* (input: QueryGenerationInput): AsyncGenerator { + yield* (input.mode === "research" ? researchGenerator : retrievalGenerator).stream(input); + }, + }; +} + /** * Selects answer synthesis without letting a configured space silently lose its reasoning model. * diff --git a/knowledge-fs/packages/api/src/research-task-job.test.ts b/knowledge-fs/packages/api/src/research-task-job.test.ts index 16df5f90a19..a47a38393be 100644 --- a/knowledge-fs/packages/api/src/research-task-job.test.ts +++ b/knowledge-fs/packages/api/src/research-task-job.test.ts @@ -1,6 +1,8 @@ import { describe, expect, it, vi } from "vitest"; import { + RESEARCH_TASK_PARTIAL_ANSWER_MAX_CHARS, + type ResearchTaskDurableDispatch, createInMemoryResearchTaskJobRepository, createInMemoryResearchTaskPartialResultRepository, createResearchTaskJobStateMachine, @@ -57,6 +59,17 @@ describe("research task job state machine", () => { await expect(machine.start(missing)).rejects.toThrow( "exactly one durable authorization binding", ); + + const { permissionSnapshot: _snapshot, ...subjectOnly } = baseStartInput(); + await expect(machine.start(subjectOnly)).rejects.toThrow( + "Research task legacy authorization binding is incomplete", + ); + await expect( + machine.start({ + ...missing, + capabilityGrantId: " ", + }), + ).rejects.toThrow("Research task capabilityGrantId is required"); }); it("starts a research task and enqueues bounded durable work", async () => { @@ -97,6 +110,38 @@ describe("research task job state machine", () => { expect(record).toHaveBeenCalledWith({ lifecycle: "queued", taskKind: "research" }); }); + it("uses the durable dispatch for both initial delivery and resume", async () => { + const repository = createInMemoryResearchTaskJobRepository({ maxJobs: 10 }); + const durableDispatch: ResearchTaskDurableDispatch = { + requestResume: vi.fn(async ({ job, resumeFromStage, updatedAt }) => + repository.update({ + ...job, + stage: resumeFromStage, + updatedAt, + }), + ), + start: vi.fn(async (job) => repository.create(job)), + }; + const machine = createResearchTaskJobStateMachine({ + durableDispatch, + generateId: () => "research-task-job-1", + jobs: new FakeJobQueue(), + now: () => 1_000, + repository, + }); + + const job = await machine.start(baseStartInput()); + await machine.advance(job.id, "planning"); + await machine.pause(job.id, { reason: "backpressure" }); + const resumed = await machine.resume(job.id); + + expect(durableDispatch.start).toHaveBeenCalledOnce(); + expect(durableDispatch.requestResume).toHaveBeenCalledWith( + expect.objectContaining({ resumeFromStage: "planning" }), + ); + expect(resumed.stage).toBe("planning"); + }); + it("persists retrieval mode and topK while queue payloads contain only the job locator", async () => { const queue = new FakeJobQueue(); const machine = createResearchTaskJobStateMachine({ @@ -253,6 +298,14 @@ describe("research task job state machine", () => { expect(() => createInMemoryResearchTaskJobRepository({ maxJobs: 0 })).toThrow( "Research task job repository maxJobs must be at least 1", ); + expect(() => + createResearchTaskJobStateMachine({ + generateId: () => "unused", + jobs: new FakeJobQueue(), + maxExecutionAttempts: 0, + repository, + }), + ).toThrow("Research task job maxExecutionAttempts must be at least 1"); await expect(machine.get("missing")).resolves.toBeNull(); await expect(machine.advance("missing", "planning")).rejects.toThrow( "Research task job missing not found", @@ -263,6 +316,27 @@ describe("research task job state machine", () => { await expect(machine.start({ ...baseStartInput(), query: " " })).rejects.toThrow( "Research task job query is required", ); + await expect( + machine.start({ + ...baseStartInput(), + permissionSnapshot: { ...basePermissionSnapshot, id: " " }, + }), + ).rejects.toThrow("Research task permission snapshot id is required"); + await expect( + machine.start({ + ...baseStartInput(), + permissionSnapshot: { + ...basePermissionSnapshot, + accessChannel: "invalid" as never, + }, + }), + ).rejects.toThrow("Research task permission snapshot access channel is invalid"); + await expect( + machine.start({ + ...baseStartInput(), + permissionSnapshot: { ...basePermissionSnapshot, revision: 0 }, + }), + ).rejects.toThrow("Research task permission snapshot revision must be at least 1"); await expect( machine.start({ ...baseStartInput(), @@ -339,6 +413,80 @@ describe("research task job state machine", () => { expect(secondPage.nextCursor).toBeUndefined(); }); + it("fails closed on unavailable listing, invalid cursors, and stale repository writes", async () => { + const fullRepository = createInMemoryResearchTaskJobRepository({ maxJobs: 10 }); + const { listBySpace: _listBySpace, ...repositoryWithoutListing } = fullRepository; + const unavailableMachine = createResearchTaskJobStateMachine({ + generateId: () => "unused", + jobs: new FakeJobQueue(), + repository: repositoryWithoutListing, + }); + const listInput = { + capabilityRequester: { + callerKind: "interactive" as const, + grantId: "grant-1", + subjectId: "subject-1", + }, + knowledgeSpaceId: "space-1", + limit: 1, + tenantId: "tenant-1", + }; + + await expect(unavailableMachine.listBySpace(listInput)).rejects.toThrow( + "Research task space listing is unavailable", + ); + + const machine = createResearchTaskJobStateMachine({ + generateId: () => "research-task-job-1", + jobs: new FakeJobQueue(), + now: () => 1_000, + repository: fullRepository, + }); + const job = await machine.start(capabilityStartInput("grant-1")); + await expect(machine.listBySpace({ ...listInput, limit: 0 })).rejects.toThrow( + "Research task list limit must be between 1 and 100", + ); + await expect( + machine.listBySpace({ + ...listInput, + cursor: { createdAt: -1, id: job.id }, + }), + ).rejects.toThrow("Research task list cursor createdAt must be a nonnegative integer"); + await expect(fullRepository.update({ ...job, rowVersion: 0 })).rejects.toThrow( + "Research task job update lost its row-version fence", + ); + }); + + it("orders equal timestamps by id and excludes legacy jobs from Capability listing", async () => { + const repository = createInMemoryResearchTaskJobRepository({ maxJobs: 10 }); + const ids = ["research-task-b", "research-task-a", "legacy-task"]; + const machine = createResearchTaskJobStateMachine({ + generateId: () => ids.shift() ?? "unexpected-task", + jobs: new FakeJobQueue(), + now: () => 1_000, + repository, + }); + const grantId = "grant-1"; + await machine.start(capabilityStartInput(grantId)); + await machine.start(capabilityStartInput(grantId)); + await machine.start(baseStartInput()); + + await expect( + machine.listBySpace({ + capabilityRequester: { + callerKind: "interactive", + grantId, + subjectId: "subject-1", + }, + knowledgeSpaceId: "space-1", + limit: 10, + tenantId: "tenant-1", + }), + ).resolves.toMatchObject({ + items: [{ id: "research-task-b" }, { id: "research-task-a" }], + }); + }); + it("records step costs and cancels when the research budget is exhausted", async () => { const queue = new FakeJobQueue(); const machine = createResearchTaskJobStateMachine({ @@ -593,11 +741,23 @@ describe("research task partial result repository", () => { }); const first = await repository.append({ + answer: " The first final answer. ", evidenceBundle: evidenceBundle("018f0d60-7a49-7cc2-9c1b-5b36f18f6a01", "first evidence"), + idempotencyKey: "final-answer", knowledgeSpaceId: "space-1", researchTaskJobId: "research-task-job-1", tenantId: "tenant-1", }); + await expect( + repository.append({ + answer: "This replay must not replace the first result.", + evidenceBundle: evidenceBundle("018f0d60-7a49-7cc2-9c1b-5b36f18f6a08", "replayed evidence"), + idempotencyKey: "final-answer", + knowledgeSpaceId: "space-1", + researchTaskJobId: "research-task-job-1", + tenantId: "tenant-1", + }), + ).resolves.toEqual(first); await repository.append({ evidenceBundle: evidenceBundle("018f0d60-7a49-7cc2-9c1b-5b36f18f6a02", "second evidence"), knowledgeSpaceId: "space-1", @@ -619,6 +779,7 @@ describe("research task partial result repository", () => { expect(firstPage).toMatchObject({ items: [ { + answer: "The first final answer.", evidenceBundle: { id: "018f0d60-7a49-7cc2-9c1b-5b36f18f6a01" }, sequence: 1, }, @@ -658,6 +819,38 @@ describe("research task partial result repository", () => { }), ]), }); + + await expect( + repository.append({ + answer: " ", + evidenceBundle: evidenceBundle("018f0d60-7a49-7cc2-9c1b-5b36f18f6a06", "invalid answer"), + knowledgeSpaceId: "space-1", + researchTaskJobId: "research-task-job-3", + tenantId: "tenant-1", + }), + ).rejects.toThrow("answer must not be empty"); + await expect( + repository.append({ + evidenceBundle: evidenceBundle( + "018f0d60-7a49-7cc2-9c1b-5b36f18f6a09", + "invalid idempotency key", + ), + idempotencyKey: " ", + knowledgeSpaceId: "space-1", + researchTaskJobId: "research-task-job-3", + tenantId: "tenant-1", + }), + ).rejects.toThrow("idempotencyKey must not be empty"); + + await expect( + repository.append({ + answer: "x".repeat(RESEARCH_TASK_PARTIAL_ANSWER_MAX_CHARS + 1), + evidenceBundle: evidenceBundle("018f0d60-7a49-7cc2-9c1b-5b36f18f6a07", "oversized answer"), + knowledgeSpaceId: "space-1", + researchTaskJobId: "research-task-job-3", + tenantId: "tenant-1", + }), + ).rejects.toThrow(`answer exceeds maxChars=${RESEARCH_TASK_PARTIAL_ANSWER_MAX_CHARS}`); }); it("rejects unbounded partial result storage and reads", async () => { diff --git a/knowledge-fs/packages/api/src/research-task-job.ts b/knowledge-fs/packages/api/src/research-task-job.ts index 08b46d5aafb..2194a4e4ecd 100644 --- a/knowledge-fs/packages/api/src/research-task-job.ts +++ b/knowledge-fs/packages/api/src/research-task-job.ts @@ -114,6 +114,7 @@ export interface ListResearchTaskJobsResult { } export interface ResearchTaskPartialResult { + answer?: string | undefined; evidenceBundle: EvidenceBundle; knowledgeSpaceId: string; researchTaskJobId: string; @@ -122,6 +123,7 @@ export interface ResearchTaskPartialResult { } export interface AppendResearchTaskPartialResultInput { + readonly answer?: string | undefined; readonly evidenceBundle: EvidenceBundle; readonly idempotencyKey?: string | undefined; readonly knowledgeSpaceId: string; @@ -146,6 +148,25 @@ export interface ResearchTaskPartialResultRepository { list(input: ListResearchTaskPartialResultsInput): Promise; } +export const RESEARCH_TASK_PARTIAL_ANSWER_MAX_CHARS = 20_000; + +export function normalizeResearchTaskPartialAnswer(answer: string | undefined): string | undefined { + if (answer === undefined) { + return undefined; + } + + const normalized = answer.trim(); + if (!normalized) { + throw new Error("Research task partial result answer must not be empty"); + } + if (normalized.length > RESEARCH_TASK_PARTIAL_ANSWER_MAX_CHARS) { + throw new Error( + `Research task partial result answer exceeds maxChars=${RESEARCH_TASK_PARTIAL_ANSWER_MAX_CHARS}`, + ); + } + return normalized; +} + export interface InMemoryResearchTaskJobRepositoryOptions { readonly capabilityGrants?: Pick | undefined; readonly maxJobs: number; @@ -678,6 +699,7 @@ export function createInMemoryResearchTaskPartialResultRepository({ return { append: async (input) => { validatePartialResultScope(input); + const answer = normalizeResearchTaskPartialAnswer(input.answer); const idempotencyKey = input.idempotencyKey?.trim(); if (input.idempotencyKey !== undefined && !idempotencyKey) { throw new Error("Research task partial result idempotencyKey must not be empty"); @@ -697,6 +719,7 @@ export function createInMemoryResearchTaskPartialResultRepository({ } const result = { + ...(answer ? { answer } : {}), evidenceBundle: EvidenceBundleSchema.parse(cloneEvidenceBundle(input.evidenceBundle)), knowledgeSpaceId: input.knowledgeSpaceId.trim(), researchTaskJobId: input.researchTaskJobId.trim(), diff --git a/knowledge-fs/packages/api/src/research-task-partial-result-database-repository.test.ts b/knowledge-fs/packages/api/src/research-task-partial-result-database-repository.test.ts index 957ee19bcb5..245e76c6ece 100644 --- a/knowledge-fs/packages/api/src/research-task-partial-result-database-repository.test.ts +++ b/knowledge-fs/packages/api/src/research-task-partial-result-database-repository.test.ts @@ -40,6 +40,7 @@ describe.each(["postgres", "tidb"] as const)( await expect( firstRepository.append({ + answer: " Final researched answer. ", evidenceBundle: evidenceBundle("bundle-1"), idempotencyKey: "partial-step-1", knowledgeSpaceId, @@ -47,6 +48,7 @@ describe.each(["postgres", "tidb"] as const)( tenantId, }), ).resolves.toMatchObject({ + answer: "Final researched answer.", evidenceBundle: { id: evidenceBundle("bundle-1").id }, knowledgeSpaceId, researchTaskJobId, @@ -62,6 +64,7 @@ describe.each(["postgres", "tidb"] as const)( 1, "partial-step-1", JSON.stringify(evidenceBundle("bundle-1")), + "Final researched answer.", now, ]); expect(insert?.sql).toContain(dialect === "postgres" ? "::jsonb" : " AS JSON"); @@ -98,7 +101,7 @@ describe.each(["postgres", "tidb"] as const)( ]); secondScript.expectDone(); - const replayRow = partialRow({ sequence: 7 }); + const replayRow = partialRow({ answer: "Previously persisted answer.", sequence: 7 }); const replayScript = scriptedDatabase(dialect, [ step("research_task_jobs", "select", [{ id: researchTaskJobId }]), step("research_task_partial_results", "select", [replayRow]), @@ -116,6 +119,7 @@ describe.each(["postgres", "tidb"] as const)( tenantId, }), ).resolves.toEqual({ + answer: "Previously persisted answer.", evidenceBundle: evidenceBundle("bundle-7"), knowledgeSpaceId, researchTaskJobId, @@ -175,6 +179,14 @@ describe.each(["postgres", "tidb"] as const)( }), ).toThrow("positive integer"); const invalidOperations = [ + () => + repository.append({ + answer: " ", + evidenceBundle: evidenceBundle("invalid"), + knowledgeSpaceId, + researchTaskJobId, + tenantId, + }), () => repository.append({ evidenceBundle: evidenceBundle("invalid"), @@ -215,6 +227,7 @@ function evidenceBundle(id: string) { function partialRow(overrides: Partial = {}): DatabaseRow { const sequence = typeof overrides.sequence === "number" ? overrides.sequence : 1; return { + answer: null, evidence_bundle: evidenceBundle(`bundle-${sequence}`), knowledge_space_id: knowledgeSpaceId, research_task_job_id: researchTaskJobId, diff --git a/knowledge-fs/packages/api/src/research-task-partial-result-database-repository.ts b/knowledge-fs/packages/api/src/research-task-partial-result-database-repository.ts index 288dfb42842..84a164e7f7a 100644 --- a/knowledge-fs/packages/api/src/research-task-partial-result-database-repository.ts +++ b/knowledge-fs/packages/api/src/research-task-partial-result-database-repository.ts @@ -8,14 +8,15 @@ import type { } from "@knowledge/core"; import { EvidenceBundleSchema } from "@knowledge/core"; -import { numberColumn, stringColumn } from "./database-row-utils"; +import { numberColumn, optionalStringColumn, stringColumn } from "./database-row-utils"; import { databasePlaceholder, quoteDatabaseIdentifier } from "./database-sql-utils"; import { jsonObjectColumn } from "./json-utils"; -import type { - AppendResearchTaskPartialResultInput, - ListResearchTaskPartialResultsInput, - ResearchTaskPartialResult, - ResearchTaskPartialResultRepository, +import { + type AppendResearchTaskPartialResultInput, + type ListResearchTaskPartialResultsInput, + type ResearchTaskPartialResult, + type ResearchTaskPartialResultRepository, + normalizeResearchTaskPartialAnswer, } from "./research-task-job"; export interface CreateDatabaseResearchTaskPartialResultRepositoryOptions { @@ -41,6 +42,7 @@ export function createDatabaseResearchTaskPartialResultRepository({ return { append: async (input) => { validateAppend(input); + const answer = normalizeResearchTaskPartialAnswer(input.answer); return database.transaction(async (transaction) => { await requireJobScope(database, transaction, input); const idempotencyKey = @@ -64,6 +66,7 @@ export function createDatabaseResearchTaskPartialResultRepository({ sequence, idempotencyKey, JSON.stringify(evidenceBundle), + answer ?? null, now(), ]; await transaction.execute({ @@ -78,6 +81,7 @@ export function createDatabaseResearchTaskPartialResultRepository({ "sequence", "idempotency_key", "evidence_bundle", + "answer", "created_at", ] .map((column) => q(database, column)) @@ -93,6 +97,7 @@ export function createDatabaseResearchTaskPartialResultRepository({ tableName: partialTable, }); return { + ...(answer ? { answer } : {}), evidenceBundle, knowledgeSpaceId: input.knowledgeSpaceId, researchTaskJobId: input.researchTaskJobId, @@ -201,7 +206,9 @@ async function getByIdempotencyKey( } function partialFromRow(row: DatabaseRow): ResearchTaskPartialResult { + const answer = normalizeResearchTaskPartialAnswer(optionalStringColumn(row, "answer")); return { + ...(answer ? { answer } : {}), evidenceBundle: EvidenceBundleSchema.parse(jsonObjectColumn(row, "evidence_bundle")), knowledgeSpaceId: stringColumn(row, "knowledge_space_id"), researchTaskJobId: stringColumn(row, "research_task_job_id"), diff --git a/knowledge-fs/packages/api/src/research-task-response-schemas.test.ts b/knowledge-fs/packages/api/src/research-task-response-schemas.test.ts index f6741d4edc0..8056d993e6b 100644 --- a/knowledge-fs/packages/api/src/research-task-response-schemas.test.ts +++ b/knowledge-fs/packages/api/src/research-task-response-schemas.test.ts @@ -45,6 +45,7 @@ describe("research-task-response-schemas", () => { ResearchTaskPartialResultListResponseSchema.parse({ items: [ { + answer: "The final researched answer.", evidenceBundle: { createdAt: "2026-05-14T00:00:00.000Z", id: UUID_A, @@ -59,7 +60,9 @@ describe("research-task-response-schemas", () => { }, ], }), - ).toMatchObject({ items: [{ sequence: 1 }] }); + ).toMatchObject({ + items: [{ answer: "The final researched answer.", sequence: 1 }], + }); }); it("rejects invalid persisted retrieval settings", () => { diff --git a/knowledge-fs/packages/api/src/research-task-response-schemas.ts b/knowledge-fs/packages/api/src/research-task-response-schemas.ts index 5a8e1a9452a..7089112550f 100644 --- a/knowledge-fs/packages/api/src/research-task-response-schemas.ts +++ b/knowledge-fs/packages/api/src/research-task-response-schemas.ts @@ -1,6 +1,8 @@ import { z } from "@hono/zod-openapi"; import { EvidenceBundleSchema } from "@knowledge/core"; +import { RESEARCH_TASK_PARTIAL_ANSWER_MAX_CHARS } from "./research-task-job"; + export const ResearchTaskJobResponseSchema = z .object({ budgetUsd: z.number().nonnegative().optional(), @@ -60,6 +62,7 @@ export const ResearchTaskJobListResponseSchema = z export const ResearchTaskPartialResultResponseSchema = z .object({ + answer: z.string().min(1).max(RESEARCH_TASK_PARTIAL_ANSWER_MAX_CHARS).optional(), evidenceBundle: EvidenceBundleSchema, knowledgeSpaceId: z.string().min(1), researchTaskJobId: z.string().min(1), diff --git a/knowledge-fs/packages/api/src/research-task-runtime.test.ts b/knowledge-fs/packages/api/src/research-task-runtime.test.ts index b73976edd02..ad33198623b 100644 --- a/knowledge-fs/packages/api/src/research-task-runtime.test.ts +++ b/knowledge-fs/packages/api/src/research-task-runtime.test.ts @@ -19,6 +19,7 @@ import type { ResearchTaskOutboxEvent, } from "./research-task-durable-repository"; import { + RESEARCH_TASK_PARTIAL_ANSWER_MAX_CHARS, type ResearchTaskJob, type ResearchTaskJobStage, createInMemoryResearchTaskPartialResultRepository, @@ -275,6 +276,8 @@ describe("research task production runtime", () => { generationInputs.push(input); yield traceStep("query.retrieve"); yield traceStep("query.answer"); + yield { delta: "The warranty ", type: "delta" as const }; + yield { delta: "is two years.", type: "delta" as const }; yield { finishReason: "retrieval-evidence", metadata: { evidenceBundle: evidenceBundle() }, @@ -324,7 +327,9 @@ describe("research task production runtime", () => { ]); await expect( partials.list({ limit: 10, researchTaskJobId: JOB_ID, tenantId: "tenant-1" }), - ).resolves.toMatchObject({ items: [{ sequence: 1 }] }); + ).resolves.toMatchObject({ + items: [{ answer: "The warranty is two years.", sequence: 1 }], + }); await expect( progress.list({ limit: 20, researchTaskJobId: JOB_ID, tenantId: "tenant-1" }), ).resolves.toMatchObject({ @@ -339,6 +344,43 @@ describe("research task production runtime", () => { }); }); + it.each([ + { + error: `Research task partial result answer exceeds maxChars=${RESEARCH_TASK_PARTIAL_ANSWER_MAX_CHARS}`, + kind: "oversized" as const, + }, + { + error: "Research task generated an answer without an evidence bundle", + kind: "missing-evidence" as const, + }, + ])("fails and retries a $kind generator result instead of persisting it", async (scenario) => { + const repository = new MemoryDurableRepository(baseJob()); + const runtime = createResearchTaskRuntime({ + ...runtimeOptions(repository), + generator: { + stream: async function* () { + yield { + delta: + scenario.kind === "oversized" + ? "x".repeat(RESEARCH_TASK_PARTIAL_ANSWER_MAX_CHARS + 1) + : "Unsupported final answer", + type: "delta" as const, + }; + }, + }, + maxRetryDelayMs: 1, + now: () => 1_000, + retryDelayMs: 1, + }); + + await expect(runtime.tick()).resolves.toMatchObject({ + failed: 0, + retryScheduled: 1, + succeeded: 0, + }); + expect(repository.job.error).toBe(scenario.error); + }); + it("reuses the frozen publication and profiles across retries without mutable reads", async () => { const frozenRuntime = publishedRuntimeSnapshot(SPACE_ID); const repository = new MemoryDurableRepository({ diff --git a/knowledge-fs/packages/api/src/research-task-runtime.ts b/knowledge-fs/packages/api/src/research-task-runtime.ts index 821f4e4e4ec..ee7bca4347d 100644 --- a/knowledge-fs/packages/api/src/research-task-runtime.ts +++ b/knowledge-fs/packages/api/src/research-task-runtime.ts @@ -38,10 +38,11 @@ import type { ResearchTaskDurableRepository, ResearchTaskExecutionFence, } from "./research-task-durable-repository"; -import type { - ResearchTaskJob, - ResearchTaskJobStage, - ResearchTaskPartialResultRepository, +import { + RESEARCH_TASK_PARTIAL_ANSWER_MAX_CHARS, + type ResearchTaskJob, + type ResearchTaskJobStage, + type ResearchTaskPartialResultRepository, } from "./research-task-job"; import type { ResearchTaskProgressEventType, @@ -602,6 +603,7 @@ async function runResearchTask({ }) : undefined); + let answer = ""; let evidenceBundle: EvidenceBundle | undefined; const iterator = generator .stream({ @@ -635,6 +637,14 @@ async function runResearchTask({ break; } const event = result.value; + if (event.type === "delta") { + if (answer.length + event.delta.length > RESEARCH_TASK_PARTIAL_ANSWER_MAX_CHARS) { + throw new Error( + `Research task partial result answer exceeds maxChars=${RESEARCH_TASK_PARTIAL_ANSWER_MAX_CHARS}`, + ); + } + answer += event.delta; + } evidenceBundle = evidenceBundleFromEvent(event) ?? evidenceBundle; if ( event.type === "trace-step" && @@ -659,9 +669,14 @@ async function runResearchTask({ await advance("generating"); } await revalidate(); + const normalizedAnswer = answer.trim(); + if (normalizedAnswer && !evidenceBundle) { + throw new Error("Research task generated an answer without an evidence bundle"); + } if (evidenceBundle) { await assertWritable(); await partials.append({ + ...(normalizedAnswer ? { answer: normalizedAnswer } : {}), evidenceBundle, idempotencyKey: `research-task:${current.id}:final-evidence`, knowledgeSpaceId: current.knowledgeSpaceId, diff --git a/knowledge-fs/packages/database/migrations/0033_research_task_final_answers.postgres.sql b/knowledge-fs/packages/database/migrations/0033_research_task_final_answers.postgres.sql new file mode 100644 index 00000000000..9fd5ffe1498 --- /dev/null +++ b/knowledge-fs/packages/database/migrations/0033_research_task_final_answers.postgres.sql @@ -0,0 +1,7 @@ +-- Knowledge Platform schema migration +-- Migration id: 0033_research_task_final_answers +-- Dialect: postgres +-- Research retrieval evidence and its final LLM synthesis are returned from the same durable row. + +ALTER TABLE "research_task_partial_results" + ADD COLUMN IF NOT EXISTS "answer" TEXT; diff --git a/knowledge-fs/packages/database/migrations/0033_research_task_final_answers.tidb.sql b/knowledge-fs/packages/database/migrations/0033_research_task_final_answers.tidb.sql new file mode 100644 index 00000000000..7e172aeef73 --- /dev/null +++ b/knowledge-fs/packages/database/migrations/0033_research_task_final_answers.tidb.sql @@ -0,0 +1,7 @@ +-- Knowledge Platform schema migration +-- Migration id: 0033_research_task_final_answers +-- Dialect: tidb +-- Research retrieval evidence and its final LLM synthesis are returned from the same durable row. + +ALTER TABLE `research_task_partial_results` + ADD COLUMN IF NOT EXISTS `answer` TEXT NULL; diff --git a/knowledge-fs/packages/database/src/migration-artifacts.generated.ts b/knowledge-fs/packages/database/src/migration-artifacts.generated.ts index 889fbdb24c0..9aa7e46e9b2 100644 --- a/knowledge-fs/packages/database/src/migration-artifacts.generated.ts +++ b/knowledge-fs/packages/database/src/migration-artifacts.generated.ts @@ -67,4 +67,6 @@ export const migrationArtifacts = [ { content: "-- Knowledge Platform schema migration\n-- Migration id: 0031_source_connection_capability_provenance\n-- Dialect: tidb\n-- Integrated source connections persist only the admitted Capability grant locator. The bearer,\n-- raw jti, Dify credential, and membership snapshot never cross this persistence boundary.\n\nALTER TABLE `source_connections`\n ADD COLUMN IF NOT EXISTS `capability_grant_id` CHAR(36) NULL;\n\nSET @source_connection_capability_fk_exists = (\n SELECT COUNT(*)\n FROM information_schema.table_constraints\n WHERE constraint_schema = DATABASE()\n AND table_name = 'source_connections'\n AND constraint_name = 'source_connections_capability_grant_fk'\n);\nSET @source_connection_capability_fk_ddl = IF(\n @source_connection_capability_fk_exists = 0,\n 'ALTER TABLE `source_connections` ADD CONSTRAINT `source_connections_capability_grant_fk` FOREIGN KEY (`tenant_id`, `knowledge_space_id`, `capability_grant_id`) REFERENCES `capability_grants` (`tenant_id`, `knowledge_space_id`, `grant_id`) ON DELETE RESTRICT',\n 'DO 0'\n);\nPREPARE source_connection_capability_fk_statement\n FROM @source_connection_capability_fk_ddl;\nEXECUTE source_connection_capability_fk_statement;\nDEALLOCATE PREPARE source_connection_capability_fk_statement;\n\nCREATE INDEX IF NOT EXISTS `source_connections_capability_grant_idx`\n ON `source_connections` (`tenant_id`, `knowledge_space_id`, `capability_grant_id`);\n", path: "packages/database/migrations/0031_source_connection_capability_provenance.tidb.sql" }, { content: "-- Knowledge Platform schema migration\n-- Migration id: 0032_capability_source_sync_policies\n-- Dialect: postgres\n-- Allows durable source sync policies to retain either a Capability grant or legacy ACL snapshot.\n\nALTER TABLE \"source_sync_policies\"\n ADD COLUMN IF NOT EXISTS \"capability_grant_id\" UUID,\n ALTER COLUMN \"requested_by_subject_id\" DROP NOT NULL,\n ALTER COLUMN \"access_channel\" DROP NOT NULL,\n ALTER COLUMN \"permission_snapshot_id\" DROP NOT NULL,\n ALTER COLUMN \"permission_snapshot_revision\" DROP NOT NULL,\n ALTER COLUMN \"required_permission_scope\" DROP NOT NULL;\n\nALTER TABLE \"source_sync_policies\"\n DROP CONSTRAINT IF EXISTS \"source_sync_policies_channel_ck\",\n DROP CONSTRAINT IF EXISTS \"source_sync_policies_revision_ck\",\n DROP CONSTRAINT IF EXISTS \"source_sync_policies_authorization_binding_ck\";\n\nALTER TABLE \"source_sync_policies\"\n ADD CONSTRAINT \"source_sync_policies_channel_ck\" CHECK (\n \"access_channel\" IS NULL\n OR \"access_channel\" IN ('interactive', 'service_api', 'mcp', 'agent')\n ),\n ADD CONSTRAINT \"source_sync_policies_revision_ck\" CHECK (\n \"revision\" >= 1 AND \"expected_source_version\" >= 1\n AND (\"capability_grant_id\" IS NOT NULL OR \"permission_snapshot_revision\" >= 1)\n ),\n ADD CONSTRAINT \"source_sync_policies_authorization_binding_ck\" CHECK (\n (\n \"capability_grant_id\" IS NOT NULL\n AND \"requested_by_subject_id\" IS NULL\n AND \"access_channel\" IS NULL\n AND \"permission_snapshot_id\" IS NULL\n AND \"permission_snapshot_revision\" IS NULL\n AND \"required_permission_scope\" IS NULL\n )\n OR (\n \"capability_grant_id\" IS NULL\n AND \"requested_by_subject_id\" IS NOT NULL\n AND \"access_channel\" IN ('interactive', 'service_api', 'mcp', 'agent')\n AND \"permission_snapshot_id\" IS NOT NULL\n AND \"permission_snapshot_revision\" >= 1\n AND \"required_permission_scope\" IS NOT NULL\n AND jsonb_typeof(\"required_permission_scope\") = 'array'\n )\n );\n\nDO $kfs_0032_source_sync_policy_capability_fk$\nBEGIN\n IF NOT EXISTS (\n SELECT 1\n FROM pg_constraint\n WHERE conname = 'source_sync_policies_capability_grant_fk'\n AND conrelid = 'source_sync_policies'::regclass\n ) THEN\n ALTER TABLE \"source_sync_policies\"\n ADD CONSTRAINT \"source_sync_policies_capability_grant_fk\"\n FOREIGN KEY (\"tenant_id\", \"knowledge_space_id\", \"capability_grant_id\")\n REFERENCES \"capability_grants\" (\"tenant_id\", \"knowledge_space_id\", \"grant_id\")\n ON DELETE RESTRICT;\n END IF;\nEND\n$kfs_0032_source_sync_policy_capability_fk$;\n\nCREATE INDEX IF NOT EXISTS \"source_sync_policies_capability_grant_idx\"\n ON \"source_sync_policies\" (\n \"tenant_id\", \"knowledge_space_id\", \"capability_grant_id\"\n );\n", path: "packages/database/migrations/0032_capability_source_sync_policies.postgres.sql" }, { content: "-- Knowledge Platform schema migration\n-- Migration id: 0032_capability_source_sync_policies\n-- Dialect: tidb\n-- Allows durable source sync policies to retain either a Capability grant or legacy ACL snapshot.\n\nALTER TABLE `source_sync_policies`\n ADD COLUMN IF NOT EXISTS `capability_grant_id` CHAR(36) NULL,\n MODIFY COLUMN `requested_by_subject_id` VARCHAR(255) NULL,\n MODIFY COLUMN `access_channel` VARCHAR(16) NULL,\n MODIFY COLUMN `permission_snapshot_id` CHAR(36) NULL,\n MODIFY COLUMN `permission_snapshot_revision` INT NULL,\n MODIFY COLUMN `required_permission_scope` JSON NULL,\n DROP CONSTRAINT `source_sync_policies_channel_ck`,\n DROP CONSTRAINT `source_sync_policies_revision_ck`,\n ADD CONSTRAINT `source_sync_policies_channel_ck` CHECK (\n `access_channel` IS NULL\n OR `access_channel` IN ('interactive', 'service_api', 'mcp', 'agent')\n ),\n ADD CONSTRAINT `source_sync_policies_revision_ck` CHECK (\n `revision` >= 1 AND `expected_source_version` >= 1\n AND (`capability_grant_id` IS NOT NULL OR `permission_snapshot_revision` >= 1)\n );\n\nSET @kfs_0032_source_sync_policy_authorization_sql = IF(\n EXISTS(\n SELECT 1\n FROM information_schema.tidb_check_constraints\n WHERE constraint_schema = DATABASE()\n AND table_name = 'source_sync_policies'\n AND constraint_name = 'source_sync_policies_authorization_binding_ck'\n ),\n 'DO 0',\n 'ALTER TABLE `source_sync_policies` ADD CONSTRAINT `source_sync_policies_authorization_binding_ck` CHECK ((`capability_grant_id` IS NOT NULL AND `requested_by_subject_id` IS NULL AND `access_channel` IS NULL AND `permission_snapshot_id` IS NULL AND `permission_snapshot_revision` IS NULL AND `required_permission_scope` IS NULL) OR (`capability_grant_id` IS NULL AND `requested_by_subject_id` IS NOT NULL AND `access_channel` IN (''interactive'', ''service_api'', ''mcp'', ''agent'') AND `permission_snapshot_id` IS NOT NULL AND `permission_snapshot_revision` >= 1 AND `required_permission_scope` IS NOT NULL AND JSON_TYPE(`required_permission_scope`) = ''ARRAY''))'\n);\nPREPARE kfs_0032_source_sync_policy_authorization_stmt\n FROM @kfs_0032_source_sync_policy_authorization_sql;\nEXECUTE kfs_0032_source_sync_policy_authorization_stmt;\nDEALLOCATE PREPARE kfs_0032_source_sync_policy_authorization_stmt;\n\nSET @kfs_0032_source_sync_policy_capability_fk_sql = IF(\n EXISTS(\n SELECT 1\n FROM information_schema.referential_constraints\n WHERE constraint_schema = DATABASE()\n AND table_name = 'source_sync_policies'\n AND constraint_name = 'source_sync_policies_capability_grant_fk'\n ),\n 'DO 0',\n 'ALTER TABLE `source_sync_policies` ADD CONSTRAINT `source_sync_policies_capability_grant_fk` FOREIGN KEY (`tenant_id`, `knowledge_space_id`, `capability_grant_id`) REFERENCES `capability_grants` (`tenant_id`, `knowledge_space_id`, `grant_id`) ON DELETE RESTRICT'\n);\nPREPARE kfs_0032_source_sync_policy_capability_fk_stmt\n FROM @kfs_0032_source_sync_policy_capability_fk_sql;\nEXECUTE kfs_0032_source_sync_policy_capability_fk_stmt;\nDEALLOCATE PREPARE kfs_0032_source_sync_policy_capability_fk_stmt;\n\nCREATE INDEX IF NOT EXISTS `source_sync_policies_capability_grant_idx`\n ON `source_sync_policies` (\n `tenant_id`, `knowledge_space_id`, `capability_grant_id`\n );\n", path: "packages/database/migrations/0032_capability_source_sync_policies.tidb.sql" }, + { content: "-- Knowledge Platform schema migration\n-- Migration id: 0033_research_task_final_answers\n-- Dialect: postgres\n-- Research retrieval evidence and its final LLM synthesis are returned from the same durable row.\n\nALTER TABLE \"research_task_partial_results\"\n ADD COLUMN IF NOT EXISTS \"answer\" TEXT;\n", path: "packages/database/migrations/0033_research_task_final_answers.postgres.sql" }, + { content: "-- Knowledge Platform schema migration\n-- Migration id: 0033_research_task_final_answers\n-- Dialect: tidb\n-- Research retrieval evidence and its final LLM synthesis are returned from the same durable row.\n\nALTER TABLE `research_task_partial_results`\n ADD COLUMN IF NOT EXISTS `answer` TEXT NULL;\n", path: "packages/database/migrations/0033_research_task_final_answers.tidb.sql" }, ] as const satisfies readonly MigrationArtifact[]; diff --git a/knowledge-fs/packages/database/src/migration-file.test.ts b/knowledge-fs/packages/database/src/migration-file.test.ts index fe41c770a99..1501c84e72b 100644 --- a/knowledge-fs/packages/database/src/migration-file.test.ts +++ b/knowledge-fs/packages/database/src/migration-file.test.ts @@ -124,6 +124,8 @@ describe("migration file rendering", () => { "packages/database/migrations/0031_source_connection_capability_provenance.tidb.sql", "packages/database/migrations/0032_capability_source_sync_policies.postgres.sql", "packages/database/migrations/0032_capability_source_sync_policies.tidb.sql", + "packages/database/migrations/0033_research_task_final_answers.postgres.sql", + "packages/database/migrations/0033_research_task_final_answers.tidb.sql", ]); expect(artifacts[2]?.content).toContain('ALTER COLUMN "dense_vector" TYPE vector'); expect(artifacts[2]?.content).not.toContain("vector(1536)"); @@ -805,6 +807,7 @@ describe("migration file rendering", () => { "packages/database/migrations/0030_bulk_operations.postgres.sql", "packages/database/migrations/0031_source_connection_capability_provenance.postgres.sql", "packages/database/migrations/0032_capability_source_sync_policies.postgres.sql", + "packages/database/migrations/0033_research_task_final_answers.postgres.sql", ]); expect( getPendingMigrationArtifacts({ @@ -841,6 +844,7 @@ describe("migration file rendering", () => { "0030_bulk_operations", "0031_source_connection_capability_provenance", "0032_capability_source_sync_policies", + "0033_research_task_final_answers", ], dialect: "postgres", }), diff --git a/knowledge-fs/packages/database/src/research-task-final-answer-migration.test.ts b/knowledge-fs/packages/database/src/research-task-final-answer-migration.test.ts new file mode 100644 index 00000000000..02876bb08ca --- /dev/null +++ b/knowledge-fs/packages/database/src/research-task-final-answer-migration.test.ts @@ -0,0 +1,24 @@ +import { readFileSync } from "node:fs"; +import { resolve } from "node:path"; +import { describe, expect, it } from "vitest"; + +const root = resolve(import.meta.dirname, "../../.."); +const postgres = readFileSync( + resolve(root, "packages/database/migrations/0033_research_task_final_answers.postgres.sql"), + "utf8", +); +const tidb = readFileSync( + resolve(root, "packages/database/migrations/0033_research_task_final_answers.tidb.sql"), + "utf8", +); + +describe("0033 Research task final answers migration", () => { + it("adds a nullable durable answer column without rewriting existing evidence rows", () => { + expect(postgres).toContain('ALTER TABLE "research_task_partial_results"'); + expect(postgres).toContain('ADD COLUMN IF NOT EXISTS "answer" TEXT'); + expect(tidb).toContain("ALTER TABLE `research_task_partial_results`"); + expect(tidb).toContain("ADD COLUMN IF NOT EXISTS `answer` TEXT NULL"); + expect(postgres).not.toMatch(/\b(?:DELETE|DROP|UPDATE)\b/u); + expect(tidb).not.toMatch(/\b(?:DELETE|DROP|UPDATE)\b/u); + }); +}); diff --git a/knowledge-fs/packages/database/src/schema.ts b/knowledge-fs/packages/database/src/schema.ts index 8f9fd46701c..96553af1fde 100644 --- a/knowledge-fs/packages/database/src/schema.ts +++ b/knowledge-fs/packages/database/src/schema.ts @@ -5313,6 +5313,7 @@ const tables = [ integerColumn("sequence"), varcharColumn("idempotency_key", 512), jsonColumn("evidence_bundle"), + textColumn("answer", true), bigintColumn("created_at"), ], },