From 1534443765627adec72acbc718a8eebb74e50169 Mon Sep 17 00:00:00 2001 From: tianxiaochannel-oss88 Date: Mon, 10 Aug 2026 13:16:01 +0800 Subject: [PATCH] feat(channels): enforce multi-agent context consistency --- ...ation-continuity-v2-implementation-plan.md | 17 +- .../channel-conversation-continuity-v2.md | 2 +- docs/design/implementation-status.md | 2 +- .../src/conversation-continuity-validation.ts | 2 + .../contracts/src/conversation-continuity.ts | 6 + packages/contracts/src/workflows.ts | 29 + .../test/conversation-continuity-v2.test.mjs | 1 + packages/core/src/handoff-intents.ts | 5 + packages/core/src/redaction.ts | 19 +- .../sqlite/channel-workflow-mappers.ts | 72 ++ .../storage/sqlite/channel-workflow-store.ts | 30 +- .../core/src/storage/sqlite/migrations.ts | 19 + .../core/src/storage/sqlite/select-sql.ts | 1 + packages/core/src/workspace-types.ts | 34 + .../test/channel-intent-v2-migration.test.mjs | 1 + .../core/test/channel-runtime-store.test.mjs | 52 ++ packages/core/test/handoff-intents.test.mjs | 20 + packages/core/test/secrets.test.mjs | 39 + .../channels/agent-context-consistency.ts | 273 +++++++ .../src/features/channels/context-packs.ts | 26 +- .../conversation-context-projector.ts | 30 +- .../context-processing/workflow-steps.ts | 17 +- .../src/features/channels/orchestrator.ts | 166 +++- .../channels/terminal-report-evidence.ts | 21 +- .../channels/workflow-coordination-routes.ts | 22 +- .../src/features/channels/workflow-fanin.ts | 84 +- .../features/channels/workflow-handoffs.ts | 195 ++++- .../channels/workflow-step-packets.ts | 29 +- .../daemon/src/features/channels/workflows.ts | 128 ++- .../daemon/src/features/runs/orchestrator.ts | 31 +- packages/daemon/src/features/runs/queue.ts | 1 + packages/daemon/src/features/runs/schemas.ts | 8 + packages/daemon/src/server/router.ts | 75 +- .../test/agent-context-consistency.test.mjs | 750 ++++++++++++++++++ .../daemon/test/approval-policy-run.test.mjs | 2 + .../daemon/test/channel-workspace.test.mjs | 241 ++++++ .../test/workflow-terminal-summary.test.mjs | 13 +- 37 files changed, 2337 insertions(+), 126 deletions(-) create mode 100644 packages/daemon/src/features/channels/agent-context-consistency.ts create mode 100644 packages/daemon/test/agent-context-consistency.test.mjs diff --git a/docs/design/channel-conversation-continuity-v2-implementation-plan.md b/docs/design/channel-conversation-continuity-v2-implementation-plan.md index 6e84e7d79..6b0d3e700 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, S4, and S5 implemented and locally verified; S6-S7 pending +Status: G0 complete; S-1 and S0 merged; S1+S2 through S6 implemented and locally verified; S7 pending Owner: conversation, channel context, Ling Agent Session, and storage maintainers Date: 2026-08-08 Last verified: 2026-08-10 @@ -527,12 +527,21 @@ snapshot reconstruction, and the one strict context-length fallback. ## Slice S6: Multi-Agent Context Version Consistency +Status: implemented on `codex/continuity-v2-agent-consistency` and locally verified. +Contracts (`35/35`), Core (`360/360`), Daemon (`902/902`), and Web (`42/42`) +tests pass; Web typecheck and `git diff --check` pass. Agent requests, output packets, +fan-in, and handoffs carry durable context stamps; stale relevant facts and direct private +evidence copies fail closed, while safe metadata-only rebases and accepted-handoff retries +remain deterministic. + ### Changes - Include projection version, segment id, item watermark, and source refs in every agent request and handoff packet. - Provide every agent the same shared core, then add role-specific evidence. - Enforce teammate-private memory and private-chat isolation. +- Keep owner-scoped execution prompts separate from durable Task, Run, and WorkUnit audit + snapshots; persist memory-use ids and policy evidence without copying private memory text. - On result fan-in, compare consumed projection with current projection: - accept if changed refs are unrelated; - rebase metadata-only results when safe; @@ -811,7 +820,7 @@ interfaces at the end of S3 and keep implementation files separate before parall - Verify: daemon regression suite, explicit file safety, clarification actionability, and eval cases. - [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. +- [x] **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. - [ ] **T8 (P2, human: ~2 days / Codex: ~3-4 hours)**: switch transcript/API/Web diagnostics and remove old runtime authorities. - Verify: package suites, channel E2E, old-symbol search, and migration smoke. @@ -861,7 +870,7 @@ schema. | Design Review | `/plan-design-review` | UI and UX gaps | 0 | not run | UI changes are limited to later advanced Context details | | DX Review | `/plan-devex-review` | Developer experience gaps | 0 | not run | Not required for the design baseline | -**VERDICT:** ENG CLEARED. S-1 and S0 are merged; S1+S2, S3, and S4 are implemented -and locally verified; S5-S7 retain the reviewed delivery order and exit criteria. +**VERDICT:** ENG CLEARED. S-1 and S0 are merged; S1+S2 through S6 are implemented +and locally verified; S7 retains the reviewed delivery order and exit criteria. NO UNRESOLVED DECISIONS diff --git a/docs/design/channel-conversation-continuity-v2.md b/docs/design/channel-conversation-continuity-v2.md index a4da37185..5cd58750a 100644 --- a/docs/design/channel-conversation-continuity-v2.md +++ b/docs/design/channel-conversation-continuity-v2.md @@ -1,6 +1,6 @@ # Channel Conversation Continuity v2 -Status: approved design baseline; S-1 and S0 merged; S1+S2, S3, and S4 implemented and locally verified; S5-S7 pending +Status: approved design baseline; S-1 and S0 merged; S1+S2 through S6 implemented and locally verified; S7 pending Date: 2026-08-08 Last verified: 2026-08-10 Scope: channel conversation persistence, Thread/Turn/Item topology, context segmentation, diff --git a/docs/design/implementation-status.md b/docs/design/implementation-status.md index 3e7652e97..df42868aa 100644 --- a/docs/design/implementation-status.md +++ b/docs/design/implementation-status.md @@ -28,7 +28,7 @@ counts, screenshots, hashes and run diaries belong in CI or Release artifacts. | Work | Current boundary | Closure condition | |---|---|---| -| [Channel Conversation Continuity v2](channel-conversation-continuity-v2-implementation-plan.md) | G0 and S0 merged; S1+S2, S3, and S4 implemented and locally verified | S5-S7, final migration, recovery, performance, security, eval, and Web E2E gates pass | +| [Channel Conversation Continuity v2](channel-conversation-continuity-v2-implementation-plan.md) | G0, S0, and S1+S2 merged; S3-S6 implemented and locally verified | S7, final migration, recovery, performance, security, eval, and Web E2E gates pass | | [Channel Context Processing](channel-context-processing-layer-implementation-plan.md) | Channel and direct-message context views exist | Condensation, quote eligibility, persistence and redaction ownership are closed | | [Permission Presets](codex-inspired-settings-permission-presets-implementation-plan.md) | Enforcement modes exist | Preset UI, effective preview and run audit agree with daemon enforcement | | [Conversation Surface](conversation-surface-contract-implementation-plan.md) | Shared primitives and initial package extraction are implemented | WorkCell, evidence, attachment, package CSS and Desktop-consumption boundaries close | diff --git a/packages/contracts/src/conversation-continuity-validation.ts b/packages/contracts/src/conversation-continuity-validation.ts index dcb19cd7f..ea0e64765 100644 --- a/packages/contracts/src/conversation-continuity-validation.ts +++ b/packages/contracts/src/conversation-continuity-validation.ts @@ -158,6 +158,7 @@ const SNAPSHOT_KEYS = new Set([ const PROJECTION_KEYS = new Set([ "projectionVersion", + "sourceStateFingerprint", "threadId", "turnId", "segmentId", @@ -393,6 +394,7 @@ export function isConversationContextProjectionMetadata( ): value is ConversationContextProjectionMetadata { return isRecordWithOnlyKeys(value, PROJECTION_KEYS) && isText(value.projectionVersion) + && isText(value.sourceStateFingerprint) && isText(value.threadId) && isText(value.turnId) && isText(value.segmentId) diff --git a/packages/contracts/src/conversation-continuity.ts b/packages/contracts/src/conversation-continuity.ts index a13c0c2fd..090744ca9 100644 --- a/packages/contracts/src/conversation-continuity.ts +++ b/packages/contracts/src/conversation-continuity.ts @@ -339,6 +339,12 @@ export interface ConversationContextSnapshot { export interface ConversationContextProjectionMetadata { readonly projectionVersion: string; + /** + * Fingerprint of the effective assignment, summaries, and item revisions that + * contributed facts to this projection. Unlike projectionVersion, this does + * not change for metadata-only segment activity. + */ + readonly sourceStateFingerprint: string; readonly threadId: string; readonly turnId: string; readonly segmentId: string; diff --git a/packages/contracts/src/workflows.ts b/packages/contracts/src/workflows.ts index c912e043b..b3a0df1f5 100644 --- a/packages/contracts/src/workflows.ts +++ b/packages/contracts/src/workflows.ts @@ -109,6 +109,29 @@ export interface WorkflowPacketSourceRef { readonly label?: string; } +export interface WorkflowAgentContextStamp { + readonly schemaVersion: 1; + readonly sourceItemId: string; + readonly consumerTeammateId?: string; + readonly contextProjection: ConversationContextProjectionMetadata; + readonly roleEvidenceRefs: readonly WorkflowPacketSourceRef[]; + readonly roleEvidenceFingerprint: string; +} + +export type WorkflowFanInContextReviewStatus = + | "accepted_current" + | "rebased_metadata" + | "rejected_stale" + | "rejected_private_leak"; + +export interface WorkflowFanInContextReview { + readonly dependencyWorkflowStepId: string; + readonly status: WorkflowFanInContextReviewStatus; + readonly consumedProjectionVersion?: string; + readonly currentProjectionVersion?: string; + readonly reason: string; +} + export type WorkflowHandoffIntentStatus = | "proposed" | "validated" @@ -148,6 +171,7 @@ export interface WorkflowHandoffIntent { readonly expectedOutput: string; readonly riskSummary?: string; readonly sourceRefs: readonly WorkflowPacketSourceRef[]; + readonly agentContext?: WorkflowAgentContextStamp; readonly status: WorkflowHandoffIntentStatus; readonly validation?: WorkflowHandoffValidation; readonly createdBy: WorkflowHandoffIntentCreatedBy; @@ -218,6 +242,7 @@ export interface ChannelWorkflowStepContextPacket { readonly dependencies: readonly WorkflowStepOutputRef[]; readonly priorOutputArtifactIds: readonly string[]; readonly sourceRefs?: readonly WorkflowPacketSourceRef[]; + readonly agentContext?: WorkflowAgentContextStamp; readonly selectorDecisionIds?: readonly string[]; readonly handoffIntentIds?: readonly string[]; readonly requestedRunnerId?: string; @@ -260,6 +285,7 @@ export interface WorkflowStepContextPackSnapshot { readonly label?: string; }[]; readonly contextProjection?: ConversationContextProjectionMetadata; + readonly agentContext?: WorkflowAgentContextStamp; } export interface ChannelWorkflowStepOutputPacket { @@ -278,6 +304,7 @@ export interface ChannelWorkflowStepOutputPacket { readonly blockers?: readonly string[]; readonly changedPaths?: readonly string[]; readonly sourceRefs?: readonly WorkflowPacketSourceRef[]; + readonly agentContext?: WorkflowAgentContextStamp; readonly selectorDecisionIds?: readonly string[]; readonly handoffIntentIds?: readonly string[]; readonly runnerAudit?: RunnerResolutionAudit; @@ -320,6 +347,8 @@ export interface ChannelWorkflowStepMetadata { readonly proposalStepSources?: readonly ChannelExecutionProposalStepSource[]; readonly fanInSummaryArtifactId?: string; readonly memoryCandidateIds?: readonly string[]; + readonly fanInContextReviews?: readonly WorkflowFanInContextReview[]; + readonly resultContextReview?: WorkflowFanInContextReview; } export type WorkflowGraphNodeKind = "step" | "join"; diff --git a/packages/contracts/test/conversation-continuity-v2.test.mjs b/packages/contracts/test/conversation-continuity-v2.test.mjs index 88b0ab79d..2d63c0e74 100644 --- a/packages/contracts/test/conversation-continuity-v2.test.mjs +++ b/packages/contracts/test/conversation-continuity-v2.test.mjs @@ -320,6 +320,7 @@ test("segment, assignment, summary, snapshot and projection contracts preserve s }), true); assert.equal(isConversationContextProjectionMetadata({ projectionVersion: "projection-1", + sourceStateFingerprint: "source-state-1", threadId: "thread-1", turnId: "turn-2", segmentId: "segment-1", diff --git a/packages/core/src/handoff-intents.ts b/packages/core/src/handoff-intents.ts index 924b4f469..d056dd0ee 100644 --- a/packages/core/src/handoff-intents.ts +++ b/packages/core/src/handoff-intents.ts @@ -4,6 +4,7 @@ import type { WorkflowHandoffIntent, WorkflowHandoffIntentStatus, WorkflowHandoffValidation, + WorkflowAgentContextStamp, WorkflowPacketSourceRef, } from "./workspace-types.js"; @@ -29,6 +30,7 @@ export interface LingHandoffInputPacket { readonly requiredInput: string; readonly expectedOutput: string; readonly sourceRefs: readonly WorkflowPacketSourceRef[]; + readonly agentContext?: WorkflowAgentContextStamp; } export interface LingHandoffIntent { @@ -51,6 +53,7 @@ export interface LingHandoffIntent { readonly riskSummary?: string; readonly status: LingHandoffIntentStatus; readonly sourceRefs: readonly WorkflowPacketSourceRef[]; + readonly agentContext?: WorkflowAgentContextStamp; readonly inputPacket: LingHandoffInputPacket; readonly validation?: WorkflowHandoffValidation; readonly createdBy: string; @@ -134,6 +137,7 @@ export function normalizeWorkflowHandoffIntentToLingIntent(intent: WorkflowHando riskSummary: intent.riskSummary, status: intent.status, sourceRefs, + agentContext: intent.agentContext, inputPacket: { schemaVersion: 1, intentId: `workflow_handoff_intent:${intent.id}`, @@ -147,6 +151,7 @@ export function normalizeWorkflowHandoffIntentToLingIntent(intent: WorkflowHando requiredInput: intent.requiredInput, expectedOutput: intent.expectedOutput, sourceRefs, + agentContext: intent.agentContext, }, validation: intent.validation, createdBy: intent.createdBy, diff --git a/packages/core/src/redaction.ts b/packages/core/src/redaction.ts index f761c715f..af96d77c2 100644 --- a/packages/core/src/redaction.ts +++ b/packages/core/src/redaction.ts @@ -17,7 +17,11 @@ function redactValue(value: unknown, seen: WeakSet, extraSecrets: readon return "[Circular]"; } seen.add(value); - return value.map((item) => redactValue(item, seen, extraSecrets)); + try { + return value.map((item) => redactValue(item, seen, extraSecrets)); + } finally { + seen.delete(value); + } } if (typeof value === "object" && value !== null) { @@ -25,12 +29,15 @@ function redactValue(value: unknown, seen: WeakSet, extraSecrets: readon return "[Circular]"; } seen.add(value); - - const result: Record = {}; - for (const [key, item] of Object.entries(value)) { - result[key] = shouldRedactKey(key) ? REDACTED : redactValue(item, seen, extraSecrets); + try { + const result: Record = {}; + for (const [key, item] of Object.entries(value)) { + result[key] = shouldRedactKey(key) ? REDACTED : redactValue(item, seen, extraSecrets); + } + return result; + } finally { + seen.delete(value); } - return result; } return value; diff --git a/packages/core/src/storage/sqlite/channel-workflow-mappers.ts b/packages/core/src/storage/sqlite/channel-workflow-mappers.ts index 5ed7e76c0..b105e6c48 100644 --- a/packages/core/src/storage/sqlite/channel-workflow-mappers.ts +++ b/packages/core/src/storage/sqlite/channel-workflow-mappers.ts @@ -1,4 +1,5 @@ import type { SQLOutputValue } from "node:sqlite"; +import { isConversationContextProjectionMetadata } from "@ling-stack/contracts"; import type { RunTool } from "../../approval-policy.js"; import type { SkillApprovalProfile } from "../../skill-types.js"; import { @@ -12,9 +13,11 @@ import { type ChannelWorkflowStepStatus, type ResponseLanguage, type WorkflowFanInPolicy, + type WorkflowFanInContextReview, type WorkflowHandoffIntent, type WorkflowHandoffIntentStatus, type WorkflowHandoffValidation, + type WorkflowAgentContextStamp, type WorkflowPacketSourceRef, type WorkflowSelectorAlternative, type WorkflowSelectorDecision, @@ -54,6 +57,7 @@ export function mapWorkflowHandoffIntent(row: Record): W expectedOutput: readString(row, "expected_output"), riskSummary: readNullableString(row, "risk_summary"), sourceRefs: parseWorkflowPacketSourceRefs(readString(row, "source_refs_json")), + agentContext: parseOptionalWorkflowAgentContextStamp(readNullableString(row, "agent_context_json")), status: readString(row, "status") as WorkflowHandoffIntentStatus, validation: parseOptionalWorkflowHandoffValidation(readNullableString(row, "validation_json")), createdBy: readString(row, "created_by") as WorkflowHandoffIntent["createdBy"], @@ -62,6 +66,36 @@ export function mapWorkflowHandoffIntent(row: Record): W }); } +function parseOptionalWorkflowAgentContextStamp(value: string | undefined): WorkflowAgentContextStamp | undefined { + if (!value) return undefined; + const parsed = parseJson(value); + if (!parsed || typeof parsed !== "object" || Array.isArray(parsed)) return undefined; + const record = parsed as Record; + if ( + record.schemaVersion !== 1 + || typeof record.sourceItemId !== "string" + || record.sourceItemId.trim().length === 0 + || (record.consumerTeammateId !== undefined + && (typeof record.consumerTeammateId !== "string" || record.consumerTeammateId.trim().length === 0)) + || !isConversationContextProjectionMetadata(record.contextProjection) + || !Array.isArray(record.roleEvidenceRefs) + || typeof record.roleEvidenceFingerprint !== "string" + || record.roleEvidenceFingerprint.trim().length === 0 + ) { + return undefined; + } + const roleEvidenceRefs = parseWorkflowPacketSourceRefs(JSON.stringify(record.roleEvidenceRefs)); + if (roleEvidenceRefs.some((ref) => ref.id.trim().length === 0)) return undefined; + return { + schemaVersion: 1, + sourceItemId: record.sourceItemId, + consumerTeammateId: record.consumerTeammateId as string | undefined, + contextProjection: record.contextProjection, + roleEvidenceRefs, + roleEvidenceFingerprint: record.roleEvidenceFingerprint, + }; +} + export function mapWorkflowSelectorDecision(row: Record): WorkflowSelectorDecision { return omitUndefined({ schemaVersion: 1 as const, @@ -179,6 +213,44 @@ function parseOptionalChannelWorkflowStepMetadata(value: string | undefined): Ch proposalStepSources: parseOptionalProposalStepSourcesValue(record.proposalStepSources), fanInSummaryArtifactId: readOptionalStringValue(record.fanInSummaryArtifactId), memoryCandidateIds: readOptionalStringListValue(record.memoryCandidateIds), + fanInContextReviews: parseOptionalWorkflowFanInContextReviews(record.fanInContextReviews), + resultContextReview: parseWorkflowFanInContextReview(record.resultContextReview), + }); +} + +function parseOptionalWorkflowFanInContextReviews(value: unknown): readonly WorkflowFanInContextReview[] | undefined { + if (!Array.isArray(value)) return undefined; + const reviews: WorkflowFanInContextReview[] = []; + for (const item of value) { + const review = parseWorkflowFanInContextReview(item); + if (!review) return undefined; + reviews.push(review); + } + return reviews; +} + +function parseWorkflowFanInContextReview(value: unknown): WorkflowFanInContextReview | undefined { + if (typeof value !== "object" || value === null || Array.isArray(value)) return undefined; + const record = value as Record; + const status = record.status; + if ( + typeof record.dependencyWorkflowStepId !== "string" + || (status !== "accepted_current" + && status !== "rebased_metadata" + && status !== "rejected_stale" + && status !== "rejected_private_leak") + || typeof record.reason !== "string" + || (record.consumedProjectionVersion !== undefined && typeof record.consumedProjectionVersion !== "string") + || (record.currentProjectionVersion !== undefined && typeof record.currentProjectionVersion !== "string") + ) { + return undefined; + } + return omitUndefined({ + dependencyWorkflowStepId: record.dependencyWorkflowStepId, + status: status as WorkflowFanInContextReview["status"], + consumedProjectionVersion: readOptionalStringValue(record.consumedProjectionVersion), + currentProjectionVersion: readOptionalStringValue(record.currentProjectionVersion), + reason: record.reason, }); } diff --git a/packages/core/src/storage/sqlite/channel-workflow-store.ts b/packages/core/src/storage/sqlite/channel-workflow-store.ts index a823238c5..0a6c0159a 100644 --- a/packages/core/src/storage/sqlite/channel-workflow-store.ts +++ b/packages/core/src/storage/sqlite/channel-workflow-store.ts @@ -1,4 +1,5 @@ import type { SQLOutputValue } from "node:sqlite"; +import { isConversationContextProjectionMetadata } from "@ling-stack/contracts"; import type { CreateStateTransitionEventInput, StateTransitionEvent } from "../../channel-runtime-types.js"; import { transitionWorkflow, @@ -119,6 +120,7 @@ export class SqliteChannelWorkflowStore { expectedOutput: input.expectedOutput, riskSummary: input.riskSummary, sourceRefs: [...(input.sourceRefs ?? [])], + agentContext: input.agentContext, status: input.status ?? "proposed", validation: input.validation, createdBy: input.createdBy, @@ -146,12 +148,13 @@ export class SqliteChannelWorkflowStore { expected_output, risk_summary, source_refs_json, + agent_context_json, status, validation_json, created_by, created_at, updated_at - ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) `) .run( intent.id, @@ -171,6 +174,7 @@ export class SqliteChannelWorkflowStore { intent.expectedOutput, intent.riskSummary ?? null, stringifyJson(intent.sourceRefs), + intent.agentContext ? stringifyJson(intent.agentContext) : null, intent.status, intent.validation ? stringifyJson(intent.validation) : null, intent.createdBy, @@ -200,6 +204,10 @@ export class SqliteChannelWorkflowStore { clauses.push("workflow_id = ?"); params.push(options.workflowId); } + if (options.channelHandoffId) { + clauses.push("channel_handoff_id = ?"); + params.push(options.channelHandoffId); + } if (options.sourceWorkflowStepId) { clauses.push("source_workflow_step_id = ?"); params.push(options.sourceWorkflowStepId); @@ -231,6 +239,9 @@ export class SqliteChannelWorkflowStore { } const sourceRefs = input.sourceRefs ? [...input.sourceRefs] : existing.sourceRefs; this.ports.assertWorkflowPacketSourceRefsInChannel(existing.channelId, sourceRefs); + const agentContext = input.agentContext === null + ? undefined + : input.agentContext ?? existing.agentContext; const next: WorkflowHandoffIntent = { ...existing, channelHandoffId, @@ -240,6 +251,7 @@ export class SqliteChannelWorkflowStore { expectedOutput: input.expectedOutput ?? existing.expectedOutput, riskSummary: nullableUpdate(input.riskSummary, existing.riskSummary), sourceRefs, + agentContext, status: input.status ?? existing.status, validation: nullableUpdate(input.validation, existing.validation), updatedAt: now, @@ -256,6 +268,7 @@ export class SqliteChannelWorkflowStore { expected_output = ?, risk_summary = ?, source_refs_json = ?, + agent_context_json = ?, status = ?, validation_json = ?, updated_at = ? @@ -269,6 +282,7 @@ export class SqliteChannelWorkflowStore { next.expectedOutput, next.riskSummary ?? null, stringifyJson(next.sourceRefs), + next.agentContext ? stringifyJson(next.agentContext) : null, next.status, next.validation ? stringifyJson(next.validation) : null, next.updatedAt, @@ -892,6 +906,20 @@ function validateWorkflowHandoffIntent(intent: WorkflowHandoffIntent): void { throw new Error("workflow handoff intent createdBy is not supported"); } validateWorkflowPacketSourceRefs(intent.sourceRefs); + if (intent.agentContext) { + validateNonEmptyString(intent.agentContext.sourceItemId, "workflow handoff agent context source item id"); + if (intent.agentContext.consumerTeammateId !== undefined) { + validateNonEmptyString(intent.agentContext.consumerTeammateId, "workflow handoff agent context consumer teammate id"); + } + if ( + intent.agentContext.schemaVersion !== 1 + || !isConversationContextProjectionMetadata(intent.agentContext.contextProjection) + ) { + throw new Error("workflow handoff agent context projection is invalid"); + } + validateWorkflowPacketSourceRefs(intent.agentContext.roleEvidenceRefs); + validateNonEmptyString(intent.agentContext.roleEvidenceFingerprint, "workflow handoff agent context role evidence fingerprint"); + } if (intent.validation) { const validation = intent.validation; if (!Array.isArray(validation.blockers)) { diff --git a/packages/core/src/storage/sqlite/migrations.ts b/packages/core/src/storage/sqlite/migrations.ts index 37e5236e3..a18b4c4f7 100644 --- a/packages/core/src/storage/sqlite/migrations.ts +++ b/packages/core/src/storage/sqlite/migrations.ts @@ -224,6 +224,8 @@ const TEAMMATE_MODEL_BINDING_MIGRATION_NAME = "add_teammate_model_binding_mode"; const REMOTE_UNREAD_ENTITY_MIGRATION_VERSION = 109; const REMOTE_UNREAD_ENTITY_MIGRATION_NAME = "deduplicate_remote_run_unread_entities"; const AI_EVOLUTION_SKILL_LIFECYCLE_MIGRATION_NAME = "repair_ai_evolution_skill_lifecycle"; +const WORKFLOW_AGENT_CONTEXT_MIGRATION_VERSION = 112; +const WORKFLOW_AGENT_CONTEXT_MIGRATION_NAME = "add_workflow_agent_context_stamp"; export function configureSqliteRunStoreDatabase(database: DatabaseSync): void { database.exec(` @@ -3560,6 +3562,23 @@ export function applySqliteRunStoreMigrations(context: SqliteRunStoreMigrationCo } } + const hasWorkflowAgentContextMigration = context.hasMigration(WORKFLOW_AGENT_CONTEXT_MIGRATION_VERSION); + if ( + !hasWorkflowAgentContextMigration + || !context.hasColumn("workflow_handoff_intents", "agent_context_json") + ) { + if (!context.hasColumn("workflow_handoff_intents", "agent_context_json")) { + database.exec("ALTER TABLE workflow_handoff_intents ADD COLUMN agent_context_json TEXT;"); + } + if (!hasWorkflowAgentContextMigration) { + context.recordMigration(WORKFLOW_AGENT_CONTEXT_MIGRATION_VERSION, WORKFLOW_AGENT_CONTEXT_MIGRATION_NAME); + } + } + database.exec(` + CREATE INDEX IF NOT EXISTS idx_workflow_handoff_intents_channel_handoff + ON workflow_handoff_intents(channel_handoff_id, updated_at); + `); + const hasChannelAutoReplyRouterMigration = context.hasMigration(CHANNEL_AUTO_REPLY_ROUTER_MIGRATION_VERSION); if ( !hasChannelAutoReplyRouterMigration || diff --git a/packages/core/src/storage/sqlite/select-sql.ts b/packages/core/src/storage/sqlite/select-sql.ts index 6398ba1ef..82c95d3d5 100644 --- a/packages/core/src/storage/sqlite/select-sql.ts +++ b/packages/core/src/storage/sqlite/select-sql.ts @@ -1301,6 +1301,7 @@ export function workflowHandoffIntentSelectSql(whereClause: string): string { expected_output, risk_summary, source_refs_json, + agent_context_json, status, validation_json, created_by, diff --git a/packages/core/src/workspace-types.ts b/packages/core/src/workspace-types.ts index 3487426df..002e0b7dc 100644 --- a/packages/core/src/workspace-types.ts +++ b/packages/core/src/workspace-types.ts @@ -2008,6 +2008,29 @@ export interface WorkflowPacketSourceRef { readonly label?: string; } +export interface WorkflowAgentContextStamp { + readonly schemaVersion: 1; + readonly sourceItemId: string; + readonly consumerTeammateId?: string; + readonly contextProjection: ConversationContextProjectionMetadata; + readonly roleEvidenceRefs: readonly WorkflowPacketSourceRef[]; + readonly roleEvidenceFingerprint: string; +} + +export type WorkflowFanInContextReviewStatus = + | "accepted_current" + | "rebased_metadata" + | "rejected_stale" + | "rejected_private_leak"; + +export interface WorkflowFanInContextReview { + readonly dependencyWorkflowStepId: string; + readonly status: WorkflowFanInContextReviewStatus; + readonly consumedProjectionVersion?: string; + readonly currentProjectionVersion?: string; + readonly reason: string; +} + export type WorkflowHandoffIntentStatus = | "proposed" | "validated" @@ -2056,6 +2079,7 @@ export interface WorkflowHandoffIntent { readonly expectedOutput: string; readonly riskSummary?: string; readonly sourceRefs: readonly WorkflowPacketSourceRef[]; + readonly agentContext?: WorkflowAgentContextStamp; readonly status: WorkflowHandoffIntentStatus; readonly validation?: WorkflowHandoffValidation; readonly createdBy: WorkflowHandoffIntentCreatedBy; @@ -2079,6 +2103,8 @@ export interface CreateWorkflowHandoffIntentInput { readonly expectedOutput: string; readonly riskSummary?: string; readonly sourceRefs?: readonly WorkflowPacketSourceRef[]; + /** Internal-only continuity stamp derived by the daemon from the source step. */ + readonly agentContext?: WorkflowAgentContextStamp; readonly status?: WorkflowHandoffIntentStatus; readonly validation?: WorkflowHandoffValidation; readonly createdBy: WorkflowHandoffIntentCreatedBy; @@ -2092,6 +2118,8 @@ export interface UpdateWorkflowHandoffIntentInput { readonly expectedOutput?: string; readonly riskSummary?: string | null; readonly sourceRefs?: readonly WorkflowPacketSourceRef[]; + /** Internal-only refreshed continuity stamp. Public PATCH routes must not accept this field. */ + readonly agentContext?: WorkflowAgentContextStamp | null; readonly status?: WorkflowHandoffIntentStatus; readonly validation?: WorkflowHandoffValidation | null; } @@ -2099,6 +2127,7 @@ export interface UpdateWorkflowHandoffIntentInput { export interface ListWorkflowHandoffIntentsOptions { readonly channelId?: string; readonly workflowId?: string; + readonly channelHandoffId?: string; readonly sourceWorkflowStepId?: string; readonly sourceRunId?: string; readonly status?: WorkflowHandoffIntentStatus; @@ -2223,6 +2252,7 @@ export interface ChannelWorkflowStepContextPacket { readonly dependencies: readonly WorkflowStepOutputRef[]; readonly priorOutputArtifactIds: readonly string[]; readonly sourceRefs?: readonly WorkflowPacketSourceRef[]; + readonly agentContext?: WorkflowAgentContextStamp; readonly selectorDecisionIds?: readonly string[]; readonly handoffIntentIds?: readonly string[]; readonly requestedRunnerId?: string; @@ -2265,6 +2295,7 @@ export interface WorkflowStepContextPackSnapshot { readonly label?: string; }[]; readonly contextProjection?: ConversationContextProjectionMetadata; + readonly agentContext?: WorkflowAgentContextStamp; } export interface ChannelWorkflowStepOutputPacket { @@ -2283,6 +2314,7 @@ export interface ChannelWorkflowStepOutputPacket { readonly blockers?: readonly string[]; readonly changedPaths?: readonly string[]; readonly sourceRefs?: readonly WorkflowPacketSourceRef[]; + readonly agentContext?: WorkflowAgentContextStamp; readonly selectorDecisionIds?: readonly string[]; readonly handoffIntentIds?: readonly string[]; readonly runnerAudit?: RunnerResolutionAudit; @@ -2325,6 +2357,8 @@ export interface ChannelWorkflowStepMetadata { readonly proposalStepSources?: readonly ChannelExecutionProposalStepSource[]; readonly fanInSummaryArtifactId?: string; readonly memoryCandidateIds?: readonly string[]; + readonly fanInContextReviews?: readonly WorkflowFanInContextReview[]; + readonly resultContextReview?: WorkflowFanInContextReview; } export interface ChannelWorkflowMetadata { diff --git a/packages/core/test/channel-intent-v2-migration.test.mjs b/packages/core/test/channel-intent-v2-migration.test.mjs index 87faa5d40..7d9e1c86a 100644 --- a/packages/core/test/channel-intent-v2-migration.test.mjs +++ b/packages/core/test/channel-intent-v2-migration.test.mjs @@ -81,6 +81,7 @@ test("Channel Intent v2 migration deletes v1 rows and preserves new v2 decisions function projectionFor(item, assignment) { return { projectionVersion: `projection:${item.id}:${assignment.revision}`, + sourceStateFingerprint: `source-state:${item.id}:${assignment.revision}`, threadId: item.threadId, turnId: item.turnId, segmentId: assignment.segmentId, diff --git a/packages/core/test/channel-runtime-store.test.mjs b/packages/core/test/channel-runtime-store.test.mjs index 37bfdf80f..34c47db05 100644 --- a/packages/core/test/channel-runtime-store.test.mjs +++ b/packages/core/test/channel-runtime-store.test.mjs @@ -217,6 +217,7 @@ test("sqlite channel auto reply mode and route decisions round trip", async () = assert.ok(sourceAssignment); const contextProjection = { projectionVersion: "ctxv2_route_test", + sourceStateFingerprint: "source-state-route-test", threadId: sourceItem.threadId, turnId: sourceItem.turnId, segmentId: sourceAssignment.segmentId, @@ -374,6 +375,7 @@ function contextProjectionForItem(store, itemId) { assert.ok(assignment); return { projectionVersion: `projection:${item.id}:${assignment.revision}`, + sourceStateFingerprint: `source-state:${item.id}:${assignment.revision}`, threadId: item.threadId, turnId: item.turnId, segmentId: assignment.segmentId, @@ -1530,6 +1532,56 @@ test("sqlite channel workflow migration self-heals missing workflow tables", asy } }); +test("sqlite workflow agent context migration self-heals its durable stamp column and handoff index", async () => { + const cwd = await mkdtemp(resolve(tmpdir(), "ling-stack-workflow-agent-context-heal-")); + const initialStore = createSqliteRunStore(cwd); + try { + const databasePath = initialStore.paths.databasePath; + initialStore.close(); + + const database = new DatabaseSync(databasePath); + try { + database.exec("ALTER TABLE workflow_handoff_intents DROP COLUMN agent_context_json;"); + assert.equal( + database.prepare("PRAGMA table_info(workflow_handoff_intents)").all().some((row) => row.name === "agent_context_json"), + false, + ); + } finally { + database.close(); + } + + createSqliteRunStore(cwd).close(); + const repaired = new DatabaseSync(databasePath); + try { + assert.equal( + repaired.prepare("PRAGMA table_info(workflow_handoff_intents)").all().some((row) => row.name === "agent_context_json"), + true, + ); + assert.ok(repaired.prepare( + "SELECT version FROM schema_migrations WHERE version = 112 AND name = 'add_workflow_agent_context_stamp'", + ).get()); + assert.ok(repaired.prepare( + "SELECT name FROM sqlite_master WHERE type = 'index' AND name = 'idx_workflow_handoff_intents_channel_handoff'", + ).get()); + repaired.exec("DROP INDEX idx_workflow_handoff_intents_channel_handoff;"); + } finally { + repaired.close(); + } + + createSqliteRunStore(cwd).close(); + const indexRepaired = new DatabaseSync(databasePath); + try { + assert.ok(indexRepaired.prepare( + "SELECT name FROM sqlite_master WHERE type = 'index' AND name = 'idx_workflow_handoff_intents_channel_handoff'", + ).get()); + } finally { + indexRepaired.close(); + } + } finally { + await rm(cwd, { recursive: true, force: true }); + } +}); + test("recoverStaleChannelWorkflows reconciles running steps from linked terminal runs", async () => { const cwd = await mkdtemp(resolve(tmpdir(), "ling-stack-workflow-recovery-")); const now = "2026-06-05T00:00:00.000Z"; diff --git a/packages/core/test/handoff-intents.test.mjs b/packages/core/test/handoff-intents.test.mjs index d58e6b6ca..0844acb56 100644 --- a/packages/core/test/handoff-intents.test.mjs +++ b/packages/core/test/handoff-intents.test.mjs @@ -75,6 +75,25 @@ test("normalizes workflow handoff intents and preserves explicit source refs", ( { type: "channel_message", id: "message_1", label: "custom label" }, { type: "artifact", id: "artifact_1", label: "implementation summary" }, ], + agentContext: { + schemaVersion: 1, + sourceItemId: "message_1", + consumerTeammateId: "reviewer", + contextProjection: { + projectionVersion: "ctxv2_handoff_1", + sourceStateFingerprint: "source-state-handoff-1", + threadId: "thread_1", + turnId: "turn_1", + segmentId: "segment_1", + throughItemSequence: 7, + summaryIds: ["summary_1"], + sourceRefs: [{ type: "conversation_item", id: "message_1" }], + policyVersion: "continuity-v2-projector-v1", + visibilityFingerprint: "channel-shared", + }, + roleEvidenceRefs: [{ type: "memory", id: "memory_reviewer" }], + roleEvidenceFingerprint: "role-reviewer-1", + }, status: "validated", validation: { targetTeammateExists: true, @@ -104,6 +123,7 @@ test("normalizes workflow handoff intents and preserves explicit source refs", ( ]); assert.equal(normalized.inputPacket.workflowId, "workflow_1"); assert.equal(normalized.inputPacket.sourceWorkflowStepId, "step_1"); + assert.deepEqual(normalized.inputPacket.agentContext, intent.agentContext); assert.equal(normalized.validation.targetTeammateExists, true); }); diff --git a/packages/core/test/secrets.test.mjs b/packages/core/test/secrets.test.mjs index 558c21a48..78a599de5 100644 --- a/packages/core/test/secrets.test.mjs +++ b/packages/core/test/secrets.test.mjs @@ -70,3 +70,42 @@ test("redaction can include resolved secret values", () => { nested: ["[REDACTED]"], }); }); + +test("redaction preserves shared references while stopping true cycles", () => { + const shared = { + projectionVersion: "projection-v1", + apiKey: "shared-reference-secret", + }; + const circular = { label: "root" }; + circular.self = circular; + + assert.deepEqual(redactSecrets({ + contextPack: { + contextProjection: shared, + agentContext: { contextProjection: shared }, + }, + contextProjection: shared, + circular, + }), { + contextPack: { + contextProjection: { + projectionVersion: "projection-v1", + apiKey: "[REDACTED]", + }, + agentContext: { + contextProjection: { + projectionVersion: "projection-v1", + apiKey: "[REDACTED]", + }, + }, + }, + contextProjection: { + projectionVersion: "projection-v1", + apiKey: "[REDACTED]", + }, + circular: { + label: "root", + self: "[Circular]", + }, + }); +}); diff --git a/packages/daemon/src/features/channels/agent-context-consistency.ts b/packages/daemon/src/features/channels/agent-context-consistency.ts new file mode 100644 index 000000000..3b975ec08 --- /dev/null +++ b/packages/daemon/src/features/channels/agent-context-consistency.ts @@ -0,0 +1,273 @@ +import { createHash } from "node:crypto"; +import { + isConversationContextProjectionMetadata, + type ConversationContextProjectionMetadata, +} from "@ling-stack/contracts"; +import type { + RunStore, + WorkflowAgentContextStamp, + WorkflowFanInContextReview, + WorkflowPacketSourceRef, +} from "@ling-stack/core"; +import { buildConversationContextProjection } from "./context-processing/conversation-context-projector.js"; + +export function createWorkflowAgentContextStamp(input: { + readonly store: RunStore; + readonly sourceItemId: string; + readonly contextProjection: ConversationContextProjectionMetadata; + readonly consumerTeammateId?: string; + readonly roleEvidenceRefs?: readonly WorkflowPacketSourceRef[]; +}): WorkflowAgentContextStamp { + const roleEvidenceRefs = uniqueWorkflowSourceRefs(input.roleEvidenceRefs ?? []); + return { + schemaVersion: 1, + sourceItemId: input.sourceItemId, + consumerTeammateId: input.consumerTeammateId, + contextProjection: input.contextProjection, + roleEvidenceRefs, + roleEvidenceFingerprint: fingerprintRoleEvidence({ + store: input.store, + consumerTeammateId: input.consumerTeammateId, + roleEvidenceRefs, + }), + }; +} + +export function isWorkflowAgentContextStamp(value: unknown): value is WorkflowAgentContextStamp { + if (!value || typeof value !== "object" || Array.isArray(value)) return false; + const record = value as Record; + if ( + !hasOnlyKeys(record, [ + "schemaVersion", + "sourceItemId", + "consumerTeammateId", + "contextProjection", + "roleEvidenceRefs", + "roleEvidenceFingerprint", + ]) + || + record.schemaVersion !== 1 + || typeof record.sourceItemId !== "string" + || record.sourceItemId.trim().length === 0 + || (record.consumerTeammateId !== undefined + && (typeof record.consumerTeammateId !== "string" || record.consumerTeammateId.trim().length === 0)) + || !isConversationContextProjectionMetadata(record.contextProjection) + || !Array.isArray(record.roleEvidenceRefs) + || typeof record.roleEvidenceFingerprint !== "string" + || record.roleEvidenceFingerprint.trim().length === 0 + ) { + return false; + } + return record.roleEvidenceRefs.every((ref) => { + if (!ref || typeof ref !== "object" || Array.isArray(ref)) return false; + const sourceRef = ref as Record; + return hasOnlyKeys(sourceRef, ["type", "id", "label"]) + && isWorkflowPacketSourceRefType(sourceRef.type) + && typeof sourceRef.id === "string" + && sourceRef.id.trim().length > 0 + && (sourceRef.label === undefined || typeof sourceRef.label === "string"); + }); +} + +function isWorkflowPacketSourceRefType(value: unknown): value is WorkflowPacketSourceRef["type"] { + return value === "channel_message" + || value === "channel_brief" + || value === "channel_decision" + || value === "context_summary" + || value === "task" + || value === "run" + || value === "workflow" + || value === "workflow_step" + || value === "workflow_handoff_intent" + || value === "workflow_selector_decision" + || value === "artifact" + || value === "approval" + || value === "memory"; +} + +function hasOnlyKeys(record: Record, allowedKeys: readonly string[]): boolean { + const allowed = new Set(allowedKeys); + return Object.keys(record).every((key) => allowed.has(key)); +} + +export function reviewWorkflowAgentContextStamp(input: { + readonly store: RunStore; + readonly dependencyWorkflowStepId: string; + readonly consumed: WorkflowAgentContextStamp; +}): { + readonly review: WorkflowFanInContextReview; + readonly current?: WorkflowAgentContextStamp; +} { + try { + const projection = buildConversationContextProjection({ + store: input.store, + sourceItemId: input.consumed.sourceItemId, + view: "workflow_step", + }); + const current = createWorkflowAgentContextStamp({ + store: input.store, + sourceItemId: input.consumed.sourceItemId, + contextProjection: projection.metadata, + consumerTeammateId: input.consumed.consumerTeammateId, + roleEvidenceRefs: input.consumed.roleEvidenceRefs, + }); + const consumedProjection = input.consumed.contextProjection; + const currentProjection = current.contextProjection; + if ( + consumedProjection.projectionVersion === currentProjection.projectionVersion + && consumedProjection.sourceStateFingerprint === currentProjection.sourceStateFingerprint + && sameProjectionIdentity(consumedProjection, currentProjection) + && sameSourceRefs(consumedProjection.sourceRefs, currentProjection.sourceRefs) + && input.consumed.roleEvidenceFingerprint === current.roleEvidenceFingerprint + ) { + return { + current, + review: { + dependencyWorkflowStepId: input.dependencyWorkflowStepId, + status: "accepted_current", + consumedProjectionVersion: consumedProjection.projectionVersion, + currentProjectionVersion: currentProjection.projectionVersion, + reason: "The dependency consumed the current shared projection and role evidence.", + }, + }; + } + if ( + input.consumed.roleEvidenceFingerprint === current.roleEvidenceFingerprint + && consumedProjection.sourceStateFingerprint === currentProjection.sourceStateFingerprint + && sameProjectionIdentity(consumedProjection, currentProjection) + && sameSourceRefs(consumedProjection.sourceRefs, currentProjection.sourceRefs) + ) { + return { + current, + review: { + dependencyWorkflowStepId: input.dependencyWorkflowStepId, + status: "rebased_metadata", + consumedProjectionVersion: consumedProjection.projectionVersion, + currentProjectionVersion: currentProjection.projectionVersion, + reason: "Only projection metadata changed; all consumed facts and role evidence remain current.", + }, + }; + } + return { + current, + review: { + dependencyWorkflowStepId: input.dependencyWorkflowStepId, + status: "rejected_stale", + consumedProjectionVersion: consumedProjection.projectionVersion, + currentProjectionVersion: currentProjection.projectionVersion, + reason: "The dependency consumed facts, visibility, or role evidence that no longer match the current projection.", + }, + }; + } catch (error) { + return { + review: { + dependencyWorkflowStepId: input.dependencyWorkflowStepId, + status: "rejected_stale", + consumedProjectionVersion: input.consumed.contextProjection.projectionVersion, + reason: `The current projection could not be reconstructed safely: ${error instanceof Error ? error.message : String(error)}`, + }, + }; + } +} + +export function privateRoleEvidenceValues(input: { + readonly store: RunStore; + readonly stamp: WorkflowAgentContextStamp; +}): readonly string[] { + const values: string[] = []; + for (const ref of input.stamp.roleEvidenceRefs) { + if (ref.type !== "memory") continue; + const memory = input.store.getMemoryObject(ref.id); + if (!memory || memory.sensitivity !== "private") continue; + values.push(memory.title, memory.body); + } + return uniqueStrings(values.map(normalizeLeakText).filter((value) => value.length >= 12)); +} + +export function textContainsPrivateRoleEvidence(text: string | undefined, privateValues: readonly string[]): boolean { + if (!text?.trim() || privateValues.length === 0) return false; + const normalized = normalizeLeakText(text); + return privateValues.some((value) => normalized.includes(value)); +} + +function fingerprintRoleEvidence(input: { + readonly store: RunStore; + readonly consumerTeammateId?: string; + readonly roleEvidenceRefs: readonly WorkflowPacketSourceRef[]; +}): string { + const states = input.roleEvidenceRefs.map((ref) => { + if (ref.type !== "memory") return [ref.type, ref.id]; + const memory = input.store.getMemoryObject(ref.id); + if (!memory) return [ref.type, ref.id, "missing"]; + if ( + memory.sensitivity === "private" + && (memory.scopeType !== "teammate" || memory.ownerId !== input.consumerTeammateId) + ) { + throw new Error(`Role evidence memory is private to another teammate: ${memory.id}`); + } + if (memory.sensitivity === "secret-adjacent") { + throw new Error(`Role evidence memory is secret-adjacent: ${memory.id}`); + } + return [ + ref.type, + ref.id, + memory.scopeType, + memory.ownerId, + memory.sensitivity, + memory.reviewState, + memory.enabled, + memory.title, + memory.body, + memory.tags, + memory.sourceRefs, + memory.verifyBeforeUse, + memory.updatedAt, + ]; + }); + return `role_${createHash("sha256").update(JSON.stringify(states)).digest("hex")}`; +} + +function sameProjectionIdentity( + left: WorkflowAgentContextStamp["contextProjection"], + right: WorkflowAgentContextStamp["contextProjection"], +): boolean { + return left.threadId === right.threadId + && left.turnId === right.turnId + && left.segmentId === right.segmentId + && left.throughItemSequence === right.throughItemSequence + && left.policyVersion === right.policyVersion + && left.visibilityFingerprint === right.visibilityFingerprint + && sameStrings(left.summaryIds, right.summaryIds); +} + +function sameSourceRefs( + left: WorkflowAgentContextStamp["contextProjection"]["sourceRefs"], + right: WorkflowAgentContextStamp["contextProjection"]["sourceRefs"], +): boolean { + return sameStrings( + left.map((ref) => `${ref.type}:${ref.id}`).sort(), + right.map((ref) => `${ref.type}:${ref.id}`).sort(), + ); +} + +function sameStrings(left: readonly string[], right: readonly string[]): boolean { + return left.length === right.length && left.every((value, index) => value === right[index]); +} + +function uniqueWorkflowSourceRefs(refs: readonly WorkflowPacketSourceRef[]): readonly WorkflowPacketSourceRef[] { + const seen = new Set(); + return refs.filter((ref) => { + const key = `${ref.type}:${ref.id}`; + if (seen.has(key)) return false; + seen.add(key); + return true; + }); +} + +function normalizeLeakText(value: string): string { + return value.replace(/\s+/gu, " ").trim().toLocaleLowerCase(); +} + +function uniqueStrings(values: readonly string[]): readonly string[] { + return [...new Set(values)]; +} diff --git a/packages/daemon/src/features/channels/context-packs.ts b/packages/daemon/src/features/channels/context-packs.ts index ade5fd958..d29bfd940 100644 --- a/packages/daemon/src/features/channels/context-packs.ts +++ b/packages/daemon/src/features/channels/context-packs.ts @@ -16,6 +16,7 @@ import type { ConversationContextProjectionMetadata, ConversationDomainRefType, } from "@ling-stack/contracts"; +import type { WorkflowAgentContextStamp } from "@ling-stack/core"; import { resolveAiTeammateReference } from "../teammates/identity.js"; import { formatChannelDiscussionPrompt, type ChannelDiscussionPromptContext, type ChannelDiscussionPromptTeammate } from "./ai-prompts.js"; import { @@ -49,6 +50,7 @@ import type { ChannelContextViewKind, } from "./context-processing/types.js"; import { providerRuntimeObservationForPreset } from "./provider-runtime-observations.js"; +import { createWorkflowAgentContextStamp } from "./agent-context-consistency.js"; export type ContextPackKind = | "channel_reply" @@ -121,6 +123,7 @@ export interface ContextPack { readonly estimatedInputTokens: number; readonly overBudget: boolean; readonly contextProjection?: ConversationContextProjectionMetadata; + readonly agentContext?: WorkflowAgentContextStamp; } interface BuildContextPackInput { @@ -130,6 +133,7 @@ interface BuildContextPackInput { readonly sourceRefs?: readonly ContextSourceRef[]; readonly compactionIds?: readonly string[]; readonly contextProjection?: ConversationContextProjectionMetadata; + readonly agentContext?: WorkflowAgentContextStamp; } interface ChannelReplyContextPackInput { @@ -257,6 +261,7 @@ export function buildContextPack(input: BuildContextPackInput): ContextPack { estimatedInputTokens, overBudget: estimatedInputTokens > budget.inputBudgetTokens, contextProjection: input.contextProjection, + agentContext: input.agentContext, }; } @@ -285,6 +290,7 @@ interface CompatContextView { readonly providerPreset?: ProviderPreset; readonly view: ChannelContextView; readonly legacySourceRefs: readonly ContextSourceRef[]; + readonly agentContext?: WorkflowAgentContextStamp; } function buildCompatContextView(input: { @@ -332,6 +338,7 @@ function buildContextPackFromContextView(input: CompatContextView): ContextPack sourceRefs: uniqueContextSourceRefs(input.legacySourceRefs), compactionIds: input.view.metadata.compactionIds, contextProjection: input.view.contextProjection, + agentContext: input.agentContext, }); } @@ -652,12 +659,29 @@ export function buildWorkflowStepContextPack(input: WorkflowStepContextPackInput capabilityConfidence: budget.capabilityConfidence, }, }); - return buildContextPackFromContextView({ + const pack = buildContextPackFromContextView({ kind: "workflow_step", providerPreset, view, legacySourceRefs: legacyContextSourceRefsFromContextView(view), }); + const sourceItemId = input.step.metadata?.sourceMessageId ?? input.workflow.metadata?.sourceMessageId; + const roleEvidenceWasKept = pack.sections.some((section) => section.id === "workflow_role_evidence"); + const roleEvidenceRefs = roleEvidenceWasKept + ? (input.roleEvidence?.sourceRefs ?? []) + .filter((ref): ref is ContextSourceRef & { readonly type: "memory" } => ref.type === "memory") + .map((ref) => ({ type: "memory" as const, id: ref.id })) + : []; + const agentContext = sourceItemId && pack.contextProjection + ? createWorkflowAgentContextStamp({ + store: input.store, + sourceItemId, + contextProjection: pack.contextProjection, + consumerTeammateId: input.consumerTeammateId ?? input.step.toParticipantId, + roleEvidenceRefs, + }) + : undefined; + return agentContext ? { ...pack, agentContext } : pack; } export function buildChannelProtocolContextSection(input: { 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 68f73f330..d5a7f5c52 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 @@ -53,7 +53,7 @@ export interface ConversationContextProjection { export function buildConversationContextProjection( input: BuildConversationContextProjectionInput, ): ConversationContextProjection { - const sourceItem = input.store.getConversationItem(input.sourceItemId); + const sourceItem = input.store.getEffectiveConversationItem(input.sourceItemId); if (!sourceItem?.turnId || sourceItem.status !== "committed") { throw new Error(`Conversation projection source must be a committed turn item: ${input.sourceItemId}`); } @@ -175,6 +175,32 @@ export function buildConversationContextProjection( ...items.flatMap((item) => item.domainRefs), ...items.map((item) => ({ type: "conversation_item" as const, id: item.id })), ]); + const sourceStateFingerprint = fingerprint({ + assignment: { + id: assignment.id, + revision: assignment.revision, + status: assignment.status, + evidenceRefs: assignment.evidenceRefs, + }, + summaries: summaries.map((summary) => ({ + id: summary.id, + version: summary.version, + status: inheritedSummaries.some((inherited) => inherited.id === summary.id) ? "snapshot" : summary.status, + coveredFromSequence: summary.coveredFromSequence, + coveredThroughSequence: summary.coveredThroughSequence, + sourceRefs: summary.sourceRefs, + })), + items: items.map((item) => ({ + id: item.id, + revision: item.revision, + status: item.status, + visibility: item.visibility, + visibilityOwnerId: item.visibilityOwnerId ?? null, + supersedesItemId: item.supersedesItemId ?? null, + domainRefs: item.domainRefs, + updatedAt: item.updatedAt, + })), + }); const projectionVersion = fingerprint({ policyVersion: CONVERSATION_CONTEXT_PROJECTOR_POLICY_VERSION, thread: { @@ -194,10 +220,12 @@ export function buildConversationContextProjection( ]), throughItemSequence: sourceItem.sequence, visibilityFingerprint, + sourceStateFingerprint, }); return { metadata: { projectionVersion, + sourceStateFingerprint, threadId: thread.id, turnId: sourceItem.turnId, segmentId: segment.id, diff --git a/packages/daemon/src/features/channels/context-processing/workflow-steps.ts b/packages/daemon/src/features/channels/context-processing/workflow-steps.ts index 3675f2ecf..b013e275a 100644 --- a/packages/daemon/src/features/channels/context-processing/workflow-steps.ts +++ b/packages/daemon/src/features/channels/context-processing/workflow-steps.ts @@ -8,6 +8,7 @@ import type { import { collectChannelStateSource, collectWorkflowEvidenceSources, collectWorkflowStepSource } from "./sources.js"; import { buildProjectedConversationContextSlice } from "./conversation-context-adapter.js"; import type { ChannelContextBudgetSnapshot, ChannelContextSection, ChannelContextSourceRef, ChannelContextView } from "./types.js"; +import type { ChannelContextDroppedSource, ChannelContextSource } from "./types.js"; import { buildChannelContextView } from "./views.js"; type WorkflowStepContextPackStore = RunStore; @@ -19,8 +20,17 @@ export interface WorkflowStepContextPackInput { readonly step: ChannelWorkflowStep; readonly prompt: string; readonly providerPresetId?: string; + readonly consumerTeammateId?: string; readonly workflowSteps: readonly ChannelWorkflowStep[]; readonly compactedDependencySummary?: WorkflowStepCompactedDependencySummary; + readonly roleEvidence?: WorkflowStepRoleEvidence; +} + +export interface WorkflowStepRoleEvidence { + readonly section?: ChannelContextSection; + readonly sourceRefs: readonly ChannelContextSourceRef[]; + readonly sources: readonly ChannelContextSource[]; + readonly dropped: readonly ChannelContextDroppedSource[]; } export interface WorkflowStepCompactedDependencySummary { @@ -43,7 +53,6 @@ export function buildWorkflowStepContextView(input: BuildWorkflowStepContextView channel: input.channel, sourceItemId, view: "workflow_step", - consumerTeammateId: input.step.toParticipantId, }) : undefined; const dependencySections = input.compactedDependencySummary ? [buildCompactedWorkflowDependencySection(input.compactedDependencySummary)] @@ -126,6 +135,7 @@ export function buildWorkflowStepContextView(input: BuildWorkflowStepContextView ...dependencySections, ...formatWorkflowDecisionAndHandoffSections(input), ...(projectedContext?.sections ?? []), + ...(input.roleEvidence?.section ? [input.roleEvidence.section] : []), ...(contextSummary ? [{ id: "workflow_context_summary", @@ -149,11 +159,12 @@ export function buildWorkflowStepContextView(input: BuildWorkflowStepContextView kind: "workflow_step_view", workspaceId: input.channel.workspaceId, channelId: input.channel.id, - consumerTeammateId: input.step.toParticipantId, + consumerTeammateId: input.consumerTeammateId ?? input.step.toParticipantId, sections, sourceRefs: uniqueContextSourceRefs([ ...compactedSourceRefs, ...(projectedContext?.sourceRefs ?? []), + ...(input.roleEvidence?.sourceRefs ?? []), ]), contextProjection: projectedContext?.projection.metadata, compactionIds: input.compactedDependencySummary ? [input.compactedDependencySummary.id] : undefined, @@ -166,6 +177,7 @@ export function buildWorkflowStepContextView(input: BuildWorkflowStepContextView }), collectWorkflowStepSource({ channel: input.channel, workflow: input.workflow, step: input.step }), ...(projectedContext?.sources ?? []), + ...(input.roleEvidence?.sources ?? []), ...collectWorkflowEvidenceSources({ store: input.store, channel: input.channel, @@ -173,6 +185,7 @@ export function buildWorkflowStepContextView(input: BuildWorkflowStepContextView step: input.step, }), ], + dropped: input.roleEvidence?.dropped, }); } diff --git a/packages/daemon/src/features/channels/orchestrator.ts b/packages/daemon/src/features/channels/orchestrator.ts index 0ef89fcd4..6bd2d6090 100644 --- a/packages/daemon/src/features/channels/orchestrator.ts +++ b/packages/daemon/src/features/channels/orchestrator.ts @@ -5,6 +5,7 @@ import { type RunSession, type RunStore, type SkillApprovalProfile, + type WorkflowFanInContextReview, type WorkflowStepFailurePolicy, type WorkflowStepGate, type WorkflowStepPolicy, @@ -19,7 +20,16 @@ import { } from "./workflows.js"; import { getWorkflowPlanModeBlocker } from "./plan-mode-guard.js"; import { maybeCreateWorkflowFanInEvidence } from "./workflow-fanin.js"; -import { createWorkflowStepOutputArtifact } from "./workflow-step-packets.js"; +import { + createWorkflowStepOutputArtifact, + readWorkflowAgentContextFromRunSnapshot, + readWorkflowStepContextPacketArtifact, +} from "./workflow-step-packets.js"; +import { + privateRoleEvidenceValues, + reviewWorkflowAgentContextStamp, + textContainsPrivateRoleEvidence, +} from "./agent-context-consistency.js"; import { ensureDeterministicWorkflowSelectorDecision } from "./workflow-selector.js"; import { resolveSmartSchedulingDecision, @@ -572,21 +582,25 @@ export function finalizeOrchestratedWorkflowStepRun( return step; } if (run.status === "succeeded") { - if (maybeRequestPostStepApproval(store, workflow, step, taskId, run, now)) { + const resultSummary = workflowStepRunResultSummary(store, run); + if (blockWorkflowStepForRejectedAgentContext(store, workflow, step, run, resultSummary, now)) { + // Preserve the rejected run as audit evidence, but never commit its stale or private result. + } else if (maybeRequestPostStepApproval(store, workflow, step, taskId, run, now)) { return store.getChannelWorkflowStep(workflowStepId); + } else { + store.transitionChannelWorkflowStepStatus({ + workflowStepId, + expectedStatus: "running", + event: "succeed", + actor: "runner", + reason: `run:${run.id}`, + now, + taskId, + runId: run.id, + resultSummary, + }); + completeWorkflowIfStepsFinished(store, step.workflowId, now); } - store.transitionChannelWorkflowStepStatus({ - workflowStepId, - expectedStatus: "running", - event: "succeed", - actor: "runner", - reason: `run:${run.id}`, - now, - taskId, - runId: run.id, - resultSummary: workflowStepRunResultSummary(store, run), - }); - completeWorkflowIfStepsFinished(store, step.workflowId, now); } else if (run.status === "cancelled") { applyWorkflowStepFailurePolicy({ store, @@ -634,6 +648,112 @@ export function finalizeOrchestratedWorkflowStepRun( return store.getChannelWorkflowStep(workflowStepId); } +function blockWorkflowStepForRejectedAgentContext( + store: RunStore, + workflow: ChannelWorkflow, + step: ChannelWorkflowStep, + run: RunSession, + resultSummary: string, + now: string, +): boolean { + const continuityBacked = Boolean(step.metadata?.sourceMessageId ?? workflow.metadata?.sourceMessageId); + if (!continuityBacked) return false; + const contextArtifactId = step.metadata?.contextArtifactId; + const agentContext = contextArtifactId + ? readWorkflowStepContextPacketArtifact({ store, artifactId: contextArtifactId })?.agentContext + : readWorkflowAgentContextFromRunSnapshot(run); + if (!agentContext) { + return blockWorkflowStepForContextReason( + store, + workflow, + step, + run, + "Continuity-backed result has no durable agent context stamp.", + { + dependencyWorkflowStepId: step.id, + status: "rejected_stale", + reason: "Continuity-backed result has no durable agent context stamp.", + }, + now, + ); + } + const privateValues = privateRoleEvidenceValues({ store, stamp: agentContext }); + if (textContainsPrivateRoleEvidence(resultSummary, privateValues)) { + return blockWorkflowStepForContextReason( + store, + workflow, + step, + run, + "Workflow result repeats private role evidence and cannot be committed.", + { + dependencyWorkflowStepId: step.id, + status: "rejected_private_leak", + consumedProjectionVersion: agentContext.contextProjection.projectionVersion, + reason: "Workflow result repeats private role evidence and cannot be committed.", + }, + now, + ); + } + const reviewed = reviewWorkflowAgentContextStamp({ + store, + dependencyWorkflowStepId: step.id, + consumed: agentContext, + }).review; + if (reviewed.status === "rejected_stale" || reviewed.status === "rejected_private_leak") { + return blockWorkflowStepForContextReason( + store, + workflow, + step, + run, + `Workflow result context is no longer current: ${reviewed.reason}`, + reviewed, + now, + ); + } + return false; +} + +function blockWorkflowStepForContextReason( + store: RunStore, + workflow: ChannelWorkflow, + step: ChannelWorkflowStep, + run: RunSession, + reason: string, + review: WorkflowFanInContextReview, + now: string, +): true { + const stepTransition = store.transitionChannelWorkflowStepStatus({ + workflowStepId: step.id, + expectedStatus: "running", + event: "blocked", + actor: "system", + reason, + now, + taskId: step.taskId ?? run.taskId, + runId: run.id, + blocker: reason, + }); + if (!stepTransition.ok) { + throw new HttpError(409, stepTransition.code, `Workflow result context transition failed: ${stepTransition.code}`); + } + const blockedStep = store.getChannelWorkflowStep(step.id) ?? step; + store.updateChannelWorkflowStep(step.id, { + metadata: { ...blockedStep.metadata, resultContextReview: review }, + }, now); + const workflowTransition = store.transitionChannelWorkflowStatus({ + workflowId: workflow.id, + expectedStatus: "running", + event: "step_blocked", + actor: "system", + reason, + now, + }); + if (!workflowTransition.ok) { + throw new HttpError(409, workflowTransition.code, `Workflow result context transition failed: ${workflowTransition.code}`); + } + return true; +} + function workflowStepRunResultSummary(store: RunStore, run: RunSession): string { return readRunAssistantTextSummary(store, run.id) ?? readRunArtifactResultSummary(store, run.id) ?? @@ -1052,6 +1172,24 @@ function reconcilePostStepApprovalDecision( } const runningStep = store.getChannelWorkflowStep(step.id) ?? latestStep; if (runningStep.status === "running") { + const run = runningStep.runId ? store.getRun(runningStep.runId) : undefined; + if (run && blockWorkflowStepForRejectedAgentContext( + store, + store.getChannelWorkflow(workflow.id) ?? workflow, + runningStep, + run, + workflowStepRunResultSummary(store, run), + now, + )) { + try { + attachWorkflowStepOutputArtifact(store, runningStep.id, run, now); + } catch { + // The rejected run is still linked durably even when its audit artifact cannot be written. + } + const blockedWorkflow = store.getChannelWorkflow(workflow.id) ?? workflow; + const blockedStep = store.getChannelWorkflowStep(step.id) ?? runningStep; + return { status: "blocked", workflow: blockedWorkflow, workflowStep: blockedStep, blocker: blockedStep.blocker ?? "Workflow result context changed." }; + } const succeed = store.transitionChannelWorkflowStepStatus({ workflowStepId: runningStep.id, expectedStatus: "running", diff --git a/packages/daemon/src/features/channels/terminal-report-evidence.ts b/packages/daemon/src/features/channels/terminal-report-evidence.ts index f9a42b9bb..6480a6930 100644 --- a/packages/daemon/src/features/channels/terminal-report-evidence.ts +++ b/packages/daemon/src/features/channels/terminal-report-evidence.ts @@ -3,6 +3,7 @@ import { createTerminalReportSnapshotId, validateTerminalReportEvidenceSnapshot, type ApprovalRequest, + type ChannelWorkflowStep, type ExecutionAction, type LingTurn, type RunArtifact, @@ -135,7 +136,11 @@ export function collectRunTerminalReportEvidence( const workspaceFileRefsByArtifactId = collectWorkspaceFileRefsByArtifactId(actions); const changedPaths = changedPathsFromActions(actions); const teammate = run.teammateId ? store.getAiTeammate(run.teammateId) : undefined; - const resultSummary = readRunAssistantSummary(store, run.id, budgets.summaryCharacters) ?? run.errorMessage; + const workflowStep = run.workflowStepId ? store.getChannelWorkflowStep(run.workflowStepId) : undefined; + const rejectedContext = workflowStepResultContextRejected(workflowStep); + const resultSummary = rejectedContext + ? undefined + : readRunAssistantSummary(store, run.id, budgets.summaryCharacters) ?? run.errorMessage; return finalizeSnapshot({ workspaceId: turn?.workspaceId ?? thread?.workspaceId ?? "workspace_default", @@ -145,7 +150,7 @@ export function collectRunTerminalReportEvidence( turnId: turn?.id, sourceMessageId: turn?.sourceItemId, source: { kind: "run", id: run.id }, - objective: boundedText(task?.prompt || task?.title || turn?.userInstruction || turn?.title || "Complete the run.", budgets.summaryCharacters), + objective: boundedText(workflowStep?.prompt || task?.prompt || task?.title || turn?.userInstruction || turn?.title || "Complete the run.", budgets.summaryCharacters), language: options.language ?? languageForRun(run, turn?.userInstruction ?? task?.prompt ?? task?.title) ?? "en-US", terminalState, terminalAt: run.endedAt ?? run.updatedAt, @@ -316,13 +321,14 @@ function mapWorkflowStep(item: WorkflowEvidence["steps"][number], maxCharacters: throw new Error(`Workflow step is not terminal: ${step.id} (${step.status})`); } const changedPaths = uniqueStrings(item.outputPacket?.changedPaths ?? []).map((path) => redactEvidenceText(path)); + const rejectedContext = workflowStepResultContextRejected(step); return { id: step.id, title: boundedText(step.title, maxCharacters), status, teammateId: step.toParticipantId, runId: step.runId, - resultSummary: step.resultSummary ? boundedText(step.resultSummary, maxCharacters) : item.outputPacket?.resultSummary ? boundedText(item.outputPacket.resultSummary, maxCharacters) : undefined, + resultSummary: rejectedContext ? undefined : step.resultSummary ? boundedText(step.resultSummary, maxCharacters) : item.outputPacket?.resultSummary ? boundedText(item.outputPacket.resultSummary, maxCharacters) : undefined, blocker: step.blocker ? boundedText(step.blocker, maxCharacters) : undefined, changedPaths, artifactIds: uniqueStrings([ @@ -488,7 +494,9 @@ function aggregateWorkflowParticipants( workflowStepId: item.step.id, runId: item.step.runId, status: participantStatusForStep(item.step.status), - summary: item.step.resultSummary ?? item.outputPacket?.resultSummary, + summary: workflowStepResultContextRejected(item.step) + ? undefined + : item.step.resultSummary ?? item.outputPacket?.resultSummary, sourceRefs: uniqueRefs([ sourceRefFor("workflow_step", item.step.id, item.step.title), ...(item.step.runId ? [sourceRefFor("run", item.step.runId, item.step.title)] : []), @@ -515,6 +523,11 @@ function aggregateWorkflowParticipants( return [...byAgent.values()]; } +function workflowStepResultContextRejected(step: ChannelWorkflowStep | undefined): boolean { + const status = step?.metadata?.resultContextReview?.status; + return status === "rejected_stale" || status === "rejected_private_leak"; +} + function aggregateRunParticipants( store: RunStore, runs: readonly RunSession[], diff --git a/packages/daemon/src/features/channels/workflow-coordination-routes.ts b/packages/daemon/src/features/channels/workflow-coordination-routes.ts index 7355ccb88..53c4dd23b 100644 --- a/packages/daemon/src/features/channels/workflow-coordination-routes.ts +++ b/packages/daemon/src/features/channels/workflow-coordination-routes.ts @@ -189,6 +189,7 @@ function parseCreateWorkflowHandoffIntentInput(value: unknown): CreateWorkflowHa if (!isRecord(value)) { throw new Error("Workflow handoff intent request must be a JSON object"); } + rejectServerOwnedWorkflowHandoffFields(value); return { channelId: readString(value, "channelId"), workflowId: readOptionalString(value, "workflowId"), @@ -213,6 +214,7 @@ function parseWorkflowHandoffIntentUpdateInput(value: unknown): UpdateWorkflowHa if (!isRecord(value)) { throw new Error("Workflow handoff intent update must be a JSON object"); } + rejectServerOwnedWorkflowHandoffFields(value); return { channelHandoffId: readNullableString(value, "channelHandoffId"), toParticipantLabel: readNullableString(value, "toParticipantLabel"), @@ -221,10 +223,17 @@ function parseWorkflowHandoffIntentUpdateInput(value: unknown): UpdateWorkflowHa expectedOutput: readOptionalString(value, "expectedOutput"), riskSummary: readNullableString(value, "riskSummary"), sourceRefs: readOptionalWorkflowPacketSourceRefs(value, "sourceRefs"), - status: readOptionalWorkflowHandoffIntentStatus(value, "status"), }; } +function rejectServerOwnedWorkflowHandoffFields(value: Record): void { + for (const key of ["agentContext", "status", "validation"] as const) { + if (key in value) { + throw new Error(`Workflow handoff ${key} is server-owned`); + } + } +} + function readOptionalWorkflowPacketSourceRefs(record: Record, key: string): readonly WorkflowPacketSourceRef[] | undefined { const value = record[key]; if (value === undefined) { @@ -245,17 +254,6 @@ function readOptionalWorkflowPacketSourceRefs(record: Record, k }); } -function readOptionalWorkflowHandoffIntentStatus(record: Record, key: string): WorkflowHandoffIntentStatus | undefined { - const value = record[key]; - if (value === undefined) { - return undefined; - } - if (!workflowHandoffIntentStatuses.includes(value as WorkflowHandoffIntentStatus)) { - throw new Error(`${key} must be a workflow handoff intent status`); - } - return value as WorkflowHandoffIntentStatus; -} - function readOptionalWorkflowHandoffIntentStatusQuery(url: URL, key: string): WorkflowHandoffIntentStatus | undefined { const value = url.searchParams.get(key); return value && workflowHandoffIntentStatuses.includes(value as WorkflowHandoffIntentStatus) ? value as WorkflowHandoffIntentStatus : undefined; diff --git a/packages/daemon/src/features/channels/workflow-fanin.ts b/packages/daemon/src/features/channels/workflow-fanin.ts index 267f4e020..a36d1df70 100644 --- a/packages/daemon/src/features/channels/workflow-fanin.ts +++ b/packages/daemon/src/features/channels/workflow-fanin.ts @@ -1,19 +1,82 @@ import { mkdirSync, writeFileSync } from "node:fs"; import { dirname, resolve as resolvePath } from "node:path"; import { - type ChannelStoreFacet, type ChannelWorkflow, type ChannelWorkflowStep, - type MemoryStoreFacet, - type RunExecutionStoreFacet, + type RunStore, type RunSession, type RunStorePathsFacet, type WorkflowStepOutputRef, + type WorkflowFanInContextReview, } from "@ling-stack/core"; import { normalizeArtifactPath } from "../../shared/path.js"; import { readWorkflowStepOutputPacketArtifact } from "./workflow-step-packets.js"; +import { + privateRoleEvidenceValues, + reviewWorkflowAgentContextStamp, + textContainsPrivateRoleEvidence, +} from "./agent-context-consistency.js"; + +type WorkflowFanInStore = RunStore & RunStorePathsFacet; -type WorkflowFanInStore = ChannelStoreFacet & RunExecutionStoreFacet & RunStorePathsFacet & MemoryStoreFacet; +export function reviewWorkflowFanInContext(input: { + readonly store: WorkflowFanInStore; + readonly workflow: ChannelWorkflow; + readonly step: ChannelWorkflowStep; +}): { + readonly reviews: readonly WorkflowFanInContextReview[]; + readonly blocker?: string; +} { + const workflowIsContinuityBacked = Boolean( + input.step.metadata?.sourceMessageId ?? input.workflow.metadata?.sourceMessageId, + ); + const dependencies = dependencySteps(input.store, input.workflow.id, input.step); + const reviews: WorkflowFanInContextReview[] = []; + for (const dependency of dependencies) { + const outputPacket = dependency.metadata?.outputArtifactId + ? readWorkflowStepOutputPacketArtifact({ + store: input.store, + artifactId: dependency.metadata.outputArtifactId, + }) + : undefined; + if (!outputPacket?.agentContext) { + if (workflowIsContinuityBacked) { + reviews.push({ + dependencyWorkflowStepId: dependency.id, + status: "rejected_stale", + reason: "A continuity-backed dependency has no consumed agent context stamp.", + }); + } + continue; + } + const privateValues = privateRoleEvidenceValues({ store: input.store, stamp: outputPacket.agentContext }); + if ( + textContainsPrivateRoleEvidence(outputPacket.resultSummary, privateValues) + || textContainsPrivateRoleEvidence(dependency.resultSummary, privateValues) + ) { + reviews.push({ + dependencyWorkflowStepId: dependency.id, + status: "rejected_private_leak", + consumedProjectionVersion: outputPacket.agentContext.contextProjection.projectionVersion, + reason: "The dependency result repeats private role evidence and cannot enter shared fan-in.", + }); + continue; + } + reviews.push(reviewWorkflowAgentContextStamp({ + store: input.store, + dependencyWorkflowStepId: dependency.id, + consumed: outputPacket.agentContext, + }).review); + } + const rejected = reviews.find((review) => + review.status === "rejected_stale" || review.status === "rejected_private_leak"); + return rejected + ? { + reviews, + blocker: `Workflow rejected dependency ${rejected.dependencyWorkflowStepId}: ${rejected.reason}`, + } + : { reviews }; +} export function maybeCreateWorkflowFanInEvidence(input: { readonly store: WorkflowFanInStore; @@ -99,6 +162,19 @@ function collectDependencyOutputs( }); } +function dependencySteps( + store: WorkflowFanInStore, + workflowId: string, + step: ChannelWorkflowStep, +): readonly ChannelWorkflowStep[] { + const stepsById = new Map( + store.listChannelWorkflowSteps({ workflowId, limit: 500 }).map((candidate) => [candidate.id, candidate] as const), + ); + return (step.metadata?.dependsOnStepIds ?? []) + .map((dependencyId) => stepsById.get(dependencyId)) + .filter((dependency): dependency is ChannelWorkflowStep => Boolean(dependency)); +} + function writeFanInSummaryArtifact(input: { readonly store: WorkflowFanInStore; readonly workflow: ChannelWorkflow; diff --git a/packages/daemon/src/features/channels/workflow-handoffs.ts b/packages/daemon/src/features/channels/workflow-handoffs.ts index 33c345b18..611d43e66 100644 --- a/packages/daemon/src/features/channels/workflow-handoffs.ts +++ b/packages/daemon/src/features/channels/workflow-handoffs.ts @@ -1,19 +1,35 @@ import { type CreateWorkflowHandoffIntentInput, type RunStore, + type WorkflowAgentContextStamp, type WorkflowHandoffIntent, type WorkflowHandoffValidation, } from "@ling-stack/core"; +import { + readWorkflowAgentContextFromRunSnapshot, + readWorkflowStepContextPacketArtifact, +} from "./workflow-step-packets.js"; +import { + createWorkflowAgentContextStamp, + privateRoleEvidenceValues, + reviewWorkflowAgentContextStamp, + textContainsPrivateRoleEvidence, +} from "./agent-context-consistency.js"; export function createValidatedWorkflowHandoffIntent(input: { readonly store: RunStore; readonly intent: CreateWorkflowHandoffIntentInput; readonly now: string; }): WorkflowHandoffIntent { - const validation = validateWorkflowHandoffIntentTarget(input.store, input.intent); + const resolvedContext = resolveHandoffAgentContext(input.store, input.intent); + const validation = withContextValidationBlocker( + validateWorkflowHandoffIntentTarget(input.store, input.intent), + resolvedContext.blocker, + ); const intent = input.store.createWorkflowHandoffIntent( { ...input.intent, + agentContext: resolvedContext.agentContext, status: validation.blockers.length > 0 ? "failed" : "validated", validation, }, @@ -23,6 +39,59 @@ export function createValidatedWorkflowHandoffIntent(input: { return intent; } +function resolveHandoffAgentContext( + store: RunStore, + intent: CreateWorkflowHandoffIntentInput, +): HandoffAgentContextResolution { + const sourceContext = readSourceStepAgentContext(store, intent) ?? intent.agentContext; + if (!sourceContext) { + return continuityBackedHandoff(store, intent) + ? { blocker: "Continuity-backed handoff has no source agent context stamp." } + : {}; + } + return resolveTargetHandoffAgentContext(store, intent, sourceContext); +} + +function resolveTargetHandoffAgentContext( + store: RunStore, + intent: Pick, + sourceContext: WorkflowAgentContextStamp, +): HandoffAgentContextResolution { + const sourceItem = store.getEffectiveConversationItem(sourceContext.sourceItemId); + if (!sourceItem || sourceItem.channelId !== intent.channelId) { + return { blocker: "Workflow handoff agent context does not belong to the target channel." }; + } + const privateValues = privateRoleEvidenceValues({ store, stamp: sourceContext }); + if ([intent.reason, intent.requiredInput, intent.expectedOutput, intent.riskSummary].some((value) => + textContainsPrivateRoleEvidence(value, privateValues))) { + return { blocker: "Workflow handoff text repeats private role evidence." }; + } + const reviewed = reviewWorkflowAgentContextStamp({ + store, + dependencyWorkflowStepId: intent.sourceWorkflowStepId ?? "workflow_handoff", + consumed: sourceContext, + }); + if (reviewed.review.status === "rejected_stale") { + return { blocker: `Workflow handoff source context is stale: ${reviewed.review.reason}` }; + } + const current = reviewed.current ?? sourceContext; + try { + return { + agentContext: createWorkflowAgentContextStamp({ + store, + sourceItemId: current.sourceItemId, + contextProjection: current.contextProjection, + consumerTeammateId: intent.toParticipantId, + roleEvidenceRefs: [], + }), + }; + } catch (error) { + return { + blocker: `Workflow handoff target context could not be created safely: ${error instanceof Error ? error.message : String(error)}`, + }; + } +} + export function acceptWorkflowHandoffIntent(input: { readonly store: RunStore; readonly intentId: string; @@ -34,9 +103,17 @@ export function acceptWorkflowHandoffIntent(input: { if (!existing) { throw new Error(`Workflow handoff intent not found: ${input.intentId}`); } - const validation = existing.validation ?? validateWorkflowHandoffIntentTarget(input.store, existing); + const refreshedContext = refreshAcceptedHandoffAgentContext(input.store, existing); + const validation = withContextValidationBlocker( + validateWorkflowHandoffIntentTarget(input.store, existing), + refreshedContext.blocker, + ); if (validation.blockers.length > 0) { - const failed = input.store.updateWorkflowHandoffIntent(existing.id, { status: "failed", validation }, input.now) ?? existing; + const failed = input.store.updateWorkflowHandoffIntent(existing.id, { + agentContext: refreshedContext.agentContext, + status: "failed", + validation, + }, input.now) ?? existing; attachHandoffIntentToSourceStep(input.store, failed, input.now); return { intent: failed }; } @@ -66,6 +143,7 @@ export function acceptWorkflowHandoffIntent(input: { { channelHandoffId: channelHandoff?.id, toParticipantLabel: channelHandoff?.toParticipantLabel ?? existing.toParticipantLabel, + agentContext: refreshedContext.agentContext, status: "accepted", validation, }, @@ -75,6 +153,117 @@ export function acceptWorkflowHandoffIntent(input: { return { intent: accepted }; } +export function resolveWorkflowHandoffExecutionContext(input: { + readonly store: RunStore; + readonly channelHandoffId: string; + readonly now: string; +}): { + readonly intent?: WorkflowHandoffIntent; + readonly sourceItemId?: string; + readonly blocker?: string; +} { + const linkedIntents = input.store + .listWorkflowHandoffIntents({ channelHandoffId: input.channelHandoffId, limit: 2 }); + if (linkedIntents.length === 0) return {}; + if (linkedIntents.length > 1) { + return { blocker: `Channel handoff is linked to multiple workflow handoff intents: ${input.channelHandoffId}` }; + } + const refreshed = acceptWorkflowHandoffIntent({ + store: input.store, + intentId: linkedIntents[0].id, + now: input.now, + }).intent; + if (refreshed.status !== "accepted" || !refreshed.agentContext) { + return { + intent: refreshed, + blocker: refreshed.validation?.blockers.join("; ") || "Workflow handoff context is not valid for execution.", + }; + } + return { + intent: refreshed, + sourceItemId: refreshed.agentContext.sourceItemId, + }; +} + +interface HandoffAgentContextResolution { + readonly agentContext?: WorkflowAgentContextStamp; + readonly blocker?: string; +} + +function refreshAcceptedHandoffAgentContext( + store: RunStore, + intent: WorkflowHandoffIntent, +): HandoffAgentContextResolution { + const sourceContext = readSourceStepAgentContext(store, intent); + if (sourceContext) { + const refreshed = resolveTargetHandoffAgentContext(store, intent, sourceContext); + return refreshed.blocker?.startsWith("Workflow handoff source context is stale:") + ? { + agentContext: refreshed.agentContext, + blocker: refreshed.blocker.replace( + "Workflow handoff source context is stale:", + "Workflow handoff context changed before acceptance:", + ), + } + : refreshed; + } + if (!intent.agentContext) { + return continuityBackedHandoff(store, intent) + ? { blocker: "Continuity-backed handoff has no persisted agent context stamp." } + : {}; + } + const reviewed = reviewWorkflowAgentContextStamp({ + store, + dependencyWorkflowStepId: intent.sourceWorkflowStepId ?? "workflow_handoff", + consumed: intent.agentContext, + }); + if (reviewed.review.status === "rejected_stale") { + return { + agentContext: intent.agentContext, + blocker: `Workflow handoff context changed before acceptance: ${reviewed.review.reason}`, + }; + } + return { agentContext: reviewed.current ?? intent.agentContext }; +} + +function readSourceStepAgentContext( + store: RunStore, + intent: Pick, +): WorkflowAgentContextStamp | undefined { + const step = intent.sourceWorkflowStepId + ? store.getChannelWorkflowStep(intent.sourceWorkflowStepId) + : undefined; + const contextArtifactId = step?.metadata?.contextArtifactId; + const artifactContext = contextArtifactId + ? readWorkflowStepContextPacketArtifact({ store, artifactId: contextArtifactId })?.agentContext + : undefined; + if (artifactContext) return artifactContext; + const run = step?.runId ? store.getRun(step.runId) : undefined; + return run ? readWorkflowAgentContextFromRunSnapshot(run) : undefined; +} + +function continuityBackedHandoff( + store: RunStore, + intent: Pick, +): boolean { + if (intent.sourceMessageId) return true; + const step = intent.sourceWorkflowStepId + ? store.getChannelWorkflowStep(intent.sourceWorkflowStepId) + : undefined; + const workflowId = intent.workflowId ?? step?.workflowId; + const workflow = workflowId ? store.getChannelWorkflow(workflowId) : undefined; + return Boolean(step?.metadata?.sourceMessageId ?? workflow?.metadata?.sourceMessageId); +} + +function withContextValidationBlocker( + validation: WorkflowHandoffValidation, + blocker: string | undefined, +): WorkflowHandoffValidation { + return blocker + ? { ...validation, blockers: [...new Set([...validation.blockers, blocker])] } + : validation; +} + export function validateWorkflowHandoffIntentTarget( store: RunStore, intent: Pick, diff --git a/packages/daemon/src/features/channels/workflow-step-packets.ts b/packages/daemon/src/features/channels/workflow-step-packets.ts index 0275aeff3..8af765a21 100644 --- a/packages/daemon/src/features/channels/workflow-step-packets.ts +++ b/packages/daemon/src/features/channels/workflow-step-packets.ts @@ -12,11 +12,13 @@ import { type RunSession, type RunStorePathsFacet, type WorkspaceStoreFacet, + type WorkflowAgentContextStamp, type WorkflowPacketSourceRef, type WorkflowStepOutputRef, } from "@ling-stack/core"; import { normalizeArtifactPath } from "../../shared/path.js"; import { isRecord } from "../../shared/validation.js"; +import { isWorkflowAgentContextStamp } from "./agent-context-consistency.js"; import type { WorkflowStepContextCompactionSummary } from "./context-compaction.js"; import type { WorkflowStepRunRequest } from "./workflows.js"; @@ -103,6 +105,7 @@ export function buildWorkflowStepContextPacket(input: { ...ref.artifactIds, ])), sourceRefs: buildWorkflowPacketSourceRefs(input), + agentContext: input.runRequest.contextPack?.agentContext, selectorDecisionIds: input.step.metadata?.selectorDecisionIds, handoffIntentIds: input.step.metadata?.handoffIntentIds, requestedRunnerId: runnerAudit?.requestedRunnerId ?? input.runRequest.runnerId, @@ -115,13 +118,19 @@ export function buildWorkflowStepContextPacket(input: { } export function buildWorkflowStepOutputPacket(input: { - readonly store: WorkflowStepPacketSourceStore; + readonly store: WorkflowStepPacketStore; readonly workflow: ChannelWorkflow; readonly step: ChannelWorkflowStep; readonly run: RunSession; readonly now: string; }): ChannelWorkflowStepOutputPacket { const artifacts = input.store.listArtifacts(input.run.id); + const contextPacket = input.step.metadata?.contextArtifactId + ? readWorkflowStepContextPacketArtifact({ + store: input.store, + artifactId: input.step.metadata.contextArtifactId, + }) + : undefined; const graphId = input.step.metadata?.graphId ?? input.workflow.metadata?.workflowGraphId; const graphNodeId = input.step.metadata?.graphNodeId ?? input.step.metadata?.templateStepId; return omitUndefined({ @@ -139,6 +148,7 @@ export function buildWorkflowStepOutputPacket(input: { blockers: uniqueStrings([input.step.blocker, input.run.errorMessage ?? undefined]), changedPaths: extractWorkflowStepChangedPaths({ artifacts }), sourceRefs: buildWorkflowPacketSourceRefs(input), + agentContext: contextPacket?.agentContext ?? readWorkflowAgentContextFromRunSnapshot(input.run), selectorDecisionIds: input.step.metadata?.selectorDecisionIds, handoffIntentIds: input.step.metadata?.handoffIntentIds, runnerAudit: readRunnerAudit(input.run), @@ -190,6 +200,9 @@ export function readWorkflowStepContextPacketArtifact(input: { if (!isRecord(packet) || packet.schemaVersion !== 1 || typeof packet.workflowId !== "string" || typeof packet.workflowStepId !== "string") { return undefined; } + if (packet.agentContext !== undefined && !isWorkflowAgentContextStamp(packet.agentContext)) { + return undefined; + } return packet as unknown as ChannelWorkflowStepContextPacket; } @@ -201,9 +214,23 @@ export function readWorkflowStepOutputPacketArtifact(input: { if (!isRecord(packet) || packet.schemaVersion !== 1 || typeof packet.workflowId !== "string" || typeof packet.workflowStepId !== "string") { return undefined; } + if (packet.agentContext !== undefined && !isWorkflowAgentContextStamp(packet.agentContext)) { + return undefined; + } return packet as unknown as ChannelWorkflowStepOutputPacket; } +export function readWorkflowAgentContextFromRunSnapshot( + run: Pick, +): WorkflowAgentContextStamp | undefined { + if (!isRecord(run.requestSnapshot)) return undefined; + const contextPack = run.requestSnapshot.contextPack; + if (!isRecord(contextPack)) return undefined; + return isWorkflowAgentContextStamp(contextPack.agentContext) + ? contextPack.agentContext + : undefined; +} + export function extractWorkflowStepChangedPaths(input: { readonly artifacts: readonly RunArtifact[]; }): readonly string[] { diff --git a/packages/daemon/src/features/channels/workflows.ts b/packages/daemon/src/features/channels/workflows.ts index 57e055a7a..9cf4a1d38 100644 --- a/packages/daemon/src/features/channels/workflows.ts +++ b/packages/daemon/src/features/channels/workflows.ts @@ -59,6 +59,10 @@ import { terminalReportsEnabledForStore, } from "./terminal-report-coordinator.js"; import { buildWorkflowEvidence } from "./workflow-evidence.js"; +import { renderUsedMemoryPromptForView } from "../memory/prompt.js"; +import { reviewWorkflowFanInContext } from "./workflow-fanin.js"; +import type { PreselectedMemoryUse } from "../runs/schemas.js"; +import { resolveWorkflowHandoffExecutionContext } from "./workflow-handoffs.js"; const WORKFLOW_STEP_RESOURCE_LEASE_TTL_MS = 6 * 60 * 60 * 1000; @@ -71,6 +75,8 @@ export interface WorkflowStepRunRequest { readonly workflowStepId?: string; readonly requestedSkillId?: string; readonly skipProviderPreflight?: boolean; + readonly skipMemoryRecall?: boolean; + readonly preselectedMemoryUse?: PreselectedMemoryUse; readonly idempotencyKey?: string; readonly approvalProfile?: SkillApprovalProfile; readonly task: { @@ -82,6 +88,7 @@ export interface WorkflowStepRunRequest { readonly tools?: readonly RunTool[]; readonly thinkingLevel?: "off" | "minimal" | "low" | "medium" | "high" | "xhigh"; readonly contextPack?: WorkflowStepRunContextPackSnapshot; + readonly contextProjection?: ContextPack["contextProjection"]; readonly contextCompactions?: readonly WorkflowStepContextCompactionSummary[]; } @@ -104,6 +111,7 @@ export interface WorkflowStepRunContextPackSnapshot { }[]; readonly sourceRefs: ContextPack["sourceRefs"]; readonly contextProjection?: ContextPack["contextProjection"]; + readonly agentContext?: ContextPack["agentContext"]; } export type ValidateWorkflowStepRunRequest = (runRequest: WorkflowStepRunRequest) => void; @@ -230,6 +238,16 @@ export async function handleChannelWorkflowFromHandoff( } const body = await readJsonBody(request, config.maxBodyBytes); const input = parseChannelWorkflowFromHandoffRequest(body); + const createdAt = new Date().toISOString(); + const continuityContext = resolveWorkflowHandoffExecutionContext({ + store, + channelHandoffId: handoff.id, + now: createdAt, + }); + if (continuityContext.blocker) { + throw new Error(`Workflow handoff cannot start: ${continuityContext.blocker}`); + } + const sourceItemId = continuityContext.sourceItemId ?? handoff.sourceMessageId; const workflow = store.createChannelWorkflow( { channelId: handoff.channelId, @@ -237,8 +255,9 @@ export async function handleChannelWorkflowFromHandoff( title: input.title ?? `Handoff to ${handoff.toParticipantLabel}`, objective: input.objective ?? (handoff.expectedOutput || handoff.reason || "Complete handoff workflow."), createdBy: input.createdBy, + metadata: sourceItemId ? { sourceMessageId: sourceItemId } : undefined, }, - new Date().toISOString(), + createdAt, ); const workflowStep = store.createChannelWorkflowStep( { @@ -249,8 +268,9 @@ export async function handleChannelWorkflowFromHandoff( prompt: formatHandoffStepPrompt(handoff), toParticipantId: handoff.toParticipantId, toParticipantLabel: handoff.toParticipantLabel, + metadata: sourceItemId ? { sourceMessageId: sourceItemId } : undefined, }, - new Date().toISOString(), + createdAt, ); writeJson(response, 201, { workflow, workflowStep }); } catch (error) { @@ -619,40 +639,6 @@ async function handleChannelWorkflowStepStart( }); return; } - const run = store.getRun(runId); - if (run?.status === "succeeded") { - store.transitionChannelWorkflowStepStatus({ - workflowStepId, - expectedStatus: "running", - event: "succeed", - actor: "runner", - reason: `run:${runId}`, - now: completedAt, - taskId: started.task.id, - runId, - resultSummary: "Run succeeded.", - }); - completeWorkflowIfStepsFinished(store, started.workflowId, completedAt); - return; - } - store.transitionChannelWorkflowStepStatus({ - workflowStepId, - expectedStatus: "running", - event: "fail", - actor: "runner", - reason: run?.errorMessage ?? `run:${runId}`, - now: completedAt, - taskId: started.task.id, - runId, - blocker: run?.errorMessage, - }); - store.transitionChannelWorkflowStatus({ - workflowId: started.workflowId, - event: "step_failed", - actor: "runner", - reason: run?.errorMessage ?? `run:${runId}`, - now: completedAt, - }); } export type PreparedWorkflowStepStart = @@ -741,6 +727,20 @@ export function prepareWorkflowStepStart( return fail(409, stepTransition.code, `Workflow step transition failed: ${stepTransition.code}`); } + const fanInContext = reviewWorkflowFanInContext({ store, workflow, step }); + if (fanInContext.reviews.length > 0) { + const persistedStep = store.getChannelWorkflowStep(step.id) ?? step; + store.updateChannelWorkflowStep(step.id, { + metadata: { + ...persistedStep.metadata, + fanInContextReviews: fanInContext.reviews, + }, + }, now); + } + if (fanInContext.blocker) { + return block(fanInContext.blocker); + } + const providerPresetId = input.providerPresetId ?? channel.defaultProviderPresetId; const workspace = assertChannelWorkspaceReady(channel, cwd); if (!workspace.ok) { @@ -748,7 +748,24 @@ export function prepareWorkflowStepStart( } const runnerId = input.runnerId; const taskTitle = input.taskTitle ?? step.title; + const assigneeTeammateId = input.assigneeTeammateId ?? step.toParticipantId; const workflowSteps = store.listChannelWorkflowSteps({ workflowId: workflow.id, limit: 500 }); + const memoryRecallQuery = `${workflow.objective}\n${step.title}\n${input.taskPrompt ?? step.prompt}`; + const memoryRecall = store.recallMemory({ + teammateId: assigneeTeammateId, + channelId: channel.id, + intent: "task-run", + query: memoryRecallQuery, + maxResults: 6, + }); + const renderedMemory = renderUsedMemoryPromptForView({ + memories: memoryRecall.memories, + recallPolicyVersion: memoryRecall.policyVersion, + viewKind: "workflow_step_view", + workspaceId: workflow.workspaceId, + channelId: channel.id, + consumerTeammateId: assigneeTeammateId, + }); const contextPackInput = { store, channel, @@ -756,7 +773,31 @@ export function prepareWorkflowStepStart( step, prompt: input.taskPrompt ?? step.prompt, providerPresetId, + consumerTeammateId: assigneeTeammateId, workflowSteps, + roleEvidence: { + section: renderedMemory.prompt ? { + id: "workflow_role_evidence", + role: "memory" as const, + priority: "medium" as const, + title: "workflow_role_evidence", + text: renderedMemory.prompt, + sourceRefIds: renderedMemory.memoryIds, + visibility: "prompt" as const, + quotePolicy: "influence_only" as const, + metadata: { + policyVersion: renderedMemory.policyVersion, + consumerTeammateId: assigneeTeammateId, + }, + } : undefined, + sourceRefs: renderedMemory.memoryIds.map((memoryId) => ({ + type: "memory", + id: memoryId, + label: "role-scoped memory", + })), + sources: renderedMemory.sources, + dropped: [], + }, }; const initialContextPack = buildWorkflowStepContextPack(contextPackInput); const compacted = maybeCompactWorkflowStepContext({ @@ -769,7 +810,6 @@ export function prepareWorkflowStepStart( return block(`Workflow step context exceeds provider budget (${contextPack.estimatedInputTokens}/${contextPack.budget.inputBudgetTokens} estimated input tokens).`); } const taskPrompt = renderWorkflowStepContextPackPrompt(contextPack); - const assigneeTeammateId = input.assigneeTeammateId ?? step.toParticipantId; const unavailableRequiredTools = workflowStepUnavailableRequiredTools(channel.allowedTools, step.metadata); if (unavailableRequiredTools.length > 0) { return block(`Workflow step requires tools not allowed in this channel: ${unavailableRequiredTools.join(", ")}.`); @@ -782,6 +822,14 @@ export function prepareWorkflowStepStart( teammateId: assigneeTeammateId, workflowStepId: step.id, skipProviderPreflight: input.skipProviderPreflight ?? false, + skipMemoryRecall: true, + preselectedMemoryUse: renderedMemory.memoryIds.length > 0 + ? { + memoryIds: renderedMemory.memoryIds, + recallQuery: memoryRecallQuery, + policyVersion: renderedMemory.policyVersion, + } + : undefined, idempotencyKey: input.idempotencyKey ?? `workflow-step:${step.id}`, task: { id: "pending", @@ -792,6 +840,7 @@ export function prepareWorkflowStepStart( tools: workflowStepRunTools(channel.allowedTools, step.metadata?.requiredTools), thinkingLevel: "medium", contextPack: snapshotWorkflowStepContextPack(contextPack), + contextProjection: contextPack.contextProjection, contextCompactions: compacted.compactions, }; @@ -810,7 +859,7 @@ export function prepareWorkflowStepStart( { channelId: channel.id, title: taskTitle, - prompt: taskPrompt, + prompt: input.taskPrompt ?? step.prompt, assigneeTeammateId, }, new Date().toISOString(), @@ -826,7 +875,7 @@ export function prepareWorkflowStepStart( task: { id: task.id, title: task.title, - prompt: task.prompt, + prompt: taskPrompt, channelId: task.channelId, }, }, @@ -1119,6 +1168,7 @@ function snapshotWorkflowStepContextPack(pack: ContextPack): WorkflowStepRunCont })), sourceRefs: pack.sourceRefs, contextProjection: pack.contextProjection, + agentContext: pack.agentContext, }; } diff --git a/packages/daemon/src/features/runs/orchestrator.ts b/packages/daemon/src/features/runs/orchestrator.ts index 87056ccf3..44761ad83 100644 --- a/packages/daemon/src/features/runs/orchestrator.ts +++ b/packages/daemon/src/features/runs/orchestrator.ts @@ -221,6 +221,8 @@ export async function executeRunRequest( let runtimeApprovalProfile: SkillApprovalProfile | undefined; let effectivePolicySnapshot: EffectiveRunPolicySnapshot | undefined; let approvalPolicy: ApprovalPolicy | undefined; + let durableTaskSnapshot: RunRequest["task"] = runRequest.task; + let durableRunRequest: RunRequest = { ...runRequest, preselectedMemoryUse: undefined }; let runId = ""; let createdAt = ""; let workspaceCwd = ""; @@ -372,6 +374,12 @@ export async function executeRunRequest( thread.conversationMode.mode, ); turn = ensureLingTurnForRun(store, thread, runRequest, createdAt); + durableTaskSnapshot = store.getTask(runRequest.task.id) ?? runRequest.task; + durableRunRequest = { + ...runRequest, + task: durableTaskSnapshot, + preselectedMemoryUse: undefined, + }; const workUnitOwner = resolveRunWorkUnitOwner(thread, runRequest, turn.id); const existingWorkUnit = runRequest.idempotencyKey ? store.findWorkUnitByIdempotency({ @@ -411,7 +419,7 @@ export async function executeRunRequest( createdByActor: "user", linkedMessageId: runRequest.linkedMessageId, request: { - ...runRequest, + ...durableRunRequest, permissionMode: effectivePolicySnapshot.mode, effectivePolicySnapshot, workspaceCwd, @@ -546,10 +554,10 @@ export async function executeRunRequest( providerPresetId: runRequest.providerPresetId, providerName: resolvedProvider.provider, providerModel: resolvedProvider.model, - taskSnapshot: runRequest.task, + taskSnapshot: durableTaskSnapshot, providerSnapshot: { providerPresetId: runRequest.providerPresetId, ...resolvedProvider }, requestSnapshot: { - ...runRequest, + ...durableRunRequest, permissionMode: effectivePolicySnapshot.mode, effectivePolicySnapshot, workspaceCwd, @@ -608,18 +616,25 @@ export async function executeRunRequest( }) : undefined, }; - if (memoryRecall && memoryRecall.memories.length > 0) { + const usedMemory = memoryRecall && memoryRecall.memories.length > 0 + ? { + memoryIds: memoryRecall.memories.map((item) => item.memory.id), + recallQuery: memoryRecall.request.query, + policyVersion: memoryRecall.policyVersion, + } + : runRequest.preselectedMemoryUse; + if (usedMemory && usedMemory.memoryIds.length > 0) { store.recordUsedMemoryEvent( { - memoryIds: memoryRecall.memories.map((item) => item.memory.id), + memoryIds: usedMemory.memoryIds, intent: "task-run", - recallQuery: memoryRecall.request.query, + recallQuery: usedMemory.recallQuery, teammateId: run.teammateId, channelId: run.channelId, taskId: run.taskId, runId: run.id, responseId: run.id, - policyVersion: memoryRecall.policyVersion, + policyVersion: usedMemory.policyVersion, }, createdAt, ); @@ -672,7 +687,7 @@ export async function executeRunRequest( run = createQueuedRunForConflict({ store, storeRegistry, - runRequest, + runRequest: durableRunRequest, runId, workspaceCwd: workspaceCwd || runRequest.cwd, workUnit, diff --git a/packages/daemon/src/features/runs/queue.ts b/packages/daemon/src/features/runs/queue.ts index dd748ea4e..f4d4b9e9d 100644 --- a/packages/daemon/src/features/runs/queue.ts +++ b/packages/daemon/src/features/runs/queue.ts @@ -29,6 +29,7 @@ export function createQueuedRunForConflict(input: { taskId: input.runRequest.task.id, channelId: input.channel?.id ?? input.runRequest.task.channelId, teammateId: input.runRequest.teammateId, + workflowStepId: input.runRequest.workflowStepId, selectedSkillId: input.selectedSkillId, providerPresetId: input.runRequest.providerPresetId, providerName: input.provider.provider, diff --git a/packages/daemon/src/features/runs/schemas.ts b/packages/daemon/src/features/runs/schemas.ts index b2ea8e9f2..e25f5128a 100644 --- a/packages/daemon/src/features/runs/schemas.ts +++ b/packages/daemon/src/features/runs/schemas.ts @@ -39,6 +39,12 @@ import { readString, } from "../../shared/validation.js"; +export interface PreselectedMemoryUse { + readonly memoryIds: readonly string[]; + readonly recallQuery: string; + readonly policyVersion: string; +} + export interface RunRequest extends Omit { readonly workspaceCwd?: string; readonly providerPresetId?: string; @@ -72,6 +78,8 @@ export interface RunRequest extends Omit { + const now = new Date().toISOString(); + const latestStep = store.getChannelWorkflowStep(workflowStepId); + const workflow = latestStep ? store.getChannelWorkflow(latestStep.workflowId) : undefined; + const run = store.getRun(runId); + if (!latestStep || !workflow || !run) return; + try { + const artifact = createWorkflowStepContextArtifact({ + store, + workflow, + step: latestStep, + run, + runRequest, + now, + }); + const stepWithLatestMetadata = store.getChannelWorkflowStep(workflowStepId) ?? latestStep; + store.updateChannelWorkflowStep( + workflowStepId, + { metadata: { ...stepWithLatestMetadata.metadata, contextArtifactId: artifact.id } }, + now, + ); + } catch { + // Context artifacts are audit evidence; a write failure must not orphan an already-started run. + } + }; + const createStartOrchestratedRun = (store: RunStore, cwd: string): StartOrchestratedWorkflowStepRun => { const startOrchestratedRun: StartOrchestratedWorkflowStepRun = ({ workflow, workflowStep, taskId, runRequest }) => new Promise((resolve) => { @@ -1492,28 +1524,7 @@ export async function routeRequest( onRunReady: (runId) => { const now = new Date().toISOString(); store.updateChannelWorkflowStep(workflowStep.id, { taskId, runId }, now); - const run = store.getRun(runId); - if (run) { - try { - const latestStep = store.getChannelWorkflowStep(workflowStep.id) ?? workflowStep; - const artifact = createWorkflowStepContextArtifact({ - store, - workflow, - step: latestStep, - run, - runRequest, - now, - }); - const stepWithLatestMetadata = store.getChannelWorkflowStep(workflowStep.id) ?? latestStep; - store.updateChannelWorkflowStep( - workflowStep.id, - { metadata: { ...stepWithLatestMetadata.metadata, contextArtifactId: artifact.id } }, - now, - ); - } catch { - // Context artifacts are audit evidence; a write failure must not orphan an already-started run. - } - } + attachWorkflowStepContextArtifact(store, workflowStep.id, runId, runRequest); resolveOnce(runId); }, afterRunTerminal: async (terminalRun) => { @@ -1873,7 +1884,25 @@ export async function routeRequest( secretResolver, mcpBroker, { - afterRunTerminal: maybeContinueRunInterventions, + onRunReady: (runId) => { + const store = storeRegistry.getStore(runRequest.workspaceCwd); + store.updateChannelWorkflowStep( + workflowRoute.workflowStepId, + { taskId: runRequest.task.id, runId }, + new Date().toISOString(), + ); + attachWorkflowStepContextArtifact(store, workflowRoute.workflowStepId, runId, runRequest); + }, + afterRunTerminal: async (terminalRun) => { + const store = storeRegistry.getStore(runRequest.workspaceCwd); + finalizeOrchestratedWorkflowStepRun( + store, + workflowRoute.workflowStepId, + terminalRun.taskId, + terminalRun, + ); + await maybeContinueRunInterventions(terminalRun); + }, controlPlaneEvents, startupState, computerUsePreviews, diff --git a/packages/daemon/test/agent-context-consistency.test.mjs b/packages/daemon/test/agent-context-consistency.test.mjs new file mode 100644 index 000000000..c43b923d6 --- /dev/null +++ b/packages/daemon/test/agent-context-consistency.test.mjs @@ -0,0 +1,750 @@ +import assert from "node:assert/strict"; +import { mkdtemp, readFile, rm, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { resolve } from "node:path"; +import test from "node:test"; +import { createSqliteRunStore } from "../../core/dist/index.js"; +import { reviewWorkflowAgentContextStamp } from "../dist/features/channels/agent-context-consistency.js"; +import { reviewWorkflowFanInContext } from "../dist/features/channels/workflow-fanin.js"; +import { + acceptWorkflowHandoffIntent, + createValidatedWorkflowHandoffIntent, + resolveWorkflowHandoffExecutionContext, +} from "../dist/features/channels/workflow-handoffs.js"; +import { + buildWorkflowStepContextPack, +} from "../dist/features/channels/context-packs.js"; +import { + createWorkflowStepContextArtifact, + createWorkflowStepOutputArtifact, + readWorkflowStepOutputPacketArtifact, +} from "../dist/features/channels/workflow-step-packets.js"; +import { prepareWorkflowStepStart } from "../dist/features/channels/workflows.js"; +import { finalizeOrchestratedWorkflowStepRun } from "../dist/features/channels/orchestrator.js"; + +const now = "2026-08-10T01:00:00.000Z"; + +test("workflow agents share one projection while private role memory stays owner-scoped", async () => { + const fixture = await createFixture(); + try { + const engineer = prepareStep(fixture, fixture.engineerStep.id); + const designer = prepareStep(fixture, fixture.designerStep.id); + + assert.ok(engineer.runRequest.contextPack.agentContext); + assert.ok(designer.runRequest.contextPack.agentContext); + assert.equal( + engineer.runRequest.contextPack.agentContext.contextProjection.projectionVersion, + designer.runRequest.contextPack.agentContext.contextProjection.projectionVersion, + ); + assert.equal( + engineer.runRequest.contextPack.agentContext.contextProjection.sourceStateFingerprint, + designer.runRequest.contextPack.agentContext.contextProjection.sourceStateFingerprint, + ); + assert.equal(engineer.runRequest.contextPack.agentContext.contextProjection.segmentId.length > 0, true); + assert.equal(engineer.runRequest.contextPack.agentContext.contextProjection.throughItemSequence, fixture.sourceItem.sequence); + assert.deepEqual(engineer.runRequest.contextPack.agentContext.roleEvidenceRefs, [ + { type: "memory", id: fixture.engineerMemory.id }, + ]); + assert.deepEqual(designer.runRequest.contextPack.agentContext.roleEvidenceRefs, [ + { type: "memory", id: fixture.designerMemory.id }, + ]); + assert.doesNotMatch(engineer.task.prompt, /private continuity-agent-memory evidence/u); + assert.doesNotMatch(designer.task.prompt, /private continuity-agent-memory evidence/u); + assert.match(engineer.runRequest.task.prompt, /Engineer private continuity-agent-memory evidence/u); + assert.doesNotMatch(engineer.runRequest.task.prompt, /Designer private continuity-agent-memory evidence/u); + assert.match(designer.runRequest.task.prompt, /Designer private continuity-agent-memory evidence/u); + assert.doesNotMatch(designer.runRequest.task.prompt, /Engineer private continuity-agent-memory evidence/u); + assert.equal(engineer.runRequest.skipMemoryRecall, true); + assert.deepEqual(engineer.runRequest.preselectedMemoryUse, { + memoryIds: [fixture.engineerMemory.id], + recallQuery: `${fixture.workflow.objective}\n${fixture.engineerStep.title}\n${fixture.engineerStep.prompt}`, + policyVersion: "ling-memory-v1-lexical+memory_use_isolation_v1", + }); + assert.equal(engineer.runRequest.contextProjection.projectionVersion, engineer.runRequest.contextPack.agentContext.contextProjection.projectionVersion); + + const forgedProjection = { + ...engineer.runRequest.contextPack.agentContext, + contextProjection: { + ...engineer.runRequest.contextPack.agentContext.contextProjection, + segmentId: "segment_forged_with_reused_version", + }, + }; + assert.equal(reviewWorkflowAgentContextStamp({ + store: fixture.store, + dependencyWorkflowStepId: fixture.engineerStep.id, + consumed: forgedProjection, + }).review.status, "rejected_stale"); + + const unstampedHandoff = createValidatedWorkflowHandoffIntent({ + store: fixture.store, + intent: { + channelId: fixture.channel.id, + workflowId: fixture.workflow.id, + sourceWorkflowStepId: fixture.designerStep.id, + sourceMessageId: fixture.sourceItem.id, + fromParticipantId: "teammate_builtin_designer", + fromParticipantLabel: "Designer", + toParticipantId: "teammate_builtin_engineer", + toParticipantLabel: "Engineer", + reason: "Attempt an unstamped continuity handoff.", + requiredInput: "Continuity-backed input.", + expectedOutput: "Safe result.", + createdBy: "orchestrator", + }, + now: "2026-08-10T01:00:30.000Z", + }); + assert.equal(unstampedHandoff.status, "failed"); + assert.match(unstampedHandoff.validation.blockers.join("\n"), /no source agent context stamp/u); + + const snapshotOnlyOutput = persistWorkflowOutputWithoutContextArtifact({ + ...fixture, + stepId: fixture.designerStep.id, + prepared: designer, + resultSummary: "Snapshot-backed designer result.", + }); + assert.ok(snapshotOnlyOutput.packet.agentContext); + assert.equal( + snapshotOnlyOutput.packet.agentContext.contextProjection.projectionVersion, + designer.runRequest.contextPack.agentContext.contextProjection.projectionVersion, + ); + const snapshotBackedHandoff = createValidatedWorkflowHandoffIntent({ + store: fixture.store, + intent: { + channelId: fixture.channel.id, + workflowId: fixture.workflow.id, + sourceWorkflowStepId: fixture.designerStep.id, + sourceRunId: snapshotOnlyOutput.run.id, + sourceMessageId: fixture.sourceItem.id, + fromParticipantId: "teammate_builtin_designer", + fromParticipantLabel: "Designer", + toParticipantId: "teammate_builtin_engineer", + toParticipantLabel: "Engineer", + reason: "Recover context from the durable run request.", + requiredInput: "Snapshot-backed output.", + expectedOutput: "Safe review.", + createdBy: "orchestrator", + }, + now: "2026-08-10T01:00:40.000Z", + }); + assert.equal(snapshotBackedHandoff.status, "validated"); + assert.ok(snapshotBackedHandoff.agentContext); + + const outputPath = resolve(fixture.store.paths.artifactsDir, snapshotOnlyOutput.outputArtifact.path); + const corruptPacket = JSON.parse(await readFile(outputPath, "utf8")); + corruptPacket.agentContext = { schemaVersion: 1 }; + await writeFile(outputPath, `${JSON.stringify(corruptPacket)}\n`, "utf8"); + assert.equal(readWorkflowStepOutputPacketArtifact({ + store: fixture.store, + artifactId: snapshotOnlyOutput.outputArtifact.id, + }), undefined); + corruptPacket.agentContext = { + ...designer.runRequest.contextPack.agentContext, + roleEvidenceRefs: [{ type: "unknown_ref_type", id: fixture.designerMemory.id }], + }; + await writeFile(outputPath, `${JSON.stringify(corruptPacket)}\n`, "utf8"); + assert.equal(readWorkflowStepOutputPacketArtifact({ + store: fixture.store, + artifactId: snapshotOnlyOutput.outputArtifact.id, + }), undefined); + + const oversizedRoleEvidence = buildWorkflowStepContextPack({ + store: fixture.store, + channel: fixture.channel, + workflow: fixture.workflow, + step: fixture.engineerStep, + prompt: fixture.engineerStep.prompt, + consumerTeammateId: "teammate_builtin_engineer", + workflowSteps: [fixture.engineerStep, fixture.designerStep], + roleEvidence: { + section: { + id: "workflow_role_evidence", + role: "memory", + priority: "medium", + title: "workflow_role_evidence", + text: "private-role-evidence ".repeat(2_000), + sourceRefIds: [fixture.engineerMemory.id], + visibility: "prompt", + quotePolicy: "influence_only", + }, + sourceRefs: [{ type: "memory", id: fixture.engineerMemory.id }], + sources: [], + dropped: [], + }, + }); + assert.equal(oversizedRoleEvidence.sections.some((section) => section.id === "workflow_role_evidence"), false); + assert.deepEqual(oversizedRoleEvidence.agentContext.roleEvidenceRefs, []); + + const otherChannel = fixture.store.createChannel({ + name: "Other channel", + status: "active", + boundPath: fixture.cwd, + allowedTools: ["read"], + }, "2026-08-10T01:00:45.000Z"); + const otherSource = fixture.store.createChannelMessage({ + channelId: otherChannel.id, + authorType: "local-user", + body: "Keep this channel isolated.", + }, "2026-08-10T01:00:46.000Z"); + const crossChannelHandoff = createValidatedWorkflowHandoffIntent({ + store: fixture.store, + intent: { + channelId: otherChannel.id, + sourceMessageId: otherSource.id, + fromParticipantId: "teammate_builtin_designer", + fromParticipantLabel: "Designer", + toParticipantId: "teammate_builtin_engineer", + toParticipantLabel: "Engineer", + reason: "Attempt cross-channel context reuse.", + requiredInput: "Isolated input.", + expectedOutput: "Safe result.", + agentContext: designer.runRequest.contextPack.agentContext, + createdBy: "orchestrator", + }, + now: "2026-08-10T01:00:47.000Z", + }); + assert.equal(crossChannelHandoff.status, "failed"); + assert.match(crossChannelHandoff.validation.blockers.join("\n"), /does not belong to the target channel/u); + + fixture.store.updateMemoryObject(fixture.engineerMemory.id, { + body: "Engineer private continuity-agent-memory evidence changed at the same timestamp.", + }, now); + assert.equal(reviewWorkflowAgentContextStamp({ + store: fixture.store, + dependencyWorkflowStepId: fixture.engineerStep.id, + consumed: engineer.runRequest.contextPack.agentContext, + }).review.status, "rejected_stale"); + } finally { + fixture.store.close(); + await rm(fixture.cwd, { recursive: true, force: true }); + } +}); + +test("fan-in rebases metadata-only stamps and rejects private leaks or relevant stale facts", async () => { + const fixture = await createFixture(); + try { + const prepared = prepareStep(fixture, fixture.engineerStep.id); + const currentStamp = prepared.runRequest.contextPack.agentContext; + assert.ok(currentStamp); + const metadataOnly = { + ...currentStamp, + contextProjection: { + ...currentStamp.contextProjection, + projectionVersion: "ctxv2_previous_metadata_only", + }, + }; + assert.equal(reviewWorkflowAgentContextStamp({ + store: fixture.store, + dependencyWorkflowStepId: fixture.engineerStep.id, + consumed: metadataOnly, + }).review.status, "rebased_metadata"); + + const output = persistWorkflowOutput({ + ...fixture, + stepId: fixture.engineerStep.id, + prepared, + resultSummary: "Safe implementation result.", + }); + const handoff = createValidatedWorkflowHandoffIntent({ + store: fixture.store, + intent: { + channelId: fixture.channel.id, + workflowId: fixture.workflow.id, + sourceWorkflowStepId: output.step.id, + sourceRunId: output.run.id, + sourceMessageId: fixture.sourceItem.id, + fromParticipantId: "teammate_builtin_engineer", + fromParticipantLabel: "Engineer", + toParticipantId: "teammate_builtin_designer", + toParticipantLabel: "Designer", + reason: "Review the source-linked result.", + requiredInput: "Implementation output.", + expectedOutput: "Review verdict.", + sourceRefs: [{ type: "workflow_step", id: output.step.id }], + createdBy: "orchestrator", + }, + now: "2026-08-10T01:01:40.000Z", + }); + assert.equal(handoff.agentContext.contextProjection.projectionVersion, currentStamp.contextProjection.projectionVersion); + assert.equal(handoff.agentContext.sourceItemId, fixture.sourceItem.id); + assert.equal(handoff.agentContext.consumerTeammateId, "teammate_builtin_designer"); + assert.deepEqual(handoff.agentContext.roleEvidenceRefs, []); + assert.deepEqual(fixture.store.getWorkflowHandoffIntent(handoff.id).agentContext, handoff.agentContext); + const acceptedHandoff = acceptWorkflowHandoffIntent({ + store: fixture.store, + intentId: handoff.id, + now: "2026-08-10T01:01:45.000Z", + }).intent; + assert.equal(acceptedHandoff.status, "accepted"); + assert.ok(acceptedHandoff.channelHandoffId); + for (const index of [1, 2]) { + fixture.store.createWorkflowHandoffIntent({ + channelId: fixture.channel.id, + workflowId: fixture.workflow.id, + fromParticipantId: "ling_agent", + fromParticipantLabel: "Ling Agent", + toParticipantId: "teammate_builtin_designer", + toParticipantLabel: "Designer", + reason: `Unrelated newer handoff ${index}.`, + requiredInput: "Unrelated input.", + expectedOutput: "Unrelated output.", + createdBy: "orchestrator", + }, `2026-08-10T01:01:4${6 + index}.000Z`); + } + assert.deepEqual( + fixture.store.listWorkflowHandoffIntents({ channelHandoffId: acceptedHandoff.channelHandoffId, limit: 2 }).map((intent) => intent.id), + [acceptedHandoff.id], + ); + assert.equal(resolveWorkflowHandoffExecutionContext({ + store: fixture.store, + channelHandoffId: acceptedHandoff.channelHandoffId, + now: "2026-08-10T01:01:46.000Z", + }).sourceItemId, fixture.sourceItem.id); + const leakingHandoff = createValidatedWorkflowHandoffIntent({ + store: fixture.store, + intent: { + channelId: fixture.channel.id, + workflowId: fixture.workflow.id, + sourceWorkflowStepId: output.step.id, + sourceRunId: output.run.id, + sourceMessageId: fixture.sourceItem.id, + fromParticipantId: "teammate_builtin_engineer", + fromParticipantLabel: "Engineer", + toParticipantId: "teammate_builtin_designer", + toParticipantLabel: "Designer", + reason: "Attempt to copy private evidence.", + requiredInput: fixture.engineerMemory.body, + expectedOutput: "Review verdict.", + createdBy: "orchestrator", + }, + now: "2026-08-10T01:01:50.000Z", + }); + assert.equal(leakingHandoff.status, "failed"); + assert.match(leakingHandoff.validation.blockers.join("\n"), /repeats private role evidence/u); + const patchedHandoff = createValidatedWorkflowHandoffIntent({ + store: fixture.store, + intent: { + channelId: fixture.channel.id, + workflowId: fixture.workflow.id, + sourceWorkflowStepId: output.step.id, + sourceRunId: output.run.id, + sourceMessageId: fixture.sourceItem.id, + fromParticipantId: "teammate_builtin_engineer", + fromParticipantLabel: "Engineer", + toParticipantId: "teammate_builtin_designer", + toParticipantLabel: "Designer", + reason: "Review another source-linked result.", + requiredInput: "Safe implementation output.", + expectedOutput: "Review verdict.", + createdBy: "orchestrator", + }, + now: "2026-08-10T01:01:55.000Z", + }); + assert.equal(patchedHandoff.status, "validated"); + fixture.store.updateWorkflowHandoffIntent(patchedHandoff.id, { + requiredInput: fixture.engineerMemory.body, + }, "2026-08-10T01:01:56.000Z"); + const patchedHandoffAcceptance = acceptWorkflowHandoffIntent({ + store: fixture.store, + intentId: patchedHandoff.id, + now: "2026-08-10T01:01:57.000Z", + }).intent; + assert.equal(patchedHandoffAcceptance.status, "failed"); + assert.equal(patchedHandoffAcceptance.channelHandoffId, undefined); + assert.match(patchedHandoffAcceptance.validation.blockers.join("\n"), /repeats private role evidence/u); + const fanIn = fixture.store.createChannelWorkflowStep({ + workflowId: fixture.workflow.id, + position: 3, + title: "Fan-in", + prompt: "Combine safe dependency results.", + toParticipantId: "teammate_builtin_engineer", + toParticipantLabel: "Engineer", + metadata: { + sourceMessageId: fixture.sourceItem.id, + dependsOnStepIds: [output.step.id], + dependencyMode: "all_succeeded", + fanInPolicy: { + mode: "compare_outputs", + requireAllBranchesTerminal: true, + includeFailedBranches: false, + createSummaryArtifact: true, + createMemoryCandidates: false, + memoryCandidateScope: "channel", + maxInputArtifacts: 8, + }, + }, + }, now); + assert.deepEqual( + reviewWorkflowFanInContext({ store: fixture.store, workflow: fixture.workflow, step: fanIn }).reviews.map((review) => review.status), + ["accepted_current"], + ); + + fixture.store.updateChannelWorkflowStep(output.step.id, { + resultSummary: fixture.engineerMemory.body, + }, "2026-08-10T01:02:00.000Z"); + const leaked = reviewWorkflowFanInContext({ store: fixture.store, workflow: fixture.workflow, step: fanIn }); + assert.equal(leaked.reviews[0].status, "rejected_private_leak"); + assert.match(leaked.blocker, /private role evidence/u); + + fixture.store.updateChannelWorkflowStep(output.step.id, { + resultSummary: "Safe implementation result.", + }, "2026-08-10T01:03:00.000Z"); + fixture.store.appendConversationItem({ + channelId: fixture.channel.id, + threadId: fixture.sourceItem.threadId, + turnId: fixture.sourceItem.turnId, + kind: "user_message", + status: "committed", + actorType: "local_user", + visibility: "channel", + payload: { + ...fixture.sourceItem.payload, + body: "Corrected continuity-agent-memory objective.", + }, + supersedesItemId: fixture.sourceItem.id, + idempotencyKey: "agent-context-corrected-source", + }, "2026-08-10T01:04:00.000Z"); + const stale = reviewWorkflowFanInContext({ store: fixture.store, workflow: fixture.workflow, step: fanIn }); + assert.equal(stale.reviews[0].status, "rejected_stale"); + assert.match(stale.blocker, /rejected dependency/u); + const blockedStart = prepareWorkflowStepStart( + fixture.store, + fanIn.id, + { mode: "create_task" }, + fixture.cwd, + () => {}, + ); + assert.equal(blockedStart.kind, "blocked"); + assert.equal(fixture.store.getChannelWorkflowStep(fanIn.id).status, "blocked"); + assert.equal(fixture.store.getChannelWorkflowStep(fanIn.id).metadata.fanInContextReviews[0].status, "rejected_stale"); + + const staleHandoffAcceptance = acceptWorkflowHandoffIntent({ + store: fixture.store, + intentId: handoff.id, + now: "2026-08-10T01:05:00.000Z", + }).intent; + assert.equal(staleHandoffAcceptance.status, "failed"); + assert.equal(staleHandoffAcceptance.channelHandoffId, acceptedHandoff.channelHandoffId); + assert.match(staleHandoffAcceptance.validation.blockers.join("\n"), /context changed before acceptance/u); + const staleExecution = resolveWorkflowHandoffExecutionContext({ + store: fixture.store, + channelHandoffId: acceptedHandoff.channelHandoffId, + now: "2026-08-10T01:05:01.000Z", + }); + assert.match(staleExecution.blocker, /context changed before acceptance/u); + } finally { + fixture.store.close(); + await rm(fixture.cwd, { recursive: true, force: true }); + } +}); + +test("sequential dependencies reject stale agent context before entering a downstream prompt", async () => { + const fixture = await createFixture(); + try { + const prepared = prepareStep(fixture, fixture.engineerStep.id); + const output = persistWorkflowOutput({ + ...fixture, + stepId: fixture.engineerStep.id, + prepared, + resultSummary: "Safe implementation result.", + }); + const downstream = fixture.store.createChannelWorkflowStep({ + workflowId: fixture.workflow.id, + position: 3, + title: "Sequential review", + prompt: "Review the implementation result.", + toParticipantId: "teammate_builtin_designer", + toParticipantLabel: "Designer", + metadata: { + sourceMessageId: fixture.sourceItem.id, + dependsOnStepIds: [output.step.id], + dependencyMode: "all_succeeded", + }, + }, now); + + assert.deepEqual( + reviewWorkflowFanInContext({ + store: fixture.store, + workflow: fixture.workflow, + step: downstream, + }).reviews.map((review) => review.status), + ["accepted_current"], + ); + + fixture.store.appendConversationItem({ + channelId: fixture.channel.id, + threadId: fixture.sourceItem.threadId, + turnId: fixture.sourceItem.turnId, + kind: "user_message", + status: "committed", + actorType: "local_user", + visibility: "channel", + payload: { + ...fixture.sourceItem.payload, + body: "Corrected objective before the sequential dependency is consumed.", + }, + supersedesItemId: fixture.sourceItem.id, + idempotencyKey: "sequential-dependency-context-correction", + }, "2026-08-10T01:06:00.000Z"); + + const stale = reviewWorkflowFanInContext({ + store: fixture.store, + workflow: fixture.workflow, + step: downstream, + }); + assert.equal(stale.reviews[0].status, "rejected_stale"); + assert.match(stale.blocker, /rejected dependency/u); + + const blockedStart = prepareWorkflowStepStart( + fixture.store, + downstream.id, + { mode: "create_task" }, + fixture.cwd, + () => {}, + ); + assert.equal(blockedStart.kind, "blocked"); + const blockedStep = fixture.store.getChannelWorkflowStep(downstream.id); + assert.equal(blockedStep.status, "blocked"); + assert.equal(blockedStep.metadata.fanInContextReviews[0].status, "rejected_stale"); + } finally { + fixture.store.close(); + await rm(fixture.cwd, { recursive: true, force: true }); + } +}); + +test("leaf workflow results fail closed when relevant context changes before commit", async () => { + const fixture = await createFixture(); + try { + const prepared = prepareStep(fixture, fixture.engineerStep.id); + const run = fixture.store.createRun({ + id: "run_leaf_stale_context", + cwd: fixture.cwd, + taskId: prepared.task.id, + channelId: fixture.channel.id, + teammateId: prepared.runRequest.teammateId, + workflowStepId: fixture.engineerStep.id, + providerName: "test-provider", + providerModel: "test-model", + taskSnapshot: prepared.runRequest.task, + providerSnapshot: { provider: "test-provider", model: "test-model" }, + requestSnapshot: prepared.runRequest, + createdAt: "2026-08-10T02:00:00.000Z", + }); + fixture.store.updateChannelWorkflowStep(fixture.engineerStep.id, { + taskId: prepared.task.id, + runId: run.id, + }, "2026-08-10T02:00:01.000Z"); + assert.equal(fixture.store.transitionRunStatus({ + runId: run.id, + expectedStatus: "queued", + event: "runner_started", + actor: "runner", + now: "2026-08-10T02:00:02.000Z", + startedAt: "2026-08-10T02:00:02.000Z", + }).ok, true); + assert.equal(fixture.store.transitionRunStatus({ + runId: run.id, + expectedStatus: "running", + event: "runner_succeeded", + actor: "runner", + now: "2026-08-10T02:00:03.000Z", + endedAt: "2026-08-10T02:00:03.000Z", + }).ok, true); + fixture.store.appendConversationItem({ + channelId: fixture.channel.id, + threadId: fixture.sourceItem.threadId, + turnId: fixture.sourceItem.turnId, + kind: "user_message", + status: "committed", + actorType: "local_user", + visibility: "channel", + payload: { ...fixture.sourceItem.payload, body: "A corrected objective before the leaf result commits." }, + supersedesItemId: fixture.sourceItem.id, + idempotencyKey: "leaf-result-context-correction", + }, "2026-08-10T02:00:04.000Z"); + + const finalized = finalizeOrchestratedWorkflowStepRun( + fixture.store, + fixture.engineerStep.id, + prepared.task.id, + fixture.store.getRun(run.id), + "2026-08-10T02:00:05.000Z", + ); + assert.equal(finalized.status, "blocked"); + assert.match(finalized.blocker, /no longer current/u); + assert.equal(fixture.store.getChannelWorkflow(fixture.workflow.id).status, "blocked"); + assert.equal(finalized.resultSummary, undefined); + assert.equal(finalized.metadata.resultContextReview.status, "rejected_stale"); + } finally { + fixture.store.close(); + await rm(fixture.cwd, { recursive: true, force: true }); + } +}); + +async function createFixture() { + const cwd = await mkdtemp(resolve(tmpdir(), "ling-agent-context-consistency-")); + const store = createSqliteRunStore(cwd); + const channel = store.createChannel({ + name: "Agent context consistency", + status: "active", + boundPath: cwd, + allowedTools: ["read"], + }, now); + const sourceMessage = store.createChannelMessage({ + channelId: channel.id, + authorType: "local-user", + body: "Use continuity-agent-memory evidence to complete this objective.", + }, now); + const sourceItem = store.getConversationItem(sourceMessage.id); + assert.ok(sourceItem?.turnId); + const workflow = store.createChannelWorkflow({ + channelId: channel.id, + title: "Agent consistency workflow", + objective: "Use continuity-agent-memory evidence without leaking private data.", + status: "ready", + createdBy: "local-user", + metadata: { sourceMessageId: sourceItem.id }, + }, now); + const engineerStep = createStep(store, workflow.id, sourceItem.id, { + id: "teammate_builtin_engineer", + label: "Engineer", + position: 1, + }); + const designerStep = createStep(store, workflow.id, sourceItem.id, { + id: "teammate_builtin_designer", + label: "Designer", + position: 2, + }); + const engineerMemory = createPrivateMemory(store, "teammate_builtin_engineer", "Engineer"); + const designerMemory = createPrivateMemory(store, "teammate_builtin_designer", "Designer"); + return { + cwd, + store, + channel, + sourceItem, + workflow, + engineerStep, + designerStep, + engineerMemory, + designerMemory, + }; +} + +function createStep(store, workflowId, sourceMessageId, teammate) { + return store.createChannelWorkflowStep({ + workflowId, + position: teammate.position, + title: `${teammate.label} continuity-agent-memory work`, + prompt: `Use continuity-agent-memory evidence as ${teammate.label}.`, + toParticipantId: teammate.id, + toParticipantLabel: teammate.label, + metadata: { + sourceMessageId, + requiredTools: ["read"], + }, + }, now); +} + +function createPrivateMemory(store, ownerId, label) { + return store.createMemoryObject({ + scopeType: "teammate", + ownerId, + title: `${label} continuity-agent-memory`, + body: `${label} private continuity-agent-memory evidence must remain influence-only.`, + tags: ["continuity-agent-memory"], + sensitivity: "private", + reviewState: "active", + enabled: true, + verifyBeforeUse: false, + createdBy: "test", + }, now); +} + +function prepareStep(fixture, workflowStepId) { + const result = prepareWorkflowStepStart( + fixture.store, + workflowStepId, + { mode: "create_task" }, + fixture.cwd, + () => {}, + ); + assert.equal(result.kind, "ready"); + return result; +} + +function persistWorkflowOutput(input) { + const run = input.store.createRun({ + id: `run_${input.stepId}`, + cwd: input.cwd, + taskId: input.prepared.task.id, + channelId: input.channel.id, + teammateId: input.prepared.runRequest.teammateId, + workflowStepId: input.stepId, + providerName: "test-provider", + providerModel: "test-model", + taskSnapshot: input.prepared.runRequest.task, + providerSnapshot: { provider: "test-provider", model: "test-model" }, + requestSnapshot: input.prepared.runRequest, + createdAt: "2026-08-10T01:01:00.000Z", + }); + let step = input.store.getChannelWorkflowStep(input.stepId); + const contextArtifact = createWorkflowStepContextArtifact({ + store: input.store, + workflow: input.workflow, + step, + run, + runRequest: input.prepared.runRequest, + now: "2026-08-10T01:01:00.000Z", + }); + step = input.store.updateChannelWorkflowStep(input.stepId, { + runId: run.id, + status: "succeeded", + resultSummary: input.resultSummary, + metadata: { ...step.metadata, contextArtifactId: contextArtifact.id }, + }, "2026-08-10T01:01:10.000Z"); + const outputArtifact = createWorkflowStepOutputArtifact({ + store: input.store, + workflow: input.workflow, + step, + run, + now: "2026-08-10T01:01:20.000Z", + }); + step = input.store.updateChannelWorkflowStep(input.stepId, { + metadata: { ...step.metadata, outputArtifactId: outputArtifact.id }, + }, "2026-08-10T01:01:30.000Z"); + return { run, step, contextArtifact, outputArtifact }; +} + +function persistWorkflowOutputWithoutContextArtifact(input) { + const run = input.store.createRun({ + id: `run_snapshot_${input.stepId}`, + cwd: input.cwd, + taskId: input.prepared.task.id, + channelId: input.channel.id, + teammateId: input.prepared.runRequest.teammateId, + workflowStepId: input.stepId, + providerName: "test-provider", + providerModel: "test-model", + taskSnapshot: input.prepared.runRequest.task, + providerSnapshot: { provider: "test-provider", model: "test-model" }, + requestSnapshot: input.prepared.runRequest, + createdAt: "2026-08-10T01:00:31.000Z", + }); + let step = input.store.updateChannelWorkflowStep(input.stepId, { + runId: run.id, + status: "succeeded", + resultSummary: input.resultSummary, + }, "2026-08-10T01:00:32.000Z"); + const outputArtifact = createWorkflowStepOutputArtifact({ + store: input.store, + workflow: input.workflow, + step, + run, + now: "2026-08-10T01:00:33.000Z", + }); + step = input.store.updateChannelWorkflowStep(input.stepId, { + metadata: { ...step.metadata, outputArtifactId: outputArtifact.id }, + }, "2026-08-10T01:00:34.000Z"); + const packet = readWorkflowStepOutputPacketArtifact({ store: input.store, artifactId: outputArtifact.id }); + assert.ok(packet); + return { run, step, outputArtifact, packet }; +} diff --git a/packages/daemon/test/approval-policy-run.test.mjs b/packages/daemon/test/approval-policy-run.test.mjs index 14a9d04bd..6b4873cd9 100644 --- a/packages/daemon/test/approval-policy-run.test.mjs +++ b/packages/daemon/test/approval-policy-run.test.mjs @@ -85,6 +85,8 @@ test("runtime runner settings api seeds Pi defaults and explicit runner requests assert.equal(run.requestSnapshot.runnerResolution.fallbackAllowed, true); assert.equal(run.requestSnapshot.runnerResolution.effectiveRunnerId, "pi"); assert.equal(run.requestSnapshot.runtimeSettings.defaultRunnerId, "pi"); + assert.equal(run.taskSnapshot.id, "task_workspace_default_runner"); + assert.equal(run.requestSnapshot.task.id, "task_workspace_default_runner"); } finally { await daemon.close(); await rm(cwd, { recursive: true, force: true }); diff --git a/packages/daemon/test/channel-workspace.test.mjs b/packages/daemon/test/channel-workspace.test.mjs index d425d1def..6caa6a803 100644 --- a/packages/daemon/test/channel-workspace.test.mjs +++ b/packages/daemon/test/channel-workspace.test.mjs @@ -7243,6 +7243,218 @@ test("channel handoff workflows start steps, create tasks, and expose blockers", } }); +test("manual workflow step start rejects a result after source context changes", async () => { + const cwd = await mkdtemp(resolve(tmpdir(), "ling-stack-manual-workflow-stale-context-")); + const runner = new DelayedRunner(); + 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 createWorkflowTestChannel(daemon.url, cwd, "Manual continuity workflow"); + const provider = (await createStaticChatProvider(daemon.url, cwd)).provider; + await postJson(`${daemon.url}/providers/${encodeURIComponent(provider.id)}/check?cwd=${encodeURIComponent(cwd)}`); + const store = stores.get(cwd); + assert.ok(store); + store.updateChannel(channel.id, { defaultProviderPresetId: provider.id }, "2026-08-10T03:00:00.000Z"); + const rootThread = store.resolveOrCreateConversationThread({ + surfaceKind: "channel_root", + channelId: channel.id, + title: "Manual continuity workflow", + objective: "Implement the original continuity objective.", + }, "2026-08-10T03:00:00.000Z").thread; + const sourceItem = store.appendConversationUserTurn({ + channelId: channel.id, + threadId: rootThread.id, + title: "Implement the original continuity objective.", + userInstruction: "Implement the original continuity objective.", + payload: { + body: "Implement the original continuity objective.", + format: "plain_text", + mentions: [], + attachmentIds: [], + linkPreviewIds: [], + fileRefs: [], + }, + idempotencyKey: "manual-workflow-stale-context-source", + }, "2026-08-10T03:00:00.000Z").item; + const workflow = store.createChannelWorkflow({ + channelId: channel.id, + title: "Manual continuity workflow", + objective: "Implement the original continuity objective.", + status: "ready", + createdBy: "local-user", + metadata: { sourceMessageId: sourceItem.id }, + }, "2026-08-10T03:00:01.000Z"); + const step = store.createChannelWorkflowStep({ + workflowId: workflow.id, + title: "Manual continuity step", + prompt: "Implement the original continuity objective.", + toParticipantId: "ling_agent", + toParticipantLabel: "Ling Agent", + metadata: { + sourceMessageId: sourceItem.id, + providerPresetId: provider.id, + requiredTools: ["read"], + }, + }, "2026-08-10T03:00:02.000Z"); + + const response = await fetch(`${daemon.url}/workflow-steps/${encodeURIComponent(step.id)}/start?cwd=${encodeURIComponent(cwd)}`, { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ mode: "create_and_run_task" }), + }); + assert.equal(response.ok, true); + await waitForCondition(() => runner.runs.length === 1, "manual workflow runner start"); + store.appendConversationItem({ + channelId: channel.id, + threadId: sourceItem.threadId, + turnId: sourceItem.turnId, + kind: "user_message", + status: "committed", + actorType: "local_user", + visibility: "channel", + payload: { ...sourceItem.payload, body: "Replace the objective before the manual result commits." }, + supersedesItemId: sourceItem.id, + idempotencyKey: "manual-workflow-stale-context-correction", + }, "2026-08-10T03:00:03.000Z"); + runner.resolve("This result consumed the old objective."); + await response.text(); + + const blockedStep = store.getChannelWorkflowStep(step.id); + assert.equal(blockedStep.status, "blocked"); + assert.ok(blockedStep.taskId); + assert.equal(blockedStep.resultSummary, undefined); + assert.equal(blockedStep.metadata.resultContextReview.status, "rejected_stale"); + assert.equal(store.getChannelWorkflow(workflow.id).status, "blocked"); + } finally { + await daemon.close(); + await rm(cwd, { recursive: true, force: true }); + } +}); + +test("queued workflow runs keep private role memory out of durable snapshots", async () => { + const cwd = await mkdtemp(resolve(tmpdir(), "ling-stack-queued-workflow-memory-isolation-")); + const runner = new DelayedRunner(); + 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 createWorkflowTestChannel(daemon.url, cwd, "Queued continuity workflow"); + const provider = (await createStaticChatProvider(daemon.url, cwd)).provider; + await postJson(`${daemon.url}/providers/${encodeURIComponent(provider.id)}/check?cwd=${encodeURIComponent(cwd)}`); + const store = stores.get(cwd); + assert.ok(store); + store.updateChannel(channel.id, { defaultProviderPresetId: provider.id }, "2026-08-10T03:10:00.000Z"); + + await postJson(`${daemon.url}/channels/${encodeURIComponent(channel.id)}/messages?cwd=${encodeURIComponent(cwd)}`, { + body: "Hold the channel runtime slot.", + requestAiReply: true, + providerPresetId: provider.id, + }); + await waitForCondition(() => runner.runs.length === 1, "blocking channel reply start"); + + const privateMemory = store.createMemoryObject({ + workspaceId: channel.workspaceId, + scopeType: "teammate", + ownerId: "ling_agent", + title: "Queued continuity objective", + body: "Queued continuity objective private role evidence must stay owner-scoped.", + tags: ["queued-continuity"], + sourceRefs: [], + reviewState: "active", + enabled: true, + sensitivity: "private", + verifyBeforeUse: false, + createdBy: "test", + }, "2026-08-10T03:10:01.000Z"); + const rootThread = store.resolveOrCreateConversationThread({ + surfaceKind: "channel_root", + channelId: channel.id, + title: "Queued continuity workflow", + objective: "Queued continuity objective", + }, "2026-08-10T03:10:02.000Z").thread; + const sourceItem = store.appendConversationUserTurn({ + channelId: channel.id, + threadId: rootThread.id, + title: "Queued continuity objective", + userInstruction: "Queued continuity objective", + payload: { + body: "Queued continuity objective", + format: "plain_text", + mentions: [], + attachmentIds: [], + linkPreviewIds: [], + fileRefs: [], + }, + idempotencyKey: "queued-workflow-memory-source", + }, "2026-08-10T03:10:02.000Z").item; + const workflow = store.createChannelWorkflow({ + channelId: channel.id, + title: "Queued continuity workflow", + objective: "Queued continuity objective", + status: "ready", + createdBy: "local-user", + metadata: { sourceMessageId: sourceItem.id }, + }, "2026-08-10T03:10:03.000Z"); + const step = store.createChannelWorkflowStep({ + workflowId: workflow.id, + title: "Queued continuity objective", + prompt: "Queued continuity objective", + toParticipantId: "ling_agent", + toParticipantLabel: "Ling Agent", + metadata: { + sourceMessageId: sourceItem.id, + providerPresetId: provider.id, + requiredTools: ["read"], + }, + }, "2026-08-10T03:10:04.000Z"); + + const response = await fetch(`${daemon.url}/workflow-steps/${encodeURIComponent(step.id)}/start?cwd=${encodeURIComponent(cwd)}`, { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ mode: "create_and_run_task" }), + }); + assert.equal(response.ok, true); + await response.text(); + + const queuedRun = store.listRuns({ status: "queued", limit: 20 }) + .find((run) => run.workflowStepId === step.id); + assert.ok(queuedRun); + assert.equal(queuedRun.requestSnapshot.workflowStepId, step.id); + const queuedStep = store.getChannelWorkflowStep(step.id); + assert.equal(queuedStep.status, "running"); + assert.equal(queuedStep.runId, queuedRun.id); + assert.equal(store.getChannelWorkflow(workflow.id).status, "running"); + assert.doesNotMatch(queuedRun.taskSnapshot.prompt, /private role evidence/u); + assert.doesNotMatch(queuedRun.requestSnapshot.task.prompt, /private role evidence/u); + assert.equal(queuedRun.requestSnapshot.preselectedMemoryUse, undefined); + const workUnit = store.getWorkUnit(queuedRun.workUnitId); + assert.ok(workUnit); + assert.doesNotMatch(workUnit.request.task.prompt, /private role evidence/u); + assert.equal(workUnit.request.preselectedMemoryUse, undefined); + assert.equal( + store.listUsedMemoryEvents({ runId: queuedRun.id, limit: 20 }).some((event) => event.memoryIds.includes(privateMemory.id)), + false, + ); + + runner.resolve("The blocking channel reply is complete."); + await waitForCondition(() => runner.completed, "blocking channel reply completion"); + } finally { + await daemon.close(); + await rm(cwd, { recursive: true, force: true }); + } +}); + test("workflow handoff intents validate, accept, and stay visible as workflow evidence", async () => { const cwd = await mkdtemp(resolve(tmpdir(), "ling-stack-channel-workflow-handoff-intents-")); const stores = new Map(); @@ -7291,6 +7503,18 @@ test("workflow handoff intents validate, accept, and stay visible as workflow ev assert.equal(created.handoffIntent.validation.targetTeammateExists, true); assert.equal(created.handoffIntent.validation.targetTeammateActive, true); + const bypassResponse = await fetch( + `${daemon.url}/workflow-handoff-intents/${encodeURIComponent(created.handoffIntent.id)}?cwd=${encodeURIComponent(cwd)}`, + { + method: "PATCH", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ status: "accepted" }), + }, + ); + assert.equal(bypassResponse.status, 400); + assert.match(await bypassResponse.text(), /status is server-owned/u); + assert.equal(store.getWorkflowHandoffIntent(created.handoffIntent.id).status, "validated"); + const stepAfterIntent = store.getChannelWorkflowStep(sourceStep.id); assert.ok(stepAfterIntent.metadata.handoffIntentIds.includes(created.handoffIntent.id)); @@ -7443,6 +7667,13 @@ test("autoplan-lite workflow compiles durable steps and orchestrates them throug assert.equal(compiled.workflowSteps[0].metadata.policy.onRunFailed, "block_for_user"); assert.equal(compiled.workflowSteps[4].metadata.policy.onRunFailed, "fail_workflow"); assert.equal(compiled.workflowSteps[1].metadata.dependsOnStepIds.length, 1); + const firstStepMemory = (await postJson(`${daemon.url}/memory?cwd=${encodeURIComponent(cwd)}`, { + scopeType: "teammate", + ownerId: compiled.workflowSteps[0].toParticipantId, + title: "Orchestrator test objective", + body: "Build the orchestrator test slice with source-linked evidence.", + sensitivity: "private", + })).memory; const firstTick = await postJson(`${daemon.url}/workflows/${encodeURIComponent(compiled.workflow.id)}/orchestrate?cwd=${encodeURIComponent(cwd)}`, {}); assert.equal(firstTick.status, "started"); @@ -7484,6 +7715,16 @@ test("autoplan-lite workflow compiles durable steps and orchestrates them throug assert.equal(firstRun.workflowStepId, firstStepAfterFirstTick.id); assert.equal(firstRun.requestSnapshot.workflowStepId, firstStepAfterFirstTick.id); assert.equal(firstRun.requestSnapshot.runnerResolution.requestSource, "workflow_step"); + assert.match(runner.runs[0].task.prompt, /Build the orchestrator test slice with source-linked evidence/u); + assert.doesNotMatch(firstRun.taskSnapshot.prompt, /source-linked evidence/u); + assert.doesNotMatch(firstRun.requestSnapshot.task.prompt, /source-linked evidence/u); + assert.equal(firstRun.requestSnapshot.preselectedMemoryUse, undefined); + const firstWorkUnit = await getJson(`${daemon.url}/work-units/${encodeURIComponent(firstRun.workUnitId)}?cwd=${encodeURIComponent(cwd)}`); + assert.doesNotMatch(firstWorkUnit.workUnit.request.task.prompt, /source-linked evidence/u); + assert.equal(firstWorkUnit.workUnit.request.preselectedMemoryUse, undefined); + const firstRunMemoryUse = await getJson(`${daemon.url}/memory/used?cwd=${encodeURIComponent(cwd)}&runId=${encodeURIComponent(firstRun.id)}`); + assert.deepEqual(firstRunMemoryUse.events[0].memoryIds, [firstStepMemory.id]); + assert.match(firstRunMemoryUse.events[0].policyVersion, /memory_use_isolation_v1/u); const runsForStep = await getJson(`${daemon.url}/runs?cwd=${encodeURIComponent(cwd)}&workflowStepId=${encodeURIComponent(firstStepAfterFirstTick.id)}`); assert.deepEqual(runsForStep.runs.map((run) => run.id), [firstRun.id]); const threadsForStep = await getJson(`${daemon.url}/threads?cwd=${encodeURIComponent(cwd)}&workflowStepId=${encodeURIComponent(firstStepAfterFirstTick.id)}`); diff --git a/packages/daemon/test/workflow-terminal-summary.test.mjs b/packages/daemon/test/workflow-terminal-summary.test.mjs index b35082e2c..c5631e1a7 100644 --- a/packages/daemon/test/workflow-terminal-summary.test.mjs +++ b/packages/daemon/test/workflow-terminal-summary.test.mjs @@ -9,6 +9,7 @@ import { validateTerminalReport, } from "../../core/dist/index.js"; import { finalizeOrchestratedWorkflowStepRun } from "../dist/features/channels/orchestrator.js"; +import { buildWorkflowStepContextPack } from "../dist/features/channels/context-packs.js"; import { collectWorkflowTerminalReportEvidence } from "../dist/features/channels/terminal-report-evidence.js"; test("workflow terminal reply preserves full step result text", async () => { @@ -53,6 +54,16 @@ test("workflow terminal reply preserves full step result text", async () => { status: "doing", assigneeTeammateId: "ling_agent", }, "2026-06-25T00:02:30.000Z"); + const contextPack = buildWorkflowStepContextPack({ + store, + channel, + workflow, + step, + prompt: step.prompt, + consumerTeammateId: "ling_agent", + workflowSteps: [step], + }); + assert.ok(contextPack.agentContext); const run = store.createRun({ id: "run_workflow_terminal_summary", cwd, @@ -64,7 +75,7 @@ test("workflow terminal reply preserves full step result text", async () => { providerModel: "test-model", taskSnapshot: { id: task.id, title: "Create index.html" }, providerSnapshot: { provider: "test-provider", model: "test-model" }, - requestSnapshot: { cwd }, + requestSnapshot: { cwd, contextPack }, createdAt: "2026-06-25T00:03:00.000Z", }); const longResult = `${Array.from({ length: 90 }, (_, index) => `结果片段${index}`).join(" ")} END_MARKER_完整结尾`;