opensource/dify
2
0
Fork 0
mirror of https://github.com/langgenius/dify.git synced 2026-09-05 08:48:10 +08:00
Code Issues Packages Projects Releases Wiki Activity

Restore Research answer synthesis routing

Browse Source
This commit is contained in:
Jyong 2026-07-29 13:32:01 -04:00
parent e653935288
commit 5869016c42
22 changed files with 447 additions and 23 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
api/knowledge-fs-contract.lock.json
View File

@ -1,7 +1,7 @@
{
"schemaVersion": 5,
"subtreeTree": "aae9105ec7afa906fe73d9f7599118caf22dafb8",
"openapiSha256": "840c3a77214d9132f2c70220fbab655d8bc791dad8b1fead6f272d4d692081fd",
"subtreeTree": "16aebf6d7f5fccf5abd2465d8a56b685d37998ae",
"openapiSha256": "a3b085216f1f4b3db7d4f024642447b87d5fd49eb85723d4e49d5890f17fc7fb",
"capabilityV2AuthManifestSha256": "fc0a47e23cce12544882f0298522b4933002e892b84ce1815df7e81d36a7a0c7",
"capabilityV2AuthTestVectorSha256": "ae0de37b1ff05c40f905cf17a7b410d8971acacf64db07d5ee3d6fecfa559ce3",
"productOperationManifestSha256": "77f2704c823d3db324c06f8f2d0109e7bda60799a6be74106ac047af3eef4492",

8
knowledge-fs/apps/api/src/answer-generation-options.test.ts
View File

@ -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 }");
});

14
knowledge-fs/apps/api/src/index.ts
View File

@ -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

8
knowledge-fs/apps/api/src/migrate.test.ts
View File

@ -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);
});

2
knowledge-fs/docs/api-reference.md
View File

@ -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`

2
knowledge-fs/packages/adapters/src/migration-runner.test.ts
View File

@ -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,

38
knowledge-fs/packages/api/src/profile-aware-query-generator.test.ts
View File

@ -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<typeof vi.fn>;

24
knowledge-fs/packages/api/src/profile-aware-query-generator.ts
View File

@ -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<QueryGenerationEvent> {
yield* (input.mode === "research" ? researchGenerator : retrievalGenerator).stream(input);
},
};
}
/**
* Selects answer synthesis without letting a configured space silently lose its reasoning model.
*

193
knowledge-fs/packages/api/src/research-task-job.test.ts
View File

@ -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 () => {

23
knowledge-fs/packages/api/src/research-task-job.ts
View File

@ -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<ListResearchTaskPartialResultsResult>;
}
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<CapabilityGrantProvenanceRepository, "get"> | 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(),

15
knowledge-fs/packages/api/src/research-task-partial-result-database-repository.test.ts
View File

@ -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> = {}): 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,

19
knowledge-fs/packages/api/src/research-task-partial-result-database-repository.ts
View File

@ -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"),

5
knowledge-fs/packages/api/src/research-task-response-schemas.test.ts
View File

@ -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", () => {

3
knowledge-fs/packages/api/src/research-task-response-schemas.ts
View File

@ -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),

44
knowledge-fs/packages/api/src/research-task-runtime.test.ts
View File

@ -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({

23
knowledge-fs/packages/api/src/research-task-runtime.ts
View File

@ -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,

7
knowledge-fs/packages/database/migrations/0033_research_task_final_answers.postgres.sql Normal file
View File

@ -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;

7
knowledge-fs/packages/database/migrations/0033_research_task_final_answers.tidb.sql Normal file
View File

@ -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;

2
knowledge-fs/packages/database/src/migration-artifacts.generated.ts
View File

@ -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[];

4
knowledge-fs/packages/database/src/migration-file.test.ts
View File

@ -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",
}),

24
knowledge-fs/packages/database/src/research-task-final-answer-migration.test.ts Normal file
View File

@ -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);
});
});

1
knowledge-fs/packages/database/src/schema.ts
View File

@ -5313,6 +5313,7 @@ const tables = [
integerColumn("sequence"),
varcharColumn("idempotency_key", 512),
jsonColumn("evidence_bundle"),
textColumn("answer", true),
bigintColumn("created_at"),
],
},