diff --git a/docs/design/channel-conversation-continuity-v2-implementation-plan.md b/docs/design/channel-conversation-continuity-v2-implementation-plan.md index 480abee71..6e84e7d79 100644 --- a/docs/design/channel-conversation-continuity-v2-implementation-plan.md +++ b/docs/design/channel-conversation-continuity-v2-implementation-plan.md @@ -1,6 +1,6 @@ # Channel Conversation Continuity v2 Implementation Plan -Status: G0 complete; S-1 and S0 merged; S1+S2, S3, and S4 implemented and locally verified; S5-S7 pending +Status: G0 complete; S-1 and S0 merged; S1+S2, S3, S4, and S5 implemented and locally verified; S6-S7 pending Owner: conversation, channel context, Ling Agent Session, and storage maintainers Date: 2026-08-08 Last verified: 2026-08-10 @@ -400,7 +400,7 @@ input accepted ## Slice S3: Context Segmentation And Shared Projection -Status: implemented on `codex/continuity-v2-context-projector` and locally verified. +Status: implemented on `codex/continuity-v2-context-projection` and locally verified. Contracts (`35/35`), Core (`355/355`), Daemon (`888/888`), Remote Protocol (`40/40`), and Web (`42/42`) tests pass; Web typecheck and `git diff --check` pass. Model-facing Channel consumers use the shared projector, including causally later queued @@ -447,7 +447,7 @@ context. Paging for UI remains allowed; context construction goes through the pr ## Slice S4: Intent v2 And Issue #421 Closure -Status: implemented on `codex/continuity-v2-context-projector`. The clean-break SQLite +Status: implemented on `codex/continuity-v2-intent-v2`. The clean-break SQLite migration, contracts, API/client filters, selector, structured clarification, typed execution, Web diagnostics, and Issue #421 regression fixtures are locally verified. The focused channel workspace suite passes 186/186, including article continuation, @@ -500,6 +500,11 @@ bad clarification -> unrelated snake game ## Slice S5: Proactive Summaries And Fork Snapshots +Status: implemented on `codex/continuity-v2-summaries-snapshots` and locally verified. +Core (`357/357`) and Daemon (`894/894`) tests pass, including provider-aware hard +summarization, stable raw-tail reconstruction, stale-result fencing, immutable fork +snapshot reconstruction, and the one strict context-length fallback. + ### Changes - Add a summary coordinator that watches committed stable ranges per segment. @@ -804,7 +809,7 @@ interfaces at the end of S3 and keep implementation files separate before parall - Verify: projection consistency, permission, correction, and consumer adapter tests. - [x] **T5 (P1, human: ~2 days / Codex: ~3-4 hours)**: cut intent decisions to v2 and satisfy issue #421 closure criteria. - Verify: daemon regression suite, explicit file safety, clarification actionability, and eval cases. -- [ ] **T6 (P2, human: ~2 days / Codex: ~3-5 hours)**: add proactive summaries, coverage watermarks, stale-result rejection, and fork snapshots. +- [x] **T6 (P2, human: ~2 days / Codex: ~3-5 hours)**: add proactive summaries, coverage watermarks, stale-result rejection, and fork snapshots. - Verify: budget, race, invalidation, reconstruction, and fallback tests. - [ ] **T7 (P2, human: ~2 days / Codex: ~3-5 hours)**: enforce multi-agent projection versions and private-memory isolation. - Verify: shared-core, stale fan-in, handoff, and security fixtures. diff --git a/packages/core/src/conversation-context-store-types.ts b/packages/core/src/conversation-context-store-types.ts index 215222787..706c6bb7a 100644 --- a/packages/core/src/conversation-context-store-types.ts +++ b/packages/core/src/conversation-context-store-types.ts @@ -1,8 +1,11 @@ import type { ContextSegment, ContextSegmentState, + ConversationContextSnapshot, ConversationContextSummary, + ConversationContextSummaryScope, ConversationDomainRef, + ConversationItemVisibilityV1, TurnContextAssignment, TurnContextAssignmentStatus, TurnContextDecisionSource, @@ -51,3 +54,71 @@ export interface ListConversationContextSummariesOptions { readonly throughSequence?: number; readonly limit?: number; } + +export interface ConversationContextSummarySourceFence { + readonly itemId: string; + readonly sequence: number; + readonly revision: number; + readonly updatedAt: string; + readonly visibility: ConversationItemVisibilityV1; + readonly visibilityOwnerId?: string; + readonly turnId: string; + readonly assignmentId: string; + readonly assignmentRevision: number; +} + +export interface CommitConversationContextSummaryInput { + readonly threadId: string; + readonly segmentId?: string; + readonly scope: ConversationContextSummaryScope; + readonly coveredFromSequence: number; + readonly coveredThroughSequence: number; + readonly objective?: string; + readonly decisions?: readonly string[]; + readonly constraints?: readonly string[]; + readonly openQuestions?: readonly string[]; + readonly activeWorkRefs?: readonly ConversationDomainRef[]; + readonly sourceRefs: readonly ConversationDomainRef[]; + readonly policyVersion: string; + readonly projectionVersion: string; + readonly visibilityFingerprint: string; + readonly expectedActiveSummaryId: string | null; + readonly sourceFence: readonly ConversationContextSummarySourceFence[]; +} + +export type ConversationContextSummaryDiscardReason = + | "active_summary_changed" + | "segment_changed" + | "source_items_changed" + | "assignments_changed" + | "visibility_changed"; + +export type CommitConversationContextSummaryResult = + | { + readonly status: "committed"; + readonly summary: ConversationContextSummary; + } + | { + readonly status: "discarded_stale"; + readonly reason: ConversationContextSummaryDiscardReason; + }; + +export interface CreateConversationContextSnapshotInput { + readonly id?: string; + readonly threadId: string; + readonly segmentId: string; + readonly projectionVersion: string; + readonly throughItemSequence: number; + readonly summaryIds?: readonly string[]; + readonly sourceRefs: readonly ConversationDomainRef[]; + readonly policyVersion: string; +} + +export interface ListConversationContextSnapshotsOptions { + readonly threadId: string; + readonly limit?: number; +} + +export interface CreateConversationContextSnapshotResult { + readonly snapshot: ConversationContextSnapshot; +} diff --git a/packages/core/src/conversation-item-store-types.ts b/packages/core/src/conversation-item-store-types.ts index f03bd9071..f7e04b749 100644 --- a/packages/core/src/conversation-item-store-types.ts +++ b/packages/core/src/conversation-item-store-types.ts @@ -68,6 +68,7 @@ export interface ListConversationItemsForSegmentOptions { readonly segmentId: string; readonly throughSequence: number; readonly afterSequence?: number; + readonly order?: "oldest" | "newest"; readonly limit?: number; } diff --git a/packages/core/src/conversation-thread-topology-types.ts b/packages/core/src/conversation-thread-topology-types.ts index b9c549b6a..ad0fea3eb 100644 --- a/packages/core/src/conversation-thread-topology-types.ts +++ b/packages/core/src/conversation-thread-topology-types.ts @@ -1,4 +1,5 @@ import type { LingThread } from "./run-types.js"; +import type { CreateConversationContextSnapshotInput } from "./conversation-context-store-types.js"; interface ConversationThreadTopologyBaseInput { readonly id?: string; @@ -19,6 +20,8 @@ export type ResolveOrCreateConversationThreadInput = readonly surfaceKind: "branch"; readonly parentThreadId: string; readonly parentTurnId: string; + readonly forkedFromRunId?: string; + readonly forkContextSnapshot: Omit; }); export interface ResolveOrCreateConversationThreadResult { diff --git a/packages/core/src/run-types.ts b/packages/core/src/run-types.ts index 0196082d4..421b0e7d2 100644 --- a/packages/core/src/run-types.ts +++ b/packages/core/src/run-types.ts @@ -1721,6 +1721,20 @@ export interface RunStore { listConversationContextSummaries( options: import("./conversation-context-store-types.js").ListConversationContextSummariesOptions, ): readonly import("@ling-stack/contracts").ConversationContextSummary[]; + commitConversationContextSummary( + input: import("./conversation-context-store-types.js").CommitConversationContextSummaryInput, + now: string, + ): import("./conversation-context-store-types.js").CommitConversationContextSummaryResult; + createConversationContextSnapshot( + input: import("./conversation-context-store-types.js").CreateConversationContextSnapshotInput, + now: string, + ): import("@ling-stack/contracts").ConversationContextSnapshot; + getConversationContextSnapshot( + id: string, + ): import("@ling-stack/contracts").ConversationContextSnapshot | undefined; + listConversationContextSnapshots( + options: import("./conversation-context-store-types.js").ListConversationContextSnapshotsOptions, + ): readonly import("@ling-stack/contracts").ConversationContextSnapshot[]; createChannelMessageRouteDecision(input: CreateChannelMessageRouteDecisionInput, now: string): ChannelMessageRouteDecision; getChannelMessageRouteDecision(id: string): ChannelMessageRouteDecision | undefined; listChannelMessageRouteDecisions(options?: ListChannelMessageRouteDecisionsOptions): ChannelMessageRouteDecision[]; @@ -2278,6 +2292,10 @@ export type ChannelStoreFacet = RunStoreFacet< | "applyTurnContextAssignment" | "getConversationContextSummary" | "listConversationContextSummaries" + | "commitConversationContextSummary" + | "createConversationContextSnapshot" + | "getConversationContextSnapshot" + | "listConversationContextSnapshots" | "createChannelMessageRouteDecision" | "getChannelMessageRouteDecision" | "listChannelMessageRouteDecisions" diff --git a/packages/core/src/storage/sqlite-run-store.ts b/packages/core/src/storage/sqlite-run-store.ts index 79990370f..843249c3b 100644 --- a/packages/core/src/storage/sqlite-run-store.ts +++ b/packages/core/src/storage/sqlite-run-store.ts @@ -746,6 +746,8 @@ export class SqliteRunStore implements RunStore { { createLingThread: (input, now) => this.lingThreads.createLingThread(input, now), getLingThread: (threadId) => this.lingThreads.getLingThread(threadId), + createConversationContextSnapshotWithinTransaction: (input, now) => + this.conversationContext.createConversationContextSnapshotWithinTransaction(input, now), }, ); this.lingThreadCleanup = new SqliteLingThreadCleanupStore( @@ -2547,6 +2549,32 @@ export class SqliteRunStore implements RunStore { return this.conversationContext.listConversationContextSummaries(options); } + commitConversationContextSummary( + input: import("../conversation-context-store-types.js").CommitConversationContextSummaryInput, + now: string, + ): import("../conversation-context-store-types.js").CommitConversationContextSummaryResult { + return this.conversationContext.commitConversationContextSummary(input, now); + } + + createConversationContextSnapshot( + input: import("../conversation-context-store-types.js").CreateConversationContextSnapshotInput, + now: string, + ): import("@ling-stack/contracts").ConversationContextSnapshot { + return this.conversationContext.createConversationContextSnapshot(input, now); + } + + getConversationContextSnapshot( + id: string, + ): import("@ling-stack/contracts").ConversationContextSnapshot | undefined { + return this.conversationContext.getConversationContextSnapshot(id); + } + + listConversationContextSnapshots( + options: import("../conversation-context-store-types.js").ListConversationContextSnapshotsOptions, + ): readonly import("@ling-stack/contracts").ConversationContextSnapshot[] { + return this.conversationContext.listConversationContextSnapshots(options); + } + createChannelMessage(input: CreateChannelMessageInput, now: string): ChannelMessage { if (this.legacyContinuityFixtureMode) return this.createLegacyChannelMessageFixture(input, now); const channel = this.getChannel(input.channelId); diff --git a/packages/core/src/storage/sqlite/conversation-context-store.ts b/packages/core/src/storage/sqlite/conversation-context-store.ts index 2bccb8a26..5d64e4a4c 100644 --- a/packages/core/src/storage/sqlite/conversation-context-store.ts +++ b/packages/core/src/storage/sqlite/conversation-context-store.ts @@ -2,10 +2,12 @@ import { randomUUID } from "node:crypto"; import type { SQLOutputValue } from "node:sqlite"; import { isContextSegment, + isConversationContextSnapshot, isConversationContextSummary, isConversationDomainRef, isTurnContextAssignment, type ContextSegment, + type ConversationContextSnapshot, type ConversationContextSummary, type ConversationDomainRef, type TurnContextAssignment, @@ -13,7 +15,13 @@ import { import type { ApplyTurnContextAssignmentInput, ApplyTurnContextAssignmentResult, + CommitConversationContextSummaryInput, + CommitConversationContextSummaryResult, + ConversationContextSummaryDiscardReason, + ConversationContextSummarySourceFence, + CreateConversationContextSnapshotInput, ListContextSegmentsOptions, + ListConversationContextSnapshotsOptions, ListConversationContextSummariesOptions, } from "../../conversation-context-store-types.js"; import { DEFAULT_WORKSPACE_ID } from "../../workspace-types.js"; @@ -167,7 +175,7 @@ export class SqliteConversationContextStore { idempotentReplay: false, }; } catch (error) { - if (this.context.database.isTransaction) this.context.database.exec("ROLLBACK"); + this.context.database.exec("ROLLBACK"); throw error; } } @@ -207,6 +215,290 @@ export class SqliteConversationContextStore { `).all(...params).map((row) => mapConversationContextSummary(row as Record)); } + commitConversationContextSummary( + input: CommitConversationContextSummaryInput, + now: string, + ): CommitConversationContextSummaryResult { + assertCanonicalTimestamp(now); + assertValidSummaryCommitInput(input); + this.context.database.exec("BEGIN IMMEDIATE"); + try { + const staleReason = this.validateSummaryCommitFence(input); + if (staleReason) { + this.context.database.exec("COMMIT"); + return { status: "discarded_stale", reason: staleReason }; + } + const active = this.readActiveSummary(input.threadId, input.segmentId, input.scope); + if (active) { + if (input.coveredFromSequence !== active.coveredFromSequence) { + throw new Error("Replacement context summary must preserve the original coverage start."); + } + if (input.coveredThroughSequence <= active.coveredThroughSequence) { + throw new Error("Replacement context summary must advance its coverage watermark."); + } + if (!input.sourceRefs.some((ref) => ref.type === "context_summary" && ref.id === active.id)) { + throw new Error(`Replacement context summary must reference its predecessor: ${active.id}`); + } + } + const summary: ConversationContextSummary = { + id: `context_summary_${randomUUID()}`, + threadId: input.threadId, + ...(input.segmentId ? { segmentId: input.segmentId } : {}), + scope: input.scope, + status: "active", + version: (active?.version ?? 0) + 1, + coveredFromSequence: input.coveredFromSequence, + coveredThroughSequence: input.coveredThroughSequence, + ...(input.objective?.trim() ? { objective: input.objective.trim() } : {}), + decisions: normalizeStrings(input.decisions), + constraints: normalizeStrings(input.constraints), + openQuestions: normalizeStrings(input.openQuestions), + activeWorkRefs: uniqueDomainRefs(input.activeWorkRefs ?? []), + sourceRefs: uniqueDomainRefs(input.sourceRefs), + ...(active ? { supersedesSummaryId: active.id } : {}), + policyVersion: input.policyVersion, + createdAt: now, + }; + if (!isConversationContextSummary(summary)) { + throw new Error("Conversation context summary failed schema validation."); + } + if (active) { + this.context.database.prepare(` + UPDATE conversation_context_summaries + SET status = 'superseded' + WHERE workspace_id = ? AND id = ? AND status = 'active' + `).run(DEFAULT_WORKSPACE_ID, active.id); + } + this.context.database.prepare(` + INSERT INTO conversation_context_summaries ( + id, workspace_id, thread_id, segment_id, scope, status, version, + covered_from_sequence, covered_through_sequence, objective, decisions_json, + constraints_json, open_questions_json, active_work_refs_json, source_refs_json, + supersedes_summary_id, policy_version, created_at + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + `).run( + summary.id, + DEFAULT_WORKSPACE_ID, + summary.threadId, + summary.segmentId ?? null, + summary.scope, + summary.status, + summary.version, + summary.coveredFromSequence, + summary.coveredThroughSequence, + summary.objective ?? null, + stringifyJson(summary.decisions), + stringifyJson(summary.constraints), + stringifyJson(summary.openQuestions), + stringifyJson(summary.activeWorkRefs), + stringifyJson(summary.sourceRefs), + summary.supersedesSummaryId ?? null, + summary.policyVersion, + summary.createdAt, + ); + this.context.database.exec("COMMIT"); + return { status: "committed", summary }; + } catch (error) { + this.context.database.exec("ROLLBACK"); + throw error; + } + } + + createConversationContextSnapshot( + input: CreateConversationContextSnapshotInput, + now: string, + ): ConversationContextSnapshot { + assertCanonicalTimestamp(now); + this.context.database.exec("BEGIN IMMEDIATE"); + try { + const snapshot = this.createConversationContextSnapshotWithinTransaction(input, now); + this.context.database.exec("COMMIT"); + return snapshot; + } catch (error) { + this.context.database.exec("ROLLBACK"); + throw error; + } + } + + createConversationContextSnapshotWithinTransaction( + input: CreateConversationContextSnapshotInput, + now: string, + ): ConversationContextSnapshot { + assertCanonicalTimestamp(now); + assertValidSnapshotInput(input); + const thread = this.context.database.prepare(` + SELECT status FROM ling_threads WHERE workspace_id = ? AND id = ? + `).get(DEFAULT_WORKSPACE_ID, input.threadId) as Record | undefined; + if (!thread) throw new Error(`Conversation snapshot thread not found: ${input.threadId}`); + const segment = this.context.database.prepare(` + SELECT thread_id, state FROM context_segments WHERE workspace_id = ? AND id = ? + `).get(DEFAULT_WORKSPACE_ID, input.segmentId) as Record | undefined; + if (!segment || readString(segment, "thread_id") !== input.threadId) { + throw new Error(`Conversation snapshot segment does not belong to thread: ${input.segmentId}`); + } + const watermark = this.context.database.prepare(` + SELECT id FROM conversation_items + WHERE workspace_id = ? AND thread_id = ? AND sequence = ? AND status = 'committed' + `).get(DEFAULT_WORKSPACE_ID, input.threadId, input.throughItemSequence); + if (!watermark) { + throw new Error(`Conversation snapshot watermark item not found: ${input.threadId}/${input.throughItemSequence}`); + } + const summaryIds = [...new Set(input.summaryIds ?? [])]; + for (const summaryId of summaryIds) { + const summary = this.getConversationContextSummary(summaryId); + if (!summary || summary.threadId !== input.threadId || summary.coveredThroughSequence > input.throughItemSequence) { + throw new Error(`Conversation snapshot summary is outside the fork boundary: ${summaryId}`); + } + } + const snapshot: ConversationContextSnapshot = { + id: input.id ?? `context_snapshot_${randomUUID()}`, + threadId: input.threadId, + segmentId: input.segmentId, + projectionVersion: input.projectionVersion, + throughItemSequence: input.throughItemSequence, + summaryIds, + sourceRefs: uniqueDomainRefs(input.sourceRefs), + policyVersion: input.policyVersion, + createdAt: now, + }; + if (!isConversationContextSnapshot(snapshot)) { + throw new Error("Conversation context snapshot failed schema validation."); + } + this.context.database.prepare(` + INSERT INTO conversation_context_snapshots ( + id, workspace_id, thread_id, segment_id, projection_version, + through_item_sequence, summary_ids_json, source_refs_json, policy_version, created_at + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + `).run( + snapshot.id, + DEFAULT_WORKSPACE_ID, + snapshot.threadId, + snapshot.segmentId, + snapshot.projectionVersion, + snapshot.throughItemSequence, + stringifyJson(snapshot.summaryIds), + stringifyJson(snapshot.sourceRefs), + snapshot.policyVersion, + snapshot.createdAt, + ); + return snapshot; + } + + getConversationContextSnapshot(snapshotId: string): ConversationContextSnapshot | undefined { + const row = this.context.database.prepare(`${conversationContextSnapshotSelectSql} WHERE id = ?`).get(snapshotId); + return row ? mapConversationContextSnapshot(row as Record) : undefined; + } + + listConversationContextSnapshots( + options: ListConversationContextSnapshotsOptions, + ): readonly ConversationContextSnapshot[] { + const limit = Math.min(Math.max(options.limit ?? 50, 1), 200); + return this.context.database.prepare(` + ${conversationContextSnapshotSelectSql} + WHERE workspace_id = ? AND thread_id = ? + ORDER BY created_at DESC, id DESC + LIMIT ? + `).all(DEFAULT_WORKSPACE_ID, options.threadId, limit) + .map((row) => mapConversationContextSnapshot(row as Record)); + } + + private validateSummaryCommitFence( + input: CommitConversationContextSummaryInput, + ): ConversationContextSummaryDiscardReason | undefined { + const active = this.readActiveSummary(input.threadId, input.segmentId, input.scope); + if ((active?.id ?? null) !== input.expectedActiveSummaryId) return "active_summary_changed"; + const thread = this.context.database.prepare(` + SELECT status FROM ling_threads WHERE workspace_id = ? AND id = ? + `).get(DEFAULT_WORKSPACE_ID, input.threadId) as Record | undefined; + if (!thread) throw new Error(`Conversation summary thread not found: ${input.threadId}`); + if (input.segmentId) { + const segment = this.context.database.prepare(` + SELECT thread_id, state FROM context_segments WHERE workspace_id = ? AND id = ? + `).get(DEFAULT_WORKSPACE_ID, input.segmentId) as Record | undefined; + if (!segment || readString(segment, "thread_id") !== input.threadId) { + throw new Error(`Conversation summary segment does not belong to thread: ${input.segmentId}`); + } + const state = readString(segment, "state"); + if (state === "archived" || state === "merged") return "segment_changed"; + } + const current = this.readSummaryFenceRows(input); + if (current.length !== input.sourceFence.length) return "source_items_changed"; + for (let index = 0; index < current.length; index += 1) { + const row = current[index]; + const expected = input.sourceFence[index]; + if (!row || !expected) return "source_items_changed"; + if ( + readString(row, "item_id") !== expected.itemId + || readNumber(row, "sequence") !== expected.sequence + || readNumber(row, "item_revision") !== expected.revision + || readString(row, "updated_at") !== expected.updatedAt + ) return "source_items_changed"; + if ( + readString(row, "turn_id") !== expected.turnId + || readString(row, "assignment_id") !== expected.assignmentId + || readNumber(row, "assignment_revision") !== expected.assignmentRevision + ) return "assignments_changed"; + if ( + readString(row, "visibility") !== expected.visibility + || readNullableString(row, "visibility_owner_id") !== expected.visibilityOwnerId + ) return "visibility_changed"; + } + return undefined; + } + + private readSummaryFenceRows( + input: CommitConversationContextSummaryInput, + ): readonly Record[] { + const segmentClause = input.segmentId ? "AND assignments.segment_id = ?" : ""; + const params: (string | number)[] = [ + DEFAULT_WORKSPACE_ID, + input.threadId, + input.sourceFence[0]?.sequence ?? input.coveredFromSequence, + input.coveredThroughSequence, + ]; + if (input.segmentId) params.push(input.segmentId); + return this.context.database.prepare(` + SELECT items.id AS item_id, items.sequence, items.revision AS item_revision, + items.updated_at, items.visibility, items.visibility_owner_id, items.turn_id, + assignments.id AS assignment_id, assignments.revision AS assignment_revision + FROM conversation_items AS items + INNER JOIN turn_context_assignments AS assignments + ON assignments.turn_id = items.turn_id + AND assignments.revision = ( + SELECT MAX(candidate.revision) + FROM turn_context_assignments AS candidate + WHERE candidate.turn_id = items.turn_id + ) + WHERE items.workspace_id = ? AND items.thread_id = ? + AND items.sequence >= ? AND items.sequence <= ? + AND items.status = 'committed' + AND items.visibility IN ('channel', 'thread_participants') + AND NOT EXISTS ( + SELECT 1 FROM conversation_items AS replacement + WHERE replacement.supersedes_item_id = items.id + ) + ${segmentClause} + ORDER BY items.sequence ASC, items.id ASC + `).all(...params) as Record[]; + } + + private readActiveSummary( + threadId: string, + segmentId: string | undefined, + scope: ConversationContextSummary["scope"], + ): ConversationContextSummary | undefined { + const rows = this.context.database.prepare(` + ${conversationContextSummarySelectSql} + WHERE workspace_id = ? AND thread_id = ? AND segment_id IS ? AND scope = ? AND status = 'active' + ORDER BY version DESC, covered_through_sequence DESC, created_at DESC + LIMIT 2 + `).all(DEFAULT_WORKSPACE_ID, threadId, segmentId ?? null, scope) as Record[]; + if (rows.length > 1) { + throw new Error(`Multiple active conversation summaries found for ${threadId}/${segmentId ?? scope}.`); + } + return rows[0] ? mapConversationContextSummary(rows[0]) : undefined; + } + private resolveAssignmentSegment( input: ApplyTurnContextAssignmentInput, threadId: string, @@ -299,6 +591,12 @@ const conversationContextSummarySelectSql = ` FROM conversation_context_summaries `; +const conversationContextSnapshotSelectSql = ` + SELECT id, workspace_id, thread_id, segment_id, projection_version, + through_item_sequence, summary_ids_json, source_refs_json, policy_version, created_at + FROM conversation_context_snapshots +`; + function mapContextSegment(row: Record): ContextSegment { const value = { id: readString(row, "id"), @@ -372,6 +670,25 @@ function mapConversationContextSummary(row: Record): Con return value; } +function mapConversationContextSnapshot(row: Record): ConversationContextSnapshot { + const summaryIds = parseJsonStringArray(readString(row, "summary_ids_json")); + const value = { + id: readString(row, "id"), + threadId: readString(row, "thread_id"), + segmentId: readString(row, "segment_id"), + projectionVersion: readString(row, "projection_version"), + throughItemSequence: readNumber(row, "through_item_sequence"), + summaryIds, + sourceRefs: parseDomainRefs(readString(row, "source_refs_json")), + policyVersion: readString(row, "policy_version"), + createdAt: readString(row, "created_at"), + } as unknown; + if (!isConversationContextSnapshot(value)) { + throw new Error(`Invalid conversation context snapshot row: ${readString(row, "id")}`); + } + return value; +} + function parseDomainRefs(value: string): ConversationDomainRef[] { const parsed = parseJson(value); if (!Array.isArray(parsed) || parsed.some((ref) => !isConversationDomainRef(ref))) { @@ -395,6 +712,102 @@ function assignmentMatches( && stringifyJson(assignment.evidenceRefs) === stringifyJson(evidenceRefs); } +function assertValidSummaryCommitInput(input: CommitConversationContextSummaryInput): void { + if (!input.threadId.trim()) throw new Error("Conversation summary thread id is required."); + if (input.scope === "segment" && !input.segmentId?.trim()) { + throw new Error("Segment context summary requires a segment id."); + } + if (!Number.isSafeInteger(input.coveredFromSequence) || input.coveredFromSequence < 1) { + throw new RangeError(`Invalid context summary coverage start: ${input.coveredFromSequence}`); + } + if ( + !Number.isSafeInteger(input.coveredThroughSequence) + || input.coveredThroughSequence < input.coveredFromSequence + ) { + throw new RangeError(`Invalid context summary coverage end: ${input.coveredThroughSequence}`); + } + if (!input.policyVersion.trim() || !input.projectionVersion.trim() || !input.visibilityFingerprint.trim()) { + throw new Error("Conversation summary policy and projection fence values are required."); + } + if (input.sourceRefs.some((ref) => !isConversationDomainRef(ref))) { + throw new Error("Conversation summary contains an invalid source ref."); + } + if ((input.activeWorkRefs ?? []).some((ref) => !isConversationDomainRef(ref))) { + throw new Error("Conversation summary contains an invalid active-work ref."); + } + if (input.sourceFence.length === 0) { + throw new Error("Conversation summary source fence must not be empty."); + } + const itemIds = new Set(); + let priorSequence = 0; + for (const source of input.sourceFence) { + assertValidSummarySourceFence(source); + if (itemIds.has(source.itemId)) throw new Error(`Duplicate conversation summary source item: ${source.itemId}`); + if (source.sequence <= priorSequence) { + throw new Error("Conversation summary source fence must be ordered by increasing sequence."); + } + itemIds.add(source.itemId); + priorSequence = source.sequence; + } + if ( + (input.sourceFence[0]?.sequence ?? 0) < input.coveredFromSequence + || input.sourceFence.at(-1)?.sequence !== input.coveredThroughSequence + ) { + throw new Error("Conversation summary coverage must contain its fenced source range."); + } + const sourceRefItemIds = new Set( + input.sourceRefs + .filter((ref) => ref.type === "conversation_item") + .map((ref) => ref.id), + ); + for (const itemId of itemIds) { + if (!sourceRefItemIds.has(itemId)) { + throw new Error(`Conversation summary source ref is missing item: ${itemId}`); + } + } +} + +function assertValidSummarySourceFence(source: ConversationContextSummarySourceFence): void { + if (!source.itemId.trim() || !source.turnId.trim() || !source.assignmentId.trim()) { + throw new Error("Conversation summary source fence ids are required."); + } + if (!Number.isSafeInteger(source.sequence) || source.sequence < 1) { + throw new RangeError(`Invalid conversation summary source sequence: ${source.sequence}`); + } + if (!Number.isSafeInteger(source.revision) || source.revision < 0) { + throw new RangeError(`Invalid conversation summary source revision: ${source.revision}`); + } + if (!Number.isSafeInteger(source.assignmentRevision) || source.assignmentRevision < 1) { + throw new RangeError(`Invalid conversation summary assignment revision: ${source.assignmentRevision}`); + } + assertCanonicalTimestamp(source.updatedAt); +} + +function assertValidSnapshotInput(input: CreateConversationContextSnapshotInput): void { + if (!input.threadId.trim() || !input.segmentId.trim()) { + throw new Error("Conversation snapshot thread and segment ids are required."); + } + if (!input.projectionVersion.trim() || !input.policyVersion.trim()) { + throw new Error("Conversation snapshot projection and policy versions are required."); + } + if (!Number.isSafeInteger(input.throughItemSequence) || input.throughItemSequence < 1) { + throw new RangeError(`Invalid conversation snapshot watermark: ${input.throughItemSequence}`); + } + if (input.sourceRefs.some((ref) => !isConversationDomainRef(ref))) { + throw new Error("Conversation snapshot contains an invalid source ref."); + } +} + +function normalizeStrings(values: readonly string[] | undefined): string[] { + return [...new Set((values ?? []).map((value) => value.trim()).filter(Boolean))]; +} + +function uniqueDomainRefs(refs: readonly ConversationDomainRef[]): ConversationDomainRef[] { + const byKey = new Map(); + for (const ref of refs) byKey.set(`${ref.type}:${ref.id}`, ref); + return [...byKey.values()]; +} + function assertValidContextSegment(value: unknown): asserts value is ContextSegment { if (!isContextSegment(value)) throw new Error("Context segment failed schema validation."); } diff --git a/packages/core/src/storage/sqlite/conversation-item-store.ts b/packages/core/src/storage/sqlite/conversation-item-store.ts index bf637a4d0..15a836340 100644 --- a/packages/core/src/storage/sqlite/conversation-item-store.ts +++ b/packages/core/src/storage/sqlite/conversation-item-store.ts @@ -372,6 +372,7 @@ export class SqliteConversationItemStore { throw new RangeError(`Invalid conversation projection lower watermark: ${afterSequence}`); } const limit = Math.min(Math.max(options.limit ?? 500, 1), 500); + const order = options.order === "oldest" ? "ASC" : "DESC"; const rows = this.context.database.prepare(` SELECT items.* FROM conversation_items AS items @@ -392,7 +393,7 @@ export class SqliteConversationItemStore { SELECT 1 FROM conversation_items AS replacement WHERE replacement.supersedes_item_id = items.id ) - ORDER BY items.sequence DESC + ORDER BY items.sequence ${order} LIMIT ? `).all( DEFAULT_WORKSPACE_ID, @@ -402,9 +403,8 @@ export class SqliteConversationItemStore { options.throughSequence, limit, ); - return rows - .map((row) => mapConversationItem(row as Record)) - .reverse(); + const items = rows.map((row) => mapConversationItem(row as Record)); + return options.order === "oldest" ? items : items.reverse(); } updatePendingConversationItem( diff --git a/packages/core/src/storage/sqlite/conversation-thread-topology-store.ts b/packages/core/src/storage/sqlite/conversation-thread-topology-store.ts index e2b6010fa..dabaab2e3 100644 --- a/packages/core/src/storage/sqlite/conversation-thread-topology-store.ts +++ b/packages/core/src/storage/sqlite/conversation-thread-topology-store.ts @@ -3,6 +3,8 @@ import type { ResolveOrCreateConversationThreadInput, ResolveOrCreateConversationThreadResult, } from "../../conversation-thread-topology-types.js"; +import type { CreateConversationContextSnapshotInput } from "../../conversation-context-store-types.js"; +import type { ConversationContextSnapshot } from "@ling-stack/contracts"; import type { CreateLingThreadInput, LingThread } from "../../run-types.js"; import { DEFAULT_WORKSPACE_ID } from "../../workspace-types.js"; import { readNullableString, readString } from "./helpers.js"; @@ -11,6 +13,10 @@ import type { SqliteStoreContext } from "./store-context.js"; export interface SqliteConversationThreadTopologyStorePorts { readonly createLingThread: (input: CreateLingThreadInput, now: string) => LingThread; readonly getLingThread: (threadId: string) => LingThread | undefined; + readonly createConversationContextSnapshotWithinTransaction: ( + input: CreateConversationContextSnapshotInput, + now: string, + ) => ConversationContextSnapshot; } export class SqliteConversationThreadTopologyStore { @@ -29,7 +35,7 @@ export class SqliteConversationThreadTopologyStore { this.context.database.exec("COMMIT"); return result; } catch (error) { - if (this.context.database.isTransaction) this.context.database.exec("ROLLBACK"); + this.context.database.exec("ROLLBACK"); throw error; } } @@ -43,7 +49,7 @@ export class SqliteConversationThreadTopologyStore { this.assertExistingMatches(input, existing); return { thread: existing, created: false }; } - const createInput = this.buildCreateInput(input); + const createInput = this.buildCreateInput(input, now); return { thread: this.ports.createLingThread(createInput, now), created: true }; } @@ -72,7 +78,7 @@ export class SqliteConversationThreadTopologyStore { return this.ports.getLingThread(readString(row as Record, "id")); } - private buildCreateInput(input: ResolveOrCreateConversationThreadInput): CreateLingThreadInput { + private buildCreateInput(input: ResolveOrCreateConversationThreadInput, now: string): CreateLingThreadInput { switch (input.surfaceKind) { case "channel_root": this.assertChannelExists(input.channelId); @@ -103,8 +109,12 @@ export class SqliteConversationThreadTopologyStore { relationKind: "topic", }; } - case "branch": + case "branch": { this.assertParentLineage(input.channelId, input.parentThreadId, input.parentTurnId); + const snapshot = this.ports.createConversationContextSnapshotWithinTransaction({ + ...input.forkContextSnapshot, + threadId: input.parentThreadId, + }, now); return { ...(input.id ? { id: input.id } : {}), kind: "channel_discussion", @@ -113,9 +123,12 @@ export class SqliteConversationThreadTopologyStore { objective: input.objective, parentThreadId: input.parentThreadId, parentTurnId: input.parentTurnId, + ...(input.forkedFromRunId ? { forkedFromRunId: input.forkedFromRunId } : {}), + forkContextSnapshotId: snapshot.id, surfaceKind: "branch", relationKind: "branch", }; + } } } diff --git a/packages/core/test/channel-runtime-store.test.mjs b/packages/core/test/channel-runtime-store.test.mjs index a4c01c231..37bfdf80f 100644 --- a/packages/core/test/channel-runtime-store.test.mjs +++ b/packages/core/test/channel-runtime-store.test.mjs @@ -604,6 +604,8 @@ test("sqlite channel message root pages can exclude channel branch lanes while k ); const rootItem = store.getConversationItem(rootMessage.id); assert.ok(rootItem?.turnId); + const rootAssignment = store.getEffectiveTurnContextAssignment(rootItem.turnId); + assert.ok(rootAssignment); const branchThread = store.resolveOrCreateConversationThread({ id: "thread_branch_lane", surfaceKind: "branch", @@ -612,6 +614,15 @@ test("sqlite channel message root pages can exclude channel branch lanes while k parentTurnId: rootItem.turnId, title: "Branch lane", objective: "Test branch isolation.", + forkContextSnapshot: { + id: "snapshot_branch_lane", + segmentId: rootAssignment.segmentId, + projectionVersion: "ctxv2_branch_lane", + throughItemSequence: rootItem.sequence, + summaryIds: [], + sourceRefs: [{ type: "conversation_item", id: rootItem.id }], + policyVersion: "continuity-v2-snapshot-v1", + }, }, "2026-05-29T00:00:02.500Z").thread; const branch = store.createChannelBranch( { diff --git a/packages/core/test/conversation-context-store.test.mjs b/packages/core/test/conversation-context-store.test.mjs index d3a2ca76d..d36f9f959 100644 --- a/packages/core/test/conversation-context-store.test.mjs +++ b/packages/core/test/conversation-context-store.test.mjs @@ -136,6 +136,113 @@ test("context assignments reject invented cross-thread segments and invalid conf }, secondAt), /between 0 and 1/); }); +test("context summaries commit with a source fence, supersede atomically, and discard stale results", async (t) => { + const cwd = await mkdtemp(join(tmpdir(), "ling-context-summary-")); + t.after(() => rm(cwd, { recursive: true, force: true })); + const store = createSqliteRunStore(cwd); + t.after(() => store.close()); + const channel = store.createChannel({ name: "Summary", boundPath: cwd }, firstAt); + const thread = store.resolveOrCreateConversationThread({ + surfaceKind: "channel_root", + channelId: channel.id, + title: "Summary", + objective: "Test summary fencing", + }, firstAt).thread; + const first = appendUserTurn(store, channel.id, thread.id, "summary first", firstAt); + const second = appendUserTurn(store, channel.id, thread.id, "summary second", secondAt); + const initialInput = summaryCommitInput(store, thread.id, first.contextSegmentId, [first, second]); + + const third = appendUserTurn(store, channel.id, thread.id, "summary third", thirdAt); + const initial = store.commitConversationContextSummary(initialInput, thirdAt); + assert.equal(initial.status, "committed"); + assert.equal(initial.summary.version, 1); + assert.equal(initial.summary.coveredThroughSequence, second.item.sequence); + + const replacementInput = summaryCommitInput( + store, + thread.id, + first.contextSegmentId, + [first, second, third], + initial.summary.id, + ); + const replacement = store.commitConversationContextSummary( + replacementInput, + "2026-08-10T01:03:00.000Z", + ); + assert.equal(replacement.status, "committed"); + assert.equal(replacement.summary.version, 2); + assert.equal(replacement.summary.supersedesSummaryId, initial.summary.id); + assert.equal(store.getConversationContextSummary(initial.summary.id)?.status, "superseded"); + assert.deepEqual( + store.listConversationContextSummaries({ threadId: thread.id }).map((summary) => summary.id), + [replacement.summary.id], + ); + + const duplicateLateResult = store.commitConversationContextSummary( + replacementInput, + "2026-08-10T01:04:00.000Z", + ); + assert.deepEqual(duplicateLateResult, { + status: "discarded_stale", + reason: "active_summary_changed", + }); +}); + +test("context summary assignment races are discarded and snapshots remain source-linked", async (t) => { + const cwd = await mkdtemp(join(tmpdir(), "ling-context-snapshot-")); + t.after(() => rm(cwd, { recursive: true, force: true })); + const store = createSqliteRunStore(cwd); + t.after(() => store.close()); + const channel = store.createChannel({ name: "Snapshot", boundPath: cwd }, firstAt); + const thread = store.resolveOrCreateConversationThread({ + surfaceKind: "channel_root", + channelId: channel.id, + title: "Snapshot", + objective: "Test snapshot reconstruction", + }, firstAt).thread; + const first = appendUserTurn(store, channel.id, thread.id, "snapshot first", firstAt); + const second = appendUserTurn(store, channel.id, thread.id, "snapshot second", secondAt); + const staleInput = summaryCommitInput(store, thread.id, first.contextSegmentId, [first, second]); + + store.applyTurnContextAssignment({ + turnId: first.turnId, + target: { kind: "existing", segmentId: first.contextSegmentId }, + status: "committed", + decisionSource: "user", + confidence: 1, + evidenceRefs: [{ type: "conversation_item", id: first.item.id }], + policyVersion, + }, thirdAt); + assert.deepEqual( + store.commitConversationContextSummary(staleInput, "2026-08-10T01:03:00.000Z"), + { status: "discarded_stale", reason: "assignments_changed" }, + ); + + const snapshot = store.createConversationContextSnapshot({ + id: "snapshot-fork-boundary", + threadId: thread.id, + segmentId: first.contextSegmentId, + projectionVersion: "ctxv2_snapshot_fixture", + throughItemSequence: second.item.sequence, + summaryIds: [], + sourceRefs: [ + { type: "thread", id: thread.id }, + { type: "conversation_item", id: first.item.id }, + { type: "conversation_item", id: second.item.id }, + ], + policyVersion: "continuity-v2-snapshot-v1", + }, "2026-08-10T01:04:00.000Z"); + assert.equal(store.getConversationContextSnapshot(snapshot.id)?.projectionVersion, "ctxv2_snapshot_fixture"); + assert.deepEqual( + store.listConversationContextSnapshots({ threadId: thread.id }).map((candidate) => candidate.id), + [snapshot.id], + ); + + appendUserTurn(store, channel.id, thread.id, "parent changed later", "2026-08-10T01:05:00.000Z"); + assert.equal(store.getConversationContextSnapshot(snapshot.id)?.throughItemSequence, second.item.sequence); + assert.deepEqual(store.getConversationContextSnapshot(snapshot.id)?.sourceRefs, snapshot.sourceRefs); +}); + function appendUserTurn(store, channelId, threadId, body, now) { return store.appendConversationUserTurn({ channelId, @@ -154,6 +261,44 @@ function appendUserTurn(store, channelId, threadId, body, now) { }, now); } +function summaryCommitInput(store, threadId, segmentId, turns, expectedActiveSummaryId = null) { + return { + threadId, + segmentId, + scope: "segment", + coveredFromSequence: turns[0].item.sequence, + coveredThroughSequence: turns.at(-1).item.sequence, + objective: "Preserve the current objective.", + decisions: ["Use fenced summaries."], + constraints: ["Do not overwrite a newer summary."], + openQuestions: [], + activeWorkRefs: [], + sourceRefs: [ + ...(expectedActiveSummaryId ? [{ type: "context_summary", id: expectedActiveSummaryId }] : []), + ...turns.map((turn) => ({ type: "conversation_item", id: turn.item.id })), + ], + policyVersion: "continuity-v2-summary-v1", + projectionVersion: `ctxv2_fixture_${turns.at(-1).item.sequence}`, + visibilityFingerprint: "visibility_fixture_channel", + expectedActiveSummaryId, + sourceFence: turns.map((turn) => { + const assignment = store.getEffectiveTurnContextAssignment(turn.turnId); + assert.ok(assignment); + return { + itemId: turn.item.id, + sequence: turn.item.sequence, + revision: turn.item.revision, + updatedAt: turn.item.updatedAt, + visibility: turn.item.visibility, + visibilityOwnerId: turn.item.visibilityOwnerId, + turnId: turn.turnId, + assignmentId: assignment.id, + assignmentRevision: assignment.revision, + }; + }), + }; +} + function insertSegmentSummary(cwd, input) { const database = new DatabaseSync(resolveRuntimePaths(cwd).databasePath); try { diff --git a/packages/core/test/conversation-thread-topology.test.mjs b/packages/core/test/conversation-thread-topology.test.mjs index 2c6fa539e..3821a2f6a 100644 --- a/packages/core/test/conversation-thread-topology.test.mjs +++ b/packages/core/test/conversation-thread-topology.test.mjs @@ -61,12 +61,24 @@ test("Conversation topology creates branch lineage atomically and rejects cross- parentTurnId: parent.turnId, title: "Branch", objective: "Explore a branch", + forkContextSnapshot: { + id: "snapshot-branch-runtime", + segmentId: parent.contextSegmentId, + projectionVersion: "ctxv2_branch_runtime", + throughItemSequence: parent.item.sequence, + summaryIds: [], + sourceRefs: [{ type: "conversation_item", id: parent.item.id }], + policyVersion: "continuity-v2-snapshot-v1", + }, }; const branch = fixture.store.resolveOrCreateConversationThread(branchInput, now); assert.equal(branch.created, true); assert.equal(branch.thread.relationKind, "branch"); assert.equal(branch.thread.parentThreadId, fixture.rootThreadId); assert.equal(branch.thread.parentTurnId, parent.turnId); + assert.equal(branch.thread.forkContextSnapshotId, "snapshot-branch-runtime"); + assert.equal(fixture.store.getConversationContextSnapshot("snapshot-branch-runtime")?.threadId, fixture.rootThreadId); + assert.equal(fixture.database.prepare("SELECT COUNT(*) AS count FROM conversation_items WHERE thread_id = ?").get(branch.thread.id).count, 0); const replay = fixture.store.resolveOrCreateConversationThread(branchInput, "2026-08-09T15:01:00.000Z"); assert.equal(replay.created, false); diff --git a/packages/daemon/src/features/channels/ai-replies.ts b/packages/daemon/src/features/channels/ai-replies.ts index f4c9e114f..26ca23bb3 100644 --- a/packages/daemon/src/features/channels/ai-replies.ts +++ b/packages/daemon/src/features/channels/ai-replies.ts @@ -307,10 +307,11 @@ async function completeAiChannelReply(input: { error: firstRunFailure, now: observedAt, }); - const compaction = compactChannelDiscussionForNoToolsRetry({ + const compaction = await compactChannelDiscussionForNoToolsRetry({ store: input.store, channel: input.channel, sourceMessage: input.sourceMessage, + inputBudgetTokens: discussionContextPack.budget.inputBudgetTokens, now: observedAt, }); discussionContextPack = buildChannelReplyContextPack({ diff --git a/packages/daemon/src/features/channels/channel-intent-selector.ts b/packages/daemon/src/features/channels/channel-intent-selector.ts index aa69b31e2..fd621e8d0 100644 --- a/packages/daemon/src/features/channels/channel-intent-selector.ts +++ b/packages/daemon/src/features/channels/channel-intent-selector.ts @@ -543,10 +543,11 @@ async function requestModelIntent(input: ChannelIntentSelectorInput): Promise pattern.test(message)); } -export function compactChannelDiscussionForNoToolsRetry(input: { +export async function compactChannelDiscussionForNoToolsRetry(input: { readonly store: RunStore; readonly channel: Channel; readonly sourceMessage: ChannelMessage; + readonly inputBudgetTokens: number; readonly now: string; -}): ChannelDiscussionCompactionResult { - const sourceMessages = selectChannelSummaryMessages(input.store, input.channel, input.sourceMessage); - const summaryText = renderChannelDiscussionSummary(input.channel, input.sourceMessage, sourceMessages); - const summary = input.store.createChannelContextSummary( - { - channelId: input.channel.id, - targetType: "thread", - targetId: input.sourceMessage.threadId ?? input.sourceMessage.id, - sourceMessageIds: sourceMessages.map((message) => message.id), - summary: summaryText, - acceptanceCriteria: [], - openQuestions: [], - authorType: "system", +}): Promise { + const result = await coordinateConversationContextSummary({ + store: input.store, + sourceItemId: input.sourceMessage.id, + inputBudgetTokens: input.inputBudgetTokens, + policy: { + softCompletedTurnCount: 2, + softInputBudgetRatio: 0.01, + hardInputBudgetRatio: 0.02, + targetInputBudgetRatio: 0.1, + rawTailCompletedTurnCount: 1, + generationChunkBudgetRatio: 0.1, }, - input.now, - ); + generate: emergencySummaryDraft, + now: input.now, + }); + const summary = result.status === "completed" && result.commit.status === "committed" + ? result.commit.summary + : selectActiveSegmentSummary(input.store, input.sourceMessage.id); + const sourceMessages = selectChannelSummaryMessages(input.store, input.channel, input.sourceMessage); + const summaryText = summary + ? renderCanonicalConversationSummary(summary) + : renderChannelDiscussionSummary(input.channel, input.sourceMessage, sourceMessages); + const sourceMessageIds = summary + ? summary.sourceRefs + .filter((ref) => ref.type === "conversation_item") + .map((ref) => ref.id) + : sourceMessages.map((message) => message.id); return { - summary, + ...(summary ? { summary } : {}), compactedSummary: { - id: summary.id, + id: summary?.id ?? `context_retry_${input.sourceMessage.id}`, text: [ "Compacted channel discussion summary for a no-tools retry:", - summary.summary, + summaryText, "", "Use this summary instead of expanding older channel history. It excludes private teammate chat, hidden prompts, provider secrets, raw artifact bodies, and protocol mutations.", ].join("\n"), - sourceMessageIds: summary.sourceMessageIds, + sourceMessageIds, }, }; } +async function emergencySummaryDraft(input: { + readonly current: ConversationSummaryDraft; + readonly items: readonly import("@ling-stack/contracts").ConversationItemV1[]; +}): Promise { + const latestUserBody = [...input.items] + .reverse() + .find((item) => item.kind === "user_message")?.payload.body; + return { + objective: input.current.objective ?? (latestUserBody ? truncateSummary(latestUserBody, 1_200) : undefined), + decisions: input.current.decisions, + constraints: input.current.constraints, + openQuestions: input.current.openQuestions, + }; +} + +function selectActiveSegmentSummary( + store: RunStore, + sourceItemId: string, +): ConversationContextSummary | undefined { + const projection = buildConversationContextProjection({ store, sourceItemId, view: "reply" }); + return projection.summaries.find((summary) => + summary.scope === "segment" && summary.segmentId === projection.metadata.segmentId); +} + +function renderCanonicalConversationSummary(summary: ConversationContextSummary): string { + return [ + summary.objective ? `Objective: ${summary.objective}` : undefined, + summary.decisions.length > 0 ? `Decisions: ${summary.decisions.join(" | ")}` : undefined, + summary.constraints.length > 0 ? `Constraints: ${summary.constraints.join(" | ")}` : undefined, + summary.openQuestions.length > 0 ? `Open questions: ${summary.openQuestions.join(" | ")}` : undefined, + `Coverage: ${summary.coveredFromSequence}-${summary.coveredThroughSequence}`, + `Source refs: ${summary.sourceRefs.length}`, + ].filter((line): line is string => Boolean(line)).join("\n"); +} + export function maybeCompactWorkflowStepContext(input: { readonly packInput: WorkflowStepContextPackInput; readonly contextPack: ContextPack; diff --git a/packages/daemon/src/features/channels/context-processing/conversation-context-projector.ts b/packages/daemon/src/features/channels/context-processing/conversation-context-projector.ts index 4e62d1f80..68f73f330 100644 --- a/packages/daemon/src/features/channels/context-processing/conversation-context-projector.ts +++ b/packages/daemon/src/features/channels/context-processing/conversation-context-projector.ts @@ -1,6 +1,7 @@ import { createHash } from "node:crypto"; import type { ContextSegment, + ConversationContextSnapshot, ConversationContextProjectionMetadata, ConversationContextSummary, ConversationDomainRef, @@ -37,6 +38,7 @@ export interface ConversationContextProjection { readonly metadata: ConversationContextProjectionMetadata; readonly segment: ContextSegment; readonly assignment: TurnContextAssignment; + readonly forkSnapshot?: ConversationContextSnapshot; readonly summaries: readonly ConversationContextSummary[]; readonly items: readonly ConversationItemV1[]; readonly sharedCore: { @@ -59,6 +61,12 @@ export function buildConversationContextProjection( if (!thread || thread.archivedAt || thread.status === "archived") { throw new Error(`Conversation projection thread is not active: ${sourceItem.threadId}`); } + const forkSnapshot = thread.forkContextSnapshotId + ? input.store.getConversationContextSnapshot(thread.forkContextSnapshotId) + : undefined; + if (thread.forkContextSnapshotId && !forkSnapshot) { + throw new Error(`Conversation projection fork snapshot is missing: ${thread.forkContextSnapshotId}`); + } const assignment = input.store.getEffectiveTurnContextAssignment(sourceItem.turnId); if (!assignment) { throw new Error(`Conversation projection turn has no context assignment: ${sourceItem.turnId}`); @@ -68,13 +76,24 @@ export function buildConversationContextProjection( throw new Error(`Conversation projection assignment references an unavailable segment: ${assignment.segmentId}`); } - const summaries = effectiveProjectionSummaries(input.store, sourceItem, segment.id); + const inheritedSummaries = forkSnapshot + ? forkSnapshot.summaryIds.map((summaryId) => { + const summary = input.store.getConversationContextSummary(summaryId); + if (!summary) throw new Error(`Conversation projection fork summary is missing: ${summaryId}`); + return { ...summary, status: "active" as const }; + }) + : []; + const summaries = uniqueSummaries([ + ...inheritedSummaries, + ...effectiveProjectionSummaries(input.store, sourceItem, segment.id), + ]); assertProjectionSummarySourceRefsVisible({ store: input.store, sourceItem, summaries, view: input.view, consumerTeammateId: input.consumerTeammateId, + inheritedSummaryIds: new Set(inheritedSummaries.map((summary) => summary.id)), }); const coveredThroughSequence = summaries .filter((summary) => summary.segmentId === segment.id && summary.scope === "segment") @@ -111,8 +130,22 @@ export function buildConversationContextProjection( return item; }); const explicitTurnItemIds = new Set(explicitTurnItems.map((item) => item.id)); + const inheritedItems = forkSnapshot + ? forkSnapshot.sourceRefs + .filter((ref) => ref.type === "conversation_item" || ref.type === "message") + .map((ref) => { + const item = input.store.getConversationItem(ref.id); + if (!item || item.status !== "committed") { + throw new Error(`Conversation projection fork item is missing: ${ref.id}`); + } + if (!canProjectItem(item, input.view, input.consumerTeammateId)) { + throw new Error(`Conversation projection fork item is not visible to this consumer: ${ref.id}`); + } + return item; + }) + : []; const historicalGuidanceItemIds = historicalRunGuidanceItemIds(input.store); - const visibleItems = uniqueItems([...tail, ...explicitItems, ...explicitTurnItems, sourceItem]) + const visibleItems = uniqueItems([...inheritedItems, ...tail, ...explicitItems, ...explicitTurnItems, sourceItem]) .filter((item) => item.id === sourceItem.id || !historicalGuidanceItemIds.has(item.id)) .filter((item) => canProjectItem(item, input.view, input.consumerTeammateId)) .filter((item) => @@ -126,12 +159,14 @@ export function buildConversationContextProjection( ); const visibilityFingerprint = fingerprint({ scope: projectionVisibilityScope(input.view, input.consumerTeammateId), + forkSnapshotId: forkSnapshot?.id ?? null, visibleItemScopes: items.map((item) => [item.id, item.visibility, item.visibilityOwnerId ?? null]), }); const sourceRefs = uniqueDomainRefs([ { type: "thread", id: thread.id }, { type: "turn", id: sourceItem.turnId }, { type: "conversation_item", id: sourceItem.id }, + ...(forkSnapshot ? [{ type: "context_snapshot" as const, id: forkSnapshot.id }] : []), ...assignment.evidenceRefs, ...summaries.flatMap((summary) => [ { type: "context_summary" as const, id: summary.id }, @@ -147,11 +182,16 @@ export function buildConversationContextProjection( surfaceKind: thread.surfaceKind, relationKind: thread.relationKind, anchorItemId: thread.anchorItemId ?? null, + forkContextSnapshotId: thread.forkContextSnapshotId ?? null, }, turnId: sourceItem.turnId, segment: { id: segment.id, revision: segment.revision }, assignment: { id: assignment.id, revision: assignment.revision }, - summaries: summaries.map((summary) => [summary.id, summary.version, summary.status]), + summaries: summaries.map((summary) => [ + summary.id, + summary.version, + inheritedSummaries.some((inherited) => inherited.id === summary.id) ? "snapshot" : summary.status, + ]), throughItemSequence: sourceItem.sequence, visibilityFingerprint, }); @@ -169,6 +209,7 @@ export function buildConversationContextProjection( }, segment, assignment, + ...(forkSnapshot ? { forkSnapshot } : {}), summaries, items, sharedCore: { @@ -231,10 +272,14 @@ function assertProjectionSummarySourceRefsVisible(input: { readonly summaries: readonly ConversationContextSummary[]; readonly view: ConversationContextProjectionView; readonly consumerTeammateId?: string; + readonly inheritedSummaryIds: ReadonlySet; }): void { - for (const ref of input.summaries.flatMap((summary) => summary.sourceRefs)) { + for (const summary of input.summaries) { + for (const ref of summary.sourceRefs) { if (ref.type === "conversation_item" || ref.type === "message") { - const item = input.store.getEffectiveConversationItem(ref.id); + const item = input.inheritedSummaryIds.has(summary.id) + ? input.store.getConversationItem(ref.id) + : input.store.getEffectiveConversationItem(ref.id); if (!item || item.status !== "committed") { throw new Error(`Conversation projection summary source ref is missing or stale: ${ref.type}:${ref.id}`); } @@ -265,6 +310,7 @@ function assertProjectionSummarySourceRefsVisible(input: { if (memory.scopeType === "channel" && memory.ownerId !== input.sourceItem.channelId) { throw new Error(`Conversation projection summary memory ref belongs to another channel: ${ref.id}`); } + } } } @@ -342,13 +388,21 @@ function trimProjectionItemsByTextBudget( const safeMax = Math.min(Math.max(Math.floor(maxTextChars), 1_024), 256_000); const selected: ConversationItemV1[] = []; let used = 0; - for (const item of [...items].sort((left, right) => right.sequence - left.sequence)) { + const sourceThreadId = items.find((item) => item.id === sourceItemId)?.threadId; + const ordered = [...items].sort((left, right) => { + const leftCurrent = left.threadId === sourceThreadId ? 1 : 0; + const rightCurrent = right.threadId === sourceThreadId ? 1 : 0; + return leftCurrent - rightCurrent + || left.sequence - right.sequence + || left.id.localeCompare(right.id); + }); + for (const item of [...ordered].reverse()) { const size = projectionItemText(item).length; if (item.id !== sourceItemId && selected.length > 0 && used + size > safeMax) continue; selected.push(item); used += size; } - return selected.sort((left, right) => left.sequence - right.sequence || left.id.localeCompare(right.id)); + return selected.reverse(); } function projectionItemText(item: ConversationItemV1): string { @@ -381,6 +435,14 @@ function uniqueItems(items: readonly ConversationItemV1[]): ConversationItemV1[] return [...byId.values()]; } +function uniqueSummaries( + summaries: readonly ConversationContextSummary[], +): ConversationContextSummary[] { + const byId = new Map(); + for (const summary of summaries) byId.set(summary.id, summary); + return [...byId.values()]; +} + function uniqueDomainRefs(refs: readonly ConversationDomainRef[]): ConversationDomainRef[] { const byKey = new Map(); for (const ref of refs) byKey.set(`${ref.type}:${ref.id}`, ref); diff --git a/packages/daemon/src/features/channels/context-processing/conversation-summary-coordinator.ts b/packages/daemon/src/features/channels/context-processing/conversation-summary-coordinator.ts new file mode 100644 index 000000000..25cbb70e9 --- /dev/null +++ b/packages/daemon/src/features/channels/context-processing/conversation-summary-coordinator.ts @@ -0,0 +1,446 @@ +import type { + CommitConversationContextSummaryResult, + ConversationContextSummarySourceFence, + RunStore, +} from "@ling-stack/core"; +import type { + ConversationContextSummary, + ConversationDomainRef, + ConversationItemV1, +} from "@ling-stack/contracts"; +import { estimateTextTokens } from "../context-packs.js"; +import { + buildConversationContextProjection, + type ConversationContextProjection, +} from "./conversation-context-projector.js"; + +export const CONVERSATION_SUMMARY_POLICY_VERSION = "continuity-v2-summary-v1"; + +export interface ConversationSummaryPolicy { + readonly softCompletedTurnCount: number; + readonly softInputBudgetRatio: number; + readonly hardInputBudgetRatio: number; + readonly targetInputBudgetRatio: number; + readonly rawTailCompletedTurnCount: number; + readonly generationChunkBudgetRatio: number; +} + +export const DEFAULT_CONVERSATION_SUMMARY_POLICY: ConversationSummaryPolicy = { + softCompletedTurnCount: 32, + softInputBudgetRatio: 0.3, + hardInputBudgetRatio: 0.7, + targetInputBudgetRatio: 0.55, + rawTailCompletedTurnCount: 8, + generationChunkBudgetRatio: 0.25, +}; + +export interface ConversationSummaryDraft { + readonly objective?: string; + readonly decisions: readonly string[]; + readonly constraints: readonly string[]; + readonly openQuestions: readonly string[]; +} + +export interface GenerateConversationSummaryInput { + readonly current: ConversationSummaryDraft; + readonly items: readonly ConversationItemV1[]; + readonly sourceRefs: readonly ConversationDomainRef[]; + readonly chunkIndex: number; + readonly chunkCount: number; + readonly chunkTokenBudget: number; + readonly outputTokenBudget: number; +} + +export type GenerateConversationSummary = ( + input: GenerateConversationSummaryInput, +) => Promise; + +export interface ConversationSummaryPlan { + readonly trigger: "soft" | "hard"; + readonly projection: ConversationContextProjection; + readonly activeSummary?: ConversationContextSummary; + readonly sourceItems: readonly ConversationItemV1[]; + readonly sourceFence: readonly ConversationContextSummarySourceFence[]; + readonly sourceRefs: readonly ConversationDomainRef[]; + readonly activeWorkRefs: readonly ConversationDomainRef[]; + readonly coveredFromSequence: number; + readonly coveredThroughSequence: number; + readonly inputBudgetTokens: number; + readonly estimatedRawTokens: number; + readonly estimatedRawTailTokens: number; + readonly estimatedProjectionTokens: number; + readonly targetExceededByRequiredTail: boolean; + readonly chunkTokenBudget: number; + readonly summaryTokenBudget: number; +} + +export type CoordinateConversationSummaryResult = + | { readonly status: "not_needed" } + | { readonly status: "generator_unavailable"; readonly plan: ConversationSummaryPlan } + | { readonly status: "generation_failed"; readonly plan: ConversationSummaryPlan; readonly error: unknown } + | { readonly status: "completed"; readonly plan: ConversationSummaryPlan; readonly commit: CommitConversationContextSummaryResult }; + +export function planConversationContextSummary(input: { + readonly store: RunStore; + readonly sourceItemId: string; + readonly inputBudgetTokens: number; + readonly policy?: ConversationSummaryPolicy; +}): ConversationSummaryPlan | undefined { + const policy = input.policy ?? DEFAULT_CONVERSATION_SUMMARY_POLICY; + assertValidPolicy(policy); + if (!Number.isSafeInteger(input.inputBudgetTokens) || input.inputBudgetTokens < 1_024) { + throw new RangeError(`Invalid conversation summary input budget: ${input.inputBudgetTokens}`); + } + const projection = buildConversationContextProjection({ + store: input.store, + sourceItemId: input.sourceItemId, + view: "reply", + maxTailTextChars: 256_000, + }); + const activeSummary = projection.summaries.find((summary) => + summary.scope === "segment" && summary.segmentId === projection.segment.id); + const unsummarizedItems = input.store.listConversationItemsForSegment({ + threadId: projection.metadata.threadId, + segmentId: projection.metadata.segmentId, + throughSequence: projection.metadata.throughItemSequence, + afterSequence: activeSummary?.coveredThroughSequence ?? 0, + order: "oldest", + limit: 500, + }).filter(isSharedSummaryItem); + const eligibleItems = stableCompletedPrefix(input.store, unsummarizedItems); + const completedTurnIds = completedTurnIdsInOrder(eligibleItems); + const estimatedRawTokens = estimateItemsTokens(eligibleItems); + const estimatedProjectionTokens = estimateProjectionTokens(projection); + const hard = estimatedProjectionTokens > input.inputBudgetTokens * policy.hardInputBudgetRatio; + const soft = completedTurnIds.length >= policy.softCompletedTurnCount + || estimatedRawTokens > input.inputBudgetTokens * policy.softInputBudgetRatio; + if (!hard && !soft) return undefined; + + const rawTailTurnIds = new Set(completedTurnIds.slice(-policy.rawTailCompletedTurnCount)); + const sourceItems = eligibleItems.filter((item) => item.turnId && !rawTailTurnIds.has(item.turnId)); + if (sourceItems.length === 0) return undefined; + const sourceFence = sourceItems.map((item) => buildSourceFence(input.store, item)); + const sourceRefs = uniqueDomainRefs([ + ...(activeSummary ? [{ type: "context_summary" as const, id: activeSummary.id }] : []), + ...sourceItems.map((item) => ({ type: "conversation_item" as const, id: item.id })), + ]); + const activeWorkRefs = uniqueDomainRefs([ + ...(activeSummary?.activeWorkRefs ?? []), + ...sourceItems.flatMap((item) => item.domainRefs.filter(isActiveWorkRef)), + ]); + const estimatedRawTailTokens = estimateItemsTokens( + eligibleItems.filter((item) => item.turnId && rawTailTurnIds.has(item.turnId)), + ); + const summaryTokenBudget = Math.max( + 128, + Math.floor(input.inputBudgetTokens * policy.targetInputBudgetRatio) - estimatedRawTailTokens, + ); + return { + trigger: hard ? "hard" : "soft", + projection, + ...(activeSummary ? { activeSummary } : {}), + sourceItems, + sourceFence, + sourceRefs, + activeWorkRefs, + coveredFromSequence: activeSummary?.coveredFromSequence ?? sourceItems[0]!.sequence, + coveredThroughSequence: sourceItems.at(-1)!.sequence, + inputBudgetTokens: input.inputBudgetTokens, + estimatedRawTokens, + estimatedRawTailTokens, + estimatedProjectionTokens, + targetExceededByRequiredTail: estimatedRawTailTokens > input.inputBudgetTokens * policy.targetInputBudgetRatio, + chunkTokenBudget: Math.max(1_024, Math.floor(input.inputBudgetTokens * policy.generationChunkBudgetRatio)), + summaryTokenBudget, + }; +} + +export async function coordinateConversationContextSummary(input: { + readonly store: RunStore; + readonly sourceItemId: string; + readonly inputBudgetTokens: number; + readonly generate?: GenerateConversationSummary; + readonly policy?: ConversationSummaryPolicy; + readonly now?: string; +}): Promise { + const plan = planConversationContextSummary(input); + if (!plan) return { status: "not_needed" }; + return executeConversationSummaryPlan({ + store: input.store, + plan, + generate: input.generate, + now: input.now, + }); +} + +export async function executeConversationSummaryPlan(input: { + readonly store: RunStore; + readonly plan: ConversationSummaryPlan; + readonly generate?: GenerateConversationSummary; + readonly now?: string; +}): Promise { + const { plan } = input; + if (!input.generate) return { status: "generator_unavailable", plan }; + let draft = normalizeDraft(draftFromSummary(plan.activeSummary), plan.summaryTokenBudget); + const chunks = chunkSummaryItems(plan.sourceItems, plan.chunkTokenBudget); + try { + for (let index = 0; index < chunks.length; index += 1) { + const chunk = chunks[index]!; + draft = normalizeDraft(await input.generate({ + current: draft, + items: chunk, + sourceRefs: uniqueDomainRefs([ + ...(plan.activeSummary + ? [{ type: "context_summary" as const, id: plan.activeSummary.id }] + : []), + ...chunk.map((item) => ({ type: "conversation_item" as const, id: item.id })), + ]), + chunkIndex: index, + chunkCount: chunks.length, + chunkTokenBudget: plan.chunkTokenBudget, + outputTokenBudget: plan.summaryTokenBudget, + }), plan.summaryTokenBudget); + } + } catch (error) { + return { status: "generation_failed", plan, error }; + } + const commit = input.store.commitConversationContextSummary({ + threadId: plan.projection.metadata.threadId, + segmentId: plan.projection.metadata.segmentId, + scope: "segment", + coveredFromSequence: plan.coveredFromSequence, + coveredThroughSequence: plan.coveredThroughSequence, + objective: draft.objective, + decisions: draft.decisions, + constraints: draft.constraints, + openQuestions: draft.openQuestions, + activeWorkRefs: plan.activeWorkRefs, + sourceRefs: plan.sourceRefs, + policyVersion: CONVERSATION_SUMMARY_POLICY_VERSION, + projectionVersion: plan.projection.metadata.projectionVersion, + visibilityFingerprint: plan.projection.metadata.visibilityFingerprint, + expectedActiveSummaryId: plan.activeSummary?.id ?? null, + sourceFence: plan.sourceFence, + }, input.now ?? new Date().toISOString()); + return { status: "completed", plan, commit }; +} + +function stableCompletedPrefix(store: RunStore, items: readonly ConversationItemV1[]): ConversationItemV1[] { + const stable: ConversationItemV1[] = []; + const terminalByTurnId = new Map(); + for (const item of items) { + if (!item.turnId) break; + let terminal = terminalByTurnId.get(item.turnId); + if (terminal === undefined) { + const turn = store.getLingTurn(item.turnId); + terminal = Boolean(turn && ( + turn.status === "succeeded" + || turn.status === "failed" + || turn.status === "cancelled" + || turn.status === "superseded" + )); + terminalByTurnId.set(item.turnId, terminal); + } + if (!terminal) break; + stable.push(item); + } + return stable; +} + +function completedTurnIdsInOrder(items: readonly ConversationItemV1[]): string[] { + const completed = new Set(); + for (const item of items) { + if (!item.turnId || completed.has(item.turnId)) continue; + completed.add(item.turnId); + } + return [...completed]; +} + +function buildSourceFence(store: RunStore, item: ConversationItemV1): ConversationContextSummarySourceFence { + if (!item.turnId) throw new Error(`Conversation summary item has no owning turn: ${item.id}`); + const assignment = store.getEffectiveTurnContextAssignment(item.turnId); + if (!assignment) throw new Error(`Conversation summary turn has no context assignment: ${item.turnId}`); + return { + itemId: item.id, + sequence: item.sequence, + revision: item.revision, + updatedAt: item.updatedAt, + visibility: item.visibility, + ...(item.visibilityOwnerId ? { visibilityOwnerId: item.visibilityOwnerId } : {}), + turnId: item.turnId, + assignmentId: assignment.id, + assignmentRevision: assignment.revision, + }; +} + +function chunkSummaryItems(items: readonly ConversationItemV1[], tokenBudget: number): ConversationItemV1[][] { + const chunks: ConversationItemV1[][] = []; + let current: ConversationItemV1[] = []; + let used = 0; + for (const item of items) { + const tokens = estimateTextTokens(renderSummaryItem(item)); + if (current.length > 0 && used + tokens > tokenBudget) { + chunks.push(current); + current = []; + used = 0; + } + current.push(item); + used += tokens; + } + if (current.length > 0) chunks.push(current); + return chunks; +} + +export function renderSummaryItem(item: ConversationItemV1, tokenBudget?: number): string { + const text = itemText(item).replace(/\s+/gu, " ").trim(); + const rendered = `[${item.sequence}] ${item.actorType} ${item.kind} ${item.id}: ${text}`; + return tokenBudget ? truncateTextToTokenBudget(rendered, tokenBudget) : rendered; +} + +function itemText(item: ConversationItemV1): string { + switch (item.kind) { + case "user_message": + case "assistant_message": + return item.payload.body; + case "tool_call": + return item.payload.inputSummary ?? item.payload.title ?? item.payload.toolName; + case "tool_result": + return item.payload.outputSummary ?? item.payload.errorCode ?? item.payload.status; + case "approval_request": + return item.payload.summary ?? item.payload.title; + case "approval_decision": + return item.payload.reason ?? item.payload.decision; + case "handoff": + return item.payload.summary; + case "operation_state": + return item.payload.summary ?? item.payload.title; + case "artifact_ref": + return item.payload.title ?? item.payload.artifactId; + case "system_event": + return item.payload.summary; + } +} + +function estimateItemsTokens(items: readonly ConversationItemV1[]): number { + return items.reduce((total, item) => total + estimateTextTokens(renderSummaryItem(item)), 0); +} + +function estimateProjectionTokens(projection: ConversationContextProjection): number { + return estimateItemsTokens(projection.items.filter(isSharedSummaryItem)) + + projection.summaries.reduce((total, summary) => total + estimateTextTokens(JSON.stringify({ + objective: summary.objective, + decisions: summary.decisions, + constraints: summary.constraints, + openQuestions: summary.openQuestions, + activeWorkRefs: summary.activeWorkRefs, + })), 0); +} + +function draftFromSummary(summary: ConversationContextSummary | undefined): ConversationSummaryDraft { + return { + ...(summary?.objective ? { objective: summary.objective } : {}), + decisions: summary?.decisions ?? [], + constraints: summary?.constraints ?? [], + openQuestions: summary?.openQuestions ?? [], + }; +} + +function normalizeDraft( + draft: ConversationSummaryDraft, + tokenBudget: number, +): ConversationSummaryDraft { + const normalized = { + ...(draft.objective?.trim() ? { objective: draft.objective.trim().slice(0, 2_000) } : {}), + decisions: normalizeStrings(draft.decisions, 100), + constraints: normalizeStrings(draft.constraints, 100), + openQuestions: normalizeStrings(draft.openQuestions, 100), + }; + const fieldBudget = Math.max(16, Math.floor(tokenBudget * 0.2)); + return { + ...(normalized.objective + ? { objective: truncateTextToTokenBudget(normalized.objective, fieldBudget) } + : {}), + decisions: fitStringsToTokenBudget(normalized.decisions, fieldBudget), + constraints: fitStringsToTokenBudget(normalized.constraints, fieldBudget), + openQuestions: fitStringsToTokenBudget(normalized.openQuestions, fieldBudget), + }; +} + +function normalizeStrings(values: readonly string[], limit: number): string[] { + return [...new Set(values.map((value) => value.trim()).filter(Boolean))] + .slice(0, limit) + .map((value) => value.slice(0, 2_000)); +} + +function fitStringsToTokenBudget(values: readonly string[], tokenBudget: number): string[] { + const fitted: string[] = []; + let remaining = tokenBudget; + for (const value of values) { + if (remaining <= 0) break; + const fittedValue = truncateTextToTokenBudget(value, remaining); + if (!fittedValue) break; + fitted.push(fittedValue); + remaining -= estimateTextTokens(fittedValue); + } + return fitted; +} + +function truncateTextToTokenBudget(value: string, tokenBudget: number): string { + if (tokenBudget < 1) return ""; + if (estimateTextTokens(value) <= tokenBudget) return value; + let low = 0; + let high = value.length; + while (low < high) { + const middle = Math.ceil((low + high) / 2); + if (estimateTextTokens(value.slice(0, middle)) <= tokenBudget) { + low = middle; + } else { + high = middle - 1; + } + } + return value.slice(0, low).trimEnd(); +} + +function isSharedSummaryItem(item: ConversationItemV1): boolean { + return item.status === "committed" + && Boolean(item.turnId) + && (item.visibility === "channel" || item.visibility === "thread_participants"); +} + +function isActiveWorkRef(ref: ConversationDomainRef): boolean { + return ref.type === "run" + || ref.type === "task" + || ref.type === "workflow" + || ref.type === "workflow_step" + || ref.type === "approval" + || ref.type === "artifact" + || ref.type === "handoff" + || ref.type === "operation" + || ref.type === "external_work" + || ref.type === "automation"; +} + +function uniqueDomainRefs(refs: readonly ConversationDomainRef[]): ConversationDomainRef[] { + const byKey = new Map(); + for (const ref of refs) byKey.set(`${ref.type}:${ref.id}`, ref); + return [...byKey.values()]; +} + +function assertValidPolicy(policy: ConversationSummaryPolicy): void { + if (!Number.isSafeInteger(policy.softCompletedTurnCount) || policy.softCompletedTurnCount < 1) { + throw new RangeError("Conversation summary soft turn count must be a positive integer."); + } + if (!Number.isSafeInteger(policy.rawTailCompletedTurnCount) || policy.rawTailCompletedTurnCount < 1) { + throw new RangeError("Conversation summary raw tail count must be a positive integer."); + } + for (const [name, value] of [ + ["softInputBudgetRatio", policy.softInputBudgetRatio], + ["hardInputBudgetRatio", policy.hardInputBudgetRatio], + ["targetInputBudgetRatio", policy.targetInputBudgetRatio], + ["generationChunkBudgetRatio", policy.generationChunkBudgetRatio], + ] as const) { + if (!Number.isFinite(value) || value <= 0 || value >= 1) { + throw new RangeError(`Conversation summary ${name} must be between 0 and 1.`); + } + } +} diff --git a/packages/daemon/src/features/channels/context-processing/conversation-summary-model.ts b/packages/daemon/src/features/channels/context-processing/conversation-summary-model.ts new file mode 100644 index 000000000..31e8fb47c --- /dev/null +++ b/packages/daemon/src/features/channels/context-processing/conversation-summary-model.ts @@ -0,0 +1,122 @@ +import type { Channel, RunStore, SecretResolver } from "@ling-stack/core"; +import type { RunnerRegistry } from "../../../runtime/runners.js"; +import { extractAssistantText } from "../../../shared/assistant-text.js"; +import { resolveProviderForRun } from "../../runs/providers.js"; +import { parseStrictJsonObject } from "../model-json.js"; +import { + renderSummaryItem, + type ConversationSummaryDraft, + type GenerateConversationSummary, +} from "./conversation-summary-coordinator.js"; + +export function createModelConversationSummaryGenerator(input: { + readonly store: RunStore; + readonly channel: Channel; + readonly runnerRegistry: RunnerRegistry | undefined; + readonly secretResolver: SecretResolver; + readonly requestedProviderPresetId?: string; +}): GenerateConversationSummary | undefined { + if (!input.runnerRegistry) return undefined; + const providerPresetId = resolveSummaryProviderPresetId(input); + if (!providerPresetId) return undefined; + return async (request) => { + const prompt = renderSummaryPrompt(request); + const taskId = `${request.items[0]?.id ?? "empty"}:summary:${request.chunkIndex}`; + const runner = input.runnerRegistry!.resolve({ tools: [], allowFallback: true }).runner; + const provider = resolveProviderForRun(input.store, { + cwd: input.store.paths.cwd, + workspaceCwd: input.store.paths.cwd, + providerPresetId, + tools: [], + thinkingLevel: "low", + task: { + id: taskId, + title: `Conversation summary in ${input.channel.name}`, + prompt, + channelId: input.channel.id, + }, + }, input.secretResolver); + const deltas: string[] = []; + const result = await runner.adapter.runTask({ + cwd: input.store.paths.cwd, + provider, + tools: [], + thinkingLevel: "low", + systemPrompt: [ + "You are Ling's conversation summary coordinator.", + "Return one strict JSON object only.", + "Preserve verified goals, decisions, constraints, and unresolved questions.", + "Do not add hidden reasoning, secrets, unsupported facts, or ids not supplied in the input.", + ].join(" "), + task: { + id: taskId, + title: `Conversation summary in ${input.channel.name}`, + prompt, + channelId: input.channel.id, + }, + onEvent: (event) => { + if (event.type === "assistant_text_delta") deltas.push(event.text); + }, + }); + const text = deltas.join("").trim() || extractAssistantText(result.messages) || ""; + return parseSummaryDraft(parseStrictJsonObject(text, "Conversation summary model output")); + }; +} + +function resolveSummaryProviderPresetId(input: { + readonly store: RunStore; + readonly channel: Channel; + readonly requestedProviderPresetId?: string; +}): string | undefined { + for (const providerPresetId of [...new Set([ + input.requestedProviderPresetId, + input.channel.defaultProviderPresetId, + ].filter((id): id is string => Boolean(id)))]) { + const preset = input.store.getProviderPreset(providerPresetId); + if (preset?.enabled && preset.capabilities.includes("chat")) return providerPresetId; + } + return undefined; +} + +function renderSummaryPrompt(input: Parameters[0]): string { + const perItemTokenBudget = Math.max( + 64, + Math.floor(input.chunkTokenBudget * 0.7 / Math.max(input.items.length, 1)), + ); + return [ + "Update the typed conversation summary with the supplied committed conversation items.", + "The current summary may already represent older source ranges. Preserve still-current facts and apply explicit corrections from newer items.", + `Keep the returned JSON within approximately ${input.outputTokenBudget} tokens. Order every array from most important to least important.`, + "Return exactly this JSON shape:", + '{"objective":"string or null","decisions":["..."],"constraints":["..."],"openQuestions":["..."]}', + `Chunk: ${input.chunkIndex + 1}/${input.chunkCount}`, + `Current typed summary JSON: ${JSON.stringify(input.current)}`, + `Allowed source refs JSON: ${JSON.stringify(input.sourceRefs)}`, + "Committed items:", + ...input.items.map((item) => renderSummaryItem(item, perItemTokenBudget)), + ].join("\n"); +} + +function parseSummaryDraft(value: Record): ConversationSummaryDraft { + const allowed = new Set(["objective", "decisions", "constraints", "openQuestions"]); + for (const key of Object.keys(value)) { + if (!allowed.has(key)) throw new Error(`Conversation summary model output contains unknown field: ${key}`); + } + const objective = value.objective; + if (objective !== undefined && objective !== null && typeof objective !== "string") { + throw new Error("Conversation summary objective must be a string or null."); + } + return { + ...(typeof objective === "string" && objective.trim() ? { objective: objective.trim() } : {}), + decisions: readStringArray(value.decisions, "decisions"), + constraints: readStringArray(value.constraints, "constraints"), + openQuestions: readStringArray(value.openQuestions, "openQuestions"), + }; +} + +function readStringArray(value: unknown, key: string): string[] { + if (!Array.isArray(value) || value.some((entry) => typeof entry !== "string")) { + throw new Error(`Conversation summary ${key} must be a string array.`); + } + return value as string[]; +} diff --git a/packages/daemon/src/features/channels/messages.ts b/packages/daemon/src/features/channels/messages.ts index 0b745e817..b1b7fdfd2 100644 --- a/packages/daemon/src/features/channels/messages.ts +++ b/packages/daemon/src/features/channels/messages.ts @@ -59,6 +59,17 @@ import { projectChannelMessageRevisionMetadata, } from "../conversations/supersession-projection.js"; import { projectConversationActiveReplyTarget } from "../conversations/active-reply-target.js"; +import { resolveChannelReplyExecutionTarget } from "./channel-reply-execution-target.js"; +import { resolveProviderContextBudget } from "./context-packs.js"; +import { createModelConversationSummaryGenerator } from "./context-processing/conversation-summary-model.js"; +import { + executeConversationSummaryPlan, + planConversationContextSummary, + type ConversationSummaryPlan, + type GenerateConversationSummary, +} from "./context-processing/conversation-summary-coordinator.js"; + +const inFlightConversationSummaries = new Map>(); export type StartChannelExecutionProposalWorkflow = (input: { readonly store: RunStore; @@ -463,6 +474,28 @@ export async function handleLingAgentChannelMessage(input: LingAgentChannelMessa sourceMessage: input.sourceMessage, requestedTeammateId: input.requestedTeammateId, }); + const summaryProviderPresetId = resolveChannelReplyExecutionTarget({ + store: input.store, + channel: input.channel, + requestedTeammateId: targetTeammateId, + requestedProviderPresetId: input.requestedProviderPresetId, + }).providerPresetId; + let summaryPlan = prepareConversationSummaryPlan(input, summaryProviderPresetId); + const summaryGenerator = createModelConversationSummaryGenerator({ + store: input.store, + channel: input.channel, + runnerRegistry: input.routeDependencies.runnerRegistry, + secretResolver: input.routeDependencies.secretResolver, + requestedProviderPresetId: summaryProviderPresetId, + }); + if (summaryPlan?.trigger === "hard") { + const summaryKey = conversationSummaryPlanKey(input.store, summaryPlan); + await inFlightConversationSummaries.get(summaryKey); + summaryPlan = prepareConversationSummaryPlan(input, summaryProviderPresetId); + if (summaryPlan?.trigger === "hard") { + await executeConversationSummaryPlanOnce(input.store, summaryPlan, summaryGenerator); + } + } const lane = await acquireChannelReplyLane({ coordinator: input.replyDispatchCoordinator, store: input.store, @@ -492,6 +525,50 @@ export async function handleLingAgentChannelMessage(input: LingAgentChannelMessa ); } finally { trackedReply.releaseIfUnused(); + if (summaryPlan?.trigger === "soft") { + void executeConversationSummaryPlanOnce(input.store, summaryPlan, summaryGenerator); + } + } +} + +function executeConversationSummaryPlanOnce( + store: RunStore, + plan: ConversationSummaryPlan, + generate: GenerateConversationSummary | undefined, +): Promise { + const key = conversationSummaryPlanKey(store, plan); + const existing = inFlightConversationSummaries.get(key); + if (existing) return existing; + const execution = executeConversationSummaryPlan({ store, plan, generate }) + .then(() => undefined) + .catch(() => undefined) + .finally(() => { + if (inFlightConversationSummaries.get(key) === execution) { + inFlightConversationSummaries.delete(key); + } + }); + inFlightConversationSummaries.set(key, execution); + return execution; +} + +function conversationSummaryPlanKey(store: RunStore, plan: ConversationSummaryPlan): string { + return `${store.paths.databasePath}:${plan.projection.metadata.threadId}:${plan.projection.metadata.segmentId}`; +} + +function prepareConversationSummaryPlan( + input: LingAgentChannelMessageInput, + providerPresetId: string | undefined, +): ConversationSummaryPlan | undefined { + try { + const providerPreset = providerPresetId ? input.store.getProviderPreset(providerPresetId) : undefined; + const budget = resolveProviderContextBudget(providerPreset, "channel_reply"); + return planConversationContextSummary({ + store: input.store, + sourceItemId: input.sourceMessage.id, + inputBudgetTokens: budget.inputBudgetTokens, + }); + } catch { + return undefined; } } diff --git a/packages/daemon/src/features/runs/thread-routes.ts b/packages/daemon/src/features/runs/thread-routes.ts index 518607296..e0219512a 100644 --- a/packages/daemon/src/features/runs/thread-routes.ts +++ b/packages/daemon/src/features/runs/thread-routes.ts @@ -41,6 +41,7 @@ import { readOptionalIntegerQuery, readOptionalJsonBody } from "../../shared/req import { redactTextWithSecrets } from "../../shared/redaction.js"; import { writeJson } from "../../shared/response.js"; import { isRecord, readOptionalBoolean, readOptionalString, readString } from "../../shared/validation.js"; +import { buildConversationContextProjection } from "../channels/context-processing/conversation-context-projector.js"; import { listExecutionActions } from "./execution-actions.js"; import { hydrateProposedPlanSourceState, hydrateProposedPlanSourceStates } from "./proposed-plan-source-state.js"; import { handleLingThreadHardDelete } from "./thread-hard-delete-route.js"; @@ -970,25 +971,38 @@ function createLingThreadFork(input: { const requestedTitle = isRecord(body) ? readOptionalString(body, "title")?.trim() : undefined; const requestedObjective = isRecord(body) ? readOptionalString(body, "objective")?.trim() : undefined; const sourceLabel = anchor.sourceTurn?.title || sourceThread.title; - const childThread = store.createLingThread( - { - id: childThreadId, - kind: sourceThread.kind, - channelId: sourceThread.channelId, - taskId: sourceThread.taskId, - teammateId: sourceThread.teammateId, - workflowId: sourceThread.workflowId, - workflowStepId: sourceThread.workflowStepId, - externalWorkId: sourceThread.externalWorkId, - title: requestedTitle || `Fork: ${sourceThread.title}`, - objective: requestedObjective || `Forked from ${sourceLabel} (${sourceThread.id}) to explore a separate work branch.`, - status: "open", - parentThreadId: sourceThread.id, - parentTurnId: anchor.sourceTurn?.id, - forkedFromRunId: anchor.sourceRun?.id, - }, - now, - ); + const title = requestedTitle || `Fork: ${sourceThread.title}`; + const objective = requestedObjective || `Forked from ${sourceLabel} (${sourceThread.id}) to explore a separate work branch.`; + const childThread = sourceThread.kind === "channel_discussion" && sourceThread.channelId + ? createConversationThreadFork({ + store, + sourceThread, + sourceTurn: anchor.sourceTurn, + sourceRun: anchor.sourceRun, + childThreadId, + title, + objective, + now, + }) + : store.createLingThread( + { + id: childThreadId, + kind: sourceThread.kind, + channelId: sourceThread.channelId, + taskId: sourceThread.taskId, + teammateId: sourceThread.teammateId, + workflowId: sourceThread.workflowId, + workflowStepId: sourceThread.workflowStepId, + externalWorkId: sourceThread.externalWorkId, + title, + objective, + status: "open", + parentThreadId: sourceThread.id, + parentTurnId: anchor.sourceTurn?.id, + forkedFromRunId: anchor.sourceRun?.id, + }, + now, + ); return { sourceTurn: anchor.sourceTurn, sourceRun: anchor.sourceRun, @@ -997,6 +1011,65 @@ function createLingThreadFork(input: { }; } +function createConversationThreadFork(input: { + readonly store: RunStore; + readonly sourceThread: LingThread; + readonly sourceTurn?: LingTurn; + readonly sourceRun?: RunSession; + readonly childThreadId?: string; + readonly title: string; + readonly objective: string; + readonly now: string; +}): LingThread { + const sourceItemId = input.sourceTurn?.sourceItemId; + if (!input.sourceTurn || !sourceItemId || !input.sourceThread.channelId) { + throw new HttpError( + 409, + "fork_context_unavailable", + `Conversation thread fork requires a source turn with a committed item: ${input.sourceThread.id}`, + ); + } + const projection = buildConversationContextProjection({ + store: input.store, + sourceItemId, + view: "reply", + }); + return input.store.resolveOrCreateConversationThread({ + ...(input.childThreadId ? { id: input.childThreadId } : {}), + surfaceKind: "branch", + channelId: input.sourceThread.channelId, + parentThreadId: input.sourceThread.id, + parentTurnId: input.sourceTurn.id, + ...(input.sourceRun ? { forkedFromRunId: input.sourceRun.id } : {}), + title: input.title, + objective: input.objective, + forkContextSnapshot: { + ...(input.childThreadId ? { id: `context_snapshot_${input.childThreadId}` } : {}), + segmentId: projection.metadata.segmentId, + projectionVersion: projection.metadata.projectionVersion, + throughItemSequence: projection.metadata.throughItemSequence, + summaryIds: projection.metadata.summaryIds, + sourceRefs: uniqueConversationForkSourceRefs([ + { type: "thread", id: input.sourceThread.id }, + { type: "turn", id: input.sourceTurn.id }, + { type: "conversation_item", id: sourceItemId }, + ...projection.metadata.summaryIds.map((id) => ({ type: "context_summary" as const, id })), + ...projection.items.map((item) => ({ type: "conversation_item" as const, id: item.id })), + ...projection.sharedCore.activeWorkRefs, + ]), + policyVersion: projection.metadata.policyVersion, + }, + }, input.now).thread; +} + +function uniqueConversationForkSourceRefs( + refs: readonly import("@ling-stack/contracts").ConversationDomainRef[], +): readonly import("@ling-stack/contracts").ConversationDomainRef[] { + const byKey = new Map(); + for (const ref of refs) byKey.set(`${ref.type}:${ref.id}`, ref); + return [...byKey.values()]; +} + function createLingThreadForkRunRequest(input: { readonly store: RunStore; readonly sourceThread: LingThread; diff --git a/packages/daemon/test/channel-reply-dispatch.test.mjs b/packages/daemon/test/channel-reply-dispatch.test.mjs index 4c39d64c5..47b79e95c 100644 --- a/packages/daemon/test/channel-reply-dispatch.test.mjs +++ b/packages/daemon/test/channel-reply-dispatch.test.mjs @@ -57,21 +57,30 @@ test("active channel replies are scoped to their conversation line", async () => title: "Reply lanes", objective: "Verify active reply scoping", }, "2026-08-06T00:00:05.000Z").thread; - const rootTurn = store.createLingTurn({ - threadId: rootThread.id, - kind: "follow_up", - title: "Branch parent", - userInstruction: "Open an independent branch", - channelId: channel.id, - }, "2026-08-06T00:00:06.000Z"); + const rootMessage = createMessage(store, channel.id, [], { + body: "Open an independent branch", + }); + const rootItem = store.getConversationItem(rootMessage.id); + assert.ok(rootItem?.turnId); + const rootAssignment = store.getEffectiveTurnContextAssignment(rootItem.turnId); + assert.ok(rootAssignment); const branchThread = store.resolveOrCreateConversationThread({ id: "thread_branch", surfaceKind: "branch", channelId: channel.id, parentThreadId: rootThread.id, - parentTurnId: rootTurn.id, + parentTurnId: rootItem.turnId, title: "Independent branch", objective: "Verify active reply scoping", + forkContextSnapshot: { + id: "snapshot_thread_branch", + segmentId: rootAssignment.segmentId, + projectionVersion: "ctxv2_thread_branch", + throughItemSequence: rootItem.sequence, + summaryIds: [], + sourceRefs: [{ type: "conversation_item", id: rootItem.id }], + policyVersion: "continuity-v2-snapshot-v1", + }, }, "2026-08-06T00:00:07.000Z").thread; const branchMessage = createMessage(store, channel.id, [], { threadId: branchThread.id }); assert.equal(findActiveChannelReplyRun({ store, channelId: channel.id, sourceMessage: branchMessage }), undefined); diff --git a/packages/daemon/test/channel-workspace.test.mjs b/packages/daemon/test/channel-workspace.test.mjs index bd23fc05d..d425d1def 100644 --- a/packages/daemon/test/channel-workspace.test.mjs +++ b/packages/daemon/test/channel-workspace.test.mjs @@ -9,6 +9,8 @@ import { DEFAULT_WORKSPACE_ID, createCompositeSecretResolver, createSqliteRunSto import { createDaemonServer } from "../dist/index.js"; import { buildContextPack, resolveProviderContextBudget } from "../dist/features/channels/context-packs.js"; import { routeChannelMessageReply } from "../dist/features/channels/channel-message-router.js"; +import { resolveChannelReplyExecutionTarget } from "../dist/features/channels/channel-reply-execution-target.js"; +import { planConversationContextSummary } from "../dist/features/channels/context-processing/conversation-summary-coordinator.js"; import { buildLingAgentSessionViewForChannel } from "../dist/features/channels/ling-agent-session-view.js"; import { buildLingAgentTurnEnvelope } from "../dist/features/channels/ling-agent-turn-envelope.js"; import { findActiveChannelReplyRun } from "../dist/features/channels/channel-reply-dispatch.js"; @@ -3612,7 +3614,14 @@ test("channel no-tools AI reply retries once with compacted channel summary afte new Error("context_length exceeded by test provider"), "Retry reply", ]); - const daemon = await startDaemonForTest(runner); + const stores = new Map(); + const daemon = await startDaemonForTest(runner, { + storeFactory: (storeCwd) => { + const store = createSqliteRunStore(storeCwd); + stores.set(storeCwd, store); + return store; + }, + }); try { const channel = await createChannel(daemon.url, cwd); const provider = (await createStaticChatProvider(daemon.url, cwd)).provider; @@ -3647,9 +3656,128 @@ test("channel no-tools AI reply retries once with compacted channel summary afte assert.equal(persistedProvider.maxTokens, undefined); const summaries = await getJson(`${daemon.url}/channels/${encodeURIComponent(channel.id)}/context-summaries?cwd=${encodeURIComponent(cwd)}`); - assert.equal(summaries.contextSummaries.length, 1); - assert.equal(summaries.contextSummaries[0].authorType, "system"); - assert.deepEqual(summaries.contextSummaries[0].sourceMessageIds, [created.message.id]); + assert.equal(summaries.contextSummaries.length, 0); + const store = stores.get(cwd); + assert.ok(store); + const incidents = store.listAvailabilityIncidents({ + providerPresetId: provider.id, + now: new Date().toISOString(), + limit: 10, + }); + assert.equal(incidents.length, 1); + assert.equal(incidents[0].kind, "timeout"); + assert.equal(incidents[0].state, "slow"); + } finally { + await daemon.close(); + await rm(cwd, { recursive: true, force: true }); + } +}); + +test("channel hard summary uses the effective workspace provider before dispatching a reply", async () => { + const cwd = await mkdtemp(resolve(tmpdir(), "ling-stack-channel-proactive-summary-")); + const runner = new SummaryAwareRunner(); + const stores = new Map(); + const daemon = await startDaemonForTest(runner, { + storeFactory: (storeCwd) => { + const store = createSqliteRunStore(storeCwd); + stores.set(storeCwd, store); + return store; + }, + }); + try { + const channel = await createChannel(daemon.url, cwd); + const provider = (await createStaticChatProvider(daemon.url, cwd, { + contextWindow: 1_800, + maxTokens: 512, + })).provider; + await postJson(`${daemon.url}/providers/${encodeURIComponent(provider.id)}/check?cwd=${encodeURIComponent(cwd)}`); + await setWorkspaceDefaultProvider(daemon.url, cwd, provider.id); + const store = stores.get(cwd); + assert.ok(store); + const thread = store.resolveOrCreateConversationThread({ + surfaceKind: "channel_root", + channelId: channel.id, + title: "Proactive summary", + objective: "Summarize before the provider rejects the prompt", + }, "2026-08-10T06:00:00.000Z").thread; + + for (let index = 0; index < 10; index += 1) { + const createdAt = new Date(Date.parse("2026-08-10T06:00:01.000Z") + index * 10_000).toISOString(); + const turn = store.appendConversationUserTurn({ + itemId: `item_proactive_summary_${index}`, + turnId: `turn_proactive_summary_${index}`, + channelId: channel.id, + threadId: thread.id, + title: `Historical fact ${index}`, + userInstruction: `Historical fact ${index}`, + payload: { + body: `Historical fact ${index}: ${"x".repeat(900)}`, + format: "plain_text", + mentions: [], + attachmentIds: [], + linkPreviewIds: [], + fileRefs: [], + }, + idempotencyKey: `proactive-summary:${index}`, + }, createdAt); + const run = store.createRun({ + id: `run_proactive_summary_${index}`, + cwd, + taskId: `task_proactive_summary_${index}`, + channelId: channel.id, + threadId: thread.id, + turnId: turn.turnId, + providerName: "test", + providerModel: "test", + taskSnapshot: { + id: `task_proactive_summary_${index}`, + title: `Historical fact ${index}`, + prompt: `Historical fact ${index}`, + }, + providerSnapshot: { provider: "test", model: "test" }, + requestSnapshot: { + cwd, + task: { + id: `task_proactive_summary_${index}`, + title: `Historical fact ${index}`, + prompt: `Historical fact ${index}`, + }, + }, + createdAt: new Date(Date.parse(createdAt) + 1_000).toISOString(), + }); + store.markRunStarted(run.id, new Date(Date.parse(createdAt) + 2_000).toISOString()); + store.markRunSucceeded(run.id, new Date(Date.parse(createdAt) + 3_000).toISOString()); + } + + const created = await postJson(`${daemon.url}/channels/${encodeURIComponent(channel.id)}/messages?cwd=${encodeURIComponent(cwd)}`, { + body: "@Engineer continue from the durable summary.", + mentions: [{ id: "teammate_builtin_engineer", label: "Engineer", kind: "ai-teammate" }], + teammateId: "teammate_builtin_engineer", + requestAiReply: true, + }); + await waitForChannelMessageStatus(daemon.url, cwd, channel.id, created.aiReply.id, "ready"); + + const target = resolveChannelReplyExecutionTarget({ + store, + channel, + requestedTeammateId: "teammate_builtin_engineer", + }); + assert.equal(target.providerPresetId, provider.id); + const currentItem = store.getConversationItem(created.message.id); + assert.ok(currentItem); + const plan = planConversationContextSummary({ + store, + sourceItemId: currentItem.id, + inputBudgetTokens: resolveProviderContextBudget(provider, "channel_reply").inputBudgetTokens, + }); + assert.equal(plan?.trigger, "hard"); + assert.equal(runner.summaryRuns.length, 1); + assert.equal(runner.replyRuns.length, 1); + const summaries = store.listConversationContextSummaries({ threadId: thread.id }); + assert.equal(summaries.length, 1); + assert.equal(summaries[0].objective, "Preserve the durable historical objective."); + assert.ok(summaries[0].coveredThroughSequence < currentItem.sequence); + assert.match(runner.replyRuns[0].task.prompt, /Preserve the durable historical objective/); } finally { await daemon.close(); await rm(cwd, { recursive: true, force: true }); @@ -6053,21 +6181,32 @@ test("default_teammate route treats channel branches and topic replies as separa title: "Root", objective: "Root conversation", }, now).thread; - const rootTurn = store.createLingTurn({ - threadId: rootThread.id, - kind: "initial_request", - title: "Branch parent", - userInstruction: "Create a branch", + const rootMessage = store.createChannelMessage({ channelId: channel.id, + authorType: "local-user", + body: "Create a branch", }, now); + const rootItem = store.getConversationItem(rootMessage.id); + assert.ok(rootItem?.turnId); + const rootAssignment = store.getEffectiveTurnContextAssignment(rootItem.turnId); + assert.ok(rootAssignment); const branchThread = store.resolveOrCreateConversationThread({ id: "thread_branch_root", surfaceKind: "branch", channelId: channel.id, parentThreadId: rootThread.id, - parentTurnId: rootTurn.id, + parentTurnId: rootItem.turnId, title: "Branch", objective: "Branch conversation", + forkContextSnapshot: { + id: "snapshot_branch_root", + segmentId: rootAssignment.segmentId, + projectionVersion: "ctxv2_branch_root", + throughItemSequence: rootItem.sequence, + summaryIds: [], + sourceRefs: [{ type: "conversation_item", id: rootItem.id }], + policyVersion: "continuity-v2-snapshot-v1", + }, }, now).thread; const branch = store.createChannelBranch( { @@ -12027,6 +12166,34 @@ test("private attachments avatar and hard delete stay scoped to teammate private } }); +class SummaryAwareRunner { + constructor() { + this.runs = []; + this.summaryRuns = []; + this.replyRuns = []; + } + + async runTask(options) { + this.runs.push(options); + const summary = /conversation summary coordinator/i.test(options.systemPrompt ?? ""); + const text = summary + ? JSON.stringify({ + objective: "Preserve the durable historical objective.", + decisions: ["Use proactive summaries before provider rejection."], + constraints: ["Keep the recent raw tail."], + openQuestions: [], + }) + : "Reply after proactive summary."; + (summary ? this.summaryRuns : this.replyRuns).push(options); + options.onEvent?.({ type: "assistant_text_delta", text }); + return { + taskId: options.task.id, + messages: [{ role: "assistant", content: text }], + events: [], + }; + } +} + class RecordingRunner { constructor() { this.runs = []; diff --git a/packages/daemon/test/conversation-summary-coordinator.test.mjs b/packages/daemon/test/conversation-summary-coordinator.test.mjs new file mode 100644 index 000000000..7e167eda9 --- /dev/null +++ b/packages/daemon/test/conversation-summary-coordinator.test.mjs @@ -0,0 +1,406 @@ +import assert from "node:assert/strict"; +import { mkdtemp, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import test from "node:test"; +import { createSqliteRunStore } from "../../core/dist/index.js"; +import { + coordinateConversationContextSummary, + planConversationContextSummary, +} from "../dist/features/channels/context-processing/conversation-summary-coordinator.js"; +import { buildConversationContextProjection } from "../dist/features/channels/context-processing/conversation-context-projector.js"; +import { estimateTextTokens } from "../dist/features/channels/context-packs.js"; + +const summaryPolicy = { + softCompletedTurnCount: 10, + softInputBudgetRatio: 0.3, + hardInputBudgetRatio: 0.7, + targetInputBudgetRatio: 0.55, + rawTailCompletedTurnCount: 2, + generationChunkBudgetRatio: 0.25, +}; + +test("summary coordinator compacts stable turns, preserves a raw tail, and advances coverage", async (t) => { + const fixture = await createFixture(t, "rolling"); + const firstBatch = []; + for (let index = 0; index < 12; index += 1) { + firstBatch.push(addCompletedTurn(fixture, `first-${index}`, `Decision ${index}`)); + } + const firstPlan = planConversationContextSummary({ + store: fixture.store, + sourceItemId: firstBatch.at(-1).item.id, + inputBudgetTokens: 16_000, + policy: summaryPolicy, + }); + assert.ok(firstPlan); + assert.equal(firstPlan.trigger, "soft"); + assert.equal(firstPlan.sourceItems.length, 10); + assert.equal(firstPlan.coveredThroughSequence, firstBatch[9].item.sequence); + assert.equal(firstPlan.estimatedRawTailTokens > 0, true); + + const first = await coordinateConversationContextSummary({ + store: fixture.store, + sourceItemId: firstBatch.at(-1).item.id, + inputBudgetTokens: 16_000, + policy: summaryPolicy, + generate: mergeDraft, + now: "2026-08-10T02:30:00.000Z", + }); + assert.equal(first.status, "completed"); + assert.equal(first.commit.status, "committed"); + assert.equal(first.commit.summary.coveredThroughSequence, firstBatch[9].item.sequence); + + const secondBatch = []; + for (let index = 0; index < 10; index += 1) { + secondBatch.push(addCompletedTurn(fixture, `second-${index}`, `Constraint ${index}`)); + } + const second = await coordinateConversationContextSummary({ + store: fixture.store, + sourceItemId: secondBatch.at(-1).item.id, + inputBudgetTokens: 16_000, + policy: summaryPolicy, + generate: mergeDraft, + now: "2026-08-10T03:30:00.000Z", + }); + assert.equal(second.status, "completed"); + assert.equal(second.commit.status, "committed"); + assert.equal(second.commit.summary.version, 2); + assert.equal(second.commit.summary.coveredFromSequence, firstBatch[0].item.sequence); + assert.equal(second.commit.summary.coveredThroughSequence, secondBatch[7].item.sequence); + assert.equal(second.commit.summary.sourceRefs.some((ref) => + ref.type === "context_summary" && ref.id === first.commit.summary.id), true); + assert.equal(fixture.store.getConversationContextSummary(first.commit.summary.id)?.status, "superseded"); +}); + +test("summary coordinator selects the hard trigger and discards a late assignment result", async (t) => { + const fixture = await createFixture(t, "race"); + const turns = []; + for (let index = 0; index < 10; index += 1) { + turns.push(addCompletedTurn(fixture, `race-${index}`, `Long fact ${index} ${"x".repeat(900)}`)); + } + const plan = planConversationContextSummary({ + store: fixture.store, + sourceItemId: turns.at(-1).item.id, + inputBudgetTokens: 1_024, + policy: { + ...summaryPolicy, + softCompletedTurnCount: 100, + }, + }); + assert.ok(plan); + assert.equal(plan.trigger, "hard"); + assert.equal(plan.sourceItems.length, 8); + + let corrected = false; + const result = await coordinateConversationContextSummary({ + store: fixture.store, + sourceItemId: turns.at(-1).item.id, + inputBudgetTokens: 1_024, + policy: { + ...summaryPolicy, + softCompletedTurnCount: 100, + }, + generate: async (input) => { + if (!corrected) { + corrected = true; + fixture.store.applyTurnContextAssignment({ + turnId: turns[0].turnId, + target: { kind: "existing", segmentId: turns[0].contextSegmentId }, + status: "committed", + decisionSource: "user", + confidence: 1, + evidenceRefs: [{ type: "conversation_item", id: turns[0].item.id }], + policyVersion: "continuity-v2-segmentation-v1", + }, "2026-08-10T04:30:00.000Z"); + } + return mergeDraft(input); + }, + now: "2026-08-10T04:31:00.000Z", + }); + assert.equal(result.status, "completed"); + assert.deepEqual(result.commit, { status: "discarded_stale", reason: "assignments_changed" }); +}); + +test("summary coordinator counts the active summary in the hard budget and bounds its replacement", async (t) => { + const fixture = await createFixture(t, "summary-budget"); + const first = addCompletedTurn(fixture, "initial", "Initial summary source"); + const initialSummary = commitSingleTurnSummary(fixture, first, { + objective: `Oversized objective ${"x".repeat(7_900)}`, + now: nextTimestamp(fixture), + }); + const laterTurns = []; + for (let index = 0; index < 10; index += 1) { + laterTurns.push(addCompletedTurn(fixture, `later-${index}`, `Fact ${index}`)); + } + const plan = planConversationContextSummary({ + store: fixture.store, + sourceItemId: laterTurns.at(-1).item.id, + inputBudgetTokens: 1_024, + policy: { + ...summaryPolicy, + softCompletedTurnCount: 100, + softInputBudgetRatio: 0.9, + }, + }); + assert.ok(plan); + assert.equal(plan.trigger, "hard"); + assert.ok(plan.estimatedProjectionTokens > plan.inputBudgetTokens * summaryPolicy.hardInputBudgetRatio); + assert.ok(plan.estimatedRawTokens < plan.inputBudgetTokens * 0.9); + + const result = await coordinateConversationContextSummary({ + store: fixture.store, + sourceItemId: laterTurns.at(-1).item.id, + inputBudgetTokens: 1_024, + policy: { + ...summaryPolicy, + softCompletedTurnCount: 100, + softInputBudgetRatio: 0.9, + }, + generate: async () => ({ + objective: `Compact objective ${"y".repeat(5_000)}`, + decisions: Array.from({ length: 100 }, (_, index) => `Decision ${index} ${"d".repeat(200)}`), + constraints: Array.from({ length: 100 }, (_, index) => `Constraint ${index} ${"c".repeat(200)}`), + openQuestions: Array.from({ length: 100 }, (_, index) => `Question ${index} ${"q".repeat(200)}`), + }), + now: nextTimestamp(fixture), + }); + assert.equal(result.status, "completed"); + assert.equal(result.commit.status, "committed"); + assert.equal(result.commit.summary.supersedesSummaryId, initialSummary.id); + const draftTokens = estimateTextTokens(JSON.stringify({ + objective: result.commit.summary.objective, + decisions: result.commit.summary.decisions, + constraints: result.commit.summary.constraints, + openQuestions: result.commit.summary.openQuestions, + })); + assert.ok(draftTokens <= result.plan.summaryTokenBudget); +}); + +test("branch projection reconstructs from its immutable snapshot without copying parent items", async (t) => { + const fixture = await createFixture(t, "branch"); + const parent = fixture.store.appendConversationUserTurn({ + itemId: "item_branch_parent", + turnId: "turn_branch_parent", + channelId: fixture.channel.id, + threadId: fixture.thread.id, + title: "Original parent fact", + userInstruction: "Original parent fact", + payload: { + body: "Original parent fact", + format: "plain_text", + mentions: [], + attachmentIds: [], + linkPreviewIds: [], + fileRefs: [], + }, + idempotencyKey: "branch-parent", + }, nextTimestamp(fixture)); + const initialSummary = commitSingleTurnSummary(fixture, parent, { + objective: "Original parent objective", + now: nextTimestamp(fixture), + }); + const parentProjection = buildConversationContextProjection({ + store: fixture.store, + sourceItemId: parent.item.id, + view: "reply", + }); + assert.deepEqual(parentProjection.metadata.summaryIds, [initialSummary.id]); + const branch = fixture.store.resolveOrCreateConversationThread({ + id: "thread_branch_snapshot", + surfaceKind: "branch", + channelId: fixture.channel.id, + parentThreadId: fixture.thread.id, + parentTurnId: parent.turnId, + title: "Branch snapshot", + objective: "Keep the fork boundary stable", + forkContextSnapshot: { + id: "snapshot_branch_projection", + segmentId: parentProjection.metadata.segmentId, + projectionVersion: parentProjection.metadata.projectionVersion, + throughItemSequence: parentProjection.metadata.throughItemSequence, + summaryIds: parentProjection.metadata.summaryIds, + sourceRefs: [ + { type: "thread", id: fixture.thread.id }, + { type: "turn", id: parent.turnId }, + { type: "conversation_item", id: parent.item.id }, + ], + policyVersion: parentProjection.metadata.policyVersion, + }, + }, nextTimestamp(fixture)).thread; + assert.equal(branch.forkContextSnapshotId, "snapshot_branch_projection"); + assert.equal(fixture.store.listConversationItemsPage({ threadId: branch.id, limit: 10 }).items.length, 0); + + const child = fixture.store.appendConversationUserTurn({ + itemId: "item_branch_child", + turnId: "turn_branch_child", + channelId: fixture.channel.id, + threadId: branch.id, + title: "Child question", + userInstruction: "Child question", + payload: { + body: "Child question", + format: "plain_text", + mentions: [], + attachmentIds: [], + linkPreviewIds: [], + fileRefs: [], + }, + idempotencyKey: "branch-child", + }, nextTimestamp(fixture)); + const childProjection = buildConversationContextProjection({ + store: fixture.store, + sourceItemId: child.item.id, + view: "reply", + }); + assert.equal(childProjection.forkSnapshot?.id, "snapshot_branch_projection"); + assert.deepEqual(childProjection.items.map((item) => item.id), [parent.item.id, child.item.id]); + assert.deepEqual(childProjection.sharedCore.objectives, ["Original parent objective"]); + + const replacement = fixture.store.appendConversationUserTurn({ + itemId: "item_branch_parent_replacement", + turnId: "turn_branch_parent_replacement", + channelId: fixture.channel.id, + threadId: fixture.thread.id, + title: "Changed parent fact", + userInstruction: "Changed parent fact", + payload: { + body: "Changed parent fact", + format: "plain_text", + mentions: [], + attachmentIds: [], + linkPreviewIds: [], + fileRefs: [], + }, + supersedesItemId: parent.item.id, + idempotencyKey: "branch-parent-replacement", + }, nextTimestamp(fixture)); + const replacementSummary = commitSingleTurnSummary(fixture, replacement, { + objective: "Changed parent objective", + coveredFromSequence: parent.item.sequence, + expectedActiveSummaryId: initialSummary.id, + now: nextTimestamp(fixture), + }); + assert.equal(fixture.store.getConversationContextSummary(initialSummary.id)?.status, "superseded"); + assert.equal(replacementSummary.objective, "Changed parent objective"); + const reconstructed = buildConversationContextProjection({ + store: fixture.store, + sourceItemId: child.item.id, + view: "reply", + }); + assert.deepEqual(reconstructed.items.map((item) => item.id), [parent.item.id, child.item.id]); + assert.equal(reconstructed.items[0].payload.body, "Original parent fact"); + assert.deepEqual(reconstructed.sharedCore.objectives, ["Original parent objective"]); +}); + +async function createFixture(t, suffix) { + const cwd = await mkdtemp(join(tmpdir(), `ling-summary-${suffix}-`)); + const store = createSqliteRunStore(cwd); + t.after(async () => { + store.close(); + await rm(cwd, { recursive: true, force: true }); + }); + const now = "2026-08-10T02:00:00.000Z"; + const channel = store.createChannel({ name: `Summary ${suffix}`, boundPath: cwd }, now); + const thread = store.resolveOrCreateConversationThread({ + surfaceKind: "channel_root", + channelId: channel.id, + title: "Summary", + objective: "Exercise proactive summaries", + }, now).thread; + return { cwd, store, channel, thread, nextTimestamp: Date.parse(now) + 1_000 }; +} + +function addCompletedTurn(fixture, suffix, body) { + const turnAt = nextTimestamp(fixture); + const result = fixture.store.appendConversationUserTurn({ + channelId: fixture.channel.id, + threadId: fixture.thread.id, + title: body.slice(0, 80), + userInstruction: body, + payload: { + body, + format: "plain_text", + mentions: [], + attachmentIds: [], + linkPreviewIds: [], + fileRefs: [], + }, + idempotencyKey: `summary:${suffix}`, + }, turnAt); + const runAt = nextTimestamp(fixture); + const run = fixture.store.createRun({ + id: `run_summary_${suffix}`, + cwd: fixture.cwd, + taskId: `task_summary_${suffix}`, + channelId: fixture.channel.id, + threadId: fixture.thread.id, + turnId: result.turnId, + providerName: "test", + providerModel: "test", + taskSnapshot: { id: `task_summary_${suffix}`, title: body, prompt: body }, + providerSnapshot: { provider: "test", model: "test" }, + requestSnapshot: { cwd: fixture.cwd, task: { id: `task_summary_${suffix}`, title: body, prompt: body } }, + createdAt: runAt, + }); + fixture.store.markRunStarted(run.id, nextTimestamp(fixture)); + fixture.store.markRunSucceeded(run.id, nextTimestamp(fixture)); + return result; +} + +function nextTimestamp(fixture) { + const value = new Date(fixture.nextTimestamp).toISOString(); + fixture.nextTimestamp += 1_000; + return value; +} + +function commitSingleTurnSummary(fixture, turn, input) { + const assignment = fixture.store.getEffectiveTurnContextAssignment(turn.turnId); + assert.ok(assignment); + const result = fixture.store.commitConversationContextSummary({ + threadId: turn.item.threadId, + segmentId: assignment.segmentId, + scope: "segment", + coveredFromSequence: input.coveredFromSequence ?? turn.item.sequence, + coveredThroughSequence: turn.item.sequence, + objective: input.objective, + decisions: [], + constraints: [], + openQuestions: [], + activeWorkRefs: [], + sourceRefs: [ + ...(input.expectedActiveSummaryId + ? [{ type: "context_summary", id: input.expectedActiveSummaryId }] + : []), + { type: "conversation_item", id: turn.item.id }, + ], + policyVersion: "continuity-v2-summary-v1", + projectionVersion: `ctxv2_branch_${turn.item.sequence}`, + visibilityFingerprint: "visibility_branch_fixture", + expectedActiveSummaryId: input.expectedActiveSummaryId ?? null, + sourceFence: [{ + itemId: turn.item.id, + sequence: turn.item.sequence, + revision: turn.item.revision, + updatedAt: turn.item.updatedAt, + visibility: turn.item.visibility, + visibilityOwnerId: turn.item.visibilityOwnerId, + turnId: turn.turnId, + assignmentId: assignment.id, + assignmentRevision: assignment.revision, + }], + }, input.now); + assert.equal(result.status, "committed"); + return result.summary; +} + +async function mergeDraft(input) { + const bodies = input.items + .filter((item) => item.kind === "user_message") + .map((item) => item.payload.body.slice(0, 80)); + return { + objective: input.current.objective ?? "Maintain durable continuity.", + decisions: [...input.current.decisions, ...bodies.filter((body) => body.startsWith("Decision"))], + constraints: [...input.current.constraints, ...bodies.filter((body) => body.startsWith("Constraint"))], + openQuestions: input.current.openQuestions, + }; +} diff --git a/packages/daemon/test/ling-turns-api.test.mjs b/packages/daemon/test/ling-turns-api.test.mjs index c64534102..106a388c2 100644 --- a/packages/daemon/test/ling-turns-api.test.mjs +++ b/packages/daemon/test/ling-turns-api.test.mjs @@ -9,6 +9,7 @@ import { createSqliteRunStore, lingThreadKinds as coreLingThreadKinds } from "@l import { createDaemonServer } from "../dist/index.js"; import { maybeQueueGoalContinuation } from "../dist/features/runs/interventions.js"; import { syncLingThreadForTerminalRun } from "../dist/features/runs/threads.js"; +import { buildConversationContextProjection } from "../dist/features/channels/context-processing/conversation-context-projector.js"; import { prepareChannelCreateBody } from "./conversation-model-test-fixture.mjs"; import { ensureGitRepoForChannelPost } from "./git-test-helpers.mjs"; @@ -298,6 +299,89 @@ test("daemon fork creates child thread lineage without copying runs", async () = } }); +test("daemon channel conversation fork atomically binds an immutable context snapshot", async () => { + const cwd = await mkdtemp(resolve(tmpdir(), "ling-stack-daemon-channel-fork-snapshot-")); + const stores = new Map(); + const daemon = await startDaemonForTest(new SuccessfulRunner(), stores); + try { + const channel = (await createChannel(daemon.url, cwd)).channel; + const store = stores.get(cwd); + assert.ok(store); + const root = store.resolveOrCreateConversationThread({ + surfaceKind: "channel_root", + channelId: channel.id, + title: "Channel root", + objective: "Fork with continuity", + }, "2026-08-10T05:00:00.000Z").thread; + const source = store.appendConversationUserTurn({ + itemId: "item_channel_fork_source", + turnId: "turn_channel_fork_source", + channelId: channel.id, + threadId: root.id, + title: "Keep this fact", + userInstruction: "Keep this fact", + payload: { + body: "Keep this fact", + format: "plain_text", + mentions: [], + attachmentIds: [], + linkPreviewIds: [], + fileRefs: [], + }, + idempotencyKey: "channel-fork-source", + }, "2026-08-10T05:00:01.000Z"); + const run = store.createRun({ + id: "run_channel_fork_source", + cwd, + taskId: "task_channel_fork_source", + channelId: channel.id, + threadId: root.id, + turnId: source.turnId, + providerName: "test", + providerModel: "test", + taskSnapshot: { id: "task_channel_fork_source", title: "Keep this fact", prompt: "Keep this fact" }, + providerSnapshot: { provider: "test", model: "test" }, + requestSnapshot: { cwd, task: { id: "task_channel_fork_source", title: "Keep this fact", prompt: "Keep this fact" } }, + createdAt: "2026-08-10T05:00:02.000Z", + }); + store.markRunStarted(run.id, "2026-08-10T05:00:03.000Z"); + store.markRunSucceeded(run.id, "2026-08-10T05:00:04.000Z"); + const parentProjection = buildConversationContextProjection({ + store, + sourceItemId: source.item.id, + view: "reply", + }); + + const forked = await postJson(`${daemon.url}/threads/${encodeURIComponent(root.id)}/fork?cwd=${encodeURIComponent(cwd)}`, { + turnId: source.turnId, + sourceRunId: run.id, + idempotencyKey: "channel-context-fork", + }); + assert.equal(forked.childThread.surfaceKind, "branch"); + assert.ok(forked.childThread.forkContextSnapshotId); + const snapshot = store.getConversationContextSnapshot(forked.childThread.forkContextSnapshotId); + assert.ok(snapshot); + assert.equal(snapshot.threadId, root.id); + assert.equal(snapshot.segmentId, parentProjection.metadata.segmentId); + assert.equal(snapshot.projectionVersion, parentProjection.metadata.projectionVersion); + assert.equal(snapshot.throughItemSequence, source.item.sequence); + assert.equal(snapshot.sourceRefs.some((ref) => ref.type === "conversation_item" && ref.id === source.item.id), true); + assert.equal(store.listConversationItemsPage({ threadId: forked.childThread.id, limit: 10 }).items.length, 0); + + const repeated = await postJson(`${daemon.url}/threads/${encodeURIComponent(root.id)}/fork?cwd=${encodeURIComponent(cwd)}`, { + turnId: source.turnId, + sourceRunId: run.id, + idempotencyKey: "channel-context-fork", + }); + assert.equal(repeated.childThread.id, forked.childThread.id); + assert.equal(repeated.childThread.forkContextSnapshotId, snapshot.id); + assert.equal(store.listConversationContextSnapshots({ threadId: root.id }).length, 1); + } finally { + await daemon.close(); + await rm(cwd, { recursive: true, force: true }); + } +}); + test("daemon fork-run creates child thread turn and run idempotently", async () => { const cwd = await mkdtemp(resolve(tmpdir(), "ling-stack-daemon-thread-fork-run-")); const stores = new Map();