diff --git a/apps/api/src/config-api.ts b/apps/api/src/config-api.ts index f4065fd..dc3beee 100644 --- a/apps/api/src/config-api.ts +++ b/apps/api/src/config-api.ts @@ -84,6 +84,7 @@ import { import { buildSessionTraceDag } from "./trace-dag.js"; import { readMultipartFiles, readMultipartUpload } from "./upload-parser.js"; import { knowledgeDocumentTextFromFile } from "./knowledge-document-text.js"; +import { resolveLiveSessionActiveRun } from "./stale-active-runs.js"; const MAX_JSON_BODY_BYTES = 1024 * 1024; const DEFAULT_WORKSPACE_ID = "default"; @@ -445,7 +446,17 @@ const handleSessionRequest = async ( ...(cursor ? { cursor } : {}) }); return ok({ - sessions: records.map(sessionListDto), + sessions: records.map((session) => + sessionListDto( + session, + resolveLiveSessionActiveRun({ + metadataStore: context.metadataStore, + runCancelRegistry: context.runCancelRegistry, + userId: context.userId, + sessionId: session.id + }) + ) + ), ...(records.length === limit ? { nextCursor: encodeSessionCursor(records.at(-1) as SessionRecord) } : {}) }); } @@ -466,6 +477,20 @@ const handleSessionRequest = async ( }); return ok(sessionTitleDto(session)); } + if (!action && request.method === "DELETE") { + const result = context.metadataStore.sessions.delete({ + user_id: context.userId, + session_id: sessionId + }); + if (!result.deleted) { + throw new Error(`Session not found: ${sessionId}`); + } + return ok({ + sessionId, + deleted: true, + deletedSessionIds: result.deletedSessionIds + }); + } if (action === "branches" && request.method === "POST") { const body = await readJsonBody(request); const runId = stringValue(body.runId ?? body.run_id); @@ -594,6 +619,13 @@ const handleSessionRequest = async ( authoritativeToolNames ); + const activeRun = resolveLiveSessionActiveRun({ + metadataStore: context.metadataStore, + runCancelRegistry: context.runCancelRegistry, + userId: context.userId, + sessionId + }); + return ok({ sessionId, title: session.title ?? "", @@ -609,7 +641,8 @@ const handleSessionRequest = async ( pendingInteractions: pendingInteractions.map(pendingInteractionDto), restorableCustomEvents: runEventGroups.flatMap(({ runId, events }) => restorableCustomEventDtos(runId, events) - ) + ), + ...(activeRun ? { activeRun: sessionActiveRunDto(activeRun) } : {}) }); }; @@ -2383,21 +2416,45 @@ const conversationMessageEvidenceRefsDto = (message: ConversationMessageRecord): return refs.length > 0 ? { evidenceRefs: refs } : {}; }; -const sessionListDto = (session: SessionRecord): Record => ({ +const truncateUserInputPreview = (value: string, maxLength = 120): string => { + const normalized = value.replace(/\s+/g, " ").trim(); + if (normalized.length <= maxLength) { + return normalized; + } + return `${normalized.slice(0, Math.max(0, maxLength - 1))}…`; +}; + +const sessionActiveRunDto = (run: RunRecord): Record => ({ + sessionId: run.session_id, + activeRunId: run.id, + status: run.status, + startedAt: run.started_at, + userInputPreview: truncateUserInputPreview(run.user_input) +}); + +const sessionListDto = ( + session: SessionRecord, + activeRun?: RunRecord | null +): Record => ({ id: session.id, threadId: session.id, title: session.title ?? "", titleSource: session.title_source ?? "fallback", createdAt: session.created_at, updatedAt: session.updated_at, - lastMessageAt: session.last_message_at ?? session.updated_at + lastMessageAt: session.last_message_at ?? session.updated_at, + ...(activeRun ? { activeRun: sessionActiveRunDto(activeRun) } : {}) }); const sessionBranchCreatedDto = ( input: { branch: SessionBranchRecord; session: SessionRecord } ): Record => ({ ...sessionBranchDto(input.branch, input.session), - session: sessionListDto(input.session) + session: sessionListDto( + input.session, + // Branch create does not need active-run lookup; omit to keep response light. + null + ) }); const sessionBranchDto = ( diff --git a/apps/api/src/evidence-reference-context.ts b/apps/api/src/evidence-reference-context.ts index cc7502b..46d9ef6 100644 --- a/apps/api/src/evidence-reference-context.ts +++ b/apps/api/src/evidence-reference-context.ts @@ -96,14 +96,22 @@ const artifactEvidenceItems = ( const maxChars = input.maxCharsPerEvidence ?? 6000; const focus = resolveSelectionFocus(ref.source.selection, preview, maxChars); const includeFullPreview = preview !== undefined && (!focus || !focus.replaceFullPreview); + // When a table subset is already inlined, omit file_id so the model is not nudged + // to open the full artifact file for a partial cite. + const fileIdLine = artifact.file_asset_ref_id && !focus?.replaceFullPreview + ? `file_id=${artifact.file_asset_ref_id}` + : undefined; const content = evidenceText({ body: [ `artifact_id=${artifact.id}`, `artifact_type=${artifact.type}`, `artifact_name=${artifact.name}`, - artifact.file_asset_ref_id ? `file_id=${artifact.file_asset_ref_id}` : undefined, + fileIdLine, includeFullPreview ? `preview=${boundJson(preview, maxChars)}` : undefined, metadata !== undefined ? `metadata=${boundJson(metadata, 1200)}` : undefined, + focus?.replaceFullPreview + ? "note: selected subset is already inlined below; do not open the full artifact file unless the user asks for broader context" + : undefined, ...(focus?.lines ?? []) ], ref diff --git a/apps/api/src/run-cancel-registry.ts b/apps/api/src/run-cancel-registry.ts index 237f912..30e88c5 100644 --- a/apps/api/src/run-cancel-registry.ts +++ b/apps/api/src/run-cancel-registry.ts @@ -41,6 +41,11 @@ export class RunCancelRegistry { return { canceled: true, runId: handle.runId, sessionId: handle.sessionId }; } + /** True when this process still owns a live cancel handle for the run. */ + has(input: { runId: string; userId: string }): boolean { + return this.handles.has(this.key(input.userId, input.runId)); + } + private key(userId: string, runId: string): string { return `${userId}:${runId}`; } diff --git a/apps/api/src/run-identity-orchestrator.ts b/apps/api/src/run-identity-orchestrator.ts index e36d2d0..69c415f 100644 --- a/apps/api/src/run-identity-orchestrator.ts +++ b/apps/api/src/run-identity-orchestrator.ts @@ -8,6 +8,8 @@ import { validateParentRun } from "./run-identity.js"; import type { EffectiveRunConfig } from "./run-input.js"; +import type { RunCancelRegistry } from "./run-cancel-registry.js"; +import { resolveLiveSessionActiveRun } from "./stale-active-runs.js"; export type RunIdentityResolution = | { @@ -25,6 +27,7 @@ type ResolveRunIdentityInput = { interactionResume?: InteractionResume | undefined; metadataStore: MetadataStore; modelName: string; + runCancelRegistry: RunCancelRegistry; runEventWriter: RunEventWriter; runInput: RunAgentInput; userId: string; @@ -68,10 +71,12 @@ export const resolveRunIdentity = (input: ResolveRunIdentityInput): RunIdentityR }; } - const activeSessionRun = input.metadataStore.runs.findActiveBySession({ - user_id: input.userId, - session_id: sessionId, - exclude_run_id: runId + const activeSessionRun = resolveLiveSessionActiveRun({ + metadataStore: input.metadataStore, + runCancelRegistry: input.runCancelRegistry, + userId: input.userId, + sessionId, + excludeRunId: runId }); if (activeSessionRun) { diff --git a/apps/api/src/run-input.ts b/apps/api/src/run-input.ts index cd249c1..ded31e9 100644 --- a/apps/api/src/run-input.ts +++ b/apps/api/src/run-input.ts @@ -1,5 +1,5 @@ import type { RunAgentInput } from "@ag-ui/client"; -import type { EvidenceKind, EvidenceRef } from "@datafoundry/contracts"; +import type { EvidenceKind, EvidenceRef, EvidenceSelection } from "@datafoundry/contracts"; import type { ConfigResourceKind, MetadataStore } from "@datafoundry/metadata"; import type { SkillMode, SkillPolicyConfig } from "@datafoundry/skills"; @@ -546,6 +546,7 @@ const evidenceRefFromUnknown = (value: unknown): EvidenceRef | undefined => { const evidenceRefSourceFromUnknown = (value: unknown): EvidenceRef["source"] => { const source = isRecord(value) ? value : {}; + const selection = evidenceSelectionFromUnknown(source.selection); return { ...optionalStringField(source, "artifactId", "artifact_id"), ...optionalStringField(source, "toolCallId", "tool_call_id"), @@ -555,10 +556,54 @@ const evidenceRefSourceFromUnknown = (value: unknown): EvidenceRef["source"] => ...optionalStringField(source, "datasourceId", "datasource_id"), ...optionalStringField(source, "tableName", "table_name"), ...optionalStringField(source, "documentId", "document_id"), - ...optionalStringField(source, "chunkId", "chunk_id") + ...optionalStringField(source, "chunkId", "chunk_id"), + ...(selection ? { selection } : {}) }; }; +/** Parses fine-grained table/text selections so partial cites survive run_config intake. */ +const evidenceSelectionFromUnknown = (value: unknown): EvidenceSelection | undefined => { + if (!isRecord(value) || typeof value.mode !== "string") { + return undefined; + } + if (value.mode === "text") { + const quote = typeof value.quote === "string" ? value.quote.trim() : ""; + if (!quote) return undefined; + const offset = typeof value.offset === "number" && Number.isFinite(value.offset) + ? Math.trunc(value.offset) + : undefined; + return offset === undefined ? { mode: "text", quote } : { mode: "text", quote, offset }; + } + if (value.mode !== "cells" && value.mode !== "rows" && value.mode !== "cols") { + return undefined; + } + const range = evidenceCellRangeFromUnknown(value.range); + if (!range) return undefined; + const columns = Array.isArray(value.columns) + ? value.columns.filter((column): column is string => typeof column === "string" && column.length > 0) + : undefined; + return columns && columns.length > 0 + ? { mode: value.mode, range, columns } + : { mode: value.mode, range }; +}; + +const evidenceCellRangeFromUnknown = ( + value: unknown +): { r0: number; c0: number; r1: number; c1: number } | undefined => { + if (!isRecord(value)) return undefined; + const r0 = finiteIndex(value.r0); + const c0 = finiteIndex(value.c0); + const r1 = finiteIndex(value.r1); + const c1 = finiteIndex(value.c1); + if (r0 === undefined || c0 === undefined || r1 === undefined || c1 === undefined) { + return undefined; + } + return { r0, c0, r1, c1 }; +}; + +const finiteIndex = (value: unknown): number | undefined => + typeof value === "number" && Number.isFinite(value) ? Math.trunc(value) : undefined; + const optionalStringField = ( record: Record, camelKey: keyof EvidenceRef["source"], diff --git a/apps/api/src/server.ts b/apps/api/src/server.ts index a7e3fca..4655d5b 100644 --- a/apps/api/src/server.ts +++ b/apps/api/src/server.ts @@ -40,6 +40,7 @@ import { Observable } from "rxjs"; import { handleConfigApiRequest } from "./config-api.js"; import { createAsyncMemoByKey, createStartupTimer } from "./async-memo.js"; import { ensureBuiltinDtcGrowthDatasource } from "./builtin-dtc-growth-datasource.js"; +import { reclaimOrphanedQueuedAndRunningRuns } from "./stale-active-runs.js"; import { loadPasswordAuthConfig, type PasswordAuthConfig } from "./auth/config.js"; import { AuthService, type AuthIdentity } from "./auth/service.js"; import { serverDefaultConnectionStatus, isServerLlmEnvConfigured } from "./model-profile-connection-status.js"; @@ -220,6 +221,17 @@ export const createServer = async (options: CreateServerOptions = {}): Promise + reclaimOrphanedQueuedAndRunningRuns({ + metadataStore, + runCancelRegistry, + }), + ); + if (reclaimedActiveRuns > 0) { + console.log(`[startup] stale active run reclaim: canceled=${reclaimedActiveRuns}`); + } + startupTimings = timer.timings(); startupTotalMs = timer.totalMs(); serverReady = true; @@ -527,6 +539,7 @@ class DataFoundryAgUiAgent extends AbstractAgent { ...(interactionResume ? { interactionResume } : {}), metadataStore: this.input.metadataStore, modelName: modelProvider.model_name, + runCancelRegistry: this.input.runCancelRegistry, runEventWriter, runInput: normalizedRunInput, userId: this.input.user.id, @@ -915,11 +928,10 @@ class DataFoundryAgUiAgent extends AbstractAgent { if (!isResume && !sessionTitleStarted) { sessionTitleStarted = true; startSessionTitleTask({ - emit: (titleEvent) => { - if (!terminalStarted) { - emit(titleEvent); - } - }, + // Title generation is async and may finish after the agent run + // terminals; still forward the event while the stream is open + // (finalizer continues emitting after RUN_FINISHED). + emit, metadataStore: this.input.metadataStore, model: modelProvider.model, modelTemperature: modelSettings?.temperature, diff --git a/apps/api/src/session-title.ts b/apps/api/src/session-title.ts index b52e007..884c423 100644 --- a/apps/api/src/session-title.ts +++ b/apps/api/src/session-title.ts @@ -3,7 +3,9 @@ import { createCustomEvent } from "@datafoundry/agent-runtime"; import type { MetadataStore, SessionRecord } from "@datafoundry/metadata"; import type { BaseEvent } from "@ag-ui/client"; -const TITLE_TIMEOUT_MS = 5000; +/** Reasoning models (e.g. deepseek-v4-pro) spend output budget on thinking first. */ +const TITLE_TIMEOUT_MS = 15_000; +const TITLE_MAX_OUTPUT_TOKENS = 256; const TITLE_MAX_CHARS = 32; export type SessionTitleTaskInput = { @@ -26,17 +28,18 @@ const generateAndPersistSessionTitle = async (input: SessionTitleTaskInput): Pro user_id: input.userId, session_id: input.sessionId }); - if (current.title_source === "user" || (current.title && current.title.trim().length > 0)) { + // Only skip manual renames / already-finalized LLM titles. Empty or fallback + // titles (and legacy rows with a prefilled title but no source) may still be replaced. + if (current.title_source === "user" || current.title_source === "llm") { return; } const title = await generateLlmTitle(input).catch(() => fallbackTitle(input.userInput)); - const source: SessionRecord["title_source"] = title.source; const updated = input.metadataStore.sessions.updateAutoTitleIfAllowed({ user_id: input.userId, session_id: input.sessionId, title: title.title, - title_source: source === "llm" ? "llm" : "fallback" + title_source: title.source === "llm" ? "llm" : "fallback" }); if (!updated) { return; @@ -61,7 +64,7 @@ const generateLlmTitle = async ( defaultOptions: { maxSteps: 1, modelSettings: { - maxOutputTokens: 32, + maxOutputTokens: TITLE_MAX_OUTPUT_TOKENS, temperature: input.modelTemperature ?? 0.2 }, providerOptions: { @@ -74,7 +77,11 @@ const generateLlmTitle = async ( const output = await agent.generate(buildTitlePrompt(input.userInput), { abortSignal: AbortSignal.timeout(TITLE_TIMEOUT_MS) }); - const title = sanitizeTitle(output.text) || fallbackTitle(input.userInput).title; + const title = sanitizeTitle(output.text); + if (!title) { + // Empty text usually means the output budget was spent on reasoning tokens. + throw new Error("SESSION_TITLE_EMPTY"); + } return { source: "llm", title }; }; diff --git a/apps/api/src/stale-active-runs.test.ts b/apps/api/src/stale-active-runs.test.ts new file mode 100644 index 0000000..73cc8ff --- /dev/null +++ b/apps/api/src/stale-active-runs.test.ts @@ -0,0 +1,167 @@ +import { describe, expect, it, vi } from "vitest"; + +import { RunCancelRegistry } from "./run-cancel-registry.js"; +import { + reclaimOrphanedQueuedAndRunningRuns, + resolveLiveSessionActiveRun, +} from "./stale-active-runs.js"; + +type FakeRun = { + id: string; + user_id: string; + session_id: string; + status: "queued" | "running" | "suspended" | "canceled" | "completed" | "failed"; + error_message?: string | null; +}; + +function createFakeStore(initial: FakeRun[]) { + const runs = [...initial]; + return { + runs: { + findActiveBySession(input: { + exclude_run_id?: string; + session_id: string; + user_id: string; + }) { + return ( + runs.find( + (run) => + run.user_id === input.user_id && + run.session_id === input.session_id && + (run.status === "queued" || + run.status === "running" || + run.status === "suspended") && + run.id !== input.exclude_run_id, + ) ?? undefined + ); + }, + listByStatuses(input: { statuses: Array }) { + return runs.filter((run) => input.statuses.includes(run.status)); + }, + updateStatus(input: { + user_id: string; + run_id: string; + status: FakeRun["status"]; + error_message?: string; + }) { + const run = runs.find( + (item) => item.user_id === input.user_id && item.id === input.run_id, + ); + if (!run) { + throw new Error(`missing ${input.run_id}`); + } + run.status = input.status; + run.error_message = input.error_message ?? null; + return run; + }, + }, + }; +} + +describe("resolveLiveSessionActiveRun", () => { + it("reclaims orphaned queued/running rows without a live cancel handle", () => { + const store = createFakeStore([ + { + id: "run-zombie", + user_id: "u1", + session_id: "s1", + status: "running", + }, + ]); + const registry = new RunCancelRegistry(); + + const active = resolveLiveSessionActiveRun({ + metadataStore: store as never, + runCancelRegistry: registry, + userId: "u1", + sessionId: "s1", + }); + + expect(active).toBeNull(); + expect(store.runs.findActiveBySession({ user_id: "u1", session_id: "s1" })).toBeUndefined(); + }); + + it("keeps live running runs and suspended HITL locks", () => { + const store = createFakeStore([ + { + id: "run-live", + user_id: "u1", + session_id: "s1", + status: "running", + }, + { + id: "run-hitl", + user_id: "u1", + session_id: "s2", + status: "suspended", + }, + ]); + const registry = new RunCancelRegistry(); + registry.register({ + cancel: vi.fn(), + runId: "run-live", + sessionId: "s1", + userId: "u1", + }); + + expect( + resolveLiveSessionActiveRun({ + metadataStore: store as never, + runCancelRegistry: registry, + userId: "u1", + sessionId: "s1", + })?.id, + ).toBe("run-live"); + + expect( + resolveLiveSessionActiveRun({ + metadataStore: store as never, + runCancelRegistry: registry, + userId: "u1", + sessionId: "s2", + })?.id, + ).toBe("run-hitl"); + }); +}); + +describe("reclaimOrphanedQueuedAndRunningRuns", () => { + it("cancels orphaned queued/running rows on startup", () => { + const store = createFakeStore([ + { + id: "run-a", + user_id: "u1", + session_id: "s1", + status: "queued", + }, + { + id: "run-b", + user_id: "u1", + session_id: "s2", + status: "running", + }, + { + id: "run-c", + user_id: "u1", + session_id: "s3", + status: "suspended", + }, + ]); + const registry = new RunCancelRegistry(); + registry.register({ + cancel: vi.fn(), + runId: "run-b", + sessionId: "s2", + userId: "u1", + }); + + const reclaimed = reclaimOrphanedQueuedAndRunningRuns({ + metadataStore: store as never, + runCancelRegistry: registry, + }); + + expect(reclaimed).toBe(1); + expect(store.runs.listByStatuses({ statuses: ["queued"] })).toHaveLength(0); + expect(store.runs.listByStatuses({ statuses: ["running"] })[0]?.id).toBe("run-b"); + expect(store.runs.listByStatuses({ statuses: ["suspended"] })[0]?.id).toBe("run-c"); + }); +}); diff --git a/apps/api/src/stale-active-runs.ts b/apps/api/src/stale-active-runs.ts new file mode 100644 index 0000000..c9dc4e3 --- /dev/null +++ b/apps/api/src/stale-active-runs.ts @@ -0,0 +1,71 @@ +import type { MetadataStore, RunRecord } from "@datafoundry/metadata"; + +import type { RunCancelRegistry } from "./run-cancel-registry.js"; + +const LIVE_CLAIM_STATUSES = new Set(["queued", "running"]); + +/** + * Active metadata rows can outlive the in-process worker after crashes/restarts. + * Suspended (HITL) runs stay locked; queued/running without a cancel handle are + * treated as orphans and canceled so a new claim can proceed. + */ +export function resolveLiveSessionActiveRun(input: { + excludeRunId?: string; + metadataStore: MetadataStore; + runCancelRegistry: RunCancelRegistry; + sessionId: string; + userId: string; +}): RunRecord | null { + for (;;) { + const active = input.metadataStore.runs.findActiveBySession({ + user_id: input.userId, + session_id: input.sessionId, + ...(input.excludeRunId ? { exclude_run_id: input.excludeRunId } : {}) + }); + if (!active) { + return null; + } + if (active.status === "suspended") { + return active; + } + if ( + LIVE_CLAIM_STATUSES.has(active.status) + && input.runCancelRegistry.has({ userId: input.userId, runId: active.id }) + ) { + return active; + } + if (!LIVE_CLAIM_STATUSES.has(active.status)) { + return active; + } + input.metadataStore.runs.updateStatus({ + user_id: input.userId, + run_id: active.id, + status: "canceled", + error_message: "STALE_ACTIVE_RUN_RECLAIMED" + }); + } +} + +/** Cancel queued/running rows that have no live cancel handle (e.g. after process restart). */ +export function reclaimOrphanedQueuedAndRunningRuns(input: { + metadataStore: MetadataStore; + runCancelRegistry: RunCancelRegistry; +}): number { + const candidates = input.metadataStore.runs.listByStatuses({ + statuses: ["queued", "running"] + }); + let reclaimed = 0; + for (const run of candidates) { + if (input.runCancelRegistry.has({ userId: run.user_id, runId: run.id })) { + continue; + } + input.metadataStore.runs.updateStatus({ + user_id: run.user_id, + run_id: run.id, + status: "canceled", + error_message: "STALE_ACTIVE_RUN_RECLAIMED" + }); + reclaimed += 1; + } + return reclaimed; +} diff --git a/apps/tui/src/config/config-client.ts b/apps/tui/src/config/config-client.ts index 9d72948..d0b3eb2 100644 --- a/apps/tui/src/config/config-client.ts +++ b/apps/tui/src/config/config-client.ts @@ -307,6 +307,14 @@ const CapabilitiesSchema = z.object({ // ==================== Session Schemas ==================== +const SessionActiveRunSchema = z.object({ + sessionId: z.string(), + activeRunId: z.string(), + status: z.enum(["queued", "running", "suspended"]), + startedAt: z.string(), + userInputPreview: z.string(), +}); + const SessionListItemSchema = z.object({ id: z.string(), threadId: z.string(), @@ -315,6 +323,7 @@ const SessionListItemSchema = z.object({ createdAt: z.string().optional(), updatedAt: z.string().optional(), lastMessageAt: z.string().optional(), + activeRun: SessionActiveRunSchema.nullish(), }); const SessionListResponseSchema = z.object({ @@ -401,6 +410,7 @@ const SessionConversationSchema = z.object({ runEventRefs: z.array(ConversationRunEventRefSchema), checkpoints: z.array(ConversationCheckpointSchema).optional(), toolCalls: z.array(ConversationToolCallSchema), + activeRun: SessionActiveRunSchema.nullish(), }); const SessionArtifactSchema = z.object({ @@ -828,6 +838,17 @@ export class ConfigClient { ); } + async deleteSession(sessionId: string): Promise<{ + sessionId: string; + deleted: boolean; + deletedSessionIds: string[]; + }> { + return this.request( + "DELETE", + `/api/v1/sessions/${encodeURIComponent(sessionId)}`, + ); + } + // ==================== Model Profile Methods ==================== async listModels(): Promise { diff --git a/apps/tui/src/state/data-task-state.ts b/apps/tui/src/state/data-task-state.ts index 07ae324..37bf671 100755 --- a/apps/tui/src/state/data-task-state.ts +++ b/apps/tui/src/state/data-task-state.ts @@ -405,6 +405,36 @@ export const DATA_SKILLS: DataSkill[] = [ name: "数据分析", description: "回答指标查询、深度分析和报告类数据问题", }, + { + id: "tabular-file-import", + name: "表格文件导入", + description: "将 CSV/Excel/JSON/Parquet 导入工作区并规范化", + }, + { + id: "data-cleaning-for-load", + name: "入库前清洗", + description: "清洗校验表格数据，处理缺失、类型与去重", + }, + { + id: "batch-file-merge", + name: "批量文件合并", + description: "多文件对齐字段、去重合并并生成合并报告", + }, + { + id: "etl-pipeline-patterns", + name: "ETL 管道模式", + description: "设计可靠的 staging、幂等与增量导入流程", + }, + { + id: "api-json-ingest", + name: "API/JSON 接入", + description: "从 API 或 JSON/JSONL 拉取并展平为表格文件", + }, + { + id: "database-load-planning", + name: "数据库入库规划", + description: "映射字段并规划只读校验后的装载策略", + }, ]; export const DEFAULT_SKILL_ID = DATA_SKILLS[0]!.id; diff --git a/apps/web/src/app/data-tasks/__tests__/config-api-adapter.test.ts b/apps/web/src/app/data-tasks/__tests__/config-api-adapter.test.ts index e061e9a..3e5f7a2 100644 --- a/apps/web/src/app/data-tasks/__tests__/config-api-adapter.test.ts +++ b/apps/web/src/app/data-tasks/__tests__/config-api-adapter.test.ts @@ -494,8 +494,18 @@ describe("config api adapter", () => { }), { headers: { "Content-Type": "application/json" }, status: 200 })); vi.stubGlobal("fetch", fetchMock); + fetchMock.mockResolvedValueOnce(new Response(JSON.stringify({ + success: true, + data: { + sessionId: "thread-1", + deleted: true, + deletedSessionIds: ["thread-1", "thread-1-branch"], + }, + }), { headers: { "Content-Type": "application/json" }, status: 200 })); + const list = await configApi.listSessions({ limit: 20, cursor: "cursor-1" }); const patched = await configApi.patchSessionTitle("thread-1", "我的复盘"); + const deleted = await configApi.deleteSession("thread-1"); expect(fetchMock).toHaveBeenNthCalledWith( 1, @@ -510,10 +520,17 @@ describe("config api adapter", () => { body: JSON.stringify({ title: "我的复盘" }), }), ); + expect(fetchMock).toHaveBeenNthCalledWith( + 3, + "http://config.test/api/v1/sessions/thread-1", + expect.objectContaining({ method: "DELETE" }), + ); expect(list.sessions[0]?.title).toBe("渠道订单分析"); expect(list.nextCursor).toBe("cursor-2"); expect(patched.sessionId).toBe("thread-1"); expect(patched.titleSource).toBe("user"); + expect(deleted.deleted).toBe(true); + expect(deleted.deletedSessionIds).toEqual(["thread-1", "thread-1-branch"]); }); it("calls datalink endpoints through the config client", async () => { diff --git a/apps/web/src/app/data-tasks/__tests__/queued-chat-runs.test.ts b/apps/web/src/app/data-tasks/__tests__/queued-chat-runs.test.ts index f049875..d758fb1 100644 --- a/apps/web/src/app/data-tasks/__tests__/queued-chat-runs.test.ts +++ b/apps/web/src/app/data-tasks/__tests__/queued-chat-runs.test.ts @@ -5,7 +5,11 @@ import { createQueuedChatPrompt, deleteQueuedChatPrompt, editQueuedChatPrompt, + isForeignSessionActiveRun, + loadQueuedChatPrompts, markQueuedChatPromptInterrupting, + parseAlreadyActiveRunId, + persistQueuedChatPrompts, queuedPromptToRunInput, resolveQueuedSubmitMode, shouldShowRunningStopControl, @@ -33,6 +37,7 @@ const forwardedProps = (datasourceId: string): RunForwardedProps => ({ fileIds: [], pinnedPaths: [], evidenceRefs: [], + activeSkillId: "", }, }); @@ -210,4 +215,65 @@ describe("queued chat runs", () => { expect(runInput.forwardedProps).not.toEqual(latestProps); expect(runInput.text).toBe("use saved context"); }); + + it("detects foreign active runs for multi-device session locks", () => { + const activeRun = { + sessionId: "thread-1", + activeRunId: "run-other", + status: "running" as const, + startedAt: "2026-07-15T00:00:00.000Z", + userInputPreview: "hello", + }; + expect(isForeignSessionActiveRun(activeRun, null)).toBe(true); + expect(isForeignSessionActiveRun(activeRun, "run-local")).toBe(true); + expect(isForeignSessionActiveRun(activeRun, "run-other")).toBe(false); + expect(isForeignSessionActiveRun(activeRun, null, true)).toBe(false); + expect(isForeignSessionActiveRun(null, null)).toBe(false); + }); + + it("parses RUN_ALREADY_ACTIVE run ids for unlock retries", () => { + expect(parseAlreadyActiveRunId("RUN_ALREADY_ACTIVE:run-abc")).toBe("run-abc"); + expect(parseAlreadyActiveRunId("RUN_ALREADY_ACTIVE run-abc")).toBe("run-abc"); + expect(parseAlreadyActiveRunId("Error: RUN_ALREADY_ACTIVE:run-xyz")).toBe("run-xyz"); + expect(parseAlreadyActiveRunId("something else")).toBeNull(); + }); + + it("persists queued prompts across refresh for a thread", () => { + const memory = new Map(); + const storage = { + getItem: (key: string) => memory.get(key) ?? null, + setItem: (key: string, value: string) => { + memory.set(key, value); + }, + removeItem: (key: string) => { + memory.delete(key); + }, + }; + Object.defineProperty(globalThis, "window", { + configurable: true, + value: { sessionStorage: storage }, + }); + Object.defineProperty(globalThis, "sessionStorage", { + configurable: true, + value: storage, + }); + + const prompt = createQueuedChatPrompt({ + id: "queued-1", + text: "persist me", + attachments: [attachment("att-1")], + forwardedProps: forwardedProps("db-1"), + createdAt: 42, + }); + persistQueuedChatPrompts("thread-1", [prompt]); + expect(loadQueuedChatPrompts("thread-1")).toEqual([ + expect.objectContaining({ + id: "queued-1", + text: "persist me", + status: "queued", + }), + ]); + persistQueuedChatPrompts("thread-1", []); + expect(loadQueuedChatPrompts("thread-1")).toEqual([]); + }); }); diff --git a/apps/web/src/app/data-tasks/__tests__/run-error-message.test.ts b/apps/web/src/app/data-tasks/__tests__/run-error-message.test.ts index e053156..d8da4b5 100644 --- a/apps/web/src/app/data-tasks/__tests__/run-error-message.test.ts +++ b/apps/web/src/app/data-tasks/__tests__/run-error-message.test.ts @@ -30,4 +30,8 @@ describe("formatRunErrorMessage", () => { it("falls back when message is empty", () => { expect(formatRunErrorMessage()).toBe("Agent run failed"); }); + + it("explains RUN_ALREADY_ACTIVE session locks", () => { + expect(formatRunErrorMessage("RUN_ALREADY_ACTIVE:run-1")).toContain("active run"); + }); }); diff --git a/apps/web/src/app/data-tasks/__tests__/session-conversation-scroll-restore.test.tsx b/apps/web/src/app/data-tasks/__tests__/session-conversation-scroll-restore.test.tsx index 7cca1bb..b021529 100644 --- a/apps/web/src/app/data-tasks/__tests__/session-conversation-scroll-restore.test.tsx +++ b/apps/web/src/app/data-tasks/__tests__/session-conversation-scroll-restore.test.tsx @@ -21,8 +21,22 @@ const agentState = { }; const restoreGate = { - isRestoringConversation: false, - setIsRestoringConversation: vi.fn(), + restoringThreadIds: new Set(), + restoredThreadIds: new Set(), + isThreadRestoring: (threadId: string | null | undefined) => + Boolean(threadId && restoreGate.restoringThreadIds.has(threadId)), + isThreadRestored: (threadId: string | null | undefined) => + Boolean(threadId && restoreGate.restoredThreadIds.has(threadId)), + setThreadRestoring: vi.fn((threadId: string, restoring: boolean) => { + if (restoring) { + restoreGate.restoringThreadIds.add(threadId); + } else { + restoreGate.restoringThreadIds.delete(threadId); + } + }), + markThreadRestored: vi.fn((threadId: string) => { + restoreGate.restoredThreadIds.add(threadId); + }), }; vi.mock("@copilotkit/react-core/v2", () => ({ @@ -75,7 +89,8 @@ describe("SessionConversationScrollRestore branch switch regression", () => { document.body.innerHTML = ""; agentState.messages = [{ id: "m1", role: "user" }]; agentState.threadId = "thread-a"; - restoreGate.isRestoringConversation = false; + restoreGate.restoringThreadIds.clear(); + restoreGate.restoredThreadIds.clear(); setConversationScrollIntent({ kind: "bottom" }); host = document.createElement("div"); document.body.appendChild(host); @@ -98,7 +113,7 @@ describe("SessionConversationScrollRestore branch switch regression", () => { setConversationScrollIntent({ kind: "preserve", scrollTop: 240 }); agentState.threadId = "thread-b"; - restoreGate.isRestoringConversation = true; + restoreGate.restoringThreadIds.add("thread-b"); await act(async () => { root!.render(createElement(SessionConversationScrollRestore, { agentId: "dataFoundry" })); @@ -107,7 +122,7 @@ describe("SessionConversationScrollRestore branch switch regression", () => { // Still restoring: must not jump yet. expect(scroll.scrollTop).toBe(240); - restoreGate.isRestoringConversation = false; + restoreGate.restoringThreadIds.clear(); agentState.messages = [ { id: "m1", role: "user" }, { id: "m2", role: "assistant" }, @@ -129,13 +144,13 @@ describe("SessionConversationScrollRestore branch switch regression", () => { const scroll = mountScrollFixture(80); setConversationScrollIntent({ kind: "bottom" }); agentState.threadId = "thread-c"; - restoreGate.isRestoringConversation = true; + restoreGate.restoringThreadIds.add("thread-c"); await act(async () => { root!.render(createElement(SessionConversationScrollRestore, { agentId: "dataFoundry" })); }); - restoreGate.isRestoringConversation = false; + restoreGate.restoringThreadIds.clear(); await act(async () => { root!.render(createElement(SessionConversationScrollRestore, { agentId: "dataFoundry" })); }); diff --git a/apps/web/src/app/data-tasks/components/chat/DataTaskChatInput.tsx b/apps/web/src/app/data-tasks/components/chat/DataTaskChatInput.tsx index 0617c12..b633322 100644 --- a/apps/web/src/app/data-tasks/components/chat/DataTaskChatInput.tsx +++ b/apps/web/src/app/data-tasks/components/chat/DataTaskChatInput.tsx @@ -66,6 +66,8 @@ type DataTaskChatInputProps = CopilotChatInputProps & { onSendQueuedPromptNow?: (id: string) => void; /** Client-side send/upload failure shown above the composer (never silent). */ submitError?: string | null; + /** Session lock / remote-busy prompt rendered above the composer. */ + sessionLockSlot?: ReactNode; }; export function DataTaskChatInput({ @@ -96,6 +98,7 @@ export function DataTaskChatInput({ onDeleteQueuedPrompt, onSendQueuedPromptNow, submitError = null, + sessionLockSlot = null, textArea, onChange, ...props @@ -161,6 +164,7 @@ export function DataTaskChatInput({ onDeleteQueuedPrompt={onDeleteQueuedPrompt} onSendQueuedPromptNow={onSendQueuedPromptNow} submitError={submitError} + sessionLockSlot={sessionLockSlot} /> )} @@ -208,6 +212,7 @@ function DataTaskChatInputLayout({ onDeleteQueuedPrompt, onSendQueuedPromptNow, submitError = null, + sessionLockSlot = null, }: { textArea: ReactNode; sendButton: ReactNode; @@ -250,6 +255,7 @@ function DataTaskChatInputLayout({ onDeleteQueuedPrompt?: (id: string) => void; onSendQueuedPromptNow?: (id: string) => void; submitError?: string | null; + sessionLockSlot?: ReactNode; }) { const t = useT(); const { chatColumnWidth, draftPromptRequest, onDraftPromptConsumed } = @@ -369,6 +375,7 @@ function DataTaskChatInputLayout({ onDelete={onDeleteQueuedPrompt} onSendNow={onSendQueuedPromptNow} /> + {sessionLockSlot} {submitError ? (
void; + onInterrupt: () => void; + onDismiss: () => void; +}; + +export function SessionBusyPrompt({ + activeRun, + busyAction = "idle", + onWait, + onInterrupt, + onDismiss, +}: SessionBusyPromptProps) { + const t = useT(); + const preview = activeRun.userInputPreview?.trim(); + const disabled = busyAction !== "idle"; + + return ( +
+
+ {t("chatInput.sessionBusyTitle")} +
+

{t("chatInput.sessionBusyBody")}

+ {preview ? ( +

+ {t("chatInput.sessionBusyPreview", { preview })} +

+ ) : null} +
+ + + +
+
+ ); +} + +type SessionRemoteBusyBannerProps = { + activeRun: SessionActiveRunDto; +}; + +export function SessionRemoteBusyBanner({ activeRun }: SessionRemoteBusyBannerProps) { + const t = useT(); + const preview = activeRun.userInputPreview?.trim(); + return ( +
+
{t("chatInput.sessionRemoteBusyBanner")}
+ {preview ? ( +
+ {t("chatInput.sessionBusyPreview", { preview })} +
+ ) : null} +
+ ); +} diff --git a/apps/web/src/app/data-tasks/components/chat/SessionConversationRestore.tsx b/apps/web/src/app/data-tasks/components/chat/SessionConversationRestore.tsx index 125ab5d..29f9a0d 100644 --- a/apps/web/src/app/data-tasks/components/chat/SessionConversationRestore.tsx +++ b/apps/web/src/app/data-tasks/components/chat/SessionConversationRestore.tsx @@ -52,19 +52,28 @@ export function SessionConversationRestore({ setLatestQuestionForThread, setSessionUsageForThread, } = useLiveRunSetters(); - const { setIsRestoringConversation } = useConversationRestoreGate(); + const { setThreadRestoring, markThreadRestored } = useConversationRestoreGate(); const fetchGenerationRef = useRef(0); const prevThreadIdRef = useRef(undefined); + const agentRef = useRef(agent); + agentRef.current = agent; const restoreRunActive = isConversationRestoreRunActive({ agentIsRunning: Boolean(agent.isRunning), liveRunStatus: liveRun.runStatus, }); useLayoutEffect(() => { - if (!threadId || !capabilitiesReady) { + if (!threadId) { + return; + } + // Gate while capabilities load so welcome cannot paint before restore starts. + if (!capabilitiesReady) { + setThreadRestoring(threadId, true); return; } if (!getRuntimeCapabilities().conversationMemory) { + setThreadRestoring(threadId, false); + markThreadRestored(threadId); return; } const threadChanged = prevThreadIdRef.current !== threadId; @@ -72,15 +81,23 @@ export function SessionConversationRestore({ if (!threadChanged) { return; } + // Always gate the UI for this thread on first mount so welcome cannot flash + // while history is loading — even if a live run later skips message rewrite. + setThreadRestoring(threadId, true); if (restoreRunActive) { return; } - setIsRestoringConversation(true); - agent.setMessages([]); + agentRef.current.setMessages([]); clearRestoredInterrupts(threadId); clearPendingCollaborationInterrupt(threadId); clearConversationBranchSnapshot(threadId); - }, [agent, capabilitiesReady, restoreRunActive, setIsRestoringConversation, threadId]); + }, [ + capabilitiesReady, + markThreadRestored, + restoreRunActive, + setThreadRestoring, + threadId, + ]); useEffect(() => { if (!threadId || !capabilitiesReady) { @@ -89,14 +106,21 @@ export function SessionConversationRestore({ const conversationMemoryEnabled = getRuntimeCapabilities().conversationMemory; if (!conversationMemoryEnabled) { + setThreadRestoring(threadId, false); + markThreadRestored(threadId); return; } if (restoreRunActive) { fetchGenerationRef.current += 1; + // Active streaming already owns the transcript; release the loading gate. + setThreadRestoring(threadId, false); + markThreadRestored(threadId); return; } + // Cover both first mount and deferred restore after a live run settles. + setThreadRestoring(threadId, true); const generation = fetchGenerationRef.current + 1; fetchGenerationRef.current = generation; let cancelled = false; @@ -109,17 +133,18 @@ export function SessionConversationRestore({ } setConversationBranchSnapshot(threadId, conversation); + const currentAgent = agentRef.current; if ( shouldRestoreConversationMessages({ conversationMemoryEnabled, isRunning: restoreRunActive, - agentMessages: agent.messages, + agentMessages: currentAgent.messages, dto: conversation, }) ) { const restored = conversationToAgentMessages(conversation); if (restored.length > 0) { - agent.setMessages(restored); + currentAgent.setMessages(restored); } } @@ -161,7 +186,8 @@ export function SessionConversationRestore({ } } finally { if (!cancelled && fetchGenerationRef.current === generation) { - setIsRestoringConversation(false); + markThreadRestored(threadId); + setThreadRestoring(threadId, false); } } })(); @@ -170,13 +196,13 @@ export function SessionConversationRestore({ cancelled = true; }; }, [ - agent, capabilitiesReady, + markThreadRestored, restoreRunActive, - setIsRestoringConversation, setLatestQuestionForThread, setLiveRunForThread, setSessionUsageForThread, + setThreadRestoring, threadId, ]); diff --git a/apps/web/src/app/data-tasks/components/chat/SessionConversationScrollRestore.tsx b/apps/web/src/app/data-tasks/components/chat/SessionConversationScrollRestore.tsx index b47e746..4117ee5 100644 --- a/apps/web/src/app/data-tasks/components/chat/SessionConversationScrollRestore.tsx +++ b/apps/web/src/app/data-tasks/components/chat/SessionConversationScrollRestore.tsx @@ -28,7 +28,8 @@ export function SessionConversationScrollRestore({ const chatConfig = useCopilotChatConfiguration(); const threadId = chatConfig?.threadId; const { agent } = useAgent({ agentId }); - const { isRestoringConversation } = useConversationRestoreGate(); + const { isThreadRestoring } = useConversationRestoreGate(); + const isRestoringConversation = isThreadRestoring(threadId); const prevRestoringRef = useRef(isRestoringConversation); const activeThreadRef = useRef(null); const pendingScrollRef = useRef(false); diff --git a/apps/web/src/app/data-tasks/components/chat/queued-chat-runs.ts b/apps/web/src/app/data-tasks/components/chat/queued-chat-runs.ts index b416ff1..5ce591e 100644 --- a/apps/web/src/app/data-tasks/components/chat/queued-chat-runs.ts +++ b/apps/web/src/app/data-tasks/components/chat/queued-chat-runs.ts @@ -1,4 +1,5 @@ import type { Attachment } from "@copilotkit/react-core/v2"; +import type { SessionActiveRunDto } from "../../../../lib/config-api/types"; import type { LiveRunStatus } from "../../live-run-state"; import type { RunForwardedProps } from "../../data-task-state"; @@ -21,6 +22,73 @@ export type QueuedChatPromptInput = { createdAt: number; }; +const QUEUED_PROMPTS_STORAGE_PREFIX = "data-tasks:queued-prompts:v1:"; + +function queuedPromptsStorageKey(threadId: string): string { + return `${QUEUED_PROMPTS_STORAGE_PREFIX}${threadId}`; +} + +function isPersistedQueuedPrompt(value: unknown): value is QueuedChatPrompt { + if (!value || typeof value !== "object") return false; + const record = value as Record; + return ( + typeof record.id === "string" && + typeof record.text === "string" && + typeof record.createdAt === "number" && + (record.status === "queued" || record.status === "interrupting") && + Array.isArray(record.attachments) && + record.forwardedProps != null && + typeof record.forwardedProps === "object" + ); +} + +/** Persist text/forwardedProps queue so refresh does not drop pending prompts. */ +export function loadQueuedChatPrompts(threadId: string | null | undefined): QueuedChatPrompt[] { + if (!threadId || typeof window === "undefined") return []; + try { + const raw = window.sessionStorage.getItem(queuedPromptsStorageKey(threadId)); + if (!raw) return []; + const parsed = JSON.parse(raw) as unknown; + if (!Array.isArray(parsed)) return []; + return parsed.filter(isPersistedQueuedPrompt).map((prompt) => ({ + ...prompt, + attachments: Array.isArray(prompt.attachments) ? prompt.attachments : [], + status: prompt.status === "interrupting" ? "queued" : prompt.status, + })); + } catch { + return []; + } +} + +export function persistQueuedChatPrompts( + threadId: string | null | undefined, + prompts: QueuedChatPrompt[], +): void { + if (!threadId || typeof window === "undefined") return; + try { + const key = queuedPromptsStorageKey(threadId); + if (prompts.length === 0) { + window.sessionStorage.removeItem(key); + return; + } + // Drop binary attachment payloads; text + run config is enough to re-send after refresh. + const serializable = prompts.map((prompt) => ({ + ...prompt, + attachments: prompt.attachments.map((attachment) => ({ + id: attachment.id, + type: attachment.type, + contentType: (attachment as { contentType?: string }).contentType, + name: (attachment as { name?: string }).name, + filename: (attachment as { filename?: string }).filename, + status: attachment.status, + })), + })); + window.sessionStorage.setItem(key, JSON.stringify(serializable)); + } catch { + // Ignore quota / private-mode failures. + } +} + export function createQueuedChatPrompt( input: QueuedChatPromptInput, ): QueuedChatPrompt { @@ -83,6 +151,29 @@ export function resolveQueuedSubmitMode(input: { return "dispatch"; } +/** True when another client owns an active run on this session. */ +export function isForeignSessionActiveRun( + activeRun: SessionActiveRunDto | null | undefined, + localRunId: string | null | undefined, + localRunActive = false, +): activeRun is SessionActiveRunDto { + if (!activeRun?.activeRunId) return false; + if (localRunActive) return false; + if (localRunId && activeRun.activeRunId === localRunId) return false; + if (localRunId) return true; + // No local run id yet: treat unknown active runs as foreign so multi-device + // lock prompts still appear; callers should pass localRunActive when this tab owns the run. + return true; +} + +/** Extract the blocking run id from a RUN_ALREADY_ACTIVE backend error. */ +export function parseAlreadyActiveRunId(message: string | null | undefined): string | null { + if (!message) return null; + const match = /RUN_ALREADY_ACTIVE[:\s]+([^\s:,]+)/u.exec(message); + const runId = match?.[1]?.trim(); + return runId || null; +} + export function shouldShowRunningStopControl(input: { agentIsRunning: boolean; liveRunStatus: LiveRunStatus; diff --git a/apps/web/src/app/data-tasks/data-task-state.ts b/apps/web/src/app/data-tasks/data-task-state.ts index 5d04226..cbe5aef 100755 --- a/apps/web/src/app/data-tasks/data-task-state.ts +++ b/apps/web/src/app/data-tasks/data-task-state.ts @@ -398,7 +398,12 @@ export function applyAutoTitle( const normalized = normalizeSessionTitle(title); if (!normalized) return sessions; return sessions.map((session) => { - if (session.id !== id || session.titleSource === "user") return session; + if ( + (session.id !== id && session.threadId !== id) || + session.titleSource === "user" + ) { + return session; + } return { ...session, title: normalized, titleSource: source }; }); } @@ -422,6 +427,8 @@ function timestampFromIso(value: string | undefined, fallback: number): number { function normalizeTitleSource(value: string | undefined): ChatSessionTitleSource { if (value === "auto-snippet" || value === "llm" || value === "user") return value; + // Backend persists non-LLM auto titles as "fallback"; treat like a snippet. + if (value === "fallback") return "auto-snippet"; return "default"; } @@ -712,6 +719,36 @@ export const DATA_SKILLS: DataSkill[] = [ name: "Data analysis", description: "Answer data questions from metric lookups to full reports", }, + { + id: "tabular-file-import", + name: "Tabular file import", + description: "Import CSV/Excel/JSON/Parquet files into the workspace", + }, + { + id: "data-cleaning-for-load", + name: "Data cleaning for load", + description: "Clean and validate tabular data before analysis or load", + }, + { + id: "batch-file-merge", + name: "Batch file merge", + description: "Merge multiple tabular files with mapping and dedup", + }, + { + id: "etl-pipeline-patterns", + name: "ETL pipeline patterns", + description: "Design reliable staging, idempotent, and incremental loads", + }, + { + id: "api-json-ingest", + name: "API / JSON ingest", + description: "Ingest API or JSON/JSONL payloads into tabular files", + }, + { + id: "database-load-planning", + name: "Database load planning", + description: "Map files to tables and plan read-only-validated loads", + }, ]; export const DEFAULT_SKILL_ID = DATA_SKILLS[0].id; diff --git a/apps/web/src/app/data-tasks/data-tasks-app.tsx b/apps/web/src/app/data-tasks/data-tasks-app.tsx index 411f59f..331b417 100755 --- a/apps/web/src/app/data-tasks/data-tasks-app.tsx +++ b/apps/web/src/app/data-tasks/data-tasks-app.tsx @@ -79,6 +79,7 @@ import { workspaceConfigItemStatusBadge, } from "./data-task-state"; import { configApi } from "../../lib/config-api/client"; +import { ConfigApiError } from "../../lib/config-api"; import { hasMeaningfulText, isOrphanPreambleMergedIntoFollowingToolStep, @@ -182,13 +183,22 @@ import { createQueuedChatPrompt, deleteQueuedChatPrompt, editQueuedChatPrompt, + isForeignSessionActiveRun, + loadQueuedChatPrompts, markQueuedChatPromptInterrupting, + parseAlreadyActiveRunId, + persistQueuedChatPrompts, queuedPromptToRunInput, resolveQueuedSubmitMode, shouldShowRunningStopControl, takeNextQueuedChatPrompt, type QueuedChatPrompt, } from "./components/chat/queued-chat-runs"; +import { + SessionBusyPrompt, + SessionRemoteBusyBanner, +} from "./components/chat/SessionBusyPrompt"; +import type { SessionActiveRunDto } from "../../lib/config-api/types"; import { scheduleChatTextareaResize } from "./components/chat/use-chat-textarea-autoresize"; import { invokeWithReportedError } from "./invoke-with-reported-error"; import { formatRunErrorMessage } from "./run-error-message"; @@ -430,6 +440,21 @@ async function uploadChatDataFile( return configApi.uploadChatFile(file, sessionId); } +type PendingBusySubmit = { + text: string; + attachments: ReturnType< + ReturnType["consumeAttachments"] + >; + forwardedProps: ReturnType< + ReturnType["getRunForwardedProps"] + >; +}; + +type SessionBusyDialogState = { + activeRun: SessionActiveRunDto; + pending: PendingBusySubmit; +}; + function StableDataTaskChatInput({ inputProps, }: { @@ -442,6 +467,9 @@ function StableDataTaskChatInput({ const [queuedPrompts, setQueuedPrompts] = useState([]); const [draftText, setDraftText] = useState(""); const [submitError, setSubmitError] = useState(null); + const [remoteActiveRun, setRemoteActiveRun] = useState(null); + const [sessionBusyDialog, setSessionBusyDialog] = useState(null); + const [busyAction, setBusyAction] = useState<"idle" | "waiting" | "interrupting">("idle"); const awaitingQueuedRunCompletionRef = useRef(false); const queuedRunSawActiveRef = useRef(false); @@ -490,6 +518,79 @@ function StableDataTaskChatInput({ reportDataFoundryRunError(error, bindings.activeThreadId, { source: "client" }); }, [bindings.activeThreadId]); + const fetchForeignActiveRun = useCallback(async (): Promise => { + const threadId = bindings.activeThreadId; + if (!threadId) return null; + const localRunActive = + Boolean(agent?.isRunning) || + bindings.liveRunStatus === "running" || + bindings.liveRunStatus === "suspended"; + try { + const response = await configApi.listSessions({ limit: 50 }); + const session = response.sessions.find( + (item) => item.id === threadId || item.threadId === threadId, + ); + const activeRun = session?.activeRun ?? null; + if (isForeignSessionActiveRun(activeRun, bindings.liveRunRunId, localRunActive)) { + return activeRun; + } + return null; + } catch { + return null; + } + }, [agent?.isRunning, bindings.activeThreadId, bindings.liveRunRunId, bindings.liveRunStatus]); + + const clearPerRunSelections = useCallback(() => { + bindings.onClearPerRunMentions(); + bindings.onClearPerRunFileMentions(); + bindings.onClearEvidenceRefs(); + requestAnimationFrame(scheduleChatTextareaResize); + }, [bindings]); + + const unlockOrReportActiveRunError = useCallback( + async (error: unknown, pending?: PendingBusySubmit) => { + const message = error instanceof Error ? error.message : String(error); + const blockingRunId = parseAlreadyActiveRunId(message); + if (!blockingRunId) { + reportRunError(error); + return; + } + + const foreign = await fetchForeignActiveRun(); + if (foreign) { + setRemoteActiveRun(foreign); + if (pending) { + setSessionBusyDialog({ + activeRun: foreign, + pending, + }); + setBusyAction("idle"); + } + return; + } + + // Metadata can retain queued/running/suspended after the worker died. + // Cancel may 404 if the server already reclaimed the orphan — still retry once. + try { + await configApi.cancelRun(blockingRunId, "stale-active-run-unlock"); + } catch { + // Ignore cancel failures; retry below if the lock is already gone. + } + setRemoteActiveRun(null); + if (pending && agent) { + // Message was already appended by the failed dispatch; only retry the run. + agent.setState(buildAgentRunStatePatch(pending.forwardedProps, agent.state)); + void copilotkit + .runAgent({ agent, forwardedProps: pending.forwardedProps }) + .catch(reportRunError); + clearPerRunSelections(); + return; + } + reportRunError(error); + }, + [agent, clearPerRunSelections, copilotkit, fetchForeignActiveRun, reportRunError], + ); + const dispatchRun = useCallback( ({ text, @@ -500,6 +601,7 @@ function StableDataTaskChatInput({ attachments: ReturnType; forwardedProps: ReturnType; }) => { + const pending: PendingBusySubmit = { text, attachments, forwardedProps }; invokeWithReportedError(() => { if (!agent) return; bindings.onUserMessageSubmitted(text); @@ -510,10 +612,14 @@ function StableDataTaskChatInput({ } else { agent.addMessage({ id: createClientId("msg"), role: "user", content: text }); } - void copilotkit.runAgent({ agent, forwardedProps }).catch(reportRunError); + void copilotkit + .runAgent({ agent, forwardedProps }) + .catch((error) => { + void unlockOrReportActiveRunError(error, pending); + }); }, reportRunError); }, - [agent, bindings, copilotkit, reportRunError], + [agent, bindings, copilotkit, reportRunError, unlockOrReportActiveRunError], ); useEffect(() => { @@ -523,6 +629,49 @@ function StableDataTaskChatInput({ bindings.stopActiveChatRunRef.current = inputProps.onStop; }, [bindings.stopActiveChatRunRef, inputProps.onStop]); + useEffect(() => { + const threadId = bindings.activeThreadId; + setQueuedPrompts(loadQueuedChatPrompts(threadId)); + setSessionBusyDialog(null); + setBusyAction("idle"); + setRemoteActiveRun(null); + awaitingQueuedRunCompletionRef.current = false; + queuedRunSawActiveRef.current = false; + }, [bindings.activeThreadId]); + + useEffect(() => { + persistQueuedChatPrompts(bindings.activeThreadId, queuedPrompts); + }, [bindings.activeThreadId, queuedPrompts]); + + useEffect(() => { + const threadId = bindings.activeThreadId; + if (!threadId || !bindings.capabilitiesReady) { + setRemoteActiveRun(null); + return; + } + let cancelled = false; + const refresh = async () => { + const foreign = await fetchForeignActiveRun(); + if (!cancelled) { + setRemoteActiveRun(foreign); + } + }; + void refresh(); + const timer = window.setInterval(() => { + void refresh(); + }, 3000); + return () => { + cancelled = true; + window.clearInterval(timer); + }; + }, [ + bindings.activeThreadId, + bindings.capabilitiesReady, + bindings.liveRunRunId, + fetchForeignActiveRun, + queuedPrompts.length, + ]); + const handleSubmitMessage = (value: string) => { setDraftText(""); setSubmitError(null); @@ -557,19 +706,76 @@ function StableDataTaskChatInput({ createdAt: Date.now(), }), ]); - } else { - dispatchRun({ text: value, attachments: ready, forwardedProps }); + clearPerRunSelections(); + return; } - } else { - inputProps.onSubmitMessage?.(value); - } - bindings.onClearPerRunMentions(); - bindings.onClearPerRunFileMentions(); - bindings.onClearEvidenceRefs(); - requestAnimationFrame(scheduleChatTextareaResize); + void (async () => { + const foreign = remoteActiveRun ?? (await fetchForeignActiveRun()); + if (foreign) { + setRemoteActiveRun(foreign); + setSessionBusyDialog({ + activeRun: foreign, + pending: { text: value, attachments: ready, forwardedProps }, + }); + setBusyAction("idle"); + return; + } + dispatchRun({ text: value, attachments: ready, forwardedProps }); + clearPerRunSelections(); + })().catch(reportRunError); + return; + } + inputProps.onSubmitMessage?.(value); + clearPerRunSelections(); }, reportRunError); }; + + const handleBusyWait = useCallback(() => { + if (!sessionBusyDialog) return; + setBusyAction("waiting"); + setQueuedPrompts((current) => [ + ...current, + createQueuedChatPrompt({ + id: createClientId("queue"), + text: sessionBusyDialog.pending.text, + attachments: sessionBusyDialog.pending.attachments, + forwardedProps: sessionBusyDialog.pending.forwardedProps, + createdAt: Date.now(), + }), + ]); + setRemoteActiveRun(sessionBusyDialog.activeRun); + setSessionBusyDialog(null); + setBusyAction("idle"); + clearPerRunSelections(); + }, [clearPerRunSelections, sessionBusyDialog]); + + const handleBusyInterrupt = useCallback(() => { + if (!sessionBusyDialog) return; + const { activeRun, pending } = sessionBusyDialog; + setBusyAction("interrupting"); + void (async () => { + try { + await configApi.cancelRun(activeRun.activeRunId, "user-requested-takeover"); + setRemoteActiveRun(null); + setSessionBusyDialog(null); + setBusyAction("idle"); + dispatchRun(pending); + clearPerRunSelections(); + } catch (error) { + setBusyAction("idle"); + reportRunError(error); + } + })(); + }, [clearPerRunSelections, dispatchRun, reportRunError, sessionBusyDialog]); + + const handleBusyDismiss = useCallback(() => { + if (!sessionBusyDialog) return; + setDraftText(sessionBusyDialog.pending.text); + setSessionBusyDialog(null); + setBusyAction("idle"); + }, [sessionBusyDialog]); + const handleInputChange = useCallback( (value: string) => { setDraftText(value); @@ -587,16 +793,13 @@ function StableDataTaskChatInput({ [bindings.onCancelRun, inputProps.onStop], ); - useEffect(() => { - setQueuedPrompts([]); - awaitingQueuedRunCompletionRef.current = false; - queuedRunSawActiveRef.current = false; - }, [bindings.activeThreadId]); - useEffect(() => { if (!agent || queuedPrompts.length === 0) { return; } + if (remoteActiveRun) { + return; + } const activeRun = Boolean(agent.isRunning) || bindings.liveRunStatus === "running" || @@ -623,7 +826,7 @@ function StableDataTaskChatInput({ setQueuedPrompts(next.queue); const runInput = queuedPromptToRunInput(next.prompt); dispatchRun(runInput); - }, [agent, bindings.liveRunStatus, dispatchRun, queuedPrompts]); + }, [agent, bindings.liveRunStatus, dispatchRun, queuedPrompts, remoteActiveRun]); const handleEditQueuedPrompt = useCallback((id: string, text: string) => { setQueuedPrompts((current) => editQueuedChatPrompt(current, id, text)); @@ -650,6 +853,18 @@ function StableDataTaskChatInput({ draftText, }); + const sessionLockSlot = sessionBusyDialog ? ( + + ) : remoteActiveRun && !showStopControl ? ( + + ) : null; + return ( ); } @@ -1248,6 +1464,7 @@ function DataTaskWorkspace({ ); const { liveRun, sessionUsage, latestQuestion, runningThreadIds } = useLiveRun(activeThreadId); + const { isThreadRestored, setThreadRestoring } = useConversationRestoreGate(); const agentRenderSnapshot = useAgentMessageRenderSnapshot(); const sessionStartedHints = useMemo( () => ({ @@ -1486,7 +1703,13 @@ function DataTaskWorkspace({ const deleteSession = useCallback( (sessionId: string) => { + const target = sessions.find((session) => session.id === sessionId); + const serverSessionId = target?.threadId ?? sessionId; + let previousSessions: ChatSession[] | null = null; + let previousActiveSessionId: string | null = activeSessionId; + setSessions((current) => { + previousSessions = current; const next = deleteChatSession(current, sessionId); if (next.length === 0) { const fallback = createChatSession(defaultSessionTitle); @@ -1499,8 +1722,43 @@ function DataTaskWorkspace({ return next; }); setSelection(null); + + void configApi.deleteSession(serverSessionId) + .then((result) => { + const deletedIds = new Set(result.deletedSessionIds ?? [serverSessionId]); + setSessions((current) => { + const filtered = current.filter( + (session) => !deletedIds.has(session.id) && !deletedIds.has(session.threadId), + ); + if (filtered.length === current.length) return current; + if (filtered.length === 0) { + const fallback = createChatSession(defaultSessionTitle); + setActiveSessionId(fallback.id); + return [fallback]; + } + setActiveSessionId((active) => { + if (active && filtered.some((session) => session.id === active)) return active; + return filtered[0]?.id ?? active; + }); + return filtered; + }); + setSessionSyncError(null); + }) + .catch((error) => { + if (error instanceof ConfigApiError && error.status === 404) { + setSessionSyncError(null); + return; + } + if (previousSessions) { + setSessions(previousSessions); + setActiveSessionId(previousActiveSessionId); + } + setSessionSyncError( + error instanceof Error ? error.message : "Failed to delete session on server", + ); + }); }, - [activeSessionId, defaultSessionTitle], + [activeSessionId, defaultSessionTitle, sessions], ); const togglePinSession = useCallback((sessionId: string) => { @@ -1872,8 +2130,13 @@ function DataTaskWorkspace({ onQueryChange={setQuery} onToggleCollapse={toggleSidebar} onSelectSession={(sessionId) => { - setConversationScrollIntent({ kind: "bottom" }); + // Apply session id first so a single click always wins over slower panel teardown. setActiveSessionId(sessionId); + // Eager loading gate for threads that have never restored in this tab. + if (!isThreadRestored(sessionId)) { + setThreadRestoring(sessionId, true); + } + setConversationScrollIntent({ kind: "bottom" }); setSelection(null); setArtifactFocusId(null); setIsTraceOpen(false); @@ -4513,8 +4776,19 @@ function SessionListItem({ return (
{ + if (event.key === "Enter" || event.key === " ") { + event.preventDefault(); + onSelect(); + } + }} + title={session.title} className={[ - "group flex items-center gap-2 rounded-lg px-2 py-1.5 transition-colors duration-150", + "group flex cursor-pointer items-center gap-2 rounded-lg px-2 py-1.5 transition-colors duration-150", active ? "bg-surface shadow-[var(--shadow-card)]" : "hover:bg-surface", ].join(" ")} > @@ -4523,17 +4797,14 @@ function SessionListItem({ ) : iconSlots.leading === "session" ? ( ) : null} - + {iconSlots.trailing === "pin" && ( void; }) { + const t = useT(); const chatConfig = useCopilotChatConfiguration(); + const threadId = chatConfig?.threadId; const { agent } = useAgent({ agentId: chatConfig?.agentId ?? runtimeAgentId }); + const { isThreadRestoring, isThreadRestored } = useConversationRestoreGate(); + const isRestoringConversation = isThreadRestoring(threadId); const hasVisibleMessages = (agent.messages?.length ?? 0) > 0 || liveRunStatus !== "idle"; + // Until the first restore cycle finishes, keep loading — never flash welcome. + const awaitingFirstRestore = Boolean(threadId) && !isThreadRestored(threadId); + + if (isRestoringConversation || awaitingFirstRestore) { + return ( +
+
+ {t("common.loading")} +
+
+ ); + } if (hasVisibleMessages) return null; @@ -7286,7 +7576,8 @@ function PendingBranchRunDispatcher({ const { agent } = useAgent({ agentId: chatConfig?.agentId ?? runtimeAgentId }); const { copilotkit } = useCopilotKit(); const restoredConversation = useConversationBranchSnapshot(chatConfig?.threadId); - const { isRestoringConversation } = useConversationRestoreGate(); + const { isThreadRestoring } = useConversationRestoreGate(); + const isRestoringConversation = isThreadRestoring(chatConfig?.threadId); useEffect(() => { if (!pending || chatConfig?.threadId !== pending.sessionId) { diff --git a/apps/web/src/app/data-tasks/run-error-message.ts b/apps/web/src/app/data-tasks/run-error-message.ts index 36a95c3..517299c 100644 --- a/apps/web/src/app/data-tasks/run-error-message.ts +++ b/apps/web/src/app/data-tasks/run-error-message.ts @@ -12,6 +12,9 @@ export function formatRunErrorMessage(message?: string): string { if (trimmed === "RUN_CANCELLED") { return "Run was cancelled."; } + if (/^RUN_ALREADY_ACTIVE\b/u.test(trimmed)) { + return "This chat already has an active run. Wait for it to finish, or interrupt it to take over."; + } if (trimmed.startsWith("RUN_SUBSCRIBER_CLOSED")) { return "Run stopped because the connection closed. Retry the query or refresh the page."; } diff --git a/apps/web/src/app/data-tasks/use-data-foundry-run.tsx b/apps/web/src/app/data-tasks/use-data-foundry-run.tsx index dcfd302..dd8fcf1 100644 --- a/apps/web/src/app/data-tasks/use-data-foundry-run.tsx +++ b/apps/web/src/app/data-tasks/use-data-foundry-run.tsx @@ -9,7 +9,6 @@ import { useMemo, useRef, useState, - type Dispatch, type ReactNode, type SetStateAction, } from "react"; @@ -67,8 +66,14 @@ type LiveRunSetters = { }; type ConversationRestoreGate = { - isRestoringConversation: boolean; - setIsRestoringConversation: Dispatch>; + /** Threads currently loading persisted conversation history. */ + restoringThreadIds: ReadonlySet; + /** Threads that completed at least one restore cycle (including empty history). */ + restoredThreadIds: ReadonlySet; + isThreadRestoring: (threadId: string | null | undefined) => boolean; + isThreadRestored: (threadId: string | null | undefined) => boolean; + setThreadRestoring: (threadId: string, restoring: boolean) => void; + markThreadRestored: (threadId: string) => void; }; const emptyLiveRun = createInitialLiveRun(); @@ -147,9 +152,49 @@ export function LiveRunProvider({ children }: { children: ReactNode }) { Record >({}); const [runningThreadIds, setRunningThreadIds] = useState>(() => new Set()); - const [isRestoringConversation, setIsRestoringConversation] = useState(false); + const [restoringThreadIds, setRestoringThreadIds] = useState>(() => new Set()); + const [restoredThreadIds, setRestoredThreadIds] = useState>(() => new Set()); const prevRunStatusByThreadRef = useRef>({}); + const setThreadRestoring = useCallback((threadId: string, restoring: boolean) => { + setRestoringThreadIds((current) => { + const has = current.has(threadId); + if (restoring === has) { + return current; + } + const next = new Set(current); + if (restoring) { + next.add(threadId); + } else { + next.delete(threadId); + } + return next; + }); + }, []); + + const markThreadRestored = useCallback((threadId: string) => { + setRestoredThreadIds((current) => { + if (current.has(threadId)) { + return current; + } + const next = new Set(current); + next.add(threadId); + return next; + }); + }, []); + + const isThreadRestoring = useCallback( + (threadId: string | null | undefined) => + Boolean(threadId && restoringThreadIds.has(threadId)), + [restoringThreadIds], + ); + + const isThreadRestored = useCallback( + (threadId: string | null | undefined) => + Boolean(threadId && restoredThreadIds.has(threadId)), + [restoredThreadIds], + ); + const setLiveRunForThread = useCallback( (threadId: string | undefined, action: SetStateAction) => { if (!threadId) return; @@ -226,10 +271,21 @@ export function LiveRunProvider({ children }: { children: ReactNode }) { const restoreGate = useMemo( () => ({ - isRestoringConversation, - setIsRestoringConversation, + restoringThreadIds, + restoredThreadIds, + isThreadRestoring, + isThreadRestored, + setThreadRestoring, + markThreadRestored, }), - [isRestoringConversation], + [ + isThreadRestored, + isThreadRestoring, + markThreadRestored, + restoredThreadIds, + restoringThreadIds, + setThreadRestoring, + ], ); useLayoutEffect(() => { @@ -333,7 +389,8 @@ export function LiveRunEventSubscriber({ setLatestQuestionForThread, syncRunningThreadStatus, } = setters; - const { isRestoringConversation } = useConversationRestoreGate(); + const { isThreadRestoring } = useConversationRestoreGate(); + const isRestoringConversation = isThreadRestoring(threadId); const isRestoringConversationRef = useRef(isRestoringConversation); isRestoringConversationRef.current = isRestoringConversation; diff --git a/apps/web/src/components/auth/auth-flow.tsx b/apps/web/src/components/auth/auth-flow.tsx index a15e87a..fb38e69 100644 --- a/apps/web/src/components/auth/auth-flow.tsx +++ b/apps/web/src/components/auth/auth-flow.tsx @@ -48,7 +48,12 @@ export function AuthFlow({ setMessage(null); }, []); - const goToLogin = useCallback(() => router.push("/login"), [router]); + // Forgot/reset/verify are in-page sub-modes of /login or /register. Pushing + // `/login` alone is a no-op when already on that route, so also reset mode. + const goToLogin = useCallback(() => { + goToSubMode("login"); + router.push("/login"); + }, [goToSubMode, router]); const goToRegister = useCallback(() => router.push("/register"), [router]); const submit = async (event: FormEvent) => { diff --git a/apps/web/src/i18n/messages/en.json b/apps/web/src/i18n/messages/en.json index 02ad09a..fe216f3 100644 --- a/apps/web/src/i18n/messages/en.json +++ b/apps/web/src/i18n/messages/en.json @@ -157,7 +157,17 @@ "deleteQueued": "Delete queued prompt", "saveQueued": "Save", "cancelQueued": "Cancel", - "dropFiles": "Drop files here to upload" + "dropFiles": "Drop files here to upload", + "sessionBusyTitle": "This chat is busy on another device", + "sessionBusyBody": "Only one active run is allowed per chat. You can wait in queue, interrupt the other run and send, or cancel.", + "sessionBusyPreview": "Other device is working on: {{preview}}", + "sessionBusyWait": "Wait in queue", + "sessionBusyWaiting": "Queued…", + "sessionBusyInterrupt": "Interrupt and send", + "sessionBusyInterrupting": "Interrupting…", + "sessionBusyDismiss": "Cancel send", + "sessionRemoteBusyBanner": "This chat is running on another device. You will be asked how to proceed before sending.", + "localQueuedNotice": "A run is still in progress, so your message was added to the send queue." }, "sessionConfig": { "agentToolsSettings": "Agent Tools session settings", diff --git a/apps/web/src/i18n/messages/zh-CN.json b/apps/web/src/i18n/messages/zh-CN.json index deb1a68..da992e0 100644 --- a/apps/web/src/i18n/messages/zh-CN.json +++ b/apps/web/src/i18n/messages/zh-CN.json @@ -157,7 +157,17 @@ "deleteQueued": "删除排队消息", "saveQueued": "保存", "cancelQueued": "取消", - "dropFiles": "将文件拖放到此处上传" + "dropFiles": "将文件拖放到此处上传", + "sessionBusyTitle": "会话正在其他端运行", + "sessionBusyBody": "同一对话同时只能有一个进行中的任务。你可以排队等待、中断对方任务后发送，或取消本次发送。", + "sessionBusyPreview": "对方正在处理：{{preview}}", + "sessionBusyWait": "排队等待", + "sessionBusyWaiting": "已加入队列…", + "sessionBusyInterrupt": "中断并发送", + "sessionBusyInterrupting": "正在中断…", + "sessionBusyDismiss": "取消发送", + "sessionRemoteBusyBanner": "此对话正在其他端运行中，发送新消息前会提示你选择处理方式。", + "localQueuedNotice": "当前任务未完成，消息已加入发送队列。" }, "sessionConfig": { "agentToolsSettings": "Agent 工具会话设置", diff --git a/apps/web/src/lib/config-api/client.ts b/apps/web/src/lib/config-api/client.ts index e04c5bf..8766f95 100644 --- a/apps/web/src/lib/config-api/client.ts +++ b/apps/web/src/lib/config-api/client.ts @@ -355,6 +355,13 @@ export const configApi = { }); }, + deleteSession(sessionId: string): Promise<{ sessionId: string; deleted: boolean; deletedSessionIds: string[] }> { + return requestEnvelope<{ sessionId: string; deleted: boolean; deletedSessionIds: string[] }>( + `/api/v1/sessions/${encodeURIComponent(sessionId)}`, + { method: "DELETE" }, + ); + }, + listDatasourceTypes(): Promise { return requestEnvelope("/api/v1/datasource-types"); }, diff --git a/apps/web/src/lib/config-api/index.ts b/apps/web/src/lib/config-api/index.ts index 7814824..f2f366b 100644 --- a/apps/web/src/lib/config-api/index.ts +++ b/apps/web/src/lib/config-api/index.ts @@ -58,6 +58,7 @@ export type { QueryHistoryListResponseDto, RunCancelDto, RunDefaultsDto, + SessionActiveRunDto, SessionConversationDto, SessionListItemDto, SessionListResponseDto, diff --git a/apps/web/src/lib/config-api/types.ts b/apps/web/src/lib/config-api/types.ts index 8ec5d12..3d2b96f 100644 --- a/apps/web/src/lib/config-api/types.ts +++ b/apps/web/src/lib/config-api/types.ts @@ -570,6 +570,14 @@ export type PendingInteractionDto = { resumeSchema?: unknown; }; +export type SessionActiveRunDto = { + sessionId: string; + activeRunId: string; + status: "queued" | "running" | "suspended"; + startedAt: string; + userInputPreview: string; +}; + export type SessionConversationDto = { sessionId: string; title?: string; @@ -584,6 +592,7 @@ export type SessionConversationDto = { toolCalls: ConversationToolCallDto[]; pendingInteractions?: PendingInteractionDto[]; restorableCustomEvents?: RestorableCustomEventDto[]; + activeRun?: SessionActiveRunDto | null; }; export type SessionListItemDto = { @@ -594,6 +603,7 @@ export type SessionListItemDto = { createdAt?: string; updatedAt?: string; lastMessageAt?: string; + activeRun?: SessionActiveRunDto | null; }; export type SessionListResponseDto = { diff --git a/docs/en/guides/web-workbench.md b/docs/en/guides/web-workbench.md index 559b07d..2703cec 100644 --- a/docs/en/guides/web-workbench.md +++ b/docs/en/guides/web-workbench.md @@ -204,6 +204,7 @@ The Web workbench restores history through server session APIs: | --- | --- | | `GET /api/v1/sessions` | Read the left session list. | | `PATCH /api/v1/sessions/:id` | Update session title. | +| `DELETE /api/v1/sessions/:id` | Permanently delete a session (including history and child branches); the sidebar delete action calls this. | | `GET /api/v1/sessions/:id/conversation` | Restore conversation, tool calls, pending interactions, and run events. | | `GET /api/v1/artifacts?sessionId=:id` | Restore outputs for that session. | diff --git a/docs/en/reference/rest-api.md b/docs/en/reference/rest-api.md index 06a7bd4..1741724 100644 --- a/docs/en/reference/rest-api.md +++ b/docs/en/reference/rest-api.md @@ -112,6 +112,7 @@ curl http://127.0.0.1:8787/api/v1/capabilities | --- | --- | --- | | GET | `/api/v1/sessions` | List server sessions. Supports `limit`, `cursor`. | | PATCH | `/api/v1/sessions/:sessionId` | Update session title. | +| DELETE | `/api/v1/sessions/:sessionId` | Permanently delete a session and its conversation, runs, artifacts, and child branches. | | GET | `/api/v1/sessions/:sessionId/conversation` | Read authoritative server conversation history. Supports `limit`. | | GET | `/api/v1/sessions/:sessionId/checkpoints` | List persisted context checkpoints. Supports `limit`. | | GET | `/api/v1/sessions/:sessionId/trace-dag` | Read the semantic run/step/tool/output graph. Supports `limit`. | diff --git a/docs/zh/guides/web-workbench.md b/docs/zh/guides/web-workbench.md index 065517c..a45a3e4 100644 --- a/docs/zh/guides/web-workbench.md +++ b/docs/zh/guides/web-workbench.md @@ -204,6 +204,7 @@ Web 工作台使用服务端会话接口恢复历史： | --- | --- | | `GET /api/v1/sessions` | 读取左侧会话列表。 | | `PATCH /api/v1/sessions/:id` | 更新会话标题。 | +| `DELETE /api/v1/sessions/:id` | 永久删除会话（含对话历史与子分支）；侧栏删除会调用此接口。 | | `GET /api/v1/sessions/:id/conversation` | 恢复对话、工具调用、pending interaction 和 run events。 | | `GET /api/v1/artifacts?sessionId=:id` | 恢复该会话的产出。 | diff --git a/docs/zh/reference/rest-api.md b/docs/zh/reference/rest-api.md index e80c753..69c2126 100644 --- a/docs/zh/reference/rest-api.md +++ b/docs/zh/reference/rest-api.md @@ -112,6 +112,7 @@ curl http://127.0.0.1:8787/api/v1/capabilities | --- | --- | --- | | GET | `/api/v1/sessions` | 列出服务端会话。支持 `limit`、`cursor`。 | | PATCH | `/api/v1/sessions/:sessionId` | 更新会话标题。 | +| DELETE | `/api/v1/sessions/:sessionId` | 永久删除会话及其对话、run、产物与子分支。 | | GET | `/api/v1/sessions/:sessionId/conversation` | 读取服务端权威对话历史。支持 `limit`。 | | GET | `/api/v1/sessions/:sessionId/checkpoints` | 列出已持久化的上下文 checkpoint。支持 `limit`。 | | GET | `/api/v1/sessions/:sessionId/trace-dag` | 读取 run/step/tool/output 语义图。支持 `limit`。 | diff --git a/packages/agent-runtime/src/index.ts b/packages/agent-runtime/src/index.ts index 5f66870..1ce2be1 100644 --- a/packages/agent-runtime/src/index.ts +++ b/packages/agent-runtime/src/index.ts @@ -717,12 +717,19 @@ const buildAgentInstructions = (input: AgentInstructionsInput): string => { const evidenceRefs = context.evidence_refs; if (evidenceRefs && evidenceRefs.length > 0) { const labels = evidenceRefs - .map((ref) => `${ref.kind}:${ref.label}`) + .map((ref) => { + const selection = ref.source.selection; + if (!selection) return `${ref.kind}:${ref.label}`; + if (selection.mode === "text") return `${ref.kind}:${ref.label} (text selection)`; + return `${ref.kind}:${ref.label} (${selection.mode} selection)`; + }) .slice(0, 12) .join("; "); toolGroups.push( - `User-selected evidence focus this run: ${labels}. Treat these references as the primary context for the ` - + "follow-up question. You may run new data tools when needed; make new queries and outputs visible in steps." + `User-selected evidence focus this run: ${labels}. Selected evidence content (including any table/text ` + + "subsets) is already provided in context — prefer that over opening the full artifact file. " + + "Treat these references as the primary context for the follow-up question. You may run new data " + + "tools when needed; make new queries and outputs visible in steps." ); } const taskTools = ["task_write", "task_update", "task_complete", "task_check"].filter(enabled); diff --git a/packages/metadata/src/index.ts b/packages/metadata/src/index.ts index 2e0cf9b..adc9330 100644 --- a/packages/metadata/src/index.ts +++ b/packages/metadata/src/index.ts @@ -1270,6 +1270,165 @@ export class SessionRepository { .run(title, input.title_source, now, input.user_id, input.session_id); return result.changes === 1 ? this.get({ user_id: input.user_id, session_id: input.session_id }) : undefined; } + + /** + * Permanently delete a session and its dependent rows (runs, conversation, + * artifacts, branches, etc.). Also cascades to descendant branch sessions. + * Idempotent: missing sessions are a no-op (returns deleted: false). + */ + delete(input: { user_id: string; session_id: string }): { deleted: boolean; deletedSessionIds: string[] } { + const existing = mapSessionRow( + this.db.prepare("SELECT * FROM sessions WHERE user_id = ? AND id = ?").get(input.user_id, input.session_id) + ); + if (!existing) { + return { deleted: false, deletedSessionIds: [] }; + } + + const sessionIds = collectDescendantSessionIds(this.db, input.user_id, input.session_id); + sessionIds.push(input.session_id); + + this.db.exec("BEGIN IMMEDIATE"); + try { + const placeholders = sessionIds.map(() => "?").join(", "); + const scope = [input.user_id, ...sessionIds]; + + this.db.prepare(` + DELETE FROM session_branches + WHERE user_id = ? + AND ( + child_session_id IN (${placeholders}) + OR parent_session_id IN (${placeholders}) + OR root_session_id IN (${placeholders}) + ) + `).run(input.user_id, ...sessionIds, ...sessionIds, ...sessionIds); + + this.db.prepare(` + DELETE FROM artifact_versions + WHERE user_id = ? + AND artifact_id IN ( + SELECT id FROM artifacts WHERE user_id = ? AND session_id IN (${placeholders}) + ) + `).run(input.user_id, input.user_id, ...sessionIds); + + this.db.prepare(` + DELETE FROM artifacts WHERE user_id = ? AND session_id IN (${placeholders}) + `).run(...scope); + + this.db.prepare(` + UPDATE checkpoints + SET parent_checkpoint_id = NULL + WHERE user_id = ? AND session_id IN (${placeholders}) + `).run(...scope); + + this.db.prepare(` + DELETE FROM checkpoints WHERE user_id = ? AND session_id IN (${placeholders}) + `).run(...scope); + + this.db.prepare(` + DELETE FROM trace_sections WHERE user_id = ? AND session_id IN (${placeholders}) + `).run(...scope); + + this.db.prepare(` + DELETE FROM context_package_snapshots WHERE user_id = ? AND session_id IN (${placeholders}) + `).run(...scope); + + this.db.prepare(` + DELETE FROM conversation_messages WHERE user_id = ? AND session_id IN (${placeholders}) + `).run(...scope); + + this.db.prepare(` + DELETE FROM conversation_summaries WHERE user_id = ? AND session_id IN (${placeholders}) + `).run(...scope); + + this.db.prepare(` + DELETE FROM interactions WHERE user_id = ? AND session_id IN (${placeholders}) + `).run(...scope); + + this.db.prepare(` + DELETE FROM query_history WHERE user_id = ? AND session_id IN (${placeholders}) + `).run(...scope); + + this.db.prepare(` + DELETE FROM long_term_memories WHERE user_id = ? AND session_id IN (${placeholders}) + `).run(...scope); + + this.db.prepare(` + DELETE FROM run_events WHERE user_id = ? AND session_id IN (${placeholders}) + `).run(...scope); + + const runIds = this.db.prepare(` + SELECT id FROM runs WHERE user_id = ? AND session_id IN (${placeholders}) + `).all(...scope) + .map((row) => (isRecord(row) && typeof row.id === "string" ? row.id : null)) + .filter((id): id is string => Boolean(id)); + + if (runIds.length > 0) { + const runPlaceholders = runIds.map(() => "?").join(", "); + this.db.prepare(` + DELETE FROM sql_audit_logs + WHERE user_id = ? AND run_id IN (${runPlaceholders}) + `).run(input.user_id, ...runIds); + this.db.prepare(` + UPDATE runs + SET parent_run_id = NULL + WHERE user_id = ? AND parent_run_id IN (${runPlaceholders}) + `).run(input.user_id, ...runIds); + } + + this.db.prepare(` + UPDATE runs + SET parent_run_id = NULL + WHERE user_id = ? AND session_id IN (${placeholders}) + `).run(...scope); + + this.db.prepare(` + DELETE FROM runs WHERE user_id = ? AND session_id IN (${placeholders}) + `).run(...scope); + + this.db.prepare(` + DELETE FROM file_asset_refs WHERE user_id = ? AND session_id IN (${placeholders}) + `).run(...scope); + + this.db.prepare(` + DELETE FROM sessions WHERE user_id = ? AND id IN (${placeholders}) + `).run(...scope); + + this.db.exec("COMMIT"); + } catch (error) { + this.db.exec("ROLLBACK"); + throw error; + } + + return { deleted: true, deletedSessionIds: sessionIds }; + } +} + +function collectDescendantSessionIds(db: DatabaseSync, userId: string, sessionId: string): string[] { + const ordered: string[] = []; + const seen = new Set(); + const stack = [sessionId]; + + while (stack.length > 0) { + const current = stack.pop() as string; + const children = db.prepare(` + SELECT child_session_id + FROM session_branches + WHERE user_id = ? AND parent_session_id = ? + ORDER BY created_at ASC, child_session_id ASC + `).all(userId, current); + + for (const row of children) { + if (typeof row !== "object" || row === null) continue; + const childId = (row as { child_session_id?: unknown }).child_session_id; + if (typeof childId !== "string") continue; + if (seen.has(childId) || childId === sessionId) continue; + seen.add(childId); + stack.push(childId); + ordered.push(childId); + } + } + + return ordered; } export class SessionBranchRepository { @@ -1525,6 +1684,28 @@ export class RunRepository { return mapRunRow(this.db.prepare(sql).get(...params)); } + listByStatuses(input: { statuses: Array; limit?: number }): RunRecord[] { + if (input.statuses.length === 0) { + return []; + } + const limit = Math.max(1, Math.min(input.limit ?? 500, 2000)); + const placeholders = input.statuses.map(() => "?").join(", "); + const rows = this.db + .prepare( + ` + SELECT * + FROM runs + WHERE status IN (${placeholders}) + ORDER BY started_at DESC + LIMIT ? + ` + ) + .all(...input.statuses, limit); + return rows + .map((row) => mapRunRow(row)) + .filter((run): run is RunRecord => run !== undefined); + } + get(input: { user_id: string; run_id: string }): RunRecord { const run = this.find(input); diff --git a/scripts/test-evidence-selection.mjs b/scripts/test-evidence-selection.mjs index 56b219e..fca7898 100644 --- a/scripts/test-evidence-selection.mjs +++ b/scripts/test-evidence-selection.mjs @@ -1,4 +1,5 @@ import { resolveSelectionFocus } from "../apps/api/dist/evidence-reference-context.js"; +import { extractEffectiveRunConfig } from "../apps/api/dist/run-input.js"; const preview = { columns: ["a", "b", "c"], @@ -19,6 +20,48 @@ const check = (name, condition) => { } }; +// 0) run_config intake must preserve source.selection (partial table cites). +{ + const selection = { mode: "cells", range: { r0: 0, c0: 1, r1: 1, c1: 2 } }; + const config = extractEffectiveRunConfig({ + threadId: "thread-1", + runId: "run-1", + messages: [], + tools: [], + context: [], + state: {}, + forwardedProps: { + run_config: { + evidenceRefs: [ + { + id: "artifact:a1:sel:cells:0,1,1,2", + kind: "table", + label: "SQL result.csv", + sessionId: "session-1", + runId: "run-1", + source: { + artifactId: "a1", + fileId: "file-1", + selection, + }, + }, + ], + }, + }, + }); + const parsed = config.evidenceRefs[0]?.source.selection; + check("run-input: keeps selection.mode", parsed?.mode === "cells"); + check( + "run-input: keeps selection.range", + parsed?.mode !== "text" + && parsed?.range?.r0 === 0 + && parsed?.range?.c0 === 1 + && parsed?.range?.r1 === 1 + && parsed?.range?.c1 === 2, + ); + check("run-input: keeps fileId alongside selection", config.evidenceRefs[0]?.source.fileId === "file-1"); +} + // 1) cell range slices only the selected sub-table and replaces the full preview. { const focus = resolveSelectionFocus(