From cad5d7b5b86635e889b852d8e6ecb90bd2903630 Mon Sep 17 00:00:00 2001 From: tusharbhardwaj-bk Date: Sat, 8 Aug 2026 00:37:05 +0530 Subject: [PATCH 01/22] feat(planreview): native versioned plan review in the right panel (#56) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds a native plan review that lives in the right panel beside Plannotator, behind Settings -> Beta features -> Native plan review (on by default). Migration 1009 stores one plan document per lineage with an append-only, attributed version history, a draft guarded by a revision token, and discussions. Approval sends a short acknowledgement instead of repeating the plan; feedback sends anchored blocks plus a diff of reviewer edits. Editing is Plate (MIT), lazily imported, with suggestion mode on by default. No orchestration command or event was added: everything rides the existing fork RPC seam, so upstream contracts are untouched and mobile needs no change. The Fork markers check is red for a pre-existing reason — the origin/main mirror it diffs against is behind what expbkmain has merged, so 79 files read as unmarked. This branch adds none of them. Written by Claude Opus 5 in Claude Code. --- .../settings/DesktopClientSettings.test.ts | 1 + apps/server/src/auth/rpcForkScopes.ts | 9 + .../src/mcp/toolkits/webUi/catalog.test.ts | 4 +- .../mcp/toolkits/webUi/registration.test.ts | 6 +- apps/server/src/persistence/Migrations.ts | 3 + .../Migrations/1009_PlanReviewDocuments.ts | 110 +++ .../src/persistence/PlanReviewDocuments.ts | 816 +++++++++++++++++ .../src/planreview/PlanIngestListener.ts | 120 +++ .../PlanReviewContextPolicy.test.ts | 76 ++ .../src/planreview/PlanReviewContextPolicy.ts | 74 ++ .../src/planreview/PlanReviewService.test.ts | 707 ++++++++++++++ .../src/planreview/PlanReviewService.ts | 859 +++++++++++++++++ .../src/planreview/planReviewDiff.test.ts | 90 ++ apps/server/src/planreview/planReviewDiff.ts | 190 ++++ apps/server/src/server.test.ts | 9 + apps/server/src/server.ts | 14 +- apps/server/src/ws.ts | 18 + apps/server/src/wsForkHandlers.ts | 176 +++- apps/web/package.json | 10 + apps/web/src/components/ChatView.tsx | 55 +- apps/web/src/components/RightPanelTabs.tsx | 4 + .../src/components/chat/MessagesTimeline.tsx | 16 + .../src/components/chat/ProposedPlanCard.tsx | 20 +- .../planreview/PlanReviewDiscussions.tsx | 116 +++ .../planreview/PlanReviewEditor.tsx | 230 +++++ .../components/planreview/PlanReviewPanel.tsx | 410 +++++++++ .../planreview/PlanReviewVersions.tsx | 172 ++++ .../planreview/planReviewMarkdown.ts | 29 + .../components/settings/BetaSettingsPanel.tsx | 18 + .../src/components/settings/settingsSearch.ts | 6 + apps/web/src/fork/planReviewSurface.tsx | 37 + apps/web/src/rightPanelStore.ts | 46 +- apps/web/src/state/planReview.ts | 6 + docs/operations/expbkt3-customizations.md | 39 +- docs/user/plan-review.md | 80 ++ packages/client-runtime/package.json | 4 + packages/client-runtime/src/rpc/client.ts | 2 + .../client-runtime/src/state/planReview.ts | 85 ++ packages/contracts/src/index.ts | 2 + packages/contracts/src/planReview.ts | 190 ++++ packages/contracts/src/rpcFork.ts | 97 ++ packages/contracts/src/settings.ts | 5 + packages/shared/package.json | 4 + packages/shared/src/planReview.test.ts | 236 +++++ packages/shared/src/planReview.ts | 231 +++++ pnpm-lock.yaml | 863 +++++++++++++++++- 46 files changed, 6252 insertions(+), 43 deletions(-) create mode 100644 apps/server/src/persistence/Migrations/1009_PlanReviewDocuments.ts create mode 100644 apps/server/src/persistence/PlanReviewDocuments.ts create mode 100644 apps/server/src/planreview/PlanIngestListener.ts create mode 100644 apps/server/src/planreview/PlanReviewContextPolicy.test.ts create mode 100644 apps/server/src/planreview/PlanReviewContextPolicy.ts create mode 100644 apps/server/src/planreview/PlanReviewService.test.ts create mode 100644 apps/server/src/planreview/PlanReviewService.ts create mode 100644 apps/server/src/planreview/planReviewDiff.test.ts create mode 100644 apps/server/src/planreview/planReviewDiff.ts create mode 100644 apps/web/src/components/planreview/PlanReviewDiscussions.tsx create mode 100644 apps/web/src/components/planreview/PlanReviewEditor.tsx create mode 100644 apps/web/src/components/planreview/PlanReviewPanel.tsx create mode 100644 apps/web/src/components/planreview/PlanReviewVersions.tsx create mode 100644 apps/web/src/components/planreview/planReviewMarkdown.ts create mode 100644 apps/web/src/fork/planReviewSurface.tsx create mode 100644 apps/web/src/state/planReview.ts create mode 100644 docs/user/plan-review.md create mode 100644 packages/client-runtime/src/state/planReview.ts create mode 100644 packages/contracts/src/planReview.ts create mode 100644 packages/shared/src/planReview.test.ts create mode 100644 packages/shared/src/planReview.ts diff --git a/apps/desktop/src/settings/DesktopClientSettings.test.ts b/apps/desktop/src/settings/DesktopClientSettings.test.ts index 69e45c0ee5e..c83b2ff4b83 100644 --- a/apps/desktop/src/settings/DesktopClientSettings.test.ts +++ b/apps/desktop/src/settings/DesktopClientSettings.test.ts @@ -31,6 +31,7 @@ const clientSettings: ClientSettings = { glassOpacity: 80, phaseGroupedSidebarEnabled: true, planModeEnabled: false, + nativePlanReviewEnabled: true, providerModelPreferences: {}, providerRateLimitsEnabled: true, resourceMonitorEnabled: false, diff --git a/apps/server/src/auth/rpcForkScopes.ts b/apps/server/src/auth/rpcForkScopes.ts index f294f3196b8..87570502d43 100644 --- a/apps/server/src/auth/rpcForkScopes.ts +++ b/apps/server/src/auth/rpcForkScopes.ts @@ -35,4 +35,13 @@ export const FORK_RPC_REQUIRED_SCOPES = { [WS_FORK_METHODS.usersRevokeSessions]: AuthOrchestrationOperateScope, [WS_FORK_METHODS.usersSourceControlProfileSet]: AuthOrchestrationOperateScope, [WS_FORK_METHODS.linearIssuesResolve]: AuthOrchestrationReadScope, + [WS_FORK_METHODS.planReviewGet]: AuthOrchestrationReadScope, + [WS_FORK_METHODS.planReviewList]: AuthOrchestrationReadScope, + [WS_FORK_METHODS.planReviewVersionDiff]: AuthOrchestrationReadScope, + [WS_FORK_METHODS.subscribePlanReview]: AuthOrchestrationReadScope, + [WS_FORK_METHODS.planReviewSaveDraft]: AuthOrchestrationOperateScope, + [WS_FORK_METHODS.planReviewCutVersion]: AuthOrchestrationOperateScope, + [WS_FORK_METHODS.planReviewUpsertDiscussion]: AuthOrchestrationOperateScope, + [WS_FORK_METHODS.planReviewResolveDiscussion]: AuthOrchestrationOperateScope, + [WS_FORK_METHODS.planReviewSubmit]: AuthOrchestrationOperateScope, } as const; diff --git a/apps/server/src/mcp/toolkits/webUi/catalog.test.ts b/apps/server/src/mcp/toolkits/webUi/catalog.test.ts index 2dd6271706a..702d52ebac8 100644 --- a/apps/server/src/mcp/toolkits/webUi/catalog.test.ts +++ b/apps/server/src/mcp/toolkits/webUi/catalog.test.ts @@ -37,8 +37,8 @@ const invocation = ( }); it("generates one unique virtual tool and complete schemas for every web RPC", () => { - expect(WEB_UI_VIRTUAL_TOOL_COUNT).toBe(102); - expect(WEB_UI_STREAM_TOOL_COUNT).toBe(19); + expect(WEB_UI_VIRTUAL_TOOL_COUNT).toBe(111); + expect(WEB_UI_STREAM_TOOL_COUNT).toBe(20); expect(WEB_UI_VIRTUAL_TOOL_COUNT).toBe(WsRpcGroup.requests.size); expect(new Set(WEB_UI_VIRTUAL_TOOLS.map((tool) => tool.name)).size).toBe( WEB_UI_VIRTUAL_TOOL_COUNT, diff --git a/apps/server/src/mcp/toolkits/webUi/registration.test.ts b/apps/server/src/mcp/toolkits/webUi/registration.test.ts index c775068c4d6..db164db9c1b 100644 --- a/apps/server/src/mcp/toolkits/webUi/registration.test.ts +++ b/apps/server/src/mcp/toolkits/webUi/registration.test.ts @@ -66,9 +66,9 @@ it.effect("registers four compact tools while listing the complete virtual surfa expect(listed.isError).toBe(false); expect(listed.structuredContent).toMatchObject({ ok: true, - rpcCount: 102, - streamCount: 19, - matchedCount: 102, + rpcCount: 111, + streamCount: 20, + matchedCount: 111, }); const schema = yield* withInvocation( diff --git a/apps/server/src/persistence/Migrations.ts b/apps/server/src/persistence/Migrations.ts index 5b6a1e4608f..96a34a9dff7 100644 --- a/apps/server/src/persistence/Migrations.ts +++ b/apps/server/src/persistence/Migrations.ts @@ -81,6 +81,8 @@ import Migration1007 from "./Migrations/1007_OwnershipBackfillFastPath.ts"; // block already occupies (ThreadExecutions). It registers at the next free ID in // the 1000+ lane instead; the file keeps its upstream name. import Migration1008 from "./Migrations/037_ProjectionTurnsKeysetIndex.ts"; +// T3-CUSTOM(expbkt3): native plan review documents, versions and discussions. +import Migration1009 from "./Migrations/1009_PlanReviewDocuments.ts"; /** * Migration loader with all migrations defined inline. @@ -168,6 +170,7 @@ const migrationEntries = [ [1006, "AuthSessionClientVersion", Migration1006], [1007, "OwnershipBackfillFastPath", Migration1007], [1008, "ProjectionTurnsKeysetIndex", Migration1008], + [1009, "PlanReviewDocuments", Migration1009], ] as const; export const migrationManifest = migrationEntries.map(([id, name]) => [id, name] as const); diff --git a/apps/server/src/persistence/Migrations/1009_PlanReviewDocuments.ts b/apps/server/src/persistence/Migrations/1009_PlanReviewDocuments.ts new file mode 100644 index 00000000000..6a52093a271 --- /dev/null +++ b/apps/server/src/persistence/Migrations/1009_PlanReviewDocuments.ts @@ -0,0 +1,110 @@ +// T3-CUSTOM(expbkt3): native plan review — versioned, attributed plan documents. +// +// A plan document is the durable lineage behind one proposed plan: version 1 is +// whatever the agent produced, every later revision (agent or human) appends a +// new immutable row. Versions are never mutated, so `revision` doubles as the +// anchor key for comments the way `checkpoint_diff_blobs` keys on turn counts. +import * as Effect from "effect/Effect"; +import * as SqlClient from "effect/unstable/sql/SqlClient"; + +export default Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + + yield* sql` + CREATE TABLE IF NOT EXISTS plan_documents ( + document_id TEXT PRIMARY KEY, + thread_id TEXT NOT NULL, + project_id TEXT NOT NULL, + title TEXT NOT NULL, + current_revision INTEGER NOT NULL, + status TEXT NOT NULL, + created_by_user_id TEXT, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL + ) + `; + + yield* sql` + CREATE INDEX IF NOT EXISTS idx_plan_documents_thread + ON plan_documents(thread_id, created_at) + `; + + // Append-only. Nothing in the application ever updates or deletes these rows. + yield* sql` + CREATE TABLE IF NOT EXISTS plan_document_versions ( + version_id TEXT PRIMARY KEY, + document_id TEXT NOT NULL, + revision INTEGER NOT NULL, + author_kind TEXT NOT NULL, + author_user_id TEXT, + origin TEXT NOT NULL, + content_markdown TEXT NOT NULL, + content_value_json TEXT, + source_plan_id TEXT, + summary TEXT, + created_at TEXT NOT NULL, + UNIQUE (document_id, revision) + ) + `; + + yield* sql` + CREATE INDEX IF NOT EXISTS idx_plan_document_versions_document + ON plan_document_versions(document_id, revision) + `; + + // Source plan ids arrive from projection_thread_proposed_plans; the lookup is + // how the ingest listener decides "already captured" without a table scan. + yield* sql` + CREATE INDEX IF NOT EXISTS idx_plan_document_versions_source_plan + ON plan_document_versions(source_plan_id) + `; + + // The live working copy: exactly one per document, holding pending Plate + // suggestions inline. `revision_token` is the optimistic-concurrency guard. + yield* sql` + CREATE TABLE IF NOT EXISTS plan_document_drafts ( + document_id TEXT PRIMARY KEY, + base_version_id TEXT NOT NULL, + content_value_json TEXT NOT NULL, + updated_by_user_id TEXT, + updated_at TEXT NOT NULL, + revision_token TEXT NOT NULL + ) + `; + + yield* sql` + CREATE TABLE IF NOT EXISTS plan_discussions ( + discussion_id TEXT PRIMARY KEY, + document_id TEXT NOT NULL, + anchor_version_id TEXT NOT NULL, + quoted_text TEXT NOT NULL, + is_resolved INTEGER NOT NULL DEFAULT 0, + resolved_by_user_id TEXT, + resolved_at TEXT, + created_by_user_id TEXT, + created_at TEXT NOT NULL + ) + `; + + yield* sql` + CREATE INDEX IF NOT EXISTS idx_plan_discussions_document + ON plan_discussions(document_id, created_at) + `; + + yield* sql` + CREATE TABLE IF NOT EXISTS plan_discussion_comments ( + comment_id TEXT PRIMARY KEY, + discussion_id TEXT NOT NULL, + author_user_id TEXT, + body_markdown TEXT NOT NULL, + is_edited INTEGER NOT NULL DEFAULT 0, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL + ) + `; + + yield* sql` + CREATE INDEX IF NOT EXISTS idx_plan_discussion_comments_discussion + ON plan_discussion_comments(discussion_id, created_at) + `; +}); diff --git a/apps/server/src/persistence/PlanReviewDocuments.ts b/apps/server/src/persistence/PlanReviewDocuments.ts new file mode 100644 index 00000000000..6a606613a86 --- /dev/null +++ b/apps/server/src/persistence/PlanReviewDocuments.ts @@ -0,0 +1,816 @@ +/** + * T3-CUSTOM(expbkt3): persistence for native plan review. + * + * Three shapes live here: the document (one per plan lineage), its append-only + * versions, and the mutable working draft plus discussion threads. Versions are + * insert-only by contract — `appendVersion` fails on a duplicate `(documentId, + * revision)` rather than overwriting, so history can never be rewritten by a + * racing writer. + */ +import { ThreadId, UserId } from "@t3tools/contracts"; +import * as Context from "effect/Context"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as Schema from "effect/Schema"; +import * as SqlClient from "effect/unstable/sql/SqlClient"; +import * as SqlSchema from "effect/unstable/sql/SqlSchema"; + +import { + type ProjectionRepositoryError, + PersistenceDecodeError, + PersistenceSqlError, +} from "./Errors.ts"; + +export const PlanDocumentStatus = Schema.Literals([ + "open", + "approved", + "changes-requested", + "discarded", +]); +export type PlanDocumentStatus = typeof PlanDocumentStatus.Type; + +export const PlanVersionAuthorKind = Schema.Literals(["agent", "user"]); +export type PlanVersionAuthorKind = typeof PlanVersionAuthorKind.Type; + +export const PlanVersionOrigin = Schema.Literals([ + "agent-proposed", + "agent-revision", + "human-edit", +]); +export type PlanVersionOrigin = typeof PlanVersionOrigin.Type; + +export const PlanDocumentRecord = Schema.Struct({ + documentId: Schema.String, + threadId: ThreadId, + projectId: Schema.String, + title: Schema.String, + currentRevision: Schema.Number, + status: PlanDocumentStatus, + createdByUserId: Schema.NullOr(UserId), + createdAt: Schema.String, + updatedAt: Schema.String, +}); +export type PlanDocumentRecord = typeof PlanDocumentRecord.Type; + +export const PlanVersionRecord = Schema.Struct({ + versionId: Schema.String, + documentId: Schema.String, + revision: Schema.Number, + authorKind: PlanVersionAuthorKind, + authorUserId: Schema.NullOr(UserId), + origin: PlanVersionOrigin, + contentMarkdown: Schema.String, + contentValueJson: Schema.NullOr(Schema.String), + sourcePlanId: Schema.NullOr(Schema.String), + summary: Schema.NullOr(Schema.String), + createdAt: Schema.String, +}); +export type PlanVersionRecord = typeof PlanVersionRecord.Type; + +export const PlanDraftRecord = Schema.Struct({ + documentId: Schema.String, + baseVersionId: Schema.String, + contentValueJson: Schema.String, + updatedByUserId: Schema.NullOr(UserId), + updatedAt: Schema.String, + revisionToken: Schema.String, +}); +export type PlanDraftRecord = typeof PlanDraftRecord.Type; + +export const PlanDiscussionRecord = Schema.Struct({ + discussionId: Schema.String, + documentId: Schema.String, + anchorVersionId: Schema.String, + quotedText: Schema.String, + isResolved: Schema.Boolean, + resolvedByUserId: Schema.NullOr(UserId), + resolvedAt: Schema.NullOr(Schema.String), + createdByUserId: Schema.NullOr(UserId), + createdAt: Schema.String, +}); +export type PlanDiscussionRecord = typeof PlanDiscussionRecord.Type; + +export const PlanDiscussionCommentRecord = Schema.Struct({ + commentId: Schema.String, + discussionId: Schema.String, + authorUserId: Schema.NullOr(UserId), + bodyMarkdown: Schema.String, + isEdited: Schema.Boolean, + createdAt: Schema.String, + updatedAt: Schema.String, +}); +export type PlanDiscussionCommentRecord = typeof PlanDiscussionCommentRecord.Type; + +/** Raised when an append lost the race for a revision number. */ +export class PlanVersionConflictError extends Schema.TaggedErrorClass()( + "PlanVersionConflictError", + { documentId: Schema.String, revision: Schema.Number }, +) {} + +/** Raised when a draft save carried a stale `revisionToken`. */ +export class PlanDraftConflictError extends Schema.TaggedErrorClass()( + "PlanDraftConflictError", + { documentId: Schema.String }, +) {} + +export type PlanReviewRepositoryError = ProjectionRepositoryError; + +const PlanDocumentRawRow = Schema.Struct({ + documentId: Schema.Unknown, + threadId: Schema.Unknown, + projectId: Schema.Unknown, + title: Schema.Unknown, + currentRevision: Schema.Unknown, + status: Schema.Unknown, + createdByUserId: Schema.Unknown, + createdAt: Schema.Unknown, + updatedAt: Schema.Unknown, +}); + +const PlanVersionRawRow = Schema.Struct({ + versionId: Schema.Unknown, + documentId: Schema.Unknown, + revision: Schema.Unknown, + authorKind: Schema.Unknown, + authorUserId: Schema.Unknown, + origin: Schema.Unknown, + contentMarkdown: Schema.Unknown, + contentValueJson: Schema.Unknown, + sourcePlanId: Schema.Unknown, + summary: Schema.Unknown, + createdAt: Schema.Unknown, +}); + +const PlanDraftRawRow = Schema.Struct({ + documentId: Schema.Unknown, + baseVersionId: Schema.Unknown, + contentValueJson: Schema.Unknown, + updatedByUserId: Schema.Unknown, + updatedAt: Schema.Unknown, + revisionToken: Schema.Unknown, +}); + +const PlanDiscussionRawRow = Schema.Struct({ + discussionId: Schema.Unknown, + documentId: Schema.Unknown, + anchorVersionId: Schema.Unknown, + quotedText: Schema.Unknown, + isResolved: Schema.Unknown, + resolvedByUserId: Schema.Unknown, + resolvedAt: Schema.Unknown, + createdByUserId: Schema.Unknown, + createdAt: Schema.Unknown, +}); + +const PlanDiscussionCommentRawRow = Schema.Struct({ + commentId: Schema.Unknown, + discussionId: Schema.Unknown, + authorUserId: Schema.Unknown, + bodyMarkdown: Schema.Unknown, + isEdited: Schema.Unknown, + createdAt: Schema.Unknown, + updatedAt: Schema.Unknown, +}); + +export interface AppendVersionInput { + readonly versionId: string; + readonly documentId: string; + readonly revision: number; + readonly authorKind: PlanVersionAuthorKind; + readonly authorUserId: UserId | null; + readonly origin: PlanVersionOrigin; + readonly contentMarkdown: string; + readonly contentValueJson: string | null; + readonly sourcePlanId: string | null; + readonly summary: string | null; + readonly createdAt: string; +} + +export interface SaveDraftInput { + readonly documentId: string; + readonly baseVersionId: string; + readonly contentValueJson: string; + readonly updatedByUserId: UserId | null; + readonly updatedAt: string; + readonly expectedRevisionToken: string | null; + readonly nextRevisionToken: string; +} + +export interface UpsertDiscussionInput { + readonly discussionId: string; + readonly documentId: string; + readonly anchorVersionId: string; + readonly quotedText: string; + readonly createdByUserId: UserId | null; + readonly createdAt: string; +} + +export interface AddDiscussionCommentInput { + readonly commentId: string; + readonly discussionId: string; + readonly documentId: string; + readonly authorUserId: UserId | null; + readonly bodyMarkdown: string; + readonly createdAt: string; +} + +export interface ResolveDiscussionInput { + readonly discussionId: string; + readonly documentId: string; + readonly isResolved: boolean; + readonly resolvedByUserId: UserId | null; + readonly resolvedAt: string | null; +} + +export class PlanReviewRepository extends Context.Service< + PlanReviewRepository, + { + readonly upsertDocument: ( + input: PlanDocumentRecord, + ) => Effect.Effect; + readonly getDocument: ( + documentId: string, + ) => Effect.Effect, PlanReviewRepositoryError>; + readonly listDocumentsForThread: ( + threadId: ThreadId, + ) => Effect.Effect, PlanReviewRepositoryError>; + readonly findDocumentBySourcePlanId: ( + sourcePlanId: string, + ) => Effect.Effect, PlanReviewRepositoryError>; + readonly setDocumentStatus: (input: { + readonly documentId: string; + readonly status: PlanDocumentStatus; + readonly updatedAt: string; + }) => Effect.Effect; + readonly appendVersion: ( + input: AppendVersionInput, + ) => Effect.Effect; + readonly listVersions: ( + documentId: string, + ) => Effect.Effect, PlanReviewRepositoryError>; + readonly getVersion: (input: { + readonly documentId: string; + readonly versionId: string; + }) => Effect.Effect, PlanReviewRepositoryError>; + readonly getLatestVersion: ( + documentId: string, + ) => Effect.Effect, PlanReviewRepositoryError>; + readonly getDraft: ( + documentId: string, + ) => Effect.Effect, PlanReviewRepositoryError>; + readonly saveDraft: ( + input: SaveDraftInput, + ) => Effect.Effect; + readonly clearDraft: (documentId: string) => Effect.Effect; + readonly upsertDiscussion: ( + input: UpsertDiscussionInput, + ) => Effect.Effect; + readonly listDiscussions: ( + documentId: string, + ) => Effect.Effect, PlanReviewRepositoryError>; + readonly resolveDiscussion: ( + input: ResolveDiscussionInput, + ) => Effect.Effect; + readonly addDiscussionComment: ( + input: AddDiscussionCommentInput, + ) => Effect.Effect; + readonly listDiscussionComments: ( + documentId: string, + ) => Effect.Effect, PlanReviewRepositoryError>; + } +>()("t3/persistence/PlanReviewDocuments/PlanReviewRepository") {} + +function mapError(operation: string) { + return (cause: unknown): PlanReviewRepositoryError => + Schema.isSchemaError(cause) + ? PersistenceDecodeError.fromSchemaError(`${operation}:decode`, cause) + : new PersistenceSqlError({ operation: `${operation}:query`, cause }); +} + +const decodeDocument = Schema.decodeUnknownEffect(PlanDocumentRecord); +const decodeVersion = Schema.decodeUnknownEffect(PlanVersionRecord); +const decodeDraft = Schema.decodeUnknownEffect(PlanDraftRecord); +const decodeDiscussion = Schema.decodeUnknownEffect(PlanDiscussionRecord); +const decodeComment = Schema.decodeUnknownEffect(PlanDiscussionCommentRecord); + +/** SQLite stores booleans as 0/1; normalise before schema decoding. */ +function withBoolean(row: Record, key: K) { + return { ...row, [key]: row[key] === 1 || row[key] === true }; +} + +export const make = Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + + const documentColumns = sql` + document_id AS "documentId", + thread_id AS "threadId", + project_id AS "projectId", + title AS "title", + current_revision AS "currentRevision", + status AS "status", + created_by_user_id AS "createdByUserId", + created_at AS "createdAt", + updated_at AS "updatedAt" + `; + + const versionColumns = sql` + version_id AS "versionId", + document_id AS "documentId", + revision AS "revision", + author_kind AS "authorKind", + author_user_id AS "authorUserId", + origin AS "origin", + content_markdown AS "contentMarkdown", + content_value_json AS "contentValueJson", + source_plan_id AS "sourcePlanId", + summary AS "summary", + created_at AS "createdAt" + `; + + const discussionColumns = sql` + discussion_id AS "discussionId", + document_id AS "documentId", + anchor_version_id AS "anchorVersionId", + quoted_text AS "quotedText", + is_resolved AS "isResolved", + resolved_by_user_id AS "resolvedByUserId", + resolved_at AS "resolvedAt", + created_by_user_id AS "createdByUserId", + created_at AS "createdAt" + `; + + const upsertDocumentRow = SqlSchema.void({ + Request: PlanDocumentRecord, + execute: (input) => sql` + INSERT INTO plan_documents ( + document_id, thread_id, project_id, title, current_revision, + status, created_by_user_id, created_at, updated_at + ) VALUES ( + ${input.documentId}, ${input.threadId}, ${input.projectId}, ${input.title}, + ${input.currentRevision}, ${input.status}, ${input.createdByUserId}, + ${input.createdAt}, ${input.updatedAt} + ) + ON CONFLICT(document_id) DO UPDATE SET + title = excluded.title, + current_revision = excluded.current_revision, + status = excluded.status, + updated_at = excluded.updated_at + `, + }); + + const getDocumentRow = SqlSchema.findOneOption({ + Request: Schema.Struct({ documentId: Schema.String }), + Result: PlanDocumentRawRow, + execute: ({ documentId }) => sql` + SELECT ${documentColumns} FROM plan_documents WHERE document_id = ${documentId} + `, + }); + + const listDocumentsForThreadRows = SqlSchema.findAll({ + Request: Schema.Struct({ threadId: ThreadId }), + Result: PlanDocumentRawRow, + execute: ({ threadId }) => sql` + SELECT ${documentColumns} FROM plan_documents + WHERE thread_id = ${threadId} + ORDER BY created_at DESC + `, + }); + + const findDocumentBySourcePlanIdRow = SqlSchema.findOneOption({ + Request: Schema.Struct({ sourcePlanId: Schema.String }), + Result: PlanDocumentRawRow, + execute: ({ sourcePlanId }) => sql` + SELECT ${documentColumns} FROM plan_documents + WHERE document_id = ( + SELECT document_id FROM plan_document_versions + WHERE source_plan_id = ${sourcePlanId} + ORDER BY revision DESC LIMIT 1 + ) + `, + }); + + const setDocumentStatusRow = SqlSchema.void({ + Request: Schema.Struct({ + documentId: Schema.String, + status: PlanDocumentStatus, + updatedAt: Schema.String, + }), + execute: ({ documentId, status, updatedAt }) => sql` + UPDATE plan_documents + SET status = ${status}, updated_at = ${updatedAt} + WHERE document_id = ${documentId} + `, + }); + + // INSERT ... SELECT WHERE NOT EXISTS keeps the duplicate check inside SQLite, + // so two concurrent appends cannot both believe they won the revision. + const appendVersionRow = SqlSchema.findAll({ + Request: Schema.Struct({ + versionId: Schema.String, + documentId: Schema.String, + revision: Schema.Number, + authorKind: PlanVersionAuthorKind, + authorUserId: Schema.NullOr(UserId), + origin: PlanVersionOrigin, + contentMarkdown: Schema.String, + contentValueJson: Schema.NullOr(Schema.String), + sourcePlanId: Schema.NullOr(Schema.String), + summary: Schema.NullOr(Schema.String), + createdAt: Schema.String, + }), + Result: Schema.Struct({ versionId: Schema.String }), + execute: (input) => sql` + INSERT INTO plan_document_versions ( + version_id, document_id, revision, author_kind, author_user_id, + origin, content_markdown, content_value_json, source_plan_id, summary, created_at + ) + SELECT + ${input.versionId}, ${input.documentId}, ${input.revision}, ${input.authorKind}, + ${input.authorUserId}, ${input.origin}, ${input.contentMarkdown}, + ${input.contentValueJson}, ${input.sourcePlanId}, ${input.summary}, ${input.createdAt} + WHERE NOT EXISTS ( + SELECT 1 FROM plan_document_versions + WHERE document_id = ${input.documentId} AND revision = ${input.revision} + ) + RETURNING version_id AS "versionId" + `, + }); + + const listVersionRows = SqlSchema.findAll({ + Request: Schema.Struct({ documentId: Schema.String }), + Result: PlanVersionRawRow, + execute: ({ documentId }) => sql` + SELECT ${versionColumns} FROM plan_document_versions + WHERE document_id = ${documentId} + ORDER BY revision ASC + `, + }); + + const getVersionRow = SqlSchema.findOneOption({ + Request: Schema.Struct({ versionId: Schema.String, documentId: Schema.String }), + Result: PlanVersionRawRow, + execute: ({ versionId, documentId }) => sql` + SELECT ${versionColumns} FROM plan_document_versions + WHERE version_id = ${versionId} AND document_id = ${documentId} + `, + }); + + const getLatestVersionRow = SqlSchema.findOneOption({ + Request: Schema.Struct({ documentId: Schema.String }), + Result: PlanVersionRawRow, + execute: ({ documentId }) => sql` + SELECT ${versionColumns} FROM plan_document_versions + WHERE document_id = ${documentId} + ORDER BY revision DESC LIMIT 1 + `, + }); + + const getDraftRow = SqlSchema.findOneOption({ + Request: Schema.Struct({ documentId: Schema.String }), + Result: PlanDraftRawRow, + execute: ({ documentId }) => sql` + SELECT + document_id AS "documentId", + base_version_id AS "baseVersionId", + content_value_json AS "contentValueJson", + updated_by_user_id AS "updatedByUserId", + updated_at AS "updatedAt", + revision_token AS "revisionToken" + FROM plan_document_drafts WHERE document_id = ${documentId} + `, + }); + + // The WHERE clause is the concurrency guard: a save whose expected token no + // longer matches the stored one touches zero rows and surfaces as a conflict. + const saveDraftRow = SqlSchema.findAll({ + Request: Schema.Struct({ + documentId: Schema.String, + baseVersionId: Schema.String, + contentValueJson: Schema.String, + updatedByUserId: Schema.NullOr(UserId), + updatedAt: Schema.String, + expectedRevisionToken: Schema.NullOr(Schema.String), + nextRevisionToken: Schema.String, + }), + Result: Schema.Struct({ documentId: Schema.String }), + execute: (input) => sql` + INSERT INTO plan_document_drafts ( + document_id, base_version_id, content_value_json, + updated_by_user_id, updated_at, revision_token + ) + SELECT + ${input.documentId}, ${input.baseVersionId}, ${input.contentValueJson}, + ${input.updatedByUserId}, ${input.updatedAt}, ${input.nextRevisionToken} + WHERE ( + ${input.expectedRevisionToken} IS NULL + AND NOT EXISTS ( + SELECT 1 FROM plan_document_drafts WHERE document_id = ${input.documentId} + ) + ) + OR ${input.expectedRevisionToken} = ( + SELECT revision_token FROM plan_document_drafts WHERE document_id = ${input.documentId} + ) + ON CONFLICT(document_id) DO UPDATE SET + base_version_id = excluded.base_version_id, + content_value_json = excluded.content_value_json, + updated_by_user_id = excluded.updated_by_user_id, + updated_at = excluded.updated_at, + revision_token = excluded.revision_token + RETURNING document_id AS "documentId" + `, + }); + + const clearDraftRow = SqlSchema.void({ + Request: Schema.Struct({ documentId: Schema.String }), + execute: ({ documentId }) => sql` + DELETE FROM plan_document_drafts WHERE document_id = ${documentId} + `, + }); + + const upsertDiscussionRow = SqlSchema.findAll({ + Result: Schema.Struct({ discussionId: Schema.String }), + Request: Schema.Struct({ + discussionId: Schema.String, + documentId: Schema.String, + anchorVersionId: Schema.String, + quotedText: Schema.String, + createdByUserId: Schema.NullOr(UserId), + createdAt: Schema.String, + }), + execute: (input) => sql` + INSERT INTO plan_discussions ( + discussion_id, document_id, anchor_version_id, quoted_text, + is_resolved, resolved_by_user_id, resolved_at, created_by_user_id, created_at + ) VALUES ( + ${input.discussionId}, ${input.documentId}, ${input.anchorVersionId}, + ${input.quotedText}, 0, NULL, NULL, ${input.createdByUserId}, ${input.createdAt} + ) + ON CONFLICT(discussion_id) DO UPDATE SET + anchor_version_id = excluded.anchor_version_id, + quoted_text = excluded.quoted_text + WHERE plan_discussions.document_id = excluded.document_id + RETURNING discussion_id AS "discussionId" + `, + }); + + const listDiscussionRows = SqlSchema.findAll({ + Request: Schema.Struct({ documentId: Schema.String }), + Result: PlanDiscussionRawRow, + execute: ({ documentId }) => sql` + SELECT ${discussionColumns} FROM plan_discussions + WHERE document_id = ${documentId} + ORDER BY created_at ASC + `, + }); + + // Every discussion statement is scoped by document_id as well as by its own + // id. Callers authorize the document, so an id that belongs to a different + // document must not be reachable through it. + const resolveDiscussionRow = SqlSchema.findAll({ + Request: Schema.Struct({ + discussionId: Schema.String, + documentId: Schema.String, + isResolved: Schema.Boolean, + resolvedByUserId: Schema.NullOr(UserId), + resolvedAt: Schema.NullOr(Schema.String), + }), + Result: Schema.Struct({ discussionId: Schema.String }), + execute: (input) => sql` + UPDATE plan_discussions + SET is_resolved = ${input.isResolved ? 1 : 0}, + resolved_by_user_id = ${input.resolvedByUserId}, + resolved_at = ${input.resolvedAt} + WHERE discussion_id = ${input.discussionId} + AND document_id = ${input.documentId} + RETURNING discussion_id AS "discussionId" + `, + }); + + const addDiscussionCommentRow = SqlSchema.findAll({ + Request: Schema.Struct({ + commentId: Schema.String, + discussionId: Schema.String, + documentId: Schema.String, + authorUserId: Schema.NullOr(UserId), + bodyMarkdown: Schema.String, + createdAt: Schema.String, + }), + Result: Schema.Struct({ commentId: Schema.String }), + execute: (input) => sql` + INSERT INTO plan_discussion_comments ( + comment_id, discussion_id, author_user_id, body_markdown, + is_edited, created_at, updated_at + ) + SELECT + ${input.commentId}, ${input.discussionId}, ${input.authorUserId}, + ${input.bodyMarkdown}, 0, ${input.createdAt}, ${input.createdAt} + WHERE EXISTS ( + SELECT 1 FROM plan_discussions + WHERE discussion_id = ${input.discussionId} AND document_id = ${input.documentId} + ) + ON CONFLICT(comment_id) DO UPDATE SET + body_markdown = excluded.body_markdown, + is_edited = 1, + updated_at = excluded.updated_at + RETURNING comment_id AS "commentId" + `, + }); + + const listDiscussionCommentRows = SqlSchema.findAll({ + Request: Schema.Struct({ documentId: Schema.String }), + Result: PlanDiscussionCommentRawRow, + execute: ({ documentId }) => sql` + SELECT + c.comment_id AS "commentId", + c.discussion_id AS "discussionId", + c.author_user_id AS "authorUserId", + c.body_markdown AS "bodyMarkdown", + c.is_edited AS "isEdited", + c.created_at AS "createdAt", + c.updated_at AS "updatedAt" + FROM plan_discussion_comments c + JOIN plan_discussions d ON d.discussion_id = c.discussion_id + WHERE d.document_id = ${documentId} + ORDER BY c.created_at ASC + `, + }); + + const decodeMany = ( + rows: ReadonlyArray, + decode: (row: unknown) => Effect.Effect, + operation: string, + ): Effect.Effect, PlanReviewRepositoryError> => + Effect.forEach(rows, (row) => decode(row).pipe(Effect.mapError(mapError(operation)))); + + const service: PlanReviewRepository["Service"] = { + upsertDocument: (input) => + upsertDocumentRow(input).pipe(Effect.mapError(mapError("PlanReview.upsertDocument"))), + + getDocument: (documentId) => + getDocumentRow({ documentId }).pipe( + Effect.mapError(mapError("PlanReview.getDocument")), + Effect.flatMap( + Option.match({ + onNone: () => Effect.succeed(Option.none()), + onSome: (row) => + decodeDocument(row).pipe( + Effect.map(Option.some), + Effect.mapError(mapError("PlanReview.getDocument")), + ), + }), + ), + ), + + listDocumentsForThread: (threadId) => + listDocumentsForThreadRows({ threadId }).pipe( + Effect.mapError(mapError("PlanReview.listDocumentsForThread")), + Effect.flatMap((rows) => + decodeMany(rows, decodeDocument, "PlanReview.listDocumentsForThread"), + ), + ), + + findDocumentBySourcePlanId: (sourcePlanId) => + findDocumentBySourcePlanIdRow({ sourcePlanId }).pipe( + Effect.mapError(mapError("PlanReview.findDocumentBySourcePlanId")), + Effect.flatMap( + Option.match({ + onNone: () => Effect.succeed(Option.none()), + onSome: (row) => + decodeDocument(row).pipe( + Effect.map(Option.some), + Effect.mapError(mapError("PlanReview.findDocumentBySourcePlanId")), + ), + }), + ), + ), + + setDocumentStatus: (input) => + setDocumentStatusRow(input).pipe(Effect.mapError(mapError("PlanReview.setDocumentStatus"))), + + appendVersion: (input) => + appendVersionRow(input).pipe( + Effect.mapError(mapError("PlanReview.appendVersion")), + Effect.flatMap((rows) => + rows.length > 0 + ? Effect.void + : Effect.fail( + new PlanVersionConflictError({ + documentId: input.documentId, + revision: input.revision, + }), + ), + ), + ), + + listVersions: (documentId) => + listVersionRows({ documentId }).pipe( + Effect.mapError(mapError("PlanReview.listVersions")), + Effect.flatMap((rows) => decodeMany(rows, decodeVersion, "PlanReview.listVersions")), + ), + + getVersion: (input) => + getVersionRow(input).pipe( + Effect.mapError(mapError("PlanReview.getVersion")), + Effect.flatMap( + Option.match({ + onNone: () => Effect.succeed(Option.none()), + onSome: (row) => + decodeVersion(row).pipe( + Effect.map(Option.some), + Effect.mapError(mapError("PlanReview.getVersion")), + ), + }), + ), + ), + + getLatestVersion: (documentId) => + getLatestVersionRow({ documentId }).pipe( + Effect.mapError(mapError("PlanReview.getLatestVersion")), + Effect.flatMap( + Option.match({ + onNone: () => Effect.succeed(Option.none()), + onSome: (row) => + decodeVersion(row).pipe( + Effect.map(Option.some), + Effect.mapError(mapError("PlanReview.getLatestVersion")), + ), + }), + ), + ), + + getDraft: (documentId) => + getDraftRow({ documentId }).pipe( + Effect.mapError(mapError("PlanReview.getDraft")), + Effect.flatMap( + Option.match({ + onNone: () => Effect.succeed(Option.none()), + onSome: (row) => + decodeDraft(row).pipe( + Effect.map(Option.some), + Effect.mapError(mapError("PlanReview.getDraft")), + ), + }), + ), + ), + + saveDraft: (input) => + saveDraftRow(input).pipe( + Effect.mapError(mapError("PlanReview.saveDraft")), + Effect.flatMap((rows) => + rows.length > 0 + ? Effect.void + : Effect.fail(new PlanDraftConflictError({ documentId: input.documentId })), + ), + ), + + clearDraft: (documentId) => + clearDraftRow({ documentId }).pipe(Effect.mapError(mapError("PlanReview.clearDraft"))), + + upsertDiscussion: (input) => + upsertDiscussionRow(input).pipe( + Effect.mapError(mapError("PlanReview.upsertDiscussion")), + Effect.asVoid, + ), + + listDiscussions: (documentId) => + listDiscussionRows({ documentId }).pipe( + Effect.mapError(mapError("PlanReview.listDiscussions")), + Effect.flatMap((rows) => + decodeMany( + rows.map((row) => withBoolean(row as Record, "isResolved")), + decodeDiscussion, + "PlanReview.listDiscussions", + ), + ), + ), + + resolveDiscussion: (input) => + resolveDiscussionRow(input).pipe( + Effect.mapError(mapError("PlanReview.resolveDiscussion")), + Effect.asVoid, + ), + + addDiscussionComment: (input) => + addDiscussionCommentRow(input).pipe( + Effect.mapError(mapError("PlanReview.addDiscussionComment")), + Effect.asVoid, + ), + + listDiscussionComments: (documentId) => + listDiscussionCommentRows({ documentId }).pipe( + Effect.mapError(mapError("PlanReview.listDiscussionComments")), + Effect.flatMap((rows) => + decodeMany( + rows.map((row) => withBoolean(row as Record, "isEdited")), + decodeComment, + "PlanReview.listDiscussionComments", + ), + ), + ), + }; + + return PlanReviewRepository.of(service); +}); + +export const layer = Layer.effect(PlanReviewRepository, make); diff --git a/apps/server/src/planreview/PlanIngestListener.ts b/apps/server/src/planreview/PlanIngestListener.ts new file mode 100644 index 00000000000..7029bd45ff7 --- /dev/null +++ b/apps/server/src/planreview/PlanIngestListener.ts @@ -0,0 +1,120 @@ +/** + * T3-CUSTOM(expbkt3): captures agent plans as plan-review documents. + * + * Subscribes to `thread.proposed-plan-upserted` and turns every proposed plan + * into a version. Plan ids are `plan:${threadId}:${turnId}`, so each revision + * turn arrives as a new id — the service resolves lineage explicitly rather + * than guessing, and redelivery of an id already captured is a no-op. + */ +import { ThreadId, type OrchestrationProposedPlan } from "@t3tools/contracts"; +import { makeKeyedCoalescingWorker } from "@t3tools/shared/KeyedCoalescingWorker"; +import { withoutPlannotatorPlanMarker } from "@t3tools/shared/plannotator"; +import * as Context from "effect/Context"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as Stream from "effect/Stream"; + +import { OrchestrationEngineService } from "../orchestration/Services/OrchestrationEngine.ts"; +import { ProjectionSnapshotQuery } from "../orchestration/Services/ProjectionSnapshotQuery.ts"; +import { derivePlanTitle, PlanReviewService } from "./PlanReviewService.ts"; + +export interface PlanIngestInput { + readonly threadId: ThreadId; + readonly proposedPlan: OrchestrationProposedPlan; +} + +export class PlanIngestListener extends Context.Service< + PlanIngestListener, + { + /** Captures one plan now. Exposed so tests can drive ingestion directly. */ + readonly ingest: (input: PlanIngestInput) => Effect.Effect; + } +>()("t3/planreview/PlanIngestListener") {} + +/** + * Reconciles plans that landed while the server was down. Reads only the newest + * unimplemented plan per active thread rather than hydrating full history. + */ +export const reconcilePlansOnStartup = Effect.fn("PlanIngestListener.reconcileOnStartup")( + function* ( + query: Pick, + schedule: (input: PlanIngestInput) => Effect.Effect, + ) { + const candidates = yield* query.listLatestProposedPlansForActiveThreads(); + yield* Effect.forEach( + candidates, + ({ threadId, proposedPlan }) => schedule({ threadId, proposedPlan }), + { concurrency: 4, discard: true }, + ); + }, +); + +export const make = Effect.gen(function* () { + const service = yield* PlanReviewService; + const query = yield* ProjectionSnapshotQuery; + const orchestrationEngine = yield* OrchestrationEngineService; + + const capture = (input: PlanIngestInput) => + Effect.gen(function* () { + // An implemented plan is history; there is nothing left to review. + if (input.proposedPlan.implementedAt !== null) return; + + const planMarkdown = withoutPlannotatorPlanMarker(input.proposedPlan.planMarkdown).trim(); + if (planMarkdown.length === 0) return; + + const threadOption = yield* query.getThreadDetailById(input.threadId); + if (Option.isNone(threadOption)) return; + + yield* service.capturePlan({ + threadId: input.threadId, + projectId: threadOption.value.projectId, + planId: input.proposedPlan.id, + planMarkdown, + title: derivePlanTitle(planMarkdown), + authorUserId: null, + }); + }).pipe( + Effect.catchCause((cause) => + Effect.logWarning("could not capture a proposed plan for review", { + threadId: input.threadId, + planId: input.proposedPlan.id, + cause: String(cause), + }), + ), + ); + + // Coalesce per plan so a burst of streaming-plan upserts captures once, with + // the newest body winning. + const worker = yield* makeKeyedCoalescingWorker({ + merge: (current, next) => + next.proposedPlan.updatedAt >= current.proposedPlan.updatedAt ? next : current, + process: (_key, value) => capture(value), + }); + + const schedule = (input: PlanIngestInput) => + worker.enqueue(`${input.threadId}:${input.proposedPlan.id}`, input); + + // Subscribe before reconciling so a plan emitted during startup cannot fall + // into the gap between the two operations. + yield* Effect.forkScoped( + Stream.runForEach(orchestrationEngine.streamDomainEvents, (event) => + event.type === "thread.proposed-plan-upserted" + ? schedule({ + threadId: event.payload.threadId, + proposedPlan: event.payload.proposedPlan, + }) + : Effect.void, + ), + ); + + yield* reconcilePlansOnStartup(query, schedule).pipe( + Effect.catchCause((cause) => + Effect.logWarning("could not reconcile proposed plans for review", { cause: String(cause) }), + ), + ); + + return PlanIngestListener.of({ ingest: capture }); +}); + +export const layer = Layer.effect(PlanIngestListener, make); diff --git a/apps/server/src/planreview/PlanReviewContextPolicy.test.ts b/apps/server/src/planreview/PlanReviewContextPolicy.test.ts new file mode 100644 index 00000000000..9fae69da0a2 --- /dev/null +++ b/apps/server/src/planreview/PlanReviewContextPolicy.test.ts @@ -0,0 +1,76 @@ +import { describe, expect, it } from "vite-plus/test"; + +import { + decidePlanResend, + shouldSendFullDocumentInsteadOfDiff, +} from "./PlanReviewContextPolicy.ts"; + +const baseSignals = { + latestCompactionAt: null, + planCreatedAt: "2026-08-07T10:00:00.000Z", + planThreadId: "thread-1", + targetThreadId: "thread-1", + providerSessionStatus: "running", +} as const; + +describe("decidePlanResend", () => { + it("keeps the prompt short when the model can still see the plan", () => { + expect(decidePlanResend(baseSignals)).toEqual({ shouldResend: false, reason: null }); + }); + + it("resends when implementation moves to another thread", () => { + const decision = decidePlanResend({ ...baseSignals, targetThreadId: "thread-2" }); + expect(decision.shouldResend).toBe(true); + expect(decision.reason).toContain("different session"); + }); + + it("resends when the thread compacted after the plan was written", () => { + const decision = decidePlanResend({ + ...baseSignals, + latestCompactionAt: "2026-08-07T11:00:00.000Z", + }); + expect(decision.shouldResend).toBe(true); + expect(decision.reason).toContain("compacted"); + }); + + it("ignores a compaction that happened before the plan", () => { + expect( + decidePlanResend({ ...baseSignals, latestCompactionAt: "2026-08-07T09:00:00.000Z" }) + .shouldResend, + ).toBe(false); + }); + + it("resends when no provider session is bound", () => { + expect(decidePlanResend({ ...baseSignals, providerSessionStatus: null }).shouldResend).toBe( + true, + ); + }); + + it("resends when the provider session has stopped", () => { + const decision = decidePlanResend({ ...baseSignals, providerSessionStatus: "stopped" }); + expect(decision.shouldResend).toBe(true); + expect(decision.reason).toContain("no longer running"); + }); + + it("prefers the cross-thread reason when several signals fire at once", () => { + const decision = decidePlanResend({ + ...baseSignals, + targetThreadId: "thread-2", + providerSessionStatus: null, + latestCompactionAt: "2026-08-07T23:00:00.000Z", + }); + expect(decision.reason).toContain("different session"); + }); +}); + +describe("shouldSendFullDocumentInsteadOfDiff", () => { + it("prefers a diff for ordinary edits", () => { + expect(shouldSendFullDocumentInsteadOfDiff(0.1)).toBe(false); + expect(shouldSendFullDocumentInsteadOfDiff(0.6)).toBe(false); + }); + + it("sends the whole document once the diff stops being smaller", () => { + expect(shouldSendFullDocumentInsteadOfDiff(0.61)).toBe(true); + expect(shouldSendFullDocumentInsteadOfDiff(1)).toBe(true); + }); +}); diff --git a/apps/server/src/planreview/PlanReviewContextPolicy.ts b/apps/server/src/planreview/PlanReviewContextPolicy.ts new file mode 100644 index 00000000000..a7e4785b7ce --- /dev/null +++ b/apps/server/src/planreview/PlanReviewContextPolicy.ts @@ -0,0 +1,74 @@ +/** + * T3-CUSTOM(expbkt3): decides whether an approval must re-send the plan body. + * + * The default answer is no — the model wrote the plan and still has it in + * context, so repeating it wastes thousands of tokens per approval. We only + * repeat it when the model demonstrably cannot see it any more. + */ + +export interface PlanResendSignals { + /** + * ISO timestamp of the most recent `context-compaction` activity on the + * thread, or null when the thread has never compacted. + */ + readonly latestCompactionAt: string | null; + /** ISO timestamp of the version the agent proposed. */ + readonly planCreatedAt: string; + /** Thread the plan was authored in. */ + readonly planThreadId: string; + /** Thread the implementation turn will start in. */ + readonly targetThreadId: string; + /** Provider session status bound to the target thread, null when unbound. */ + readonly providerSessionStatus: string | null; +} + +export interface PlanResendDecision { + readonly shouldResend: boolean; + /** + * Human-readable clause completing "The full plan is repeated because …". + * Null when nothing is resent. + */ + readonly reason: string | null; +} + +/** + * Returns whether the approval prompt must carry the whole plan. + * + * Three signals force a resend; anything else keeps the prompt to one line. + */ +export function decidePlanResend(signals: PlanResendSignals): PlanResendDecision { + if (signals.targetThreadId !== signals.planThreadId) { + return { + shouldResend: true, + reason: "it is being implemented in a different session from the one that planned it", + }; + } + + if (signals.latestCompactionAt !== null && signals.latestCompactionAt > signals.planCreatedAt) { + return { + shouldResend: true, + reason: "this session compacted its context after the plan was written", + }; + } + + // A stopped or absent provider session means the next turn boots a fresh + // process, which will not have replayed the planning turn. + if (signals.providerSessionStatus === null || signals.providerSessionStatus === "stopped") { + return { + shouldResend: true, + reason: "the planning session is no longer running", + }; + } + + return { shouldResend: false, reason: null }; +} + +/** + * Guards the feedback path: a diff that rewrites most of the document is + * neither smaller than the document nor easier to read. + */ +export const MAX_DIFF_CHANGE_RATIO = 0.6; + +export function shouldSendFullDocumentInsteadOfDiff(changeRatio: number): boolean { + return changeRatio > MAX_DIFF_CHANGE_RATIO; +} diff --git a/apps/server/src/planreview/PlanReviewService.test.ts b/apps/server/src/planreview/PlanReviewService.test.ts new file mode 100644 index 00000000000..cafafc4dcc2 --- /dev/null +++ b/apps/server/src/planreview/PlanReviewService.test.ts @@ -0,0 +1,707 @@ +/** + * T3-CUSTOM(expbkt3): round-trip coverage for the native plan review service. + * + * The prompts are the product here — the whole point of the feature is that an + * approval stops re-sending the plan and feedback carries anchors instead of a + * document — so every test asserts the exact text handed to the agent. + */ +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { ThreadId, UserId, type OrchestrationCommand } from "@t3tools/contracts"; +import { describe, expect, it } from "@effect/vitest"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as Ref from "effect/Ref"; + +import { OrchestrationCommandDispatcher } from "../orchestration/dispatchCommand.ts"; +import { ProjectionSnapshotQuery } from "../orchestration/Services/ProjectionSnapshotQuery.ts"; +import { MigrationsLive } from "../persistence/Migrations.ts"; +import { SqlitePersistenceMemory } from "../persistence/Layers/Sqlite.ts"; +import * as PlanReviewDocuments from "../persistence/PlanReviewDocuments.ts"; +import * as PlanReviewServiceModule from "./PlanReviewService.ts"; +import { derivePlanTitle, PlanReviewService } from "./PlanReviewService.ts"; + +const threadId = ThreadId.make("thread-plan-review"); +const otherThreadId = ThreadId.make("thread-other"); +const reviewerId = UserId.make("user_reviewer"); + +const PLAN = [ + "# Auth rewrite", + "", + "## Steps", + "", + "1. Add the migration", + "2. Backfill the rows", + "3. Flip the flag", +].join("\n"); + +interface ThreadStub { + readonly sessionStatus: string | null; + readonly compactionAt: string | null; +} + +/** + * Captures dispatched commands so a test can assert what reached the thread, + * and stubs the one projection read the service performs. + */ +const makeHarness = (thread: ThreadStub) => + Effect.gen(function* () { + const dispatched = yield* Ref.make>([]); + + const dispatcherLayer = Layer.succeed( + OrchestrationCommandDispatcher, + OrchestrationCommandDispatcher.of({ + dispatch: (command) => + Ref.update(dispatched, (current) => [...current, command]).pipe( + Effect.as({ sequence: 1 }), + ), + }), + ); + + const queryLayer = Layer.succeed( + ProjectionSnapshotQuery, + ProjectionSnapshotQuery.of({ + getThreadDetailById: () => + Effect.succeed( + Option.some({ + id: threadId, + projectId: "project-1", + modelSelection: undefined, + runtimeMode: "local", + session: thread.sessionStatus === null ? null : { status: thread.sessionStatus }, + activities: + thread.compactionAt === null + ? [] + : [{ kind: "context-compaction", createdAt: thread.compactionAt }], + } as never), + ), + } as never), + ); + + return { dispatched, dispatcherLayer, queryLayer }; + }); + +const runWithService = ( + thread: ThreadStub, + body: (input: { + readonly service: PlanReviewService["Service"]; + readonly dispatched: Ref.Ref>; + }) => Effect.Effect, +) => + Effect.gen(function* () { + const harness = yield* makeHarness(thread); + const layer = PlanReviewServiceModule.layer.pipe( + Layer.provide(PlanReviewDocuments.layer), + Layer.provide(harness.dispatcherLayer), + Layer.provide(harness.queryLayer), + Layer.provide(MigrationsLive), + Layer.provide(SqlitePersistenceMemory), + Layer.provide(NodeServices.layer), + ); + + return yield* Effect.gen(function* () { + const service = yield* PlanReviewService; + return yield* body({ service, dispatched: harness.dispatched }); + }).pipe(Effect.provide(layer)); + }); + +const capturePlan = ( + service: PlanReviewService["Service"], + planId: string, + markdown = PLAN, + onThread: ThreadId = threadId, +) => + service.capturePlan({ + threadId: onThread, + projectId: "project-1", + planId: planId as never, + planMarkdown: markdown, + title: derivePlanTitle(markdown), + authorUserId: null, + }); + +const turnText = (commands: ReadonlyArray): string => { + const turn = commands.find((command) => command.type === "thread.turn.start"); + if (turn === undefined || turn.type !== "thread.turn.start") { + throw new Error("no turn was started"); + } + return turn.message.text; +}; + +describe("derivePlanTitle", () => { + it("uses the first heading", () => { + expect(derivePlanTitle(PLAN)).toBe("Auth rewrite"); + }); + + it("falls back to the first non-empty line", () => { + expect(derivePlanTitle("\n\nJust do the thing\n")).toBe("Just do the thing"); + }); + + it("falls back to a constant for an empty plan", () => { + expect(derivePlanTitle(" \n\n")).toBe("Plan"); + }); +}); + +describe("PlanReviewService capture", () => { + it.effect("captures the agent plan as version 1", () => + runWithService({ sessionStatus: "running", compactionAt: null }, ({ service }) => + Effect.gen(function* () { + const document = yield* capturePlan(service, "plan:a"); + const snapshot = yield* service.getReview(document.documentId); + + expect(snapshot.versions).toHaveLength(1); + expect(snapshot.versions[0]?.revision).toBe(1); + expect(snapshot.versions[0]?.authorKind).toBe("agent"); + expect(snapshot.versions[0]?.origin).toBe("agent-proposed"); + expect(snapshot.document.title).toBe("Auth rewrite"); + expect(snapshot.document.status).toBe("open"); + }), + ), + ); + + it.effect("treats a redelivered plan id as a no-op", () => + runWithService({ sessionStatus: "running", compactionAt: null }, ({ service }) => + Effect.gen(function* () { + const document = yield* capturePlan(service, "plan:a"); + yield* capturePlan(service, "plan:a"); + const snapshot = yield* service.getReview(document.documentId); + + expect(snapshot.versions).toHaveLength(1); + }), + ), + ); + + it.effect("appends an agent revision to the same lineage", () => + runWithService({ sessionStatus: "running", compactionAt: null }, ({ service }) => + Effect.gen(function* () { + const document = yield* capturePlan(service, "plan:a"); + yield* capturePlan(service, "plan:b", `${PLAN}\n4. Announce it`); + const snapshot = yield* service.getReview(document.documentId); + + expect(snapshot.versions).toHaveLength(2); + expect(snapshot.versions[1]?.origin).toBe("agent-revision"); + expect(snapshot.versions[1]?.revision).toBe(2); + expect(snapshot.document.currentRevision).toBe(2); + }), + ), + ); + + it.effect("ignores a revision whose body did not change", () => + runWithService({ sessionStatus: "running", compactionAt: null }, ({ service }) => + Effect.gen(function* () { + const document = yield* capturePlan(service, "plan:a"); + yield* capturePlan(service, "plan:b", PLAN); + const snapshot = yield* service.getReview(document.documentId); + + expect(snapshot.versions).toHaveLength(1); + }), + ), + ); +}); + +describe("PlanReviewService approval", () => { + it.effect("sends a short ack instead of the plan body", () => + runWithService({ sessionStatus: "running", compactionAt: null }, ({ service, dispatched }) => + Effect.gen(function* () { + const document = yield* capturePlan(service, "plan:a"); + const result = yield* service.submit({ + documentId: document.documentId, + decision: "approved", + globalComment: "", + editedMarkdown: null, + actorUserId: reviewerId, + actorLabel: "Tushar", + }); + + expect(result.resentPlan).toBe(false); + expect(result.prompt).toBe( + "Plan approved. Implement the plan you proposed above, exactly as written.", + ); + + const commands = yield* Ref.get(dispatched); + expect(turnText(commands)).not.toContain("Flip the flag"); + + // Only approval may leave Plan mode. + const modeCommand = commands.find( + (command) => command.type === "thread.interaction-mode.set", + ); + expect(modeCommand).toBeDefined(); + + const turn = commands.find((command) => command.type === "thread.turn.start"); + expect(turn?.type === "thread.turn.start" && turn.interactionMode).toBe("default"); + expect(turn?.type === "thread.turn.start" && turn.sourceProposedPlan?.planId).toBe( + "plan:a", + ); + }), + ), + ); + + it.effect("re-sends the plan when the thread compacted after it", () => + runWithService( + { sessionStatus: "running", compactionAt: "2099-01-01T00:00:00.000Z" }, + ({ service, dispatched }) => + Effect.gen(function* () { + const document = yield* capturePlan(service, "plan:a"); + const result = yield* service.submit({ + documentId: document.documentId, + decision: "approved", + globalComment: "", + editedMarkdown: null, + actorUserId: reviewerId, + actorLabel: "Tushar", + }); + + expect(result.resentPlan).toBe(true); + const text = turnText(yield* Ref.get(dispatched)); + expect(text).toContain("PLEASE IMPLEMENT THIS APPROVED PLAN:"); + expect(text).toContain("3. Flip the flag"); + expect(text).toContain("compacted its context"); + }), + ), + ); + + it.effect("re-sends the plan when no provider session is bound", () => + runWithService({ sessionStatus: null, compactionAt: null }, ({ service }) => + Effect.gen(function* () { + const document = yield* capturePlan(service, "plan:a"); + const result = yield* service.submit({ + documentId: document.documentId, + decision: "approved", + globalComment: "", + editedMarkdown: null, + actorUserId: reviewerId, + actorLabel: "Tushar", + }); + + expect(result.resentPlan).toBe(true); + }), + ), + ); + + it.effect("carries reviewer edits as a diff and records a human version", () => + runWithService({ sessionStatus: "running", compactionAt: null }, ({ service, dispatched }) => + Effect.gen(function* () { + const document = yield* capturePlan(service, "plan:a"); + yield* service.submit({ + documentId: document.documentId, + decision: "approved", + globalComment: "", + editedMarkdown: PLAN.replace("Flip the flag", "Flip the flag behind a kill switch"), + actorUserId: reviewerId, + actorLabel: "Tushar", + }); + + const text = turnText(yield* Ref.get(dispatched)); + expect(text).toContain("The reviewer edited the plan before approving."); + expect(text).toContain("+3. Flip the flag behind a kill switch"); + expect(text).not.toContain("1. Add the migration\n2. Backfill"); + + const snapshot = yield* service.getReview(document.documentId); + expect(snapshot.versions).toHaveLength(2); + expect(snapshot.versions[1]?.authorKind).toBe("user"); + expect(snapshot.versions[1]?.authorUserId).toBe(reviewerId); + expect(snapshot.document.status).toBe("approved"); + }), + ), + ); +}); + +describe("PlanReviewService feedback", () => { + it.effect("sends anchored comments without the plan body", () => + runWithService({ sessionStatus: "running", compactionAt: null }, ({ service, dispatched }) => + Effect.gen(function* () { + const document = yield* capturePlan(service, "plan:a"); + yield* service.upsertDiscussion({ + documentId: document.documentId, + discussionId: "discussion-1", + quotedText: "2. Backfill the rows", + bodyMarkdown: "Split this into its own migration.", + actorUserId: reviewerId, + }); + + yield* service.submit({ + documentId: document.documentId, + decision: "changes-requested", + globalComment: "Too broad overall.", + editedMarkdown: null, + actorUserId: reviewerId, + actorLabel: "Tushar", + }); + + const commands = yield* Ref.get(dispatched); + const text = turnText(commands); + + expect(text).toContain("Revise the plan you proposed."); + expect(text).toContain("Too broad overall."); + expect(text).toContain(" command.type === "thread.interaction-mode.set")).toBe( + false, + ); + const turn = commands.find((command) => command.type === "thread.turn.start"); + expect(turn?.type === "thread.turn.start" && turn.interactionMode).toBe("plan"); + + const snapshot = yield* service.getReview(document.documentId); + expect(snapshot.document.status).toBe("changes-requested"); + }), + ), + ); + + it.effect("omits a resolved discussion from the feedback", () => + runWithService({ sessionStatus: "running", compactionAt: null }, ({ service, dispatched }) => + Effect.gen(function* () { + const document = yield* capturePlan(service, "plan:a"); + yield* service.upsertDiscussion({ + documentId: document.documentId, + discussionId: "discussion-1", + quotedText: "2. Backfill the rows", + bodyMarkdown: "Split this into its own migration.", + actorUserId: reviewerId, + }); + yield* service.resolveDiscussion({ + documentId: document.documentId, + discussionId: "discussion-1", + isResolved: true, + actorUserId: reviewerId, + }); + + yield* service.submit({ + documentId: document.documentId, + decision: "changes-requested", + globalComment: "Still too broad.", + editedMarkdown: null, + actorUserId: reviewerId, + actorLabel: "Tushar", + }); + + const text = turnText(yield* Ref.get(dispatched)); + expect(text).toContain("Still too broad."); + expect(text).not.toContain(" + runWithService({ sessionStatus: "running", compactionAt: null }, ({ service, dispatched }) => + Effect.gen(function* () { + const document = yield* capturePlan(service, "plan:a"); + const result = yield* service.submit({ + documentId: document.documentId, + decision: "discarded", + globalComment: "", + editedMarkdown: null, + actorUserId: reviewerId, + actorLabel: "Tushar", + }); + + expect(result.turnStarted).toBe(false); + expect(result.prompt).toBeNull(); + + const commands = yield* Ref.get(dispatched); + expect(commands.some((command) => command.type === "thread.turn.start")).toBe(false); + + const snapshot = yield* service.getReview(document.documentId); + expect(snapshot.document.status).toBe("discarded"); + }), + ), + ); +}); + +describe("PlanReviewService regressions", () => { + it.effect("keeps the lineage when the agent answers feedback", () => + runWithService({ sessionStatus: "running", compactionAt: null }, ({ service }) => + Effect.gen(function* () { + const document = yield* capturePlan(service, "plan:a"); + yield* service.upsertDiscussion({ + documentId: document.documentId, + discussionId: "discussion-1", + quotedText: "2. Backfill the rows", + bodyMarkdown: "Split this.", + actorUserId: reviewerId, + }); + yield* service.submit({ + documentId: document.documentId, + decision: "changes-requested", + globalComment: "", + editedMarkdown: null, + actorUserId: reviewerId, + actorLabel: "Tushar", + }); + + // The agent's answer must append to the same document, not start a new + // history that orphans the comments that asked for it. + const revised = yield* capturePlan(service, "plan:b", `${PLAN}\n4. Announce it`); + expect(revised.documentId).toBe(document.documentId); + expect(revised.status).toBe("open"); + + const snapshot = yield* service.getReview(document.documentId); + expect(snapshot.versions).toHaveLength(2); + expect(snapshot.versions[1]?.origin).toBe("agent-revision"); + }), + ), + ); + + it.effect("does not re-send comments that already reached the agent", () => + runWithService({ sessionStatus: "running", compactionAt: null }, ({ service, dispatched }) => + Effect.gen(function* () { + const document = yield* capturePlan(service, "plan:a"); + yield* service.upsertDiscussion({ + documentId: document.documentId, + discussionId: "discussion-1", + quotedText: "2. Backfill the rows", + bodyMarkdown: "Split this.", + actorUserId: reviewerId, + }); + yield* service.submit({ + documentId: document.documentId, + decision: "changes-requested", + globalComment: "", + editedMarkdown: null, + actorUserId: reviewerId, + actorLabel: "Tushar", + }); + yield* capturePlan(service, "plan:b", `${PLAN}\n4. Announce it`); + + yield* service.submit({ + documentId: document.documentId, + decision: "changes-requested", + globalComment: "Second round.", + editedMarkdown: null, + actorUserId: reviewerId, + actorLabel: "Tushar", + }); + + const turns = (yield* Ref.get(dispatched)).filter( + (command) => command.type === "thread.turn.start", + ); + expect(turns).toHaveLength(2); + const second = turns[1]; + const secondText = second?.type === "thread.turn.start" ? second.message.text : ""; + expect(secondText).toContain("Second round."); + expect(secondText).not.toContain("Split this."); + }), + ), + ); + + it.effect("refuses a second decision on the same review", () => + runWithService({ sessionStatus: "running", compactionAt: null }, ({ service, dispatched }) => + Effect.gen(function* () { + const document = yield* capturePlan(service, "plan:a"); + const approve = { + documentId: document.documentId, + decision: "approved" as const, + globalComment: "", + editedMarkdown: null, + actorUserId: reviewerId, + actorLabel: "Tushar", + }; + + yield* service.submit(approve); + const second = yield* service.submit(approve).pipe(Effect.exit); + expect(second._tag).toBe("Failure"); + + // The agent must not be told to implement the plan twice. + const turns = (yield* Ref.get(dispatched)).filter( + (command) => command.type === "thread.turn.start", + ); + expect(turns).toHaveLength(1); + }), + ), + ); + + it.effect("refuses to edit a review that is no longer open", () => + runWithService({ sessionStatus: "running", compactionAt: null }, ({ service }) => + Effect.gen(function* () { + const document = yield* capturePlan(service, "plan:a"); + yield* service.submit({ + documentId: document.documentId, + decision: "discarded", + globalComment: "", + editedMarkdown: null, + actorUserId: reviewerId, + actorLabel: "Tushar", + }); + + const saved = yield* service + .saveDraft({ + documentId: document.documentId, + contentValueJson: '{"markdown":"late"}', + expectedRevisionToken: null, + actorUserId: reviewerId, + }) + .pipe(Effect.exit); + expect(saved._tag).toBe("Failure"); + }), + ), + ); + + it.effect("does not reach a discussion through a document the caller owns", () => + runWithService({ sessionStatus: "running", compactionAt: null }, ({ service }) => + Effect.gen(function* () { + const victim = yield* capturePlan(service, "plan:victim"); + yield* service.upsertDiscussion({ + documentId: victim.documentId, + discussionId: "discussion-victim", + quotedText: "2. Backfill the rows", + bodyMarkdown: "Split this.", + actorUserId: reviewerId, + }); + const attacker = yield* capturePlan(service, "plan:attacker", PLAN, otherThreadId); + + // The caller authorizes their own document, then names a discussion id + // from a thread they cannot read. Both writes must miss. + yield* service.resolveDiscussion({ + documentId: attacker.documentId, + discussionId: "discussion-victim", + isResolved: true, + actorUserId: reviewerId, + }); + yield* service.upsertDiscussion({ + documentId: attacker.documentId, + discussionId: "discussion-victim", + quotedText: "injected quote", + bodyMarkdown: "injected body", + actorUserId: reviewerId, + }); + + const snapshot = yield* service.getReview(victim.documentId); + expect(snapshot.discussions).toHaveLength(1); + expect(snapshot.discussions[0]?.isResolved).toBe(false); + expect(snapshot.discussions[0]?.quotedText).toBe("2. Backfill the rows"); + expect(snapshot.comments.map((comment) => comment.bodyMarkdown)).toEqual(["Split this."]); + }), + ), + ); + + it.effect("does not diff versions belonging to another document", () => + runWithService({ sessionStatus: "running", compactionAt: null }, ({ service }) => + Effect.gen(function* () { + const victim = yield* capturePlan(service, "plan:victim"); + const attacker = yield* capturePlan(service, "plan:attacker", PLAN, otherThreadId); + const victimVersions = yield* service.getReview(victim.documentId); + const versionId = victimVersions.versions[0]!.versionId; + + const result = yield* service + .getVersionDiff({ + documentId: attacker.documentId, + fromVersionId: versionId, + toVersionId: versionId, + }) + .pipe(Effect.exit); + expect(result._tag).toBe("Failure"); + }), + ), + ); +}); + +describe("PlanReviewService drafts", () => { + it.effect("rejects a save that carried a stale revision token", () => + runWithService({ sessionStatus: "running", compactionAt: null }, ({ service }) => + Effect.gen(function* () { + const document = yield* capturePlan(service, "plan:a"); + + const first = yield* service.saveDraft({ + documentId: document.documentId, + contentValueJson: '{"markdown":"one"}', + expectedRevisionToken: null, + actorUserId: reviewerId, + }); + + // A second writer who never saw `first` still holds the old token. + const conflict = yield* service + .saveDraft({ + documentId: document.documentId, + contentValueJson: '{"markdown":"two"}', + expectedRevisionToken: null, + actorUserId: reviewerId, + }) + .pipe(Effect.exit); + + expect(conflict._tag).toBe("Failure"); + + const accepted = yield* service.saveDraft({ + documentId: document.documentId, + contentValueJson: '{"markdown":"three"}', + expectedRevisionToken: first.revisionToken, + actorUserId: reviewerId, + }); + expect(accepted.revisionToken).not.toBe(first.revisionToken); + }), + ), + ); + + it.effect("rejects a stale token even after the draft was cleared", () => + runWithService({ sessionStatus: "running", compactionAt: null }, ({ service }) => + Effect.gen(function* () { + const document = yield* capturePlan(service, "plan:a"); + const first = yield* service.saveDraft({ + documentId: document.documentId, + contentValueJson: '{"markdown":"one"}', + expectedRevisionToken: null, + actorUserId: reviewerId, + }); + + // An agent revision invalidates the draft it was based on. + yield* capturePlan(service, "plan:b", `${PLAN}\n4. Announce it`); + + // Resurrecting it with the pre-revision token would record content + // against a version it was never derived from. + const stale = yield* service + .saveDraft({ + documentId: document.documentId, + contentValueJson: '{"markdown":"one"}', + expectedRevisionToken: first.revisionToken, + actorUserId: reviewerId, + }) + .pipe(Effect.exit); + expect(stale._tag).toBe("Failure"); + }), + ), + ); + + it.effect("clears the draft when an agent revision lands", () => + runWithService({ sessionStatus: "running", compactionAt: null }, ({ service }) => + Effect.gen(function* () { + const document = yield* capturePlan(service, "plan:a"); + yield* service.saveDraft({ + documentId: document.documentId, + contentValueJson: '{"markdown":"mine"}', + expectedRevisionToken: null, + actorUserId: reviewerId, + }); + + yield* capturePlan(service, "plan:b", `${PLAN}\n4. Announce it`); + + const snapshot = yield* service.getReview(document.documentId); + expect(snapshot.draft).toBeNull(); + }), + ), + ); +}); + +describe("PlanReviewService version diff", () => { + it.effect("renders a diff between two versions", () => + runWithService({ sessionStatus: "running", compactionAt: null }, ({ service }) => + Effect.gen(function* () { + const document = yield* capturePlan(service, "plan:a"); + yield* capturePlan(service, "plan:b", `${PLAN}\n4. Announce it`); + const snapshot = yield* service.getReview(document.documentId); + + const diff = yield* service.getVersionDiff({ + documentId: document.documentId, + fromVersionId: snapshot.versions[0]!.versionId, + toVersionId: snapshot.versions[1]!.versionId, + }); + + expect(diff.diff).toContain("diff --git a/Auth rewrite.md"); + expect(diff.diff).toContain("+4. Announce it"); + }), + ), + ); +}); diff --git a/apps/server/src/planreview/PlanReviewService.ts b/apps/server/src/planreview/PlanReviewService.ts new file mode 100644 index 00000000000..86924ee2760 --- /dev/null +++ b/apps/server/src/planreview/PlanReviewService.ts @@ -0,0 +1,859 @@ +/** + * T3-CUSTOM(expbkt3): native plan review service. + * + * Owns the plan document lifecycle: capture the agent's plan as version 1, + * accumulate attributed human edits and discussions, cut new versions, and + * feed the outcome back into the thread as a normal turn. Nothing here is an + * orchestration aggregate — the review lives in fork-owned tables and reaches + * the thread only through the existing `thread.activity.append` and + * `thread.turn.start` commands, so upstream contracts are untouched. + */ +import { + CommandId, + EventId, + MessageId, + ThreadId, + type OrchestrationCommand, + type OrchestrationProposedPlanId, + type UserId, +} from "@t3tools/contracts"; +import { + buildPlanReviewApprovalPrompt, + buildPlanReviewFeedbackPrompt, + locateQuotedLineRange, + type PlanReviewAnchoredComment, +} from "@t3tools/shared/planReview"; +import * as Context from "effect/Context"; +import * as Crypto from "effect/Crypto"; +import * as DateTime from "effect/DateTime"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as PubSub from "effect/PubSub"; +import * as Schema from "effect/Schema"; +import * as Stream from "effect/Stream"; + +import { OrchestrationCommandDispatcher } from "../orchestration/dispatchCommand.ts"; +import { ProjectionSnapshotQuery } from "../orchestration/Services/ProjectionSnapshotQuery.ts"; +import { + PlanDraftConflictError, + PlanReviewRepository, + PlanVersionConflictError, + type PlanDiscussionCommentRecord, + type PlanDiscussionRecord, + type PlanDocumentRecord, + type PlanDocumentStatus, + type PlanVersionRecord, +} from "../persistence/PlanReviewDocuments.ts"; +import { buildUnifiedDiff, toRenderableFileDiff } from "./planReviewDiff.ts"; +import { + decidePlanResend, + shouldSendFullDocumentInsteadOfDiff, +} from "./PlanReviewContextPolicy.ts"; + +export class PlanReviewNotFoundError extends Schema.TaggedErrorClass()( + "PlanReviewNotFoundError", + { documentId: Schema.String }, +) {} + +export class PlanReviewInvariantError extends Schema.TaggedErrorClass()( + "PlanReviewInvariantError", + { operation: Schema.String, detail: Schema.String }, +) {} + +export type PlanReviewServiceError = + | PlanReviewNotFoundError + | PlanReviewInvariantError + | PlanDraftConflictError + | PlanVersionConflictError; + +export interface PlanReviewSnapshot { + readonly document: PlanDocumentRecord; + readonly versions: ReadonlyArray; + readonly draft: { + readonly contentValueJson: string; + readonly baseVersionId: string; + readonly revisionToken: string; + readonly updatedByUserId: UserId | null; + readonly updatedAt: string; + } | null; + readonly discussions: ReadonlyArray; + readonly comments: ReadonlyArray; +} + +export interface CapturePlanInput { + readonly threadId: ThreadId; + readonly projectId: string; + readonly planId: OrchestrationProposedPlanId; + readonly planMarkdown: string; + readonly title: string; + /** Null for agent-authored versions. */ + readonly authorUserId: UserId | null; +} + +export interface SubmitReviewInput { + readonly documentId: string; + readonly decision: "approved" | "changes-requested" | "discarded"; + readonly globalComment: string; + /** Reviewer-edited markdown, when the plan was edited. */ + readonly editedMarkdown: string | null; + readonly actorUserId: UserId | null; + readonly actorLabel: string | null; +} + +export interface SubmitReviewResult { + readonly documentId: string; + readonly status: PlanDocumentStatus; + /** The exact text handed to the agent, so tests and the UI can assert it. */ + readonly prompt: string | null; + readonly turnStarted: boolean; + readonly resentPlan: boolean; +} + +export class PlanReviewService extends Context.Service< + PlanReviewService, + { + readonly capturePlan: ( + input: CapturePlanInput, + ) => Effect.Effect; + readonly getReview: ( + documentId: string, + ) => Effect.Effect; + readonly listForThread: ( + threadId: ThreadId, + ) => Effect.Effect, PlanReviewServiceError>; + readonly saveDraft: (input: { + readonly documentId: string; + readonly contentValueJson: string; + readonly expectedRevisionToken: string | null; + readonly actorUserId: UserId | null; + }) => Effect.Effect<{ readonly revisionToken: string }, PlanReviewServiceError>; + readonly cutVersion: (input: { + readonly documentId: string; + readonly contentMarkdown: string; + readonly contentValueJson: string | null; + readonly summary: string | null; + readonly actorUserId: UserId | null; + }) => Effect.Effect; + readonly upsertDiscussion: (input: { + readonly documentId: string; + readonly discussionId: string; + readonly quotedText: string; + readonly bodyMarkdown: string; + readonly actorUserId: UserId | null; + }) => Effect.Effect; + readonly resolveDiscussion: (input: { + readonly documentId: string; + readonly discussionId: string; + readonly isResolved: boolean; + readonly actorUserId: UserId | null; + }) => Effect.Effect; + readonly getVersionDiff: (input: { + readonly documentId: string; + readonly fromVersionId: string; + readonly toVersionId: string; + }) => Effect.Effect<{ readonly diff: string }, PlanReviewServiceError>; + readonly submit: ( + input: SubmitReviewInput, + ) => Effect.Effect; + /** + * Emits a snapshot for `documentId` on subscribe and again after every + * mutation from any client, so open panels converge without polling. + */ + readonly watch: ( + documentId: string, + ) => Stream.Stream; + } +>()("t3/planreview/PlanReviewService") {} + +const nowIso = Effect.map(DateTime.now, DateTime.formatIso); + +/** First heading, else first non-empty line, capped so titles stay tab-sized. */ +export function derivePlanTitle(markdown: string): string { + for (const rawLine of markdown.split("\n")) { + const line = rawLine.trim(); + if (line.length === 0) continue; + const heading = line.match(/^#{1,6}\s+(.*)$/); + const candidate = (heading?.[1] ?? line).trim(); + if (candidate.length === 0) continue; + return candidate.length > 80 ? `${candidate.slice(0, 77)}…` : candidate; + } + return "Plan"; +} + +export const make = Effect.gen(function* () { + const repository = yield* PlanReviewRepository; + const dispatcher = yield* OrchestrationCommandDispatcher; + const query = yield* ProjectionSnapshotQuery; + const crypto = yield* Crypto.Crypto; + + // A failing CSPRNG is a defect, not something a caller can recover from. + const uuid = crypto.randomUUIDv4.pipe(Effect.orDie); + + // Mutations announce the document they touched; `watch` re-reads from there. + const changes = yield* PubSub.unbounded(); + const announce = (documentId: string) => PubSub.publish(changes, documentId).pipe(Effect.ignore); + + /** + * Repository and platform failures are infrastructure detail the caller + * cannot act on, so they collapse into one invariant error. The two conflict + * errors are the exception: callers retry or surface them to the reviewer. + */ + const asInvariant = + (operation: string) => + ( + effect: Effect.Effect, + ): Effect.Effect => + effect.pipe( + Effect.mapError( + (cause): PlanReviewServiceError => + cause._tag === "PlanDraftConflictError" || cause._tag === "PlanVersionConflictError" + ? (cause as unknown as PlanDraftConflictError | PlanVersionConflictError) + : new PlanReviewInvariantError({ operation, detail: cause.message }), + ), + ); + + const requireDocument = (documentId: string) => + repository.getDocument(documentId).pipe( + asInvariant("getDocument"), + Effect.flatMap( + Option.match({ + onNone: (): Effect.Effect => + Effect.fail(new PlanReviewNotFoundError({ documentId })), + onSome: Effect.succeed, + }), + ), + ); + + const appendActivity = (input: { + readonly threadId: ThreadId; + readonly summary: string; + readonly tone: "info" | "approval" | "error"; + readonly payload: unknown; + }) => + Effect.gen(function* () { + const [commandUuid, eventUuid, createdAt] = yield* Effect.all([uuid, uuid, nowIso]); + return yield* dispatcher.dispatch({ + type: "thread.activity.append", + commandId: CommandId.make(`plan-review:activity:${commandUuid}`), + threadId: input.threadId, + activity: { + id: EventId.make(`plan-review:${eventUuid}`), + tone: input.tone, + kind: "plan-review", + summary: input.summary, + payload: input.payload, + turnId: null, + createdAt, + }, + createdAt, + }); + }).pipe(Effect.ignore); + + const capturePlan: PlanReviewService["Service"]["capturePlan"] = (input) => + Effect.gen(function* () { + // A plan id we have already captured means this is a redelivery, not a + // new revision — return the existing document untouched. + const existingForPlan = yield* repository + .findDocumentBySourcePlanId(input.planId) + .pipe(asInvariant("capturePlan.findBySourcePlan")); + if (Option.isSome(existingForPlan)) return existingForPlan.value; + + // An open document on the same thread is the lineage this plan revises. + // A lineage awaiting a revision is still the lineage this plan belongs + // to. Only approved and discarded documents are closed for good — without + // "changes-requested" here, the agent's answer to feedback would start a + // brand-new history and orphan the comments that asked for it. + const threadDocuments = yield* repository + .listDocumentsForThread(input.threadId) + .pipe(asInvariant("capturePlan.listForThread")); + const openDocument = threadDocuments.find( + (document) => document.status === "open" || document.status === "changes-requested", + ); + + const createdAt = yield* nowIso; + + if (openDocument === undefined) { + const documentUuid = yield* uuid; + const documentId = `plan-doc:${documentUuid}`; + const versionUuid = yield* uuid; + + const document: PlanDocumentRecord = { + documentId, + threadId: input.threadId, + projectId: input.projectId, + title: input.title, + currentRevision: 1, + status: "open", + createdByUserId: input.authorUserId, + createdAt, + updatedAt: createdAt, + }; + + yield* repository.upsertDocument(document).pipe(asInvariant("capturePlan.upsertDocument")); + yield* repository + .appendVersion({ + versionId: `plan-ver:${versionUuid}`, + documentId, + revision: 1, + authorKind: "agent", + authorUserId: null, + origin: "agent-proposed", + contentMarkdown: input.planMarkdown, + contentValueJson: null, + sourcePlanId: input.planId, + summary: null, + createdAt, + }) + .pipe(asInvariant("capturePlan.appendVersion")); + + return document; + } + + // Revision of an existing lineage: skip when the content is unchanged so + // a redelivered projection event cannot inflate the history. + const latest = yield* repository + .getLatestVersion(openDocument.documentId) + .pipe(asInvariant("capturePlan.getLatestVersion")); + if ( + Option.isSome(latest) && + latest.value.contentMarkdown.trim() === input.planMarkdown.trim() + ) { + return openDocument; + } + + const nextRevision = openDocument.currentRevision + 1; + const versionUuid = yield* uuid; + yield* repository + .appendVersion({ + versionId: `plan-ver:${versionUuid}`, + documentId: openDocument.documentId, + revision: nextRevision, + authorKind: "agent", + authorUserId: null, + origin: "agent-revision", + contentMarkdown: input.planMarkdown, + contentValueJson: null, + sourcePlanId: input.planId, + summary: null, + createdAt, + }) + .pipe(asInvariant("capturePlan.appendRevision")); + + const updated: PlanDocumentRecord = { + ...openDocument, + title: input.title, + currentRevision: nextRevision, + // The revision answers the feedback, so the review is live again. + status: "open", + updatedAt: createdAt, + }; + yield* repository.upsertDocument(updated).pipe(asInvariant("capturePlan.updateDocument")); + + // A new agent revision invalidates the human draft it was based on. + yield* repository + .clearDraft(openDocument.documentId) + .pipe(asInvariant("capturePlan.clearDraft")); + + yield* announce(openDocument.documentId); + return updated; + }); + + const getReview: PlanReviewService["Service"]["getReview"] = (documentId) => + Effect.gen(function* () { + const document = yield* requireDocument(documentId); + const [versions, draftOption, discussions, comments] = yield* Effect.all([ + repository.listVersions(documentId).pipe(asInvariant("getReview.versions")), + repository.getDraft(documentId).pipe(asInvariant("getReview.draft")), + repository.listDiscussions(documentId).pipe(asInvariant("getReview.discussions")), + repository.listDiscussionComments(documentId).pipe(asInvariant("getReview.comments")), + ]); + + return { + document, + versions, + draft: Option.isSome(draftOption) + ? { + contentValueJson: draftOption.value.contentValueJson, + baseVersionId: draftOption.value.baseVersionId, + revisionToken: draftOption.value.revisionToken, + updatedByUserId: draftOption.value.updatedByUserId, + updatedAt: draftOption.value.updatedAt, + } + : null, + discussions, + comments, + } satisfies PlanReviewSnapshot; + }); + + const listForThread: PlanReviewService["Service"]["listForThread"] = (threadId) => + repository.listDocumentsForThread(threadId).pipe(asInvariant("listForThread")); + + const saveDraft: PlanReviewService["Service"]["saveDraft"] = (input) => + Effect.gen(function* () { + const document = yield* requireDocument(input.documentId); + if (document.status !== "open") { + return yield* new PlanReviewInvariantError({ + operation: "saveDraft", + detail: `This review is ${document.status} and can no longer be edited.`, + }); + } + const latest = yield* repository + .getLatestVersion(document.documentId) + .pipe(asInvariant("saveDraft.getLatestVersion")); + if (Option.isNone(latest)) { + return yield* new PlanReviewInvariantError({ + operation: "saveDraft", + detail: "The plan has no versions yet.", + }); + } + + const [tokenUuid, updatedAt] = yield* Effect.all([uuid, nowIso]); + const nextRevisionToken = `draft:${tokenUuid}`; + + yield* repository + .saveDraft({ + documentId: document.documentId, + baseVersionId: latest.value.versionId, + contentValueJson: input.contentValueJson, + updatedByUserId: input.actorUserId, + updatedAt, + expectedRevisionToken: input.expectedRevisionToken, + nextRevisionToken, + }) + .pipe(asInvariant("saveDraft")); + + // Deliberately not announced: a draft is one reviewer's working copy, and + // broadcasting it would push the whole version history on every keystroke. + return { revisionToken: nextRevisionToken }; + }); + + const cutVersion: PlanReviewService["Service"]["cutVersion"] = (input) => + Effect.gen(function* () { + const document = yield* requireDocument(input.documentId); + const latest = yield* repository + .getLatestVersion(document.documentId) + .pipe(asInvariant("cutVersion.getLatestVersion")); + + if ( + Option.isSome(latest) && + latest.value.contentMarkdown.trim() === input.contentMarkdown.trim() + ) { + return latest.value; + } + + const [versionUuid, createdAt] = yield* Effect.all([uuid, nowIso]); + const revision = document.currentRevision + 1; + const version: PlanVersionRecord = { + versionId: `plan-ver:${versionUuid}`, + documentId: document.documentId, + revision, + authorKind: "user", + authorUserId: input.actorUserId, + origin: "human-edit", + contentMarkdown: input.contentMarkdown, + contentValueJson: input.contentValueJson, + sourcePlanId: null, + summary: input.summary, + createdAt, + }; + + yield* repository.appendVersion(version).pipe(asInvariant("cutVersion.appendVersion")); + yield* repository + .upsertDocument({ ...document, currentRevision: revision, updatedAt: createdAt }) + .pipe(asInvariant("cutVersion.updateDocument")); + + yield* announce(document.documentId); + return version; + }); + + const upsertDiscussion: PlanReviewService["Service"]["upsertDiscussion"] = (input) => + Effect.gen(function* () { + const document = yield* requireDocument(input.documentId); + const latest = yield* repository + .getLatestVersion(document.documentId) + .pipe(asInvariant("upsertDiscussion.getLatestVersion")); + if (Option.isNone(latest)) { + return yield* new PlanReviewInvariantError({ + operation: "upsertDiscussion", + detail: "The plan has no versions yet.", + }); + } + + const [commentUuid, createdAt] = yield* Effect.all([uuid, nowIso]); + yield* repository + .upsertDiscussion({ + discussionId: input.discussionId, + documentId: document.documentId, + anchorVersionId: latest.value.versionId, + quotedText: input.quotedText, + createdByUserId: input.actorUserId, + createdAt, + }) + .pipe(asInvariant("upsertDiscussion")); + + yield* repository + .addDiscussionComment({ + commentId: `plan-comment:${commentUuid}`, + discussionId: input.discussionId, + documentId: document.documentId, + authorUserId: input.actorUserId, + bodyMarkdown: input.bodyMarkdown, + createdAt, + }) + .pipe(asInvariant("upsertDiscussion.addComment")); + + yield* announce(document.documentId); + }); + + const resolveDiscussion: PlanReviewService["Service"]["resolveDiscussion"] = (input) => + Effect.gen(function* () { + yield* requireDocument(input.documentId); + const resolvedAt = yield* nowIso; + yield* repository + .resolveDiscussion({ + discussionId: input.discussionId, + documentId: input.documentId, + isResolved: input.isResolved, + resolvedByUserId: input.isResolved ? input.actorUserId : null, + resolvedAt: input.isResolved ? resolvedAt : null, + }) + .pipe(asInvariant("resolveDiscussion")); + + yield* announce(input.documentId); + }); + + const getVersionDiff: PlanReviewService["Service"]["getVersionDiff"] = (input) => + Effect.gen(function* () { + const document = yield* requireDocument(input.documentId); + const [fromOption, toOption] = yield* Effect.all([ + repository + .getVersion({ documentId: document.documentId, versionId: input.fromVersionId }) + .pipe(asInvariant("getVersionDiff.from")), + repository + .getVersion({ documentId: document.documentId, versionId: input.toVersionId }) + .pipe(asInvariant("getVersionDiff.to")), + ]); + if (Option.isNone(fromOption) || Option.isNone(toOption)) { + return yield* new PlanReviewInvariantError({ + operation: "getVersionDiff", + detail: "One of the requested versions does not exist.", + }); + } + + const { diff } = buildUnifiedDiff( + fromOption.value.contentMarkdown, + toOption.value.contentMarkdown, + ); + return { diff: toRenderableFileDiff(`${document.title}.md`, diff) }; + }); + + /** Builds the anchored comment payloads the prompt embeds. */ + const buildAnchoredComments = ( + baseMarkdown: string, + discussions: ReadonlyArray, + comments: ReadonlyArray, + resolveLabel: (userId: UserId | null) => string | null, + ): { + readonly comments: ReadonlyArray; + readonly discussionIds: ReadonlyArray; + } => { + const byDiscussion = new Map(); + for (const comment of comments) { + const bucket = byDiscussion.get(comment.discussionId); + if (bucket) bucket.push(comment); + else byDiscussion.set(comment.discussionId, [comment]); + } + + const anchored: PlanReviewAnchoredComment[] = []; + const discussionIds: string[] = []; + for (const discussion of discussions) { + if (discussion.isResolved) continue; + const bucket = byDiscussion.get(discussion.discussionId) ?? []; + const body = bucket.map((comment) => comment.bodyMarkdown.trim()).join("\n\n"); + if (body.length === 0) continue; + + // A quote we cannot find still carries its text, just without a range. + const located = locateQuotedLineRange(baseMarkdown, discussion.quotedText); + anchored.push({ + startIndex: located?.startIndex ?? null, + endIndex: located?.endIndex ?? null, + quotedText: discussion.quotedText, + body, + authorLabel: resolveLabel(bucket[0]?.authorUserId ?? discussion.createdByUserId), + }); + discussionIds.push(discussion.discussionId); + } + return { comments: anchored, discussionIds }; + }; + + const submit: PlanReviewService["Service"]["submit"] = (input) => + Effect.gen(function* () { + const snapshot = yield* getReview(input.documentId); + const document = snapshot.document; + + // Two tabs, or a replayed request, must not start implementation twice. + if (document.status !== "open") { + return yield* new PlanReviewInvariantError({ + operation: "submit", + detail: `This review was already ${document.status}.`, + }); + } + + const latestVersion = snapshot.versions.at(-1); + if (latestVersion === undefined) { + return yield* new PlanReviewInvariantError({ + operation: "submit", + detail: "The plan has no versions yet.", + }); + } + + const resolveLabel = (userId: UserId | null) => + userId === null ? input.actorLabel : userId === input.actorUserId ? input.actorLabel : null; + + // Reviewer edits become a real version before anything is sent, so the + // history always explains what the agent was told. + const agentBaseline = + snapshot.versions.toReversed().find((version) => version.authorKind === "agent") ?? + latestVersion; + + let approvedVersion = latestVersion; + if ( + input.editedMarkdown !== null && + input.editedMarkdown.trim() !== latestVersion.contentMarkdown.trim() + ) { + approvedVersion = yield* cutVersion({ + documentId: document.documentId, + contentMarkdown: input.editedMarkdown, + contentValueJson: null, + summary: input.decision === "approved" ? "Edited before approval" : "Reviewer edit", + actorUserId: input.actorUserId, + }); + } + + const editResult = buildUnifiedDiff( + agentBaseline.contentMarkdown, + approvedVersion.contentMarkdown, + ); + + if (input.decision === "discarded") { + const discardedAt = yield* nowIso; + yield* repository + .setDocumentStatus({ + documentId: document.documentId, + status: "discarded", + updatedAt: discardedAt, + }) + .pipe(asInvariant("submit.discard")); + yield* appendActivity({ + threadId: document.threadId, + summary: "Plan review was discarded.", + tone: "error", + payload: { documentId: document.documentId, decision: "discarded" }, + }); + yield* announce(document.documentId); + return { + documentId: document.documentId, + status: "discarded", + prompt: null, + turnStarted: false, + resentPlan: false, + } satisfies SubmitReviewResult; + } + + const threadOption = yield* query + .getThreadDetailById(document.threadId) + .pipe(asInvariant("submit.getThread")); + if (Option.isNone(threadOption)) { + return yield* new PlanReviewInvariantError({ + operation: "submit", + detail: `Thread ${document.threadId} was not found.`, + }); + } + const thread = threadOption.value; + + let prompt: string; + let resentPlan = false; + let sentDiscussionIds: ReadonlyArray = []; + + if (input.decision === "approved") { + const latestCompactionAt = + [...thread.activities] + .filter((activity) => activity.kind === "context-compaction") + .map((activity) => activity.createdAt) + .sort() + .at(-1) ?? null; + + const resend = decidePlanResend({ + latestCompactionAt, + planCreatedAt: agentBaseline.createdAt, + planThreadId: document.threadId, + targetThreadId: document.threadId, + providerSessionStatus: thread.session?.status ?? null, + }); + resentPlan = resend.shouldResend; + + prompt = buildPlanReviewApprovalPrompt({ + notes: input.globalComment, + resendPlanMarkdown: resend.shouldResend ? approvedVersion.contentMarkdown : null, + resendReason: resend.reason, + approvedEditDiff: editResult.diff, + }); + } else { + const anchored = buildAnchoredComments( + agentBaseline.contentMarkdown, + snapshot.discussions, + snapshot.comments, + resolveLabel, + ); + sentDiscussionIds = anchored.discussionIds; + const sendFullDocument = shouldSendFullDocumentInsteadOfDiff(editResult.stats.changeRatio); + + prompt = buildPlanReviewFeedbackPrompt({ + documentId: document.documentId, + planTitle: document.title, + globalComment: input.globalComment, + comments: anchored.comments, + editDiff: editResult.diff, + fromRevision: agentBaseline.revision, + toRevision: approvedVersion.revision, + editAuthorLabel: input.actorLabel, + fullDocument: + sendFullDocument && editResult.diff.trim().length > 0 + ? approvedVersion.contentMarkdown + : null, + }); + } + + const [commandUuid, messageUuid, modeUuid, createdAt] = yield* Effect.all([ + uuid, + uuid, + uuid, + nowIso, + ]); + + // Only approval may leave Plan mode; feedback keeps the thread planning. + if (input.decision === "approved") { + const modeCommand: OrchestrationCommand = { + type: "thread.interaction-mode.set", + commandId: CommandId.make(`plan-review:mode:${modeUuid}`), + threadId: document.threadId, + interactionMode: "default", + createdAt, + }; + yield* dispatcher.dispatch(modeCommand).pipe(asInvariant("submit.setMode")); + } + + yield* dispatcher + .dispatch({ + type: "thread.turn.start", + commandId: CommandId.make(`plan-review:turn:${commandUuid}`), + threadId: document.threadId, + message: { + messageId: MessageId.make(`plan-review:${messageUuid}`), + role: "user", + text: prompt, + attachments: [], + }, + modelSelection: thread.modelSelection, + runtimeMode: thread.runtimeMode, + interactionMode: input.decision === "approved" ? "default" : "plan", + ...(input.decision === "approved" && approvedVersion.sourcePlanId !== null + ? { + sourceProposedPlan: { + threadId: document.threadId, + planId: approvedVersion.sourcePlanId as OrchestrationProposedPlanId, + }, + } + : input.decision === "approved" && agentBaseline.sourcePlanId !== null + ? { + sourceProposedPlan: { + threadId: document.threadId, + planId: agentBaseline.sourcePlanId as OrchestrationProposedPlanId, + }, + } + : {}), + createdAt, + }) + .pipe(asInvariant("submit.startTurn")); + + const status: PlanDocumentStatus = + input.decision === "approved" ? "approved" : "changes-requested"; + yield* repository + .setDocumentStatus({ + documentId: document.documentId, + status, + updatedAt: createdAt, + }) + .pipe(asInvariant("submit.setStatus")); + yield* repository.clearDraft(document.documentId).pipe(asInvariant("submit.clearDraft")); + + // Anything already handed to the agent is spent. Leaving it open would + // re-send the same comments on every later round. + if (input.decision === "changes-requested") { + yield* Effect.forEach( + sentDiscussionIds, + (discussionId) => + repository + .resolveDiscussion({ + discussionId, + documentId: document.documentId, + isResolved: true, + resolvedByUserId: input.actorUserId, + resolvedAt: createdAt, + }) + .pipe(asInvariant("submit.consumeDiscussions")), + { discard: true }, + ); + } + + yield* announce(document.documentId); + + yield* appendActivity({ + threadId: document.threadId, + summary: + input.decision === "approved" + ? "Plan approved; implementation was started." + : "Plan feedback was sent to the planning agent.", + tone: input.decision === "approved" ? "approval" : "info", + payload: { + documentId: document.documentId, + decision: input.decision, + revision: approvedVersion.revision, + resentPlan, + }, + }); + + return { + documentId: document.documentId, + status, + prompt, + turnStarted: true, + resentPlan, + } satisfies SubmitReviewResult; + }); + + const watch: PlanReviewService["Service"]["watch"] = (documentId) => + Stream.concat( + Stream.fromEffect(getReview(documentId)), + Stream.fromPubSub(changes).pipe( + Stream.filter((changed) => changed === documentId), + Stream.mapEffect(() => getReview(documentId)), + ), + ); + + return PlanReviewService.of({ + capturePlan, + getReview, + listForThread, + saveDraft, + cutVersion, + upsertDiscussion, + resolveDiscussion, + getVersionDiff, + submit, + watch, + }); +}); + +export const layer = Layer.effect(PlanReviewService, make); diff --git a/apps/server/src/planreview/planReviewDiff.test.ts b/apps/server/src/planreview/planReviewDiff.test.ts new file mode 100644 index 00000000000..75082148a9d --- /dev/null +++ b/apps/server/src/planreview/planReviewDiff.test.ts @@ -0,0 +1,90 @@ +import { describe, expect, it } from "vite-plus/test"; + +import { buildUnifiedDiff, toRenderableFileDiff } from "./planReviewDiff.ts"; + +describe("buildUnifiedDiff", () => { + it("returns an empty diff for identical documents", () => { + const result = buildUnifiedDiff("# Plan\n\nStep one.\n", "# Plan\n\nStep one.\n"); + expect(result.diff).toBe(""); + expect(result.stats).toEqual({ added: 0, removed: 0, changeRatio: 0 }); + }); + + it("ignores a trailing-newline-only difference", () => { + expect(buildUnifiedDiff("a\nb\n", "a\nb").diff).toBe(""); + }); + + it("emits a hunk with surrounding context for a single changed line", () => { + const before = ["one", "two", "three", "four", "five"].join("\n"); + const after = ["one", "two", "THREE", "four", "five"].join("\n"); + const result = buildUnifiedDiff(before, after); + + expect(result.diff).toBe( + ["@@ -1,5 +1,5 @@", " one", " two", "-three", "+THREE", " four", " five"].join("\n"), + ); + expect(result.stats.added).toBe(1); + expect(result.stats.removed).toBe(1); + }); + + it("counts pure additions without removals", () => { + const result = buildUnifiedDiff("one\ntwo", "one\ntwo\nthree"); + expect(result.stats.added).toBe(1); + expect(result.stats.removed).toBe(0); + expect(result.diff).toContain("+three"); + }); + + it("splits distant changes into separate hunks", () => { + const before = Array.from({ length: 40 }, (_, index) => `line ${index}`).join("\n"); + const after = before.replace("line 2", "CHANGED 2").replace("line 35", "CHANGED 35"); + const result = buildUnifiedDiff(before, after); + + expect(result.diff.match(/^@@ /gm)).toHaveLength(2); + }); + + it("counts a modified line once, not as an add plus a remove", () => { + const before = ["a", "b", "c", "d"].join("\n"); + const after = ["a", "B", "C", "D"].join("\n"); + // 3 of 4 lines changed — not 6 of 4. + expect(buildUnifiedDiff(before, after).stats.changeRatio).toBeCloseTo(0.75); + + const light = buildUnifiedDiff( + Array.from({ length: 20 }, (_, index) => `line ${index}`).join("\n"), + Array.from({ length: 20 }, (_, index) => (index === 5 ? "changed" : `line ${index}`)).join( + "\n", + ), + ); + expect(light.stats.changeRatio).toBeCloseTo(0.05); + }); + + it("keeps a half-rewritten document under the full-document threshold", () => { + const before = Array.from({ length: 40 }, (_, index) => `line ${index}`).join("\n"); + const after = Array.from({ length: 40 }, (_, index) => + index < 15 ? `rewritten ${index}` : `line ${index}`, + ).join("\n"); + // 15 of 40 lines reworded is a diff worth sending, not a rewrite. + expect(buildUnifiedDiff(before, after).stats.changeRatio).toBeCloseTo(0.375); + }); + + it("handles an empty document on either side", () => { + expect(buildUnifiedDiff("", "new line").stats.added).toBe(1); + expect(buildUnifiedDiff("old line", "").stats.removed).toBe(1); + }); + + it("normalizes CRLF so a line-ending change is not a diff", () => { + expect(buildUnifiedDiff("a\r\nb", "a\nb").diff).toBe(""); + }); +}); + +describe("toRenderableFileDiff", () => { + it("wraps a diff in git headers the diff viewer understands", () => { + const wrapped = toRenderableFileDiff("Auth rewrite.md", "@@ -1,1 +1,1 @@\n-a\n+b"); + expect(wrapped.split("\n").slice(0, 3)).toEqual([ + "diff --git a/Auth rewrite.md b/Auth rewrite.md", + "--- a/Auth rewrite.md", + "+++ b/Auth rewrite.md", + ]); + }); + + it("returns an empty string when there is nothing to render", () => { + expect(toRenderableFileDiff("Plan.md", "")).toBe(""); + }); +}); diff --git a/apps/server/src/planreview/planReviewDiff.ts b/apps/server/src/planreview/planReviewDiff.ts new file mode 100644 index 00000000000..100cc03d270 --- /dev/null +++ b/apps/server/src/planreview/planReviewDiff.ts @@ -0,0 +1,190 @@ +/** + * T3-CUSTOM(expbkt3): line diff for plan versions. + * + * Deliberately dependency-free. jsdiff is BSD-3 and `@pierre/diffs` parses + * diffs rather than producing them, so the ~80 lines of LCS here buy us a + * unified diff the existing diff viewer can render without a new licence. + */ + +export interface UnifiedDiffStats { + readonly added: number; + readonly removed: number; + /** Changed lines as a fraction of the larger side, 0–1. */ + readonly changeRatio: number; +} + +export interface UnifiedDiffResult { + readonly diff: string; + readonly stats: UnifiedDiffStats; +} + +type Op = { readonly kind: "context" | "add" | "remove"; readonly line: string }; + +function splitLines(value: string): ReadonlyArray { + const normalized = value.replaceAll("\r\n", "\n"); + const lines = normalized.split("\n"); + // A trailing newline yields a final empty element that is not a real line. + return lines.length > 1 && lines.at(-1) === "" ? lines.slice(0, -1) : lines; +} + +/** + * Longest common subsequence over whole lines. Plans are at most a few hundred + * lines, so the O(n*m) table is fine and keeps the implementation obvious. + */ +function diffOps(before: ReadonlyArray, after: ReadonlyArray): ReadonlyArray { + const rows = before.length; + const cols = after.length; + const table: number[][] = Array.from({ length: rows + 1 }, () => + Array.from({ length: cols + 1 }).fill(0), + ); + + for (let i = rows - 1; i >= 0; i -= 1) { + for (let j = cols - 1; j >= 0; j -= 1) { + table[i]![j] = + before[i] === after[j] + ? table[i + 1]![j + 1]! + 1 + : Math.max(table[i + 1]![j]!, table[i]![j + 1]!); + } + } + + const ops: Op[] = []; + let i = 0; + let j = 0; + while (i < rows && j < cols) { + if (before[i] === after[j]) { + ops.push({ kind: "context", line: before[i]! }); + i += 1; + j += 1; + } else if (table[i + 1]![j]! >= table[i]![j + 1]!) { + ops.push({ kind: "remove", line: before[i]! }); + i += 1; + } else { + ops.push({ kind: "add", line: after[j]! }); + j += 1; + } + } + while (i < rows) { + ops.push({ kind: "remove", line: before[i]! }); + i += 1; + } + while (j < cols) { + ops.push({ kind: "add", line: after[j]! }); + j += 1; + } + return ops; +} + +interface Hunk { + readonly beforeStart: number; + readonly beforeCount: number; + readonly afterStart: number; + readonly afterCount: number; + readonly lines: ReadonlyArray; +} + +const CONTEXT_LINES = 3; + +function buildHunks(ops: ReadonlyArray): ReadonlyArray { + const changedIndices = ops.flatMap((op, index) => (op.kind === "context" ? [] : [index])); + if (changedIndices.length === 0) return []; + + // Group changes that sit within 2*CONTEXT_LINES of each other into one hunk. + const groups: Array<{ start: number; end: number }> = []; + for (const index of changedIndices) { + const last = groups.at(-1); + if (last && index - last.end <= CONTEXT_LINES * 2) { + last.end = index; + continue; + } + groups.push({ start: index, end: index }); + } + + const hunks: Hunk[] = []; + for (const group of groups) { + const from = Math.max(0, group.start - CONTEXT_LINES); + const to = Math.min(ops.length - 1, group.end + CONTEXT_LINES); + + let beforeLine = 1; + let afterLine = 1; + for (let index = 0; index < from; index += 1) { + const op = ops[index]!; + if (op.kind !== "add") beforeLine += 1; + if (op.kind !== "remove") afterLine += 1; + } + + let beforeCount = 0; + let afterCount = 0; + const lines: string[] = []; + for (let index = from; index <= to; index += 1) { + const op = ops[index]!; + if (op.kind === "context") { + beforeCount += 1; + afterCount += 1; + lines.push(` ${op.line}`); + } else if (op.kind === "remove") { + beforeCount += 1; + lines.push(`-${op.line}`); + } else { + afterCount += 1; + lines.push(`+${op.line}`); + } + } + + hunks.push({ + beforeStart: beforeLine, + beforeCount, + afterStart: afterLine, + afterCount, + lines, + }); + } + return hunks; +} + +/** Builds a unified diff between two markdown documents. Empty when identical. */ +export function buildUnifiedDiff(before: string, after: string): UnifiedDiffResult { + const beforeLines = splitLines(before); + const afterLines = splitLines(after); + const ops = diffOps(beforeLines, afterLines); + + const added = ops.filter((op) => op.kind === "add").length; + const removed = ops.filter((op) => op.kind === "remove").length; + const denominator = Math.max(beforeLines.length, afterLines.length, 1); + const stats: UnifiedDiffStats = { + added, + removed, + // A modified line shows up as one add and one remove, so summing them would + // report twice the fraction of the document that actually moved — and fire + // the "send the whole document" guard at half its stated threshold. + changeRatio: Math.min(1, Math.max(added, removed) / denominator), + }; + + const hunks = buildHunks(ops); + if (hunks.length === 0) return { diff: "", stats }; + + const body = hunks + .map((hunk) => + [ + `@@ -${hunk.beforeStart},${hunk.beforeCount} +${hunk.afterStart},${hunk.afterCount} @@`, + ...hunk.lines, + ].join("\n"), + ) + .join("\n"); + + return { diff: body, stats }; +} + +/** + * Wraps a unified diff in git headers so `@pierre/diffs` can render it as a + * file diff. `fileName` is cosmetic — plans have no path on disk. + */ +export function toRenderableFileDiff(fileName: string, diff: string): string { + if (diff.trim().length === 0) return ""; + const safeName = fileName.replaceAll("\\", "/"); + return [ + `diff --git a/${safeName} b/${safeName}`, + `--- a/${safeName}`, + `+++ b/${safeName}`, + diff, + ].join("\n"); +} diff --git a/apps/server/src/server.test.ts b/apps/server/src/server.test.ts index 00d1f5da7e9..5e4ca456594 100644 --- a/apps/server/src/server.test.ts +++ b/apps/server/src/server.test.ts @@ -141,6 +141,9 @@ import * as TerminalManager from "./terminal/Manager.ts"; import * as PreviewManager from "./preview/Manager.ts"; import * as PortScanner from "./preview/PortScanner.ts"; import * as PlannotatorManager from "./plannotator/PlannotatorManager.ts"; +// T3-CUSTOM(expbkt3): native plan review. +import * as PlanReviewDocuments from "./persistence/PlanReviewDocuments.ts"; +import * as PlanReviewServiceLayer from "./planreview/PlanReviewService.ts"; import * as BrowserTraceCollector from "./observability/BrowserTraceCollector.ts"; import * as ProjectFaviconResolver from "./project/ProjectFaviconResolver.ts"; import * as T3ProjectFileLoader from "./project/T3ProjectFileLoader.ts"; @@ -690,6 +693,12 @@ const buildAppUnderTest = (options?: { }, ).pipe( Layer.provide(PlannotatorManager.layer), + // T3-CUSTOM(expbkt3): native plan review service for the fork RPC handlers. + Layer.provide( + PlanReviewServiceLayer.layer.pipe( + Layer.provide(PlanReviewDocuments.layer.pipe(Layer.provide(SqlitePersistenceMemory))), + ), + ), Layer.provide( OrchestrationCommandDispatcher.layerWithBootstrapRepository.pipe( Layer.provide( diff --git a/apps/server/src/server.ts b/apps/server/src/server.ts index 9b37d2ce23d..8d4a359cffa 100644 --- a/apps/server/src/server.ts +++ b/apps/server/src/server.ts @@ -50,6 +50,10 @@ import { mcpUpstreamProxyRouteLayer } from "./mcp/McpUpstreamProxy.ts"; import * as PreviewAutomationBroker from "./mcp/PreviewAutomationBroker.ts"; // T3-CUSTOM(expbkt3): BEGIN — experimental native-plan review runtime. import * as PlannotatorManager from "./plannotator/PlannotatorManager.ts"; +// T3-CUSTOM(expbkt3): native plan review. +import * as PlanIngestListener from "./planreview/PlanIngestListener.ts"; +import * as PlanReviewServiceLayer from "./planreview/PlanReviewService.ts"; +import * as PlanReviewDocuments from "./persistence/PlanReviewDocuments.ts"; import { plannotatorProxyRouteLayer } from "./plannotator/http.ts"; // T3-CUSTOM(expbkt3): END import * as PreviewManager from "./preview/Manager.ts"; @@ -768,9 +772,17 @@ export const makeServerLayer = Layer.unwrap( runtimeBaseServicesLive, OrchestrationCommandDispatcher.layer.pipe(Layer.provide(runtimeBaseServicesLive)), ); - const runtimeServicesLive = PlannotatorManager.layer.pipe( + // T3-CUSTOM(expbkt3): native plan review sits beside Plannotator; both read + // the same proposed-plan events and neither depends on the other. + const planReviewServicesLive = PlanReviewServiceLayer.layer.pipe( + Layer.provide(PlanReviewDocuments.layer), Layer.provideMerge(runtimeServicesWithoutPlannotatorLive), ); + const runtimeServicesLive = Layer.mergeAll( + PlannotatorManager.layer.pipe(Layer.provideMerge(runtimeServicesWithoutPlannotatorLive)), + PlanIngestListener.layer.pipe(Layer.provideMerge(planReviewServicesLive)), + planReviewServicesLive, + ); const routesLayer = HttpRouter.serve(makeRoutesLayer.pipe(Layer.provide(launcherLayer)), { disableLogger: !config.logWebSocketEvents, diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 0b499ee1a68..c1bbdc8aa1b 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -109,6 +109,8 @@ import * as ServerSettings from "./serverSettings.ts"; import * as TerminalManager from "./terminal/Manager.ts"; import * as PreviewAutomationBroker from "./mcp/PreviewAutomationBroker.ts"; import * as UserMcpProfileStore from "./mcp/UserMcpProfileStore.ts"; +// T3-CUSTOM(expbkt3): native plan review service. +import { PlanReviewService } from "./planreview/PlanReviewService.ts"; import * as PreviewManager from "./preview/Manager.ts"; import { issueAssetUrl } from "./assets/AssetAccess.ts"; import * as PortScanner from "./preview/PortScanner.ts"; @@ -428,6 +430,20 @@ const makeWsRpcLayer = ( ); const serverAuth = yield* EnvironmentAuth.EnvironmentAuth; const environmentUsers = yield* EnvironmentUserService.EnvironmentUserService; + // T3-CUSTOM(expbkt3): BEGIN native plan review connection state. + const planReview = yield* PlanReviewService; + // Resolved once per connection: it only labels this actor's own comments. + const actorLabel = + actorUserId === null + ? null + : yield* clerkDirectory.listOrgMembers().pipe( + Effect.map((members) => { + const member = members.find((user) => user.id === actorUserId); + return member?.name ?? member?.email ?? null; + }), + Effect.orElseSucceed(() => null), + ); + // T3-CUSTOM(expbkt3): END native plan review connection state. const sourceControlDiscovery = yield* SourceControlDiscovery.SourceControlDiscovery; const automaticGitFetchInterval = serverSettings.getSettings.pipe( Effect.map( @@ -1120,6 +1136,8 @@ const makeWsRpcLayer = ( httpClient, sourceControlProfiles, environmentUsers, + planReview, + actorLabel, systemResourceMonitor, providerRateLimits, projectionSnapshotQuery, diff --git a/apps/server/src/wsForkHandlers.ts b/apps/server/src/wsForkHandlers.ts index b658df4ebe7..995499eb160 100644 --- a/apps/server/src/wsForkHandlers.ts +++ b/apps/server/src/wsForkHandlers.ts @@ -18,6 +18,7 @@ import { WsRpcGroup, EnvironmentAuthorizationError, OrchestrationGetSnapshotError, + PlanReviewError, SourceControlProfileError, type AuthSessionId, type OrchestrationEvent, @@ -33,6 +34,7 @@ import type * as EnvironmentUserService from "./auth/EnvironmentUserService.ts"; import type * as GitVcsDriver from "./vcs/GitVcsDriver.ts"; import type * as UserMcpProfileStore from "./mcp/UserMcpProfileStore.ts"; import type * as OrchestrationEngine from "./orchestration/Services/OrchestrationEngine.ts"; +import type * as PlanReviewService from "./planreview/PlanReviewService.ts"; import type * as ProjectionSnapshotQuery from "./orchestration/Services/ProjectionSnapshotQuery.ts"; import type * as SystemResourceMonitor from "./observability/SystemResourceMonitor.ts"; import type { ProviderRateLimitsShape } from "./provider/ProviderRateLimits.ts"; @@ -58,6 +60,10 @@ export interface ForkWsHandlerDeps { readonly httpClient: HttpClient.HttpClient; readonly sourceControlProfiles: SourceControlProfileService.SourceControlProfileService["Service"]; readonly environmentUsers: EnvironmentUserService.EnvironmentUserService["Service"]; + // T3-CUSTOM(expbkt3): native plan review. + readonly planReview: PlanReviewService.PlanReviewService["Service"]; + /** Display name for the acting user, stamped onto their review comments. */ + readonly actorLabel: string | null; readonly systemResourceMonitor: SystemResourceMonitor.SystemResourceMonitor["Service"]; readonly providerRateLimits: ProviderRateLimitsShape; readonly projectionSnapshotQuery: ProjectionSnapshotQuery.ProjectionSnapshotQuery["Service"]; @@ -103,6 +109,8 @@ export const makeForkWsHandlers = ({ httpClient, sourceControlProfiles, environmentUsers, + planReview, + actorLabel, systemResourceMonitor, providerRateLimits, projectionSnapshotQuery, @@ -115,8 +123,47 @@ export const makeForkWsHandlers = ({ observeRpcStream, requireThreadAccess, visibleAggregateIdsForActor, -}: ForkWsHandlerDeps) => - ({ +}: ForkWsHandlerDeps) => { + // T3-CUSTOM(expbkt3): BEGIN native plan review helpers. + const planReviewAccessError = (cause: OrchestrationGetSnapshotError) => + new PlanReviewError({ operation: "access", reason: "not-found", detail: cause.message }); + + const toPlanReviewError = + (operation: string) => (cause: { readonly _tag: string; readonly message: string }) => + cause._tag === "PlanReviewError" + ? (cause as unknown as PlanReviewError) + : new PlanReviewError({ + operation, + reason: + cause._tag === "PlanReviewNotFoundError" + ? "not-found" + : cause._tag === "PlanDraftConflictError" + ? "draft-conflict" + : cause._tag === "PlanVersionConflictError" + ? "version-conflict" + : "invalid", + detail: cause.message, + }); + + /** + * Reviews are reachable only through the thread that owns them, so every + * entry point resolves the document first and then applies thread access. + * A denial reads as "not found" so the check cannot leak existence. + */ + const guardDocument = (documentId: string) => + planReview.getReview(documentId).pipe( + Effect.mapError(toPlanReviewError("access")), + Effect.tap((snapshot) => + requireThreadAccess(snapshot.document.threadId).pipe( + Effect.mapError(planReviewAccessError), + ), + ), + ); + + const guardedReview = (documentId: string) => guardDocument(documentId); + // T3-CUSTOM(expbkt3): END native plan review helpers. + + return { [WS_METHODS.personalMcpGetProfile]: (_input) => observeRpcEffect( WS_METHODS.personalMcpGetProfile, @@ -332,4 +379,127 @@ export const makeForkWsHandlers = ({ observeRpcStream(WS_METHODS.subscribeProviderRateLimits, providerRateLimits.stream, { "rpc.aggregate": "server", }), - }) satisfies ForkWsHandlers; + // T3-CUSTOM(expbkt3): BEGIN native plan review. + [WS_METHODS.planReviewGet]: (input) => + observeRpcEffect(WS_METHODS.planReviewGet, guardedReview(input.documentId), { + "rpc.aggregate": "plan-review", + }), + [WS_METHODS.planReviewList]: (input) => + observeRpcEffect( + WS_METHODS.planReviewList, + requireThreadAccess(input.threadId).pipe( + Effect.mapError(planReviewAccessError), + Effect.andThen(planReview.listForThread(input.threadId)), + Effect.map((documents) => ({ documents })), + Effect.mapError(toPlanReviewError("list")), + ), + { "rpc.aggregate": "plan-review" }, + ), + [WS_METHODS.planReviewSaveDraft]: (input) => + observeRpcEffect( + WS_METHODS.planReviewSaveDraft, + guardDocument(input.documentId).pipe( + Effect.andThen( + planReview.saveDraft({ + documentId: input.documentId, + contentValueJson: input.contentValueJson, + expectedRevisionToken: input.expectedRevisionToken, + actorUserId, + }), + ), + Effect.mapError(toPlanReviewError("saveDraft")), + ), + { "rpc.aggregate": "plan-review" }, + ), + [WS_METHODS.planReviewCutVersion]: (input) => + observeRpcEffect( + WS_METHODS.planReviewCutVersion, + guardDocument(input.documentId).pipe( + Effect.andThen( + planReview.cutVersion({ + documentId: input.documentId, + contentMarkdown: input.contentMarkdown, + contentValueJson: input.contentValueJson, + summary: input.summary, + actorUserId, + }), + ), + Effect.andThen(planReview.getReview(input.documentId)), + Effect.mapError(toPlanReviewError("cutVersion")), + ), + { "rpc.aggregate": "plan-review" }, + ), + [WS_METHODS.planReviewUpsertDiscussion]: (input) => + observeRpcEffect( + WS_METHODS.planReviewUpsertDiscussion, + guardDocument(input.documentId).pipe( + Effect.andThen( + planReview.upsertDiscussion({ + documentId: input.documentId, + discussionId: input.discussionId, + quotedText: input.quotedText, + bodyMarkdown: input.bodyMarkdown, + actorUserId, + }), + ), + Effect.andThen(planReview.getReview(input.documentId)), + Effect.mapError(toPlanReviewError("upsertDiscussion")), + ), + { "rpc.aggregate": "plan-review" }, + ), + [WS_METHODS.planReviewResolveDiscussion]: (input) => + observeRpcEffect( + WS_METHODS.planReviewResolveDiscussion, + guardDocument(input.documentId).pipe( + Effect.andThen( + planReview.resolveDiscussion({ + documentId: input.documentId, + discussionId: input.discussionId, + isResolved: input.isResolved, + actorUserId, + }), + ), + Effect.andThen(planReview.getReview(input.documentId)), + Effect.mapError(toPlanReviewError("resolveDiscussion")), + ), + { "rpc.aggregate": "plan-review" }, + ), + [WS_METHODS.planReviewVersionDiff]: (input) => + observeRpcEffect( + WS_METHODS.planReviewVersionDiff, + guardDocument(input.documentId).pipe( + Effect.andThen(planReview.getVersionDiff(input)), + Effect.mapError(toPlanReviewError("versionDiff")), + ), + { "rpc.aggregate": "plan-review" }, + ), + [WS_METHODS.planReviewSubmit]: (input) => + observeRpcEffect( + WS_METHODS.planReviewSubmit, + guardDocument(input.documentId).pipe( + Effect.andThen( + planReview.submit({ + documentId: input.documentId, + decision: input.decision, + globalComment: input.globalComment, + editedMarkdown: input.editedMarkdown, + actorUserId, + actorLabel, + }), + ), + Effect.mapError(toPlanReviewError("submit")), + ), + { "rpc.aggregate": "plan-review" }, + ), + [WS_METHODS.subscribePlanReview]: (input) => + observeRpcStream( + WS_METHODS.subscribePlanReview, + Stream.fromEffect(guardDocument(input.documentId)).pipe( + Stream.flatMap(() => planReview.watch(input.documentId)), + Stream.mapError(toPlanReviewError("watch")), + ), + { "rpc.aggregate": "plan-review" }, + ), + // T3-CUSTOM(expbkt3): END native plan review. + } satisfies ForkWsHandlers; +}; diff --git a/apps/web/package.json b/apps/web/package.json index f0e5a076e84..1c8fe5d1a57 100644 --- a/apps/web/package.json +++ b/apps/web/package.json @@ -26,16 +26,26 @@ "@lexical/react": "^0.41.0", "@pierre/diffs": "catalog:", "@pierre/trees": "1.0.0-beta.4", + "@platejs/basic-nodes": "53.0.0", + "@platejs/code-block": "53.0.0", + "@platejs/comment": "53.0.0", + "@platejs/link": "53.3.1", + "@platejs/list": "53.1.3", + "@platejs/markdown": "53.3.3", + "@platejs/suggestion": "53.2.3", + "@platejs/table": "53.0.9", "@t3tools/client-runtime": "workspace:*", "@t3tools/contracts": "workspace:*", "@t3tools/shared": "workspace:*", "@tanstack/react-pacer": "^0.19.4", "@tanstack/react-router": "^1.160.2", "class-variance-authority": "^0.7.1", + "date-fns": "^4.4.0", "effect": "catalog:", "jose": "catalog:", "lexical": "^0.41.0", "lucide-react": "^0.564.0", + "platejs": "53.3.3", "react": "19.2.6", "react-dom": "19.2.6", "react-markdown": "^10.1.0", diff --git a/apps/web/src/components/ChatView.tsx b/apps/web/src/components/ChatView.tsx index 31e13f1faf7..982ed4a3ac3 100644 --- a/apps/web/src/components/ChatView.tsx +++ b/apps/web/src/components/ChatView.tsx @@ -147,6 +147,8 @@ import { type RightPanelSurface, useRightPanelStore, } from "../rightPanelStore"; +// T3-CUSTOM(expbkt3): native plan review surface. +import { PlanReviewPanel, useOpenPlanReviewDocumentId } from "../fork/planReviewSurface"; import { isPreviewSupportedInRuntime, setActivePreviewTab, @@ -177,6 +179,8 @@ import { AlarmClockIcon, CheckCircle2Icon, ChevronDownIcon, + // T3-CUSTOM(expbkt3): icon for the native plan review pill. + ClipboardListIcon, GitBranchIcon, WifiOffIcon, } from "lucide-react"; @@ -439,6 +443,7 @@ const PreviewPanel = lazy(() => ); const DiffPanel = lazy(() => import("./DiffPanel")); const FilePreviewPanel = lazy(() => import("./files/FilePreviewPanel")); +// T3-CUSTOM(expbkt3): native plan review surface (lazy: Plate is ~200 kB gzip). const EMPTY_PENDING_FILE_SURFACE_IDS: ReadonlySet = new Set(); const TYPE_TO_FOCUS_EDITABLE_SELECTOR = [ "input", @@ -3553,6 +3558,17 @@ function ChatViewContent(props: ChatViewProps) { }, [activeThreadRef], ); + const openPlanReviewSurface = useCallback( + (documentId: string) => { + if (!activeThreadRef) return; + useRightPanelStore.getState().openPlanReview(activeThreadRef, documentId); + }, + [activeThreadRef], + ); + const planReviewDocumentId = useOpenPlanReviewDocumentId( + activeThreadRef?.environmentId ?? null, + activeThreadRef?.threadId ?? null, + ); // T3-CUSTOM(expbkt3): END const togglePreviewPanel = useCallback(() => { if (!activeThreadRef || !isPreviewSupportedInRuntime()) return; @@ -6621,7 +6637,20 @@ function ChatViewContent(props: ChatViewProps) { initialGitScope={initialDiffPanelGitScope} /> - ) : activeRightPanelSurface?.kind === "agents" ? ( + ) : /* T3-CUSTOM(expbkt3): BEGIN — native plan review panel. */ + activeRightPanelSurface?.kind === "planReview" ? ( + + + useRightPanelStore.getState().closeSurface(activeThreadRef, activeRightPanelSurface.id) + } + /> + + ) : /* T3-CUSTOM(expbkt3): END */ + activeRightPanelSurface?.kind === "agents" ? ( )} + {/* T3-CUSTOM(expbkt3): BEGIN — floating entry point for a plan awaiting review. */} + {planReviewDocumentId !== null && + !showScrollToBottom && + activeRightPanelSurface?.kind !== "planReview" ? ( +
+ +
+ ) : null} + {/* T3-CUSTOM(expbkt3): END */} {/* Input bar — centered hero while a draft has no messages, docked at the bottom otherwise */} diff --git a/apps/web/src/components/RightPanelTabs.tsx b/apps/web/src/components/RightPanelTabs.tsx index a9cd7ea4703..163c7cc1273 100644 --- a/apps/web/src/components/RightPanelTabs.tsx +++ b/apps/web/src/components/RightPanelTabs.tsx @@ -218,6 +218,8 @@ function surfaceTitle( // T3-CUSTOM(expbkt3): BEGIN — label the experimental review surface. case "plannotator": return "Plannotator"; + case "planReview": + return "Plan review"; // T3-CUSTOM(expbkt3): END case "agents": return "Agents"; @@ -283,6 +285,8 @@ function SurfaceIcon({ // T3-CUSTOM(expbkt3): BEGIN — icon for the experimental review surface. case "plannotator": return ; + case "planReview": + return ; // T3-CUSTOM(expbkt3): END case "agents": return ; diff --git a/apps/web/src/components/chat/MessagesTimeline.tsx b/apps/web/src/components/chat/MessagesTimeline.tsx index 935130c1b79..5d9aa7f5dbb 100644 --- a/apps/web/src/components/chat/MessagesTimeline.tsx +++ b/apps/web/src/components/chat/MessagesTimeline.tsx @@ -157,6 +157,10 @@ interface TimelineRowSharedState { onOpenTurnDiff: (turnId: TurnId, filePath?: string) => void; // T3-CUSTOM(expbkt3): Native proposed-plan cards open the focused review surface. onOpenPlannotator: (url: `/plannotator/${string}/`) => void; + // T3-CUSTOM(expbkt3): native plan review entry point. Optional so upstream + // fixtures that predate it keep compiling. + onOpenPlanReview?: ((documentId: string) => void) | undefined; + planReviewDocumentId?: string | null | undefined; onRegenerateCatchupSummary?: ((turnId: TurnId) => void) | undefined; onToggleTurnFold: (turnId: TurnId) => void; onToggleWorkGroup: (groupId: string, anchorElement?: HTMLElement) => void; @@ -237,6 +241,10 @@ interface MessagesTimelineProps { routeThreadKey: string; onOpenTurnDiff: (turnId: TurnId, filePath?: string) => void; onOpenPlannotator: (url: `/plannotator/${string}/`) => void; + // T3-CUSTOM(expbkt3): native plan review entry point. Optional so upstream + // fixtures that predate it keep compiling. + onOpenPlanReview?: ((documentId: string) => void) | undefined; + planReviewDocumentId?: string | null | undefined; onRegenerateCatchupSummary?: ((turnId: TurnId) => void) | undefined; revertTurnCountByUserMessageId: Map; onRevertUserMessage: (messageId: MessageId) => void; @@ -288,6 +296,8 @@ export const MessagesTimeline = memo(function MessagesTimeline({ routeThreadKey, onOpenTurnDiff, onOpenPlannotator, + onOpenPlanReview, + planReviewDocumentId = null, onRegenerateCatchupSummary, revertTurnCountByUserMessageId, onRevertUserMessage, @@ -552,6 +562,8 @@ export const MessagesTimeline = memo(function MessagesTimeline({ onImageExpand, onOpenTurnDiff, onOpenPlannotator, + onOpenPlanReview, + planReviewDocumentId, onRegenerateCatchupSummary, onToggleTurnFold, onToggleWorkGroup, @@ -572,6 +584,8 @@ export const MessagesTimeline = memo(function MessagesTimeline({ onImageExpand, onOpenTurnDiff, onOpenPlannotator, + onOpenPlanReview, + planReviewDocumentId, onRegenerateCatchupSummary, onToggleTurnFold, onToggleWorkGroup, @@ -1242,6 +1256,8 @@ function ProposedPlanTimelineRow({ cwd={ctx.markdownCwd} workspaceRoot={ctx.workspaceRoot} onOpenPlannotator={ctx.onOpenPlannotator} + onOpenPlanReview={ctx.onOpenPlanReview} + planReviewDocumentId={ctx.planReviewDocumentId} reviewable={row.proposedPlan.implementedAt === null} /> diff --git a/apps/web/src/components/chat/ProposedPlanCard.tsx b/apps/web/src/components/chat/ProposedPlanCard.tsx index 654d187982a..3d4b6113fb1 100644 --- a/apps/web/src/components/chat/ProposedPlanCard.tsx +++ b/apps/web/src/components/chat/ProposedPlanCard.tsx @@ -41,6 +41,8 @@ export const ProposedPlanCard = memo(function ProposedPlanCard({ cwd, workspaceRoot, onOpenPlannotator, + onOpenPlanReview, + planReviewDocumentId, reviewable = false, }: { planMarkdown: string; @@ -49,6 +51,8 @@ export const ProposedPlanCard = memo(function ProposedPlanCard({ cwd: string | undefined; workspaceRoot: string | undefined; onOpenPlannotator?: ((url: `/plannotator/${string}/`) => void) | undefined; + onOpenPlanReview?: ((documentId: string) => void) | undefined; + planReviewDocumentId?: string | null | undefined; reviewable?: boolean | undefined; }) { const [expanded, setExpanded] = useState(false); @@ -196,7 +200,9 @@ export const ProposedPlanCard = memo(function ProposedPlanCard({
) : null}
- {canCollapse || (reviewable && onOpenPlannotator) ? ( + {canCollapse || + (reviewable && onOpenPlannotator) || + (reviewable && onOpenPlanReview && planReviewDocumentId) ? (
{canCollapse ? ( + ) : null} {reviewable && onOpenPlannotator ? ( plannotatorUrl ? ( +
+ + ); + })} + + ); +} + +export const PlanReviewDiscussions = memo(PlanReviewDiscussionsImpl); diff --git a/apps/web/src/components/planreview/PlanReviewEditor.tsx b/apps/web/src/components/planreview/PlanReviewEditor.tsx new file mode 100644 index 00000000000..f464954a6fc --- /dev/null +++ b/apps/web/src/components/planreview/PlanReviewEditor.tsx @@ -0,0 +1,230 @@ +/** + * T3-CUSTOM(expbkt3): the Plate editing surface for a plan document. + * + * Suggestion mode is on by default, so every human edit is an attributed + * insert/delete the reviewer can accept or reject before it becomes a version. + * The editor owns no persistence: it reports serialized markdown upward and + * the panel decides when that becomes a draft or a version. + */ +import { CommentPlugin } from "@platejs/comment/react"; +import { MarkdownPlugin } from "@platejs/markdown"; +import { SuggestionPlugin } from "@platejs/suggestion/react"; +import { + BlockquotePlugin, + BoldPlugin, + CodePlugin, + H1Plugin, + H2Plugin, + H3Plugin, + H4Plugin, + H5Plugin, + H6Plugin, + HorizontalRulePlugin, + ItalicPlugin, + StrikethroughPlugin, + UnderlinePlugin, +} from "@platejs/basic-nodes/react"; +import { CodeBlockPlugin, CodeLinePlugin } from "@platejs/code-block/react"; +import { LinkPlugin } from "@platejs/link/react"; +import { ListPlugin } from "@platejs/list/react"; +import { + TableCellHeaderPlugin, + TableCellPlugin, + TablePlugin, + TableRowPlugin, +} from "@platejs/table/react"; +import remarkGfm from "remark-gfm"; +import { MessageSquarePlusIcon } from "lucide-react"; +import { Plate, PlateContent, usePlateEditor } from "platejs/react"; +import { memo, useCallback, useEffect, useRef, useState } from "react"; + +import { Button } from "../ui/button"; +import { cn } from "../../lib/utils"; +import { normalizeQuotedText } from "./planReviewMarkdown"; + +/** + * Scoped deliberately to what agent plans actually contain — headings, marks, + * lists, code, tables, links, quotes — plus comments and suggestions. Every + * extra node type is bundle weight on a lazily loaded panel and one more + * markdown round trip to keep honest. Kept module-local and unexported: an + * exported plugin array would force TypeScript to name Plate's internal option + * types across package boundaries. + */ +const PLAN_REVIEW_PLUGINS = [ + // Blocks + H1Plugin, + H2Plugin, + H3Plugin, + H4Plugin, + H5Plugin, + H6Plugin, + BlockquotePlugin, + HorizontalRulePlugin, + ListPlugin, + CodeBlockPlugin, + CodeLinePlugin, + TablePlugin, + TableRowPlugin, + TableCellPlugin, + TableCellHeaderPlugin, + LinkPlugin, + // Marks + BoldPlugin, + ItalicPlugin, + UnderlinePlugin, + StrikethroughPlugin, + CodePlugin, + // Review + CommentPlugin, + SuggestionPlugin, + // Markdown last: it reads the node types the plugins above registered. + MarkdownPlugin.configure({ options: { remarkPlugins: [remarkGfm] } }), +]; + +interface PlanReviewEditorProps { + /** Canonical markdown for the version being reviewed. */ + readonly markdown: string; + readonly readOnly: boolean; + readonly suggestionMode: boolean; + /** Fires on every change with freshly serialized markdown. */ + readonly onMarkdownChange: (markdown: string) => void; + /** Fires when the reviewer comments on a selection. */ + readonly onAddComment: (quotedText: string, body: string) => void; + /** Reports whether Plate's markdown round trip reached a fixed point. */ + readonly onRoundTripUnstable: () => void; +} + +function PlanReviewEditorImpl({ + markdown, + readOnly, + suggestionMode, + onMarkdownChange, + onAddComment, + onRoundTripUnstable, +}: PlanReviewEditorProps) { + const editor = usePlateEditor({ plugins: PLAN_REVIEW_PLUGINS }); + const [pendingQuote, setPendingQuote] = useState(null); + const [commentBody, setCommentBody] = useState(""); + const loadedMarkdownRef = useRef(null); + const commentInputRef = useRef(null); + + // Load canonical markdown into the editor whenever the reviewed version + // changes. Guarded by the last-loaded value so our own edits do not reload. + useEffect(() => { + if (loadedMarkdownRef.current === markdown) return; + loadedMarkdownRef.current = markdown; + + try { + const value = editor.api.markdown.deserialize(markdown); + editor.tf.setValue(value); + + // Round-trip check: an unstable document would make every later diff + // full of formatting noise the reviewer never typed. + const once = editor.api.markdown.serialize({ value }); + const twice = editor.api.markdown.serialize({ + value: editor.api.markdown.deserialize(once), + }); + if (once !== twice) onRoundTripUnstable(); + } catch { + onRoundTripUnstable(); + } + }, [editor, markdown, onRoundTripUnstable]); + + useEffect(() => { + // Suggestion mode is a plugin option rather than editor state, so it can be + // toggled without rebuilding the editor and losing the selection. + editor.setOption(SuggestionPlugin, "isSuggesting", suggestionMode && !readOnly); + }, [editor, suggestionMode, readOnly]); + + const handleChange = useCallback(() => { + try { + onMarkdownChange(editor.api.markdown.serialize()); + } catch { + // A transient invalid tree during typing is not worth surfacing; the + // next keystroke serializes again. + } + }, [editor, onMarkdownChange]); + + const startComment = useCallback(() => { + const selected = window.getSelection()?.toString() ?? ""; + const quote = normalizeQuotedText(selected); + if (quote.length === 0) return; + setPendingQuote(quote); + setCommentBody(""); + requestAnimationFrame(() => commentInputRef.current?.focus()); + }, []); + + const submitComment = useCallback(() => { + if (pendingQuote === null) return; + const body = commentBody.trim(); + if (body.length === 0) return; + onAddComment(pendingQuote, body); + setPendingQuote(null); + setCommentBody(""); + }, [commentBody, onAddComment, pendingQuote]); + + return ( +
+
+ + {suggestionMode && !readOnly ? ( + + Suggesting — edits are tracked + + ) : null} +
+ +
+ + + +
+ + {pendingQuote !== null ? ( +
+
+ {pendingQuote} +
+