[{"_1":2,"_3":-5,"_4":-5},"loaderData",{"_5":6,"_7":8,"_9":10,"_11":12},"actionData","errors","root",{"_229":-5,"_232":233,"_234":235},"routes/shell",{"_42":43,"_228":-5,"_229":-5,"_230":231},"routes/repo",{"_33":-7,"_34":35,"_36":37,"_38":39,"_40":41,"_42":43,"_44":31,"_45":31,"_46":31,"_47":31,"_48":31,"_49":31,"_50":51,"_52":51,"_53":31,"_54":-7,"_55":-7,"_56":57,"_58":59,"_60":61,"_62":-5,"_63":64,"_65":66,"_67":68,"_69":70},"routes/repo.blob",{"_13":14,"_15":16,"_17":18,"_19":20,"_21":14,"_22":23,"_24":-5,"_25":26,"_27":28,"_28":29,"_30":31,"_32":14},"ref","e90155ab65e241fd5b4d9e10184794e9fdd72d6c","path","apps/server/src/plan/service.ts","oid","877f90520298d8fe8389721dc22fec64e133ff15","size",70900,"commit","mounts",[],"source","plain",{"_13":14,"_15":16},"kind","text","/**\n * The plan, as a room offers it.\n *\n * Sits between sockets and the authoritative document: batches incoming\n * updates, decides whether they may be applied, acknowledges the sender and\n * relays to everyone else, and keeps the disk snapshot current.\n *\n * Acknowledgement is the contract worth being careful about. A client is told\n * its update was accepted only once the document has taken it and still\n * validates, because an ack is what lets the client stop holding the bytes.\n */\n\nimport * as Y from \"yjs\";\nimport { createHash } from \"node:crypto\";\nimport { isDeepStrictEqual } from \"node:util\";\n\nimport { assertIntroducedUrls, parse } from \"@chopin/dialect\";\nimport { MENTION } from \"@chopin/protocol/address\";\nimport * as Question from \"@chopin/question\";\n\nimport { validateSource } from \"../conversation-plan/sources\";\nimport { restoreState as restoreConversationPlan } from \"../conversation-plan/domain\";\nimport { assertEventCapacity } from \"../conversation-plan/events\";\nimport { assertOptionCapacity } from \"../conversation-plan/option-capacity\";\nimport { type Effect, restoreEffectOutbox } from \"../conversation-plan/effects\";\nimport * as Jobs from \"../conversation-plan/jobs\";\nimport { restorePendingCardActions } from \"../questions/card-actions\";\nimport { validateNotice, validateScopedNotices } from \"./notice-validation\";\n\nimport * as presence from \"./presence\";\nimport * as edit from \"./edit\";\nimport * as room from \"./room\";\nimport * as Chat from \"../chat/service\";\nimport { restoreReferences } from \"../chat/references\";\nimport * as Comments from \"../comments/service\";\nimport * as Questions from \"../questions/service\";\nimport { sidecarUnansweredDecisions } from \"../questions/unanswered\";\nimport { claim, restore as restoreGraph, restoreRun } from \"../tasks/graphs\";\nimport { claimEligibility, restoreLifecycle, transition } from \"../tasks/lifecycle\";\nimport { broadcast, broadcastRepository, fail, relay, reply, tell } from \"../wire\";\nimport { documentUrl } from \"../channels/document-url\";\n\nimport type { Server } from \"bun\";\nimport type { ConversationPlan, Plan as Wire, Request, Session } from \"@chopin/protocol\";\nimport type { Socket, SocketData } from \"../wire\";\nimport type { Presence } from \"./presence\";\nimport type { Document } from \"./room\";\nimport type { Block } from \"./edit\";\nimport type { Brief, CreationOrigin } from \"../mcp\";\nimport { researchProjectionAllowed, ResearchProjectionConflict } from \"../storage/model\";\nimport type {\n\tChannelRecord,\n\tInitialChannel,\n\tJsonValue,\n\tLease,\n\tResearchProjectionChange,\n\tStoredChannel,\n} from \"../storage/model\";\nimport type { StorageAdapter } from \"../storage/port\";\nimport type { PendingCardAction } from \"../questions/card-actions\";\nimport type { ClaimInput, ClaimResult, Graph, Run } from \"../tasks/graphs\";\nimport type { Lifecycle, LifecycleInput, LifecycleResult } from \"../tasks/lifecycle\";\n\n/** Updates are grouped for this long before being applied together. */\nconst GROUP_MS = 5;\n\n/** Per-connection ceiling, generous enough that typing never reaches it. */\nconst RATE_LIMIT = 200;\nconst RATE_WINDOW_MS = 1_000;\n/**\n * Consecutive windows over the limit before the connection is closed.\n *\n * A dropped update strands every later one from the same client, because Yjs\n * holds an update until its predecessor arrives. A current client resends what\n * was never acknowledged; an older bundle only replays its outbox when it\n * reopens, so closing is what gets its edits through.\n */\nconst RATE_STRIKES = 3;\n\n/** Invalid batches tolerated from one connection before it is disconnected. */\nconst INVALID_LIMIT = 3;\nconst INVALID_WINDOW_MS = 10 * 60 * 1_000;\n\n/** Close code for a client that keeps sending updates the document rejects. */\nconst ABUSIVE = 4003;\n\n/** Close code for a client that keeps sending updates faster than the limit. */\nconst TOO_FAST = 4429;\n\ntype Queued = {\n\tws: Socket;\n\trid: string;\n\tid: string;\n\tupdate: Uint8Array;\n};\n\ntype Meter = {\n\t/** Timestamps of recent updates, for the rate limit. */\n\trecent: number[];\n\t/** Timestamps of recent rejections, for the strike count. */\n\tinvalid: number[];\n\t/** When the current window of dropped updates began. */\n\tlimitedAt?: number;\n\t/** Consecutive windows in which updates were dropped. */\n\tlimitedWindows?: number;\n};\n\nexport type Backend = {\n\tstorage: StorageAdapter;\n\tlease: () => Lease;\n\tfatal: (error: unknown) => void;\n\tonDocumentPersisted?: (target: DocumentTarget) => void;\n};\n\nexport type DocumentTarget = {\n\tchannelId: string;\n\trevision: number;\n\tsource: string;\n\tsourceHash: string;\n};\n\n/** Durable MCP context for a document created through the hosted surface. */\nexport type CreationMetadata = {\n\tbrief: Brief;\n\torigin: CreationOrigin;\n};\n\n/** Durable result of one accepted MCP document rewrite. */\nexport type McpUpdateRecord = {\n\tidempotencyKey: string;\n\tfingerprint: string;\n\tfromRevision: number;\n\tclient: { name: string; version: string };\n\tdocument: {\n\t\tsource: string;\n\t\trevision: number;\n\t\ttitle: string;\n\t\turl: string;\n\t\tdescription?: string;\n\t};\n};\n\ntype Persistence = Backend & {\n\tchannelId: string;\n\trepositoryId: string;\n\tcommittedUnanswered: number;\n\trevision: number;\n\tsequence: number;\n\tlastSidecar: string;\n\tcheckpointTimer: ReturnType | undefined;\n\tcommittedEpoch: string;\n\tcommittedSource: string;\n\tcommittedDocument: Uint8Array;\n\tcommittedSidecar: JsonValue;\n\tclosing: boolean;\n};\n\ntype Captured = {\n\trevision: number;\n\tepoch: string;\n\tsource: string;\n\tsourceHash: string;\n\tdocument: Uint8Array;\n\tsidecar: JsonValue;\n\tsidecarText: string;\n};\n\nexport type Plan = {\n\tid: string;\n\t/** Context retained for plans created through MCP. */\n\tcreation?: CreationMetadata;\n\t/** Idempotent MCP rewrites retained so later edits do not change a replay. */\n\tmcpUpdates: McpUpdateRecord[];\n\tserver: Server;\n\tdocument: Document;\n\tpresence: Presence;\n\t/** Open questionnaires and their shared answer drafts. */\n\tquestions: Questions.Questions;\n\t/** Resolutions in flight and who is typing. Nothing durable. */\n\tcomments: Comments.Threads;\n\t/** The conversation driving the agent. */\n\tchat: Chat.Chat;\n\t/** Replayable conversation-derived cards and durable inference queue. */\n\tconversationPlan: ConversationPlan.State;\n\t/** Durable request IDs for explicit analysis retries. */\n\tconversationPlanRetries: Array<{ id: string; messageId: string }>;\n\t/** Background Planner work retained across restarts. */\n\tconversationPlanJobs: ConversationPlan.Job[];\n\t/** Completed delivery keys and unfinished durable conversation work. */\n\tconversationPlanEffects: string[];\n\tconversationPlanPendingEffects: Effect[];\n\t/** Card actions waiting to be mirrored into their conversation thread. */\n\tpendingCardActions: PendingCardAction[];\n\n\t/**\n\t * Every questionnaire this plan has ever held, answered or not.\n\t *\n\t * Kept beside the document rather than in it: the plan shows a decision,\n\t * this owns it. An agent rewriting the prose cannot change what was decided.\n\t */\n\trecords: Map;\n\t/**\n\t * Every comment thread this plan has ever held.\n\t *\n\t * Beside the document for the same reason a questionnaire record is: a\n\t * comment must not be undoable with the plan, and the agent must not be\n\t * able to rewrite what somebody said about its work.\n\t */\n\tthreads: Map;\n\t/** Bumped on every committed change; the agent's concurrency token. */\n\trevision: number;\n\t/** Implementation work, stored beside rather than inside the plan. */\n\tgraph?: Graph;\n\t/** The external implementation run that freezes this plan. */\n\texecution?: Run;\n\t/** Mutable task progress and prior runs, separate from claim identity. */\n\tlifecycle: Lifecycle;\n\t/** A claim has closed mutation ingress while accepted work drains. */\n\tclaiming: boolean;\n\t/**\n\t * Block outlines by revision.\n\t *\n\t * Kept so a batch aimed at a revision that has moved can be told which\n\t * blocks moved, rather than only that it is too late.\n\t */\n\toutlines: Map;\n\tqueue: Queued[];\n\ttimer: ReturnType | undefined;\n\t/** Repeats the agent's cursor while it has one, so peers do not drop it. */\n\tattention: ReturnType | undefined;\n\t/** Serialises commits so two batches cannot interleave. */\n\tflushing: Promise;\n\tmeters: WeakMap;\n\tpersistence: Persistence;\n};\n\nfunction decode(value: string): Uint8Array {\n\treturn new Uint8Array(Buffer.from(value, \"base64\"));\n}\n\nfunction encode(value: Uint8Array): string {\n\treturn Buffer.from(value).toString(\"base64\");\n}\n\nfunction recent(stamps: number[], window: number): number[] {\n\tlet cutoff = Date.now() - window;\n\treturn stamps.filter(at => at > cutoff);\n}\n\ntype Sidecar = {\n\tversion: 1;\n\trevision: number;\n\tdocumentSeq: number;\n\tcreation?: CreationMetadata;\n\tgraph?: Graph;\n\texecution?: Run;\n\tlifecycle?: Lifecycle;\n\tmcpUpdates?: McpUpdateRecord[];\n\tquestions: Questions.Record[];\n\topenQuestions: Questions.StoredOpen[];\n\tthreads: Comments.Record[];\n\ttranscript: Chat.Chat[\"entries\"];\n\tconversationPlan?: ConversationPlan.State;\n\tconversationPlanRetries?: Array<{ id: string; messageId: string }>;\n\tconversationPlanJobs?: ConversationPlan.Job[];\n\tconversationPlanEffects?: string[];\n\tconversationPlanPendingEffects?: Effect[];\n\tpendingCardActions?: PendingCardAction[];\n\tworkflowRuns?: NonNullable;\n};\n\ntype ChatView = Pick;\n\nfunction state(plan: Plan, chat: ChatView = plan.chat): Sidecar {\n\treturn {\n\t\tversion: 1,\n\t\trevision: plan.revision,\n\t\tdocumentSeq: plan.document.seq,\n\t\t...(plan.creation ? { creation: plan.creation } : {}),\n\t\t...(plan.graph ? { graph: plan.graph } : {}),\n\t\t...(plan.execution ? { execution: plan.execution } : {}),\n\t\t...(plan.lifecycle.events?.length || plan.lifecycle.history.length > 0\n\t\t\t? { lifecycle: plan.lifecycle }\n\t\t\t: {}),\n\t\t...(plan.mcpUpdates.length > 0 ? { mcpUpdates: plan.mcpUpdates } : {}),\n\t\tquestions: [...plan.records.values()],\n\t\topenQuestions: Questions.dump(plan.questions),\n\t\tthreads: [...plan.threads.values()],\n\t\ttranscript: chat.entries,\n\t\tconversationPlan: plan.conversationPlan,\n\t\t...(plan.conversationPlanRetries.length\n\t\t\t? { conversationPlanRetries: plan.conversationPlanRetries }\n\t\t\t: {}),\n\t\t...(plan.conversationPlanJobs.length\n\t\t\t? { conversationPlanJobs: plan.conversationPlanJobs }\n\t\t\t: {}),\n\t\t...(plan.conversationPlanEffects.length\n\t\t\t? { conversationPlanEffects: plan.conversationPlanEffects }\n\t\t\t: {}),\n\t\t...(plan.conversationPlanPendingEffects.length\n\t\t\t? { conversationPlanPendingEffects: plan.conversationPlanPendingEffects }\n\t\t\t: {}),\n\t\t...(plan.pendingCardActions.length ? { pendingCardActions: plan.pendingCardActions } : {}),\n\t\t...(chat.runs?.length ? { workflowRuns: chat.runs } : {}),\n\t};\n}\n\nfunction jsonState(plan: Plan, chat?: ChatView): { value: JsonValue; text: string } {\n\tlet text = JSON.stringify(state(plan, chat));\n\treturn { value: JSON.parse(text) as JsonValue, text };\n}\n\nfunction capture(plan: Plan, chat?: ChatView): Captured {\n\tassertEventCapacity(plan.conversationPlan, plan.pendingCardActions.length);\n\tif (plan.persistence) assertOptionCapacity(plan);\n\tlet sidecar = jsonState(plan, chat);\n\tlet source = room.project(plan.document);\n\treturn {\n\t\trevision: plan.revision,\n\t\tepoch: plan.document.epoch,\n\t\tsource,\n\t\tsourceHash: sourceHash(source),\n\t\tdocument: Y.encodeStateAsUpdate(plan.document.doc),\n\t\tsidecar: sidecar.value,\n\t\tsidecarText: sidecar.text,\n\t};\n}\n\nfunction objects(value: JsonValue[], label: string): Array> {\n\tlet seen = new Set();\n\treturn value.map(entry => {\n\t\tif (!entry || typeof entry !== \"object\" || Array.isArray(entry)) {\n\t\t\tthrow new Error(`hosted channel has an invalid ${label}`);\n\t\t}\n\t\tlet item = entry as Record;\n\t\tif (typeof item.id !== \"string\" || !item.id || seen.has(item.id)) {\n\t\t\tthrow new Error(`hosted channel has an invalid or duplicate ${label} id`);\n\t\t}\n\t\tseen.add(item.id);\n\t\treturn item;\n\t});\n}\n\nfunction strings(value: JsonValue | undefined): string[] | undefined {\n\treturn Array.isArray(value) && value.every(item => typeof item === \"string\")\n\t\t? value\n\t\t: undefined;\n}\n\nfunction creation(value: JsonValue | undefined): CreationMetadata | undefined {\n\tif (value === undefined) return undefined;\n\tif (\n\t\t!value\n\t\t|| typeof value !== \"object\"\n\t\t|| Array.isArray(value)\n\t) throw new Error(\"hosted channel has invalid creation metadata\");\n\tlet metadata = value as Record;\n\tif (\n\t\tObject.keys(metadata).length !== 2\n\t\t|| !Object.hasOwn(metadata, \"brief\")\n\t\t|| !Object.hasOwn(metadata, \"origin\")\n\t\t|| !metadata.brief\n\t\t|| typeof metadata.brief !== \"object\"\n\t\t|| Array.isArray(metadata.brief)\n\t\t|| !metadata.origin\n\t\t|| typeof metadata.origin !== \"object\"\n\t\t|| Array.isArray(metadata.origin)\n\t) throw new Error(\"hosted channel has invalid creation metadata\");\n\tlet brief = metadata.brief as Record;\n\tlet origin = metadata.origin as Record;\n\tlet briefKeys = [\n\t\t\"constraints\",\n\t\t\"goal\",\n\t\t\"openQuestions\",\n\t\t\"repositoryFindings\",\n\t\t\"settledDecisions\",\n\t];\n\tlet originKeys = [\n\t\t\"baseBranch\",\n\t\t\"baseCommit\",\n\t\t\"fingerprint\",\n\t\t\"idempotencyKey\",\n\t\t\"repository\",\n\t\t\"title\",\n\t];\n\tlet constraints = strings(brief.constraints);\n\tlet settledDecisions = strings(brief.settledDecisions);\n\tlet openQuestions = strings(brief.openQuestions);\n\tlet repositoryFindings = strings(brief.repositoryFindings);\n\tif (\n\t\tObject.keys(brief).sort().some((key, index) => key !== briefKeys[index])\n\t\t|| Object.keys(brief).length !== briefKeys.length\n\t\t|| typeof brief.goal !== \"string\"\n\t\t|| !brief.goal.trim()\n\t\t|| !constraints\n\t\t|| !settledDecisions\n\t\t|| !openQuestions\n\t\t|| !repositoryFindings\n\t\t|| Object.keys(origin).sort().some((key, index) => key !== originKeys[index])\n\t\t|| Object.keys(origin).length !== originKeys.length\n\t\t|| originKeys.some(key => typeof origin[key] !== \"string\" || !origin[key].trim())\n\t) throw new Error(\"hosted channel has invalid creation metadata\");\n\treturn {\n\t\tbrief: {\n\t\t\tgoal: brief.goal,\n\t\t\tconstraints,\n\t\t\tsettledDecisions,\n\t\t\topenQuestions,\n\t\t\trepositoryFindings,\n\t\t},\n\t\torigin: origin as CreationOrigin,\n\t};\n}\n\nfunction legacyCreation(\n\tbrief: JsonValue | undefined,\n\torigin: JsonValue | undefined,\n): CreationMetadata | undefined {\n\tif (brief === undefined && origin === undefined) return undefined;\n\tif (brief === undefined || origin === undefined) {\n\t\tthrow new Error(\"hosted channel has invalid creation metadata\");\n\t}\n\treturn creation({ brief, origin });\n}\n\nfunction definitionShape(value: Question.Definition) {\n\treturn value.questions.map(question => [\n\t\tquestion.id,\n\t\tquestion.header,\n\t\tquestion.question,\n\t\tquestion.multiple,\n\t\tquestion.options.map(option => [option.id, option.label, option.description]),\n\t]);\n}\n\nfunction restoredState(\n\tvalue: JsonValue,\n\tpristine: boolean,\n\tscope?: { channelId: string; repositoryId: string },\n): Sidecar {\n\tif (value === null && pristine) {\n\t\treturn {\n\t\t\tversion: 1,\n\t\t\trevision: 0,\n\t\t\tdocumentSeq: 0,\n\t\t\tquestions: [],\n\t\t\topenQuestions: [],\n\t\t\tthreads: [],\n\t\t\ttranscript: [],\n\t\t};\n\t}\n\tif (!value || typeof value !== \"object\" || Array.isArray(value)) {\n\t\tthrow new Error(\"hosted channel has an invalid sidecar\");\n\t}\n\tlet item = value as Record;\n\tlet keys = Object.keys(item).sort();\n\tlet legacy = item.creation === undefined\n\t\t&& (item.brief !== undefined || item.origin !== undefined);\n\tlet created = item.creation === undefined\n\t\t? legacyCreation(item.brief, item.origin)\n\t\t: creation(item.creation);\n\tlet graph = restoreGraph(item.graph);\n\tlet expected = [\n\t\t\"documentSeq\",\n\t\t\"openQuestions\",\n\t\t\"questions\",\n\t\t\"revision\",\n\t\t\"threads\",\n\t\t\"transcript\",\n\t\t\"version\",\n\t];\n\tif (created) expected.push(...(legacy ? [\"brief\", \"origin\"] : [\"creation\"]));\n\tif (Object.hasOwn(item, \"graph\")) expected.push(\"graph\");\n\tif (Object.hasOwn(item, \"execution\")) expected.push(\"execution\");\n\tif (Object.hasOwn(item, \"lifecycle\")) expected.push(\"lifecycle\");\n\tif (Object.hasOwn(item, \"mcpUpdates\")) expected.push(\"mcpUpdates\");\n\tif (Object.hasOwn(item, \"conversationPlan\")) expected.push(\"conversationPlan\");\n\tif (Object.hasOwn(item, \"conversationPlanRetries\")) expected.push(\"conversationPlanRetries\");\n\tif (Object.hasOwn(item, \"conversationPlanJobs\")) expected.push(\"conversationPlanJobs\");\n\tif (Object.hasOwn(item, \"conversationPlanEffects\")) expected.push(\"conversationPlanEffects\");\n\tif (Object.hasOwn(item, \"conversationPlanPendingEffects\")) {\n\t\texpected.push(\"conversationPlanPendingEffects\");\n\t}\n\tif (Object.hasOwn(item, \"pendingCardActions\")) expected.push(\"pendingCardActions\");\n\tif (Object.hasOwn(item, \"workflowRuns\")) expected.push(\"workflowRuns\");\n\texpected.sort();\n\tif (\n\t\tkeys.length !== expected.length\n\t\t|| keys.some((key, index) => key !== expected[index])\n\t\t|| item.version !== 1\n\t\t|| typeof item.revision !== \"number\"\n\t\t|| !Number.isSafeInteger(item.revision)\n\t\t|| item.revision < 0\n\t\t|| typeof item.documentSeq !== \"number\"\n\t\t|| !Number.isSafeInteger(item.documentSeq)\n\t\t|| item.documentSeq < 0\n\t\t|| !Array.isArray(item.questions)\n\t\t|| !Array.isArray(item.openQuestions)\n\t\t|| !Array.isArray(item.threads)\n\t\t|| !Array.isArray(item.transcript)\n\t) throw new Error(\"hosted channel has an invalid sidecar\");\n\tlet execution = Object.hasOwn(item, \"execution\")\n\t\t? restoreRun(item.execution, graph, item.revision)\n\t\t: undefined;\n\tif (Object.hasOwn(item, \"execution\") && !execution) {\n\t\tthrow new Error(\"hosted channel has an invalid implementation run\");\n\t}\n\tlet hasLifecycle = Object.hasOwn(item, \"lifecycle\");\n\tlet restoredLifecycle = graph\n\t\t? restoreLifecycle(hasLifecycle ? item.lifecycle : { history: [] }, graph, execution)\n\t\t: undefined;\n\tif (graph && !restoredLifecycle || hasLifecycle && !graph) {\n\t\tthrow new Error(\"hosted channel has an invalid implementation lifecycle\");\n\t}\n\tlet lifecycle = hasLifecycle ? restoredLifecycle : undefined;\n\tlet questions = objects(item.questions, \"question record\");\n\tlet records = questions.map(question => Questions.normalizeRecord(question));\n\tlet openQuestions = objects(item.openQuestions, \"open questionnaire\");\n\tlet recordIds = new Set(records.map(record => record.id));\n\tlet openIds = new Set(openQuestions.map(entry => entry.id as string));\n\tlet definitions = new Map(\n\t\trecords.map(record => [record.id, Question.identified(record.definition)]),\n\t);\n\tif (\n\t\t[...openIds].some(id => !recordIds.has(id))\n\t\t|| records.some(record => Questions.isOpenStatus(record.status) !== openIds.has(record.id))\n\t\t|| openQuestions.some(entry => {\n\t\t\tlet record = definitions.get(entry.id as string);\n\t\t\tif (!record) return true;\n\t\t\tlet draft = Question.identified(entry.definition);\n\t\t\treturn JSON.stringify(definitionShape(record))\n\t\t\t\t!== JSON.stringify(definitionShape(draft));\n\t\t})\n\t) {\n\t\tthrow new Error(\"hosted channel question records disagree with their drafts\");\n\t}\n\tlet threads = objects(item.threads, \"comment thread\");\n\tfor (let thread of threads) {\n\t\tif (\n\t\t\t(thread.status !== \"open\" && thread.status !== \"accepted\" && thread.status !== \"dismissed\")\n\t\t\t|| !thread.passage\n\t\t\t|| typeof thread.passage !== \"object\"\n\t\t\t|| Array.isArray(thread.passage)\n\t\t\t|| !Array.isArray(thread.notes)\n\t\t) throw new Error(\"hosted channel has an invalid comment thread\");\n\t}\n\tlet transcript = objects(item.transcript, \"transcript entry\");\n\tlet savedMessages = new Map(transcript.map(entry => [entry.id, entry]));\n\tfor (let record of records) {\n\t\tfor (let origin of Object.values(record.optionOrigins)) {\n\t\t\tif (!origin.source) continue;\n\t\t\tlet message = savedMessages.get(origin.source.messageId);\n\t\t\tif (!message) throw new Error(\"hosted channel has option source without a message\");\n\t\t\ttry {\n\t\t\t\tvalidateSource(origin.source, message as Chat.Chat[\"entries\"][number]);\n\t\t\t} catch (err) {\n\t\t\t\tthrow new Error(\"hosted channel has invalid option source\", { cause: err });\n\t\t\t}\n\t\t}\n\t}\n\tlet referenceIds = new Set();\n\tfor (let entry of transcript) {\n\t\tif (\n\t\t\ttypeof entry.text !== \"string\"\n\t\t\t|| typeof entry.ts !== \"number\"\n\t\t\t|| !entry.author\n\t\t\t|| typeof entry.author !== \"object\"\n\t\t\t|| Array.isArray(entry.author)\n\t\t) throw new Error(\"hosted channel has an invalid transcript entry\");\n\t\tvalidateNotice(entry, savedMessages);\n\t\tif (Object.hasOwn(entry, \"references\")) {\n\t\t\tlet author = entry.author as Record;\n\t\t\tif (author.kind !== \"member\") {\n\t\t\t\tthrow new Error(\"hosted channel has a non-member transcript reference\");\n\t\t\t}\n\t\t\ttry {\n\t\t\t\tentry.references = restoreReferences(\n\t\t\t\t\tentry.references,\n\t\t\t\t\tentry.text,\n\t\t\t\t\treferenceIds,\n\t\t\t\t\tscope,\n\t\t\t\t) as never;\n\t\t\t} catch (err) {\n\t\t\t\tthrow new Error(\"hosted channel has an invalid transcript reference\", { cause: err });\n\t\t\t}\n\t\t}\n\t\ttry {\n\t\t\tChat.validateDelivery(entry as unknown as Chat.Chat[\"entries\"][number]);\n\t\t} catch (err) {\n\t\t\tthrow new Error(\"hosted channel has invalid chat delivery metadata\", { cause: err });\n\t\t}\n\t}\n\tlet mcpUpdates = restoreMcpUpdates(item.mcpUpdates);\n\tlet conversationPlan = restoreConversationPlan(\n\t\titem.conversationPlan,\n\t\ttranscript as unknown as Chat.Chat[\"entries\"],\n\t);\n\tvalidateScopedNotices(transcript as unknown as Chat.Chat[\"entries\"], conversationPlan, records);\n\tfor (let record of records) {\n\t\tfor (let [optionId, origin] of Object.entries(record.optionOrigins)) {\n\t\t\tif (\n\t\t\t\torigin.origin === \"chat\" && origin.source\n\t\t\t\t&& !conversationPlan.events.some(event =>\n\t\t\t\t\tevent.type === \"option.added\" && event.threadId === record.threadId\n\t\t\t\t\t&& event.contribution.id === optionId\n\t\t\t\t\t&& isDeepStrictEqual(event.source, origin.source)\n\t\t\t\t)\n\t\t\t) {\n\t\t\t\tthrow new Error(\"hosted channel has option source outside its card thread or evidence\");\n\t\t\t}\n\t\t\tif (origin.source?.role !== \"question\") continue;\n\t\t\tlet thread = conversationPlan.threads.find(item =>\n\t\t\t\titem.id === record.threadId && item.questionnaireId === record.id\n\t\t\t);\n\t\t\tlet option = record.definition.questions.flatMap(item => item.options)\n\t\t\t\t.find(item => item.id === optionId);\n\t\t\tif (\n\t\t\t\t!Questions.matchesQuestionSource(origin.source, thread)\n\t\t\t\t|| !option || !Questions.questionMentionsOption(origin.source.quote, option.label)\n\t\t\t) {\n\t\t\t\tthrow new Error(\n\t\t\t\t\t\"hosted channel has option question source outside its card thread or evidence\",\n\t\t\t\t);\n\t\t\t}\n\t\t}\n\t}\n\tlet conversationPlanRetries = restoreConversationPlanRetries(item.conversationPlanRetries);\n\tlet conversationPlanJobs = Jobs.restore(item.conversationPlanJobs);\n\tlet outbox = restoreEffectOutbox(\n\t\titem.conversationPlanPendingEffects,\n\t\titem.conversationPlanEffects,\n\t);\n\tlet pendingCardActions = restorePendingCardActions(item.pendingCardActions);\n\tlet messageIds = new Set(transcript.map(entry => entry.id as string));\n\tif (\n\t\tconversationPlan.queue.some(entry => !messageIds.has(entry.messageId))\n\t\t|| conversationPlan.analysis.some(entry => !messageIds.has(entry.messageId))\n\t\t|| conversationPlanRetries.some(entry => !messageIds.has(entry.messageId))\n\t) throw new Error(\"hosted channel has conversation analysis without a source message\");\n\tlet workflowRuns = restoreWorkflowRuns(item.workflowRuns);\n\treturn {\n\t\tversion: 1,\n\t\trevision: item.revision,\n\t\tdocumentSeq: item.documentSeq,\n\t\t...(created ? { creation: created } : {}),\n\t\t...(graph ? { graph } : {}),\n\t\t...(execution ? { execution } : {}),\n\t\t...(lifecycle ? { lifecycle } : {}),\n\t\t...(mcpUpdates.length > 0 ? { mcpUpdates } : {}),\n\t\tquestions: records,\n\t\topenQuestions: openQuestions as unknown as Questions.StoredOpen[],\n\t\tthreads: threads as never[],\n\t\ttranscript: transcript as unknown as Chat.Chat[\"entries\"],\n\t\tconversationPlan,\n\t\t...(conversationPlanRetries.length ? { conversationPlanRetries } : {}),\n\t\t...(conversationPlanJobs.length ? { conversationPlanJobs } : {}),\n\t\t...(outbox.receipts.length ? { conversationPlanEffects: outbox.receipts } : {}),\n\t\t...(outbox.pending.length ? { conversationPlanPendingEffects: outbox.pending } : {}),\n\t\t...(pendingCardActions.length ? { pendingCardActions } : {}),\n\t\t...(workflowRuns.length > 0 ? { workflowRuns } : {}),\n\t};\n}\n\nfunction restoreConversationPlanRetries(\n\tvalue: JsonValue | undefined,\n): Array<{ id: string; messageId: string }> {\n\tif (value === undefined) return [];\n\tif (!Array.isArray(value) || value.length > 4096) {\n\t\tthrow new Error(\"hosted channel has invalid conversation analysis retries\");\n\t}\n\tlet seen = new Set();\n\treturn value.map(raw => {\n\t\tif (!raw || typeof raw !== \"object\" || Array.isArray(raw)) {\n\t\t\tthrow new Error(\"hosted channel has invalid conversation analysis retry\");\n\t\t}\n\t\tlet item = raw as Record;\n\t\tif (\n\t\t\tObject.keys(item).sort().join(\",\") !== \"id,messageId\"\n\t\t\t|| typeof item.id !== \"string\" || !item.id || item.id.length > 250\n\t\t\t|| typeof item.messageId !== \"string\" || !item.messageId || item.messageId.length > 200\n\t\t\t|| seen.has(item.id)\n\t\t) throw new Error(\"hosted channel has invalid conversation analysis retry\");\n\t\tseen.add(item.id);\n\t\treturn { id: item.id, messageId: item.messageId };\n\t});\n}\n\nconst RUN_STATUSES = new Set([\n\t\"running\",\n\t\"waiting\",\n\t\"paused\",\n\t\"finished\",\n\t\"blocked\",\n\t\"failed\",\n\t\"stopped\",\n]);\nconst RUN_STAGE_STATUSES = new Set([\n\t\"pending\",\n\t\"running\",\n\t\"awaiting_input\",\n\t\"paused\",\n\t\"blocked\",\n\t\"completed\",\n\t\"failed\",\n\t\"skipped\",\n]);\n\nfunction seconds(value: JsonValue | undefined, optional = false): boolean {\n\tif (value === undefined) return optional;\n\treturn typeof value === \"number\" && Number.isSafeInteger(value) && value >= 0;\n}\n\n/** Workflow run cards Chat showed, as `Chat.Run` values. */\nfunction restoreWorkflowRuns(value: JsonValue | undefined): NonNullable {\n\tif (value === undefined) return [];\n\tif (!Array.isArray(value)) throw new Error(\"hosted channel has invalid workflow runs\");\n\tlet runs = objects(value, \"workflow run\");\n\tfor (let run of runs) {\n\t\tlet stages = run.stages;\n\t\tif (\n\t\t\ttypeof run.name !== \"string\"\n\t\t\t|| !RUN_STATUSES.has(run.status as string)\n\t\t\t|| !seconds(run.started)\n\t\t\t|| !seconds(run.updated)\n\t\t\t|| !seconds(run.ended, true)\n\t\t\t|| !seconds(run.earlierStages, true)\n\t\t\t|| !seconds(run.waiting)\n\t\t\t|| !Array.isArray(stages)\n\t\t\t|| stages.some(stage =>\n\t\t\t\t!stage || typeof stage !== \"object\" || Array.isArray(stage)\n\t\t\t\t|| typeof stage.id !== \"string\" || typeof stage.name !== \"string\"\n\t\t\t\t|| stage.kind !== undefined && stage.kind !== \"tool\"\n\t\t\t\t|| !RUN_STAGE_STATUSES.has(stage.status as string)\n\t\t\t\t|| !seconds(stage.started, true) || !seconds(stage.ended, true)\n\t\t\t)\n\t\t) throw new Error(\"hosted channel has invalid workflow runs\");\n\t}\n\treturn runs as unknown as NonNullable;\n}\n\nfunction restoreMcpUpdates(value: JsonValue | undefined): McpUpdateRecord[] {\n\tif (value === undefined) return [];\n\tif (!Array.isArray(value)) throw new Error(\"hosted channel has invalid MCP updates\");\n\tlet seen = new Set();\n\treturn value.map(entry => {\n\t\tif (!entry || typeof entry !== \"object\" || Array.isArray(entry)) {\n\t\t\tthrow new Error(\"hosted channel has invalid MCP updates\");\n\t\t}\n\t\tlet item = entry as Record;\n\t\tlet document = item.document;\n\t\tif (\n\t\t\t!document\n\t\t\t|| typeof document !== \"object\"\n\t\t\t|| Array.isArray(document)\n\t\t\t|| typeof item.idempotencyKey !== \"string\"\n\t\t\t|| !item.idempotencyKey.trim()\n\t\t\t|| seen.has(item.idempotencyKey)\n\t\t\t|| typeof item.fingerprint !== \"string\"\n\t\t\t|| !item.fingerprint.trim()\n\t\t\t|| typeof item.fromRevision !== \"number\"\n\t\t\t|| !Number.isSafeInteger(item.fromRevision)\n\t\t\t|| item.fromRevision < 0\n\t\t\t|| !item.client\n\t\t\t|| typeof item.client !== \"object\"\n\t\t\t|| Array.isArray(item.client)\n\t\t) throw new Error(\"hosted channel has invalid MCP updates\");\n\t\tseen.add(item.idempotencyKey);\n\t\tlet client = item.client as Record;\n\t\tlet recorded = document as Record;\n\t\tif (\n\t\t\ttypeof client.name !== \"string\"\n\t\t\t|| !client.name.trim()\n\t\t\t|| typeof client.version !== \"string\"\n\t\t\t|| !client.version.trim()\n\t\t\t|| typeof recorded.source !== \"string\"\n\t\t\t|| typeof recorded.revision !== \"number\"\n\t\t\t|| !Number.isSafeInteger(recorded.revision)\n\t\t\t|| recorded.revision < 0\n\t\t\t|| typeof recorded.title !== \"string\"\n\t\t\t|| typeof recorded.url !== \"string\"\n\t\t\t|| (recorded.description !== undefined && typeof recorded.description !== \"string\")\n\t\t) throw new Error(\"hosted channel has invalid MCP updates\");\n\t\treturn {\n\t\t\tidempotencyKey: item.idempotencyKey,\n\t\t\tfingerprint: item.fingerprint,\n\t\t\tfromRevision: item.fromRevision,\n\t\t\tclient: { name: client.name, version: client.version },\n\t\t\tdocument: {\n\t\t\t\tsource: recorded.source,\n\t\t\t\trevision: recorded.revision,\n\t\t\t\ttitle: recorded.title,\n\t\t\t\turl: recorded.url,\n\t\t\t\t...(recorded.description ? { description: recorded.description } : {}),\n\t\t\t},\n\t\t};\n\t});\n}\n\nexport function sourceHash(source: string): string {\n\treturn `sha256:${createHash(\"sha256\").update(source).digest(\"hex\")}`;\n}\n\n/** Build the complete revision-zero state published with a newly created channel. */\nexport async function initial(\n\tsource: string,\n\tcreation?: CreationMetadata,\n): Promise {\n\t// A new document has no stored links to spare from the newer URL rules.\n\tassertIntroducedUrls([], parse(source).children);\n\tlet document = await room.create(source);\n\ttry {\n\t\tlet canonical = room.project(document);\n\t\tlet sidecar: Sidecar = {\n\t\t\tversion: 1,\n\t\t\trevision: 0,\n\t\t\tdocumentSeq: document.seq,\n\t\t\t...(creation ? { creation } : {}),\n\t\t\tquestions: [],\n\t\t\topenQuestions: [],\n\t\t\tthreads: [],\n\t\t\ttranscript: [],\n\t\t};\n\t\treturn {\n\t\t\tgeneration: crypto.randomUUID(),\n\t\t\tepoch: document.epoch,\n\t\t\tsource: canonical,\n\t\t\tsourceHash: sourceHash(canonical),\n\t\t\tdocument: Y.encodeStateAsUpdate(document.doc),\n\t\t\tsidecar: JSON.parse(JSON.stringify(sidecar)) as JsonValue,\n\t\t};\n\t} finally {\n\t\tdocument.doc.destroy();\n\t}\n}\n\nfunction scheduleCheckpoint(plan: Plan): void {\n\tlet durable = plan.persistence;\n\tif (durable.closing || durable.checkpointTimer) return;\n\tdurable.checkpointTimer = setTimeout(() => {\n\t\tdurable.checkpointTimer = undefined;\n\t\tlet checkpoint = () => checkpointHosted(plan);\n\t\tplan.flushing = plan.flushing.then(checkpoint, checkpoint);\n\t}, 500);\n}\n\nasync function commitHosted(\n\tplan: Plan,\n\tupdate: Uint8Array | undefined,\n\toperationId: string,\n\tcaptured: Captured,\n\tallowArchived = false,\n\tresearchProjections: ResearchProjectionChange[] = [],\n\tnotifyDocumentPersisted = true,\n): Promise {\n\tlet durable = plan.persistence;\n\tif (!update && captured.sidecarText === durable.lastSidecar) {\n\t\tscheduleCheckpoint(plan);\n\t\treturn;\n\t}\n\ttry {\n\t\tlet sourceChanged = captured.source !== durable.committedSource;\n\t\tlet result = await durable.storage.collaboration.commit({\n\t\t\tchannelId: durable.channelId,\n\t\t\tlease: durable.lease(),\n\t\t\texpectedRevision: durable.revision,\n\t\t\toperationId,\n\t\t\tepoch: captured.epoch,\n\t\t\t...(update ? { update } : {}),\n\t\t\tsidecar: captured.sidecar,\n\t\t\tevents: [],\n\t\t\tnow: new Date(),\n\t\t\t...(researchProjections.length > 0 ? { researchProjections } : {}),\n\t\t\t...(allowArchived ? { allowArchived: true } : {}),\n\t\t});\n\t\tif (!result.repeated) {\n\t\t\tdurable.revision = result.revision;\n\t\t\tdurable.sequence = result.sequence;\n\t\t}\n\t\tif (result.repeated && captured.sidecarText !== durable.lastSidecar) {\n\t\t\tlet stateResult = await durable.storage.collaboration.commit({\n\t\t\t\tchannelId: durable.channelId,\n\t\t\t\tlease: durable.lease(),\n\t\t\t\texpectedRevision: durable.revision,\n\t\t\t\toperationId: `state:${crypto.randomUUID()}`,\n\t\t\t\tepoch: captured.epoch,\n\t\t\t\tsidecar: captured.sidecar,\n\t\t\t\tevents: [],\n\t\t\t\tnow: new Date(),\n\t\t\t\t...(allowArchived ? { allowArchived: true } : {}),\n\t\t\t});\n\t\t\tdurable.revision = stateResult.revision;\n\t\t\tdurable.sequence = stateResult.sequence;\n\t\t}\n\t\tdurable.lastSidecar = captured.sidecarText;\n\t\tdurable.committedEpoch = captured.epoch;\n\t\tdurable.committedSource = captured.source;\n\t\tdurable.committedDocument = captured.document;\n\t\tdurable.committedSidecar = captured.sidecar;\n\t\tannounceUnanswered(plan);\n\t\tif (plan.document.epoch === captured.epoch) {\n\t\t\tplan.document.checkpoint = new Uint8Array(captured.document);\n\t\t}\n\t\tif (update && sourceChanged && notifyDocumentPersisted && durable.onDocumentPersisted) {\n\t\t\ttry {\n\t\t\t\tdurable.onDocumentPersisted({\n\t\t\t\t\tchannelId: durable.channelId,\n\t\t\t\t\trevision: captured.revision,\n\t\t\t\t\tsource: captured.source,\n\t\t\t\t\tsourceHash: captured.sourceHash,\n\t\t\t\t});\n\t\t\t} catch (err) {\n\t\t\t\tconsole.warn(`[plan] could not schedule derived work for ${durable.channelId}:`, err);\n\t\t\t}\n\t\t}\n\t\tscheduleCheckpoint(plan);\n\t} catch (err) {\n\t\tif (!(err instanceof ResearchProjectionConflict)) durable.fatal(err);\n\t\tthrow err;\n\t}\n}\n\ntype UnansweredCounts = {\n\tchannelId: string;\n\trepositoryId: string;\n\tunanswered: number;\n\trevision: number;\n};\n\nlet repositoryAnnouncements = new Map>();\n\n/**\n * Each frame carries a repository total read when its turn comes, so frames for one\n * repository must be read and sent one at a time to never let an older total arrive last.\n */\nfunction inRepositoryOrder(repositoryId: string, send: () => Promise): Promise {\n\tlet sent = (repositoryAnnouncements.get(repositoryId) ?? Promise.resolve()).then(send);\n\tlet settled = sent.catch(() => {});\n\trepositoryAnnouncements.set(repositoryId, settled);\n\tvoid settled.then(() => {\n\t\tif (repositoryAnnouncements.get(repositoryId) === settled) {\n\t\t\trepositoryAnnouncements.delete(repositoryId);\n\t\t}\n\t});\n\treturn sent;\n}\n\nasync function unansweredFrame(\n\tstorage: StorageAdapter,\n\tcounts: UnansweredCounts,\n): Promise {\n\tlet repositoryUnanswered = await storage.channels.unansweredDecisions(counts.repositoryId);\n\treturn { kind: \"session:decisions\", ts: 0, ...counts, repositoryUnanswered };\n}\n\nfunction announceCounts(\n\tserver: Server,\n\tstorage: StorageAdapter,\n\tcounts: UnansweredCounts,\n): Promise {\n\treturn inRepositoryOrder(counts.repositoryId, async () => {\n\t\tbroadcastRepository(server, counts.repositoryId, await unansweredFrame(storage, counts));\n\t}).catch(err => {\n\t\tconsole.warn(`[plan] could not announce decision counts for ${counts.channelId}:`, err);\n\t});\n}\n\nfunction announceUnanswered(plan: Plan): void {\n\tlet durable = plan.persistence;\n\tlet unanswered = sidecarUnansweredDecisions(durable.committedSidecar);\n\tif (unanswered === durable.committedUnanswered) return;\n\tdurable.committedUnanswered = unanswered;\n\tvoid announceCounts(plan.server, durable.storage, {\n\t\tchannelId: durable.channelId,\n\t\trepositoryId: durable.repositoryId,\n\t\tunanswered,\n\t\trevision: durable.revision,\n\t});\n}\n\n/** Refresh every sidebar in the repository after a document leaves or rejoins its active catalogue. */\nexport function announceCatalogueUnanswered(\n\tserver: Server,\n\tstorage: StorageAdapter,\n\tchannel: ChannelRecord,\n): Promise {\n\treturn announceCounts(server, storage, {\n\t\tchannelId: channel.id,\n\t\trepositoryId: channel.repositoryId,\n\t\tunanswered: channel.unansweredDecisions,\n\t\trevision: channel.revision,\n\t});\n}\n\n/** Give a joining socket the committed counts the sidebar shows for this document. */\nexport function tellUnanswered(plan: Plan, ws: Socket): Promise {\n\tlet durable = plan.persistence;\n\treturn inRepositoryOrder(durable.repositoryId, async () => {\n\t\tlet frame = await unansweredFrame(durable.storage, {\n\t\t\tchannelId: durable.channelId,\n\t\t\trepositoryId: durable.repositoryId,\n\t\t\tunanswered: durable.committedUnanswered,\n\t\t\trevision: durable.revision,\n\t\t});\n\t\ttell(ws, frame);\n\t});\n}\n\nasync function checkpointHosted(plan: Plan): Promise {\n\tlet durable = plan.persistence;\n\ttry {\n\t\tawait durable.storage.collaboration.checkpoint({\n\t\t\tchannelId: durable.channelId,\n\t\t\tlease: durable.lease(),\n\t\t\texpectedRevision: durable.revision,\n\t\t\tgeneration: crypto.randomUUID(),\n\t\t\trevision: durable.revision,\n\t\t\tthroughSequence: durable.sequence,\n\t\t\tepoch: durable.committedEpoch,\n\t\t\tsource: durable.committedSource,\n\t\t\tsourceHash: sourceHash(durable.committedSource),\n\t\t\tdocument: durable.committedDocument,\n\t\t\tsidecar: durable.committedSidecar,\n\t\t\tcreatedAt: new Date(),\n\t\t});\n\t\tplan.document.checkpoint = new Uint8Array(durable.committedDocument);\n\t} catch (err) {\n\t\tdurable.fatal(err);\n\t\tthrow err;\n\t}\n}\n\nasync function replaceHosted(plan: Plan, operationId: string, captured: Captured): Promise {\n\tlet durable = plan.persistence;\n\ttry {\n\t\tlet result = await durable.storage.collaboration.replace({\n\t\t\tchannelId: durable.channelId,\n\t\t\tlease: durable.lease(),\n\t\t\texpectedRevision: durable.revision,\n\t\t\toperationId,\n\t\t\tgeneration: crypto.randomUUID(),\n\t\t\tepoch: captured.epoch,\n\t\t\tsource: captured.source,\n\t\t\tsourceHash: sourceHash(captured.source),\n\t\t\tdocument: captured.document,\n\t\t\tsidecar: captured.sidecar,\n\t\t\tnow: new Date(),\n\t\t});\n\t\tif (!result.repeated) {\n\t\t\tdurable.revision = result.revision;\n\t\t\tdurable.sequence = result.sequence;\n\t\t}\n\t\tdurable.lastSidecar = captured.sidecarText;\n\t\tdurable.committedEpoch = captured.epoch;\n\t\tdurable.committedSource = captured.source;\n\t\tdurable.committedDocument = captured.document;\n\t\tdurable.committedSidecar = captured.sidecar;\n\t\tannounceUnanswered(plan);\n\t\tplan.document.checkpoint = new Uint8Array(captured.document);\n\t\tif (durable.checkpointTimer) clearTimeout(durable.checkpointTimer);\n\t\tdurable.checkpointTimer = undefined;\n\t} catch (err) {\n\t\tdurable.fatal(err);\n\t\tthrow err;\n\t}\n}\n\n/**\n * Persist a sidecar-only state transition. `chat` is read when the commit runs\n * and stored in place of the live transcript and run cards, so a change can be\n * saved before it becomes visible in shared chat state.\n */\nexport function persist(plan: Plan, chat?: () => ChatView): Promise {\n\tlet commit = () =>\n\t\tcommitHosted(plan, undefined, `state:${crypto.randomUUID()}`, capture(plan, chat?.()));\n\tlet pending = plan.flushing.then(commit, commit);\n\tplan.flushing = pending;\n\treturn pending;\n}\n\n/** Reserve the same queue used by client batches for one complete server operation. */\nexport function exclusive(plan: Plan, action: () => Promise): Promise {\n\tlet operation = plan.flushing.then(action, action);\n\tplan.flushing = operation.then(() => {}, () => {});\n\treturn operation;\n}\n\n/** Read one canonical document target in the same queue as live mutations. */\nexport function readCurrentDocument(plan: Plan): Promise {\n\treturn exclusive(plan, async () => {\n\t\tlet source = room.project(plan.document);\n\t\treturn {\n\t\t\tchannelId: plan.id,\n\t\t\trevision: plan.revision,\n\t\t\tsource,\n\t\t\tsourceHash: sourceHash(source),\n\t\t};\n\t});\n}\n\n/** Drain mutations already admitted before a claim closes the plan to new work. */\nexport async function drain(plan: Plan): Promise {\n\tif (plan.timer) clearTimeout(plan.timer);\n\tplan.timer = undefined;\n\tlet pending = () => commit(plan);\n\tplan.flushing = plan.flushing.then(pending, pending);\n\tawait plan.flushing;\n}\n\n/** One gate for every path that can mutate a plan during implementation. */\nexport function implementationActive(plan: Plan): boolean {\n\treturn plan.claiming || !!plan.execution;\n}\n\n/** Sidecar-only commit for a caller already holding `exclusive`. */\nexport function persistExclusive(plan: Plan, allowArchived = false): Promise {\n\treturn commitHosted(\n\t\tplan,\n\t\tundefined,\n\t\t`state:${crypto.randomUUID()}`,\n\t\tcapture(plan),\n\t\tallowArchived,\n\t);\n}\n\ntype RestoredHosted = {\n\tdocument: Document;\n\tneedsInitialCheckpoint: boolean;\n\tsidecar: Sidecar;\n\t/** Exact saved sidecar, before interrupted jobs are normalized in memory. */\n\tpersistedSidecar: JsonValue;\n\tinterruptedJobs: boolean;\n};\n\n/** Prepare the sidecar half of one atomic claim for a plan that is not live. */\nexport function claimStored(\n\tloaded: StoredChannel,\n\tinput: ClaimInput,\n): { result: ClaimResult; sidecar?: JsonValue } {\n\tlet pristine = loaded.channel.revision === 0 && loaded.latestSequence === 0 && !loaded.snapshot;\n\tlet sidecar = restoredState(\n\t\tloaded.sidecar === null && loaded.snapshot && loaded.channel.revision === 0\n\t\t\t? loaded.snapshot.sidecar\n\t\t\t: loaded.sidecar,\n\t\tpristine,\n\t\t{ channelId: loaded.channel.id, repositoryId: loaded.channel.repositoryId },\n\t);\n\tlet version = sidecar.graph?.versions.at(-1);\n\tlet eligibility = version\n\t\t&& claimEligibility(sidecar.lifecycle ?? { history: [] }, version, input.run.id);\n\tif (eligibility && !eligibility.ok) {\n\t\treturn { result: { kind: \"refused\", reason: eligibility.reason } };\n\t}\n\tlet result = claim({\n\t\tgraph: sidecar.graph,\n\t\trevision: sidecar.revision,\n\t\texecution: sidecar.execution,\n\t}, input);\n\tif (result.kind !== \"started\") return { result };\n\treturn {\n\t\tresult,\n\t\tsidecar: JSON.parse(JSON.stringify({\n\t\t\t...sidecar,\n\t\t\tgraph: result.graph,\n\t\t\texecution: result.run,\n\t\t})) as JsonValue,\n\t};\n}\n\n/** Prepare one atomic lifecycle sidecar change for a plan that is not live. */\nexport function lifecycleStored(\n\tloaded: StoredChannel,\n\tinput: LifecycleInput,\n): { result: LifecycleResult; sidecar?: JsonValue } {\n\tlet pristine = loaded.channel.revision === 0 && loaded.latestSequence === 0 && !loaded.snapshot;\n\tlet sidecar = restoredState(\n\t\tloaded.sidecar === null && loaded.snapshot && loaded.channel.revision === 0\n\t\t\t? loaded.snapshot.sidecar\n\t\t\t: loaded.sidecar,\n\t\tpristine,\n\t\t{ channelId: loaded.channel.id, repositoryId: loaded.channel.repositoryId },\n\t);\n\tif (!sidecar.graph) return { result: { kind: \"refused\", reason: \"inactive\" } };\n\tlet result = transition({\n\t\tgraph: sidecar.graph,\n\t\texecution: sidecar.execution,\n\t\tlifecycle: sidecar.lifecycle ?? { history: [] },\n\t}, input);\n\tif (result.kind !== \"accepted\") return { result };\n\tlet { graph: _graph, execution: _execution, lifecycle: _lifecycle, ...rest } = sidecar;\n\treturn {\n\t\tresult,\n\t\tsidecar: JSON.parse(JSON.stringify({\n\t\t\t...rest,\n\t\t\tgraph: result.state.graph,\n\t\t\t...(result.state.execution ? { execution: result.state.execution } : {}),\n\t\t\tlifecycle: result.state.lifecycle,\n\t\t})) as JsonValue,\n\t};\n}\n\nasync function restoreHosted(id: string, loaded: StoredChannel): Promise {\n\tlet document: Document;\n\tlet needsInitialCheckpoint = false;\n\tlet pristine = loaded.channel.revision === 0\n\t\t&& loaded.latestSequence === 0\n\t\t&& !loaded.snapshot;\n\tlet persistedSidecar = loaded.sidecar === null && loaded.snapshot\n\t\t\t&& loaded.channel.revision === 0\n\t\t? loaded.snapshot.sidecar\n\t\t: loaded.sidecar;\n\tlet sidecar = restoredState(\n\t\tpersistedSidecar,\n\t\tpristine,\n\t\t{ channelId: loaded.channel.id, repositoryId: loaded.channel.repositoryId },\n\t);\n\tlet persistedJobs = persistedSidecar && typeof persistedSidecar === \"object\"\n\t\t\t&& !Array.isArray(persistedSidecar)\n\t\t? (persistedSidecar as Record).conversationPlanJobs\n\t\t: undefined;\n\tlet interruptedJobs = Array.isArray(persistedJobs)\n\t\t&& persistedJobs.some(job => (job as { status: string }).status === \"running\");\n\tif (loaded.snapshot) {\n\t\tif (\n\t\t\tloaded.snapshot.revision > loaded.channel.revision\n\t\t\t|| loaded.snapshot.throughSequence > loaded.latestSequence\n\t\t) throw new Error(`channel ${id} has an invalid checkpoint position`);\n\t\tlet previous = loaded.snapshot.throughSequence;\n\t\tfor (let update of loaded.updates) {\n\t\t\tif (\n\t\t\t\tupdate.sequence <= previous\n\t\t\t\t|| update.sequence > loaded.latestSequence\n\t\t\t\t|| update.revision > loaded.channel.revision\n\t\t\t\t|| update.epoch !== loaded.snapshot.epoch\n\t\t\t) throw new Error(`channel ${id} has an invalid update journal`);\n\t\t\tprevious = update.sequence;\n\t\t}\n\t\tif (sourceHash(loaded.snapshot.source) !== loaded.snapshot.sourceHash) {\n\t\t\tthrow new Error(`channel ${id} has a corrupt source hash`);\n\t\t}\n\t\tdocument = await room.restore(\n\t\t\tloaded.snapshot.epoch,\n\t\t\tloaded.snapshot.document,\n\t\t\tloaded.snapshot.source,\n\t\t\tloaded.updates.map(update => ({ epoch: update.epoch, update: update.update })),\n\t\t);\n\t} else {\n\t\tif (loaded.updates.length > 0) {\n\t\t\tthrow new Error(`channel ${id} has updates without a checkpoint`);\n\t\t}\n\t\tdocument = await room.create();\n\t\tneedsInitialCheckpoint = true;\n\t}\n\tdocument.seq = sidecar.documentSeq;\n\treturn { document, needsInitialCheckpoint, sidecar, persistedSidecar, interruptedJobs };\n}\n\n/** Project a closed channel without attaching it to the live room registry. */\nexport async function readStored(\n\tloaded: StoredChannel,\n): Promise<{\n\tsource: string;\n\trevision: number;\n\tcreation?: CreationMetadata;\n\tgraph?: Graph;\n\texecution?: Run;\n\tlifecycle?: Lifecycle;\n}> {\n\tlet restored = await restoreHosted(loaded.channel.id, loaded);\n\ttry {\n\t\tQuestions.shutdown(Questions.restore(restored.sidecar.openQuestions));\n\t\treturn {\n\t\t\tsource: room.project(restored.document),\n\t\t\trevision: restored.sidecar.revision,\n\t\t\t...(restored.sidecar.creation ? { creation: restored.sidecar.creation } : {}),\n\t\t\t...(restored.sidecar.graph ? { graph: restored.sidecar.graph } : {}),\n\t\t\t...(restored.sidecar.execution ? { execution: restored.sidecar.execution } : {}),\n\t\t\t...(restored.sidecar.lifecycle ? { lifecycle: restored.sidecar.lifecycle } : {}),\n\t\t};\n\t} finally {\n\t\trestored.document.doc.destroy();\n\t}\n}\n\n/** Restore one durable channel into its authoritative in-memory document. */\nexport async function open(\n\tid: string,\n\tbackend: Backend,\n\tserver: Server,\n): Promise {\n\tlet loaded = await backend.storage.collaboration.load(id, new Date());\n\tif (!loaded) throw new Error(`channel ${id} does not exist`);\n\tlet { document, needsInitialCheckpoint, sidecar, persistedSidecar, interruptedJobs } =\n\t\tawait restoreHosted(id, loaded);\n\n\tlet plan: Plan = {\n\t\tid,\n\t\t...(sidecar.creation ? { creation: sidecar.creation } : {}),\n\t\tmcpUpdates: sidecar.mcpUpdates ?? [],\n\t\tserver,\n\t\tdocument,\n\t\tpresence: presence.create(),\n\t\tquestions: Questions.restore(sidecar.openQuestions),\n\t\tcomments: Comments.create(),\n\t\tchat: Chat.restore(sidecar.transcript, sidecar.workflowRuns),\n\t\tconversationPlan: sidecar.conversationPlan ?? restoreConversationPlan(undefined),\n\t\tconversationPlanRetries: sidecar.conversationPlanRetries ?? [],\n\t\tconversationPlanJobs: sidecar.conversationPlanJobs ?? [],\n\t\tconversationPlanEffects: sidecar.conversationPlanEffects ?? [],\n\t\tconversationPlanPendingEffects: sidecar.conversationPlanPendingEffects ?? [],\n\t\tpendingCardActions: sidecar.pendingCardActions ?? [],\n\t\toutlines: new Map(),\n\t\trecords: new Map(\n\t\t\tsidecar.questions.map(record => [record.id, Questions.normalizeRecord(record)]),\n\t\t),\n\t\tthreads: new Map(sidecar.threads.map(record => [record.id, record])),\n\t\trevision: sidecar.revision,\n\t\tgraph: sidecar.graph,\n\t\texecution: sidecar.execution,\n\t\tlifecycle: sidecar.lifecycle ?? { history: [] },\n\t\tclaiming: false,\n\t\tqueue: [],\n\t\ttimer: undefined,\n\t\tattention: undefined,\n\t\tflushing: Promise.resolve(),\n\t\tmeters: new WeakMap(),\n\t\tpersistence: undefined as unknown as Persistence,\n\t};\n\n\tlet committed = capture(plan);\n\tplan.persistence = {\n\t\t...backend,\n\t\tchannelId: id,\n\t\trepositoryId: loaded.channel.repositoryId,\n\t\tcommittedUnanswered: sidecarUnansweredDecisions(committed.sidecar),\n\t\trevision: loaded.channel.revision,\n\t\tsequence: loaded.latestSequence,\n\t\tlastSidecar: committed.sidecarText,\n\t\tcheckpointTimer: undefined,\n\t\tcommittedEpoch: committed.epoch,\n\t\tcommittedSource: committed.source,\n\t\tcommittedDocument: committed.document,\n\t\tcommittedSidecar: committed.sidecar,\n\t\tclosing: false,\n\t};\n\tif (interruptedJobs) {\n\t\tplan.persistence.lastSidecar = JSON.stringify(persistedSidecar);\n\t\tplan.persistence.committedSidecar = persistedSidecar;\n\t}\n\ttry {\n\t\tif (interruptedJobs) await persistExclusive(plan, true);\n\t\tif (needsInitialCheckpoint) await checkpointHosted(plan);\n\t} catch (error) {\n\t\tif (plan.persistence.checkpointTimer) clearTimeout(plan.persistence.checkpointTimer);\n\t\tQuestions.shutdown(plan.questions);\n\t\tpresence.destroy(plan.presence);\n\t\tplan.document.doc.destroy();\n\t\tthrow error;\n\t}\n\n\t// Guarded because this is the last thing between a channel and being open. A\n\t// plan whose highlights are stale is worth having; one that refuses to open\n\t// because a decision could not be placed is not.\n\ttry {\n\t\tQuestions.rebase(plan);\n\t\tComments.rebase(plan);\n\t} catch (err) {\n\t\tconsole.error(`[plan] could not carry anchors into ${id}:`, err);\n\t}\n\n\treturn plan;\n}\n\n/** Cheap identity of the whole relationship snapshot, for spotting a change. */\nfunction signature(plan: Plan): string {\n\treturn JSON.stringify([Questions.anchors(plan), Comments.anchors(plan), Questions.prose(plan)]);\n}\n\nfunction proseOrphans(plan: Plan): Map {\n\treturn new Map(Questions.prose(plan).map(item => [item.widget, item.orphaned]));\n}\n\nfunction announceProseChanges(plan: Plan, before: Map): void {\n\tfor (let item of Questions.prose(plan)) {\n\t\tif (before.get(item.widget) === item.orphaned) continue;\n\t\ttry {\n\t\t\tQuestions.announce(plan, plan.server, plan.id, item.widget);\n\t\t} catch (err) {\n\t\t\tconsole.error(\"[plan] could not announce decided prose metadata:\", err);\n\t\t}\n\t}\n}\n\n/** Everything a joining client needs to start from. */\nexport function greet(plan: Plan, ws: Socket, msg: Request): void {\n\tlet resume = msg.epoch === plan.document.epoch && msg.vector ? decode(msg.vector) : undefined;\n\tlet hello = presence.snapshot(plan.presence);\n\n\treply(ws, msg.rid, {\n\t\tkind: \"plan:open\",\n\t\tts: 0,\n\t\tepoch: plan.document.epoch,\n\t\tseq: plan.document.seq,\n\t\tupdate: encode(room.sync(plan.document, resume)),\n\t\trevision: plan.revision,\n\t\tanchors: Questions.anchors(plan),\n\t\tthreads: Comments.anchors(plan),\n\t\tprose: Questions.prose(plan),\n\t\tlimits: room.LIMITS,\n\t\t...(hello ? { awareness: encode(hello) } : {}),\n\t});\n}\n\nfunction meter(plan: Plan, ws: Socket): Meter {\n\tlet existing = plan.meters.get(ws);\n\tif (existing) return existing;\n\tlet created: Meter = { recent: [], invalid: [] };\n\tplan.meters.set(ws, created);\n\treturn created;\n}\n\n/** Accept an update for the next batch, or say why not. */\nexport function submit(plan: Plan, ws: Socket, msg: Request): void {\n\tif (implementationActive(plan)) return fail(ws, msg.rid, \"implementation is active\");\n\tif (msg.epoch !== plan.document.epoch) {\n\t\t// Nothing to correct: the client is describing a history that no longer\n\t\t// exists and needs to re-open, which the reset already told it to do.\n\t\treturn;\n\t}\n\n\tlet update = decode(msg.update);\n\tif (update.byteLength > room.LIMITS.update) {\n\t\treturn tell(ws, {\n\t\t\tkind: \"plan:reset\",\n\t\t\tts: 0,\n\t\t\tepoch: plan.document.epoch,\n\t\t\treason: \"rebuilt\",\n\t\t});\n\t}\n\n\tlet gauge = meter(plan, ws);\n\tgauge.recent = recent(gauge.recent, RATE_WINDOW_MS);\n\tif (gauge.recent.length >= RATE_LIMIT) return limited(plan, ws, msg.rid, gauge);\n\tgauge.recent.push(Date.now());\n\n\tplan.queue.push({ ws, rid: msg.rid, id: msg.id, update });\n\tschedule(plan);\n}\n\n/** Refuse an update over the rate limit, and close a connection that keeps it up. */\nfunction limited(plan: Plan, ws: Socket, rid: string, gauge: Meter): void {\n\tfail(ws, rid, \"rate limited\");\n\tlet now = Date.now();\n\tlet since = gauge.limitedAt === undefined ? Infinity : now - gauge.limitedAt;\n\tif (since < RATE_WINDOW_MS) return;\n\n\tgauge.limitedWindows = since < 2 * RATE_WINDOW_MS ? (gauge.limitedWindows ?? 0) + 1 : 1;\n\tgauge.limitedAt = now;\n\tconsole.warn(\n\t\t`[plan] dropping updates from ${ws.data.handle} in ${plan.id}: over ${RATE_LIMIT} a second`,\n\t);\n\tif (gauge.limitedWindows >= RATE_STRIKES) {\n\t\tws.close(TOO_FAST, \"plan updates too fast\");\n\t}\n}\n\nfunction schedule(plan: Plan): void {\n\tif (plan.timer) return;\n\tplan.timer = setTimeout(() => {\n\t\tplan.timer = undefined;\n\t\tplan.flushing = plan.flushing.then(() => commit(plan), () => commit(plan));\n\t}, GROUP_MS);\n}\n\nasync function rejectBatch(plan: Plan, batch: Queued[], issues: string[]): Promise {\n\tconsole.warn(\"[plan] rejected batch:\", issues.join(\", \"));\n\tlet proseBefore = proseOrphans(plan);\n\n\tlet now = Date.now();\n\tfor (let item of batch) {\n\t\tlet gauge = meter(plan, item.ws);\n\t\tgauge.invalid = [...recent(gauge.invalid, INVALID_WINDOW_MS), now];\n\t\tif (gauge.invalid.length >= INVALID_LIMIT) {\n\t\t\titem.ws.close(ABUSIVE, \"repeated invalid plan updates\");\n\t\t}\n\t}\n\n\tlet rebuilt = await room.rebuild(plan.document);\n\tplan.document = rebuilt;\n\t// Cursors describe positions in a history that no longer exists. The\n\t// agent's is in there too, and the interval repeating it would outlive\n\t// the presence it repeats.\n\tclearInterval(plan.attention);\n\tplan.attention = undefined;\n\tpresence.destroy(plan.presence);\n\tplan.presence = presence.create();\n\t// So do anchors and passages, and unlike a cursor nobody re-announces\n\t// them. Without this every highlight in the room stays dark until the\n\t// agent happens to edit.\n\t//\n\t// Guarded because the `plan:reset` below is what tells everyone to\n\t// re-open. A throw here would strand the whole room on an epoch that no\n\t// longer exists, to save some highlights that are already stale.\n\ttry {\n\t\tQuestions.rebase(plan);\n\t\tComments.rebase(plan);\n\t} catch (err) {\n\t\tconsole.error(\"[plan] could not carry anchors onto the rebuilt document:\", err);\n\t}\n\tawait replaceHosted(plan, `epoch:${rebuilt.epoch}`, capture(plan));\n\n\tbroadcast(plan.server, plan.id, {\n\t\tkind: \"plan:reset\",\n\t\tts: 0,\n\t\tepoch: rebuilt.epoch,\n\t\treason: \"rebuilt\",\n\t});\n\tannounceProseChanges(plan, proseBefore);\n}\n\n/**\n * Apply one batch.\n *\n * On rejection the document is rebuilt from its last known-good state and\n * everyone re-opens. Yjs cannot undo a transaction, so there is no narrower\n * remedy — which is why the senders in the batch are the ones charged for it.\n */\nasync function commit(plan: Plan): Promise {\n\tlet batch = plan.queue;\n\tif (batch.length === 0) return;\n\tplan.queue = [];\n\n\tlet outcome = await room.apply(\n\t\tplan.document,\n\t\tbatch.map(item => item.update),\n\t\tasync (id, action) => {\n\t\t\tlet request = await plan.persistence.storage.research.get(plan.id, id);\n\t\t\tlet initial = request?.turns.find(turn => turn.kind === \"initial\");\n\t\t\tlet jobId = initial?.answerJobId ?? initial?.evidenceJobId;\n\t\t\tlet job = jobId ? await plan.persistence.storage.jobs.get(plan.id, jobId) : undefined;\n\t\t\treturn researchProjectionAllowed(\n\t\t\t\tplan.id,\n\t\t\t\t{ id, action },\n\t\t\t\trequest?.workspace,\n\t\t\t\tinitial,\n\t\t\t\tjob?.job,\n\t\t\t);\n\t\t},\n\t\tplan.questions.open.size,\n\t);\n\n\tif (!outcome.ok) {\n\t\tawait rejectBatch(plan, batch, outcome.issues);\n\t\treturn;\n\t}\n\n\tlet relationshipsChanged = false;\n\tlet before = signature(plan);\n\tlet previousRecords = new Map(plan.records);\n\tlet previousThreads = new Map(plan.threads);\n\ttry {\n\t\tQuestions.rebase(plan);\n\t\tComments.rebase(plan);\n\t\trelationshipsChanged = signature(plan) !== before;\n\t} catch (err) {\n\t\tconsole.error(\"[plan] could not carry anchors forward:\", err);\n\t}\n\tlet previousRevision = plan.revision;\n\tif (room.project(plan.document) !== plan.persistence.committedSource) plan.revision++;\n\tlet merged = Y.mergeUpdates(batch.map(item => item.update));\n\tlet operationId = `plan:${plan.document.epoch}:${\n\t\tcreateHash(\"sha256\").update(merged).digest(\"hex\")\n\t}`;\n\ttry {\n\t\tawait commitHosted(\n\t\t\tplan,\n\t\t\tmerged,\n\t\t\toperationId,\n\t\t\tcapture(plan),\n\t\t\tfalse,\n\t\t\toutcome.researchProjections,\n\t\t);\n\t} catch (err) {\n\t\tif (!(err instanceof ResearchProjectionConflict)) throw err;\n\t\tplan.revision = previousRevision;\n\t\tplan.records = previousRecords;\n\t\tplan.threads = previousThreads;\n\t\tawait rejectBatch(plan, batch, [\"research-reference-conflict\"]);\n\t\treturn;\n\t}\n\n\tfor (let item of batch) {\n\t\treply(item.ws, item.rid, {\n\t\t\tkind: \"plan:ack\",\n\t\t\tts: 0,\n\t\t\tepoch: plan.document.epoch,\n\t\t\tid: item.id,\n\t\t\tseq: outcome.seq,\n\t\t});\n\t\t// Peers need the bytes; the sender already applied them locally.\n\t\trelay(item.ws, {\n\t\t\tkind: \"plan:update\",\n\t\t\tts: 0,\n\t\t\tepoch: plan.document.epoch,\n\t\t\tupdate: encode(item.update),\n\t\t\tseq: outcome.seq,\n\t\t});\n\t}\n\n\tif (relationshipsChanged) anchors(plan, plan.server, plan.id);\n}\n\n/**\n * Relay a change the server made, as an ordinary update.\n *\n * Agent edits and answer projections reach clients the same way a keystroke\n * does: as a delta against the document they already hold. That is what keeps\n * an agent rewriting a paragraph from costing everybody else their cursor.\n */\nexport async function publish(\n\tplan: Plan,\n\tserver: Server,\n\troomId: string,\n\tmutation: { update: Uint8Array; source: string },\n\toptions?: { agent?: boolean },\n): Promise {\n\tif (implementationActive(plan)) throw new Error(\"implementation is active\");\n\tplan.document.seq++;\n\tif (room.project(plan.document) !== plan.persistence.committedSource) plan.revision++;\n\tlet operationId = `server:${plan.document.epoch}:${\n\t\tcreateHash(\"sha256\").update(mutation.update).digest(\"hex\")\n\t}`;\n\tawait commitHosted(plan, mutation.update, operationId, capture(plan));\n\ttry {\n\t\tbroadcast(server, roomId, {\n\t\t\tkind: \"plan:update\",\n\t\t\tts: 0,\n\t\t\tepoch: plan.document.epoch,\n\t\t\tupdate: encode(mutation.update),\n\t\t\tseq: plan.document.seq,\n\t\t\t...(options?.agent ? { agent: true as const } : {}),\n\t\t});\n\t} catch (err) {\n\t\tconsole.error(\"[plan] could not broadcast a persisted update:\", err);\n\t}\n}\n\nclass ImplementationActiveError extends Error {\n\tconstructor() {\n\t\tsuper(\"implementation is active\");\n\t}\n}\n\n/** Commit a staged document and sidecar before bringing the live room forward. */\nexport async function publishStaged(\n\tplan: Plan,\n\tserver: Server,\n\troomId: string,\n\tcandidate: Plan,\n\tmutation?: room.Mutation,\n\toptions?: { notifyDocumentPersisted?: boolean; agent?: boolean },\n): Promise {\n\tif (implementationActive(plan)) throw new ImplementationActiveError();\n\tlet source = room.project(candidate.document);\n\tif (!mutation && source !== room.project(plan.document)) {\n\t\tthrow new Error(\"staged document changed without a mutation\");\n\t}\n\tif (mutation) candidate.document.seq++;\n\tif (source !== plan.persistence.committedSource) candidate.revision++;\n\tlet operationId = mutation\n\t\t? `server:${candidate.document.epoch}:${\n\t\t\tcreateHash(\"sha256\").update(mutation.update).digest(\"hex\")\n\t\t}`\n\t\t: `state:${crypto.randomUUID()}`;\n\tawait commitHosted(\n\t\tplan,\n\t\tmutation?.update,\n\t\toperationId,\n\t\tcapture(candidate),\n\t\tfalse,\n\t\t[],\n\t\toptions?.notifyDocumentPersisted !== false,\n\t);\n\tif (mutation) {\n\t\tY.applyUpdate(plan.document.doc, mutation.update);\n\t\tawait room.settle();\n\t}\n\tplan.document.seq = candidate.document.seq;\n\tplan.revision = candidate.revision;\n\tplan.records = candidate.records;\n\tplan.threads = candidate.threads;\n\tplan.questions = candidate.questions;\n\tplan.conversationPlan = candidate.conversationPlan;\n\tplan.pendingCardActions = candidate.pendingCardActions;\n\tplan.conversationPlanPendingEffects = candidate.conversationPlanPendingEffects;\n\tplan.conversationPlanEffects = candidate.conversationPlanEffects;\n\tif (mutation) {\n\t\ttry {\n\t\t\tbroadcast(server, roomId, {\n\t\t\t\tkind: \"plan:update\",\n\t\t\t\tts: 0,\n\t\t\t\tepoch: plan.document.epoch,\n\t\t\t\tupdate: encode(mutation.update),\n\t\t\t\tseq: plan.document.seq,\n\t\t\t\t...(options?.agent ? { agent: true as const } : {}),\n\t\t\t});\n\t\t} catch (err) {\n\t\t\tconsole.error(\"[plan] could not broadcast a persisted update:\", err);\n\t\t}\n\t}\n}\n\nexport async function rewrite(\n\tplan: Plan,\n\tnextSource: string,\n\trecord: (source: string, revision: number) => McpUpdateRecord,\n): Promise {\n\tif (implementationActive(plan)) throw new Error(\"implementation is active\");\n\tlet before = room.project(plan.document);\n\tlet document = await room.restore(\n\t\tplan.document.epoch,\n\t\tY.encodeStateAsUpdate(plan.document.doc),\n\t\tbefore,\n\t\t[],\n\t);\n\tdocument.seq = plan.document.seq;\n\tlet candidate: Plan = {\n\t\t...plan,\n\t\tdocument,\n\t\trecords: new Map(plan.records),\n\t\tthreads: new Map(plan.threads),\n\t\toutlines: new Map(plan.outlines),\n\t\tmcpUpdates: [...plan.mcpUpdates],\n\t};\n\ttry {\n\t\tQuestions.rebase(candidate);\n\t\tComments.rebase(candidate);\n\t\tlet outcome = edit.replace(candidate, plan.revision, nextSource);\n\t\tif (!outcome.ok) return outcome;\n\t\tlet source = room.project(document);\n\t\tlet changed = source !== before;\n\t\tif (changed) {\n\t\t\tQuestions.rebase(candidate);\n\t\t\tQuestions.invalidate(candidate, \"plan_changed\");\n\t\t\tComments.rebase(candidate);\n\t\t\tComments.invalidate(candidate, \"plan_changed\");\n\t\t} else {\n\t\t\tcandidate.records = plan.records;\n\t\t\tcandidate.threads = plan.threads;\n\t\t}\n\t\tif (outcome.mutation) document.seq++;\n\t\tif (source !== plan.persistence.committedSource) candidate.revision++;\n\t\tlet recorded = record(source, candidate.revision);\n\t\tcandidate.mcpUpdates.push(recorded);\n\t\tlet operationId = outcome.mutation\n\t\t\t? `server:${document.epoch}:${\n\t\t\t\tcreateHash(\"sha256\").update(outcome.mutation.update).digest(\"hex\")\n\t\t\t}`\n\t\t\t: `state:${crypto.randomUUID()}`;\n\t\tawait commitHosted(plan, outcome.mutation?.update, operationId, capture(candidate));\n\t\tif (outcome.mutation) {\n\t\t\tY.applyUpdate(plan.document.doc, outcome.mutation.update);\n\t\t\tawait room.settle();\n\t\t}\n\t\tplan.document.seq = document.seq;\n\t\tplan.revision = candidate.revision;\n\t\tplan.mcpUpdates.push(recorded);\n\t\tplan.outlines = candidate.outlines;\n\t\tif (changed) {\n\t\t\tQuestions.rebase(plan, before);\n\t\t\tQuestions.invalidate(plan, \"plan_changed\");\n\t\t\tComments.rebase(plan);\n\t\t\tComments.invalidate(plan, \"plan_changed\");\n\t\t}\n\t\tif (outcome.mutation) {\n\t\t\ttry {\n\t\t\t\tbroadcast(plan.server, plan.id, {\n\t\t\t\t\tkind: \"plan:update\",\n\t\t\t\t\tts: 0,\n\t\t\t\t\tepoch: plan.document.epoch,\n\t\t\t\t\tupdate: encode(outcome.mutation.update),\n\t\t\t\t\tseq: plan.document.seq,\n\t\t\t\t\tagent: true,\n\t\t\t\t});\n\t\t\t} catch (err) {\n\t\t\t\tconsole.error(\"[plan] could not broadcast a persisted update:\", err);\n\t\t\t}\n\t\t}\n\t\treturn outcome;\n\t} finally {\n\t\tdocument.doc.destroy();\n\t}\n}\n\n/**\n * Relay what the agent just did, so a reader can be shown where.\n *\n * Indices become anchors here rather than in the edit engine: only the live\n * document can say where a block is in the collaborative history, and only\n * these survive somebody else editing between this frame being sent and the\n * browser painting it.\n *\n * Sent after the update that created the blocks it names, which is what makes\n * it resolvable at the other end. Guarded whole, and silent on failure: this\n * is decoration, and a room that dropped an edit over a mark nobody would\n * have noticed would be a poor trade.\n */\nexport function changes(\n\tplan: Plan,\n\tserver: Server,\n\troomId: string,\n\tfound: edit.Change[],\n\toptions?: { cursor?: boolean; attribution?: Wire.ChangeAttribution },\n): void {\n\tif (found.length === 0) return;\n\n\ttry {\n\t\tlet digests = room.digests(plan.document);\n\t\tlet anchor = (index: number): Wire.Anchor | undefined => {\n\t\t\tlet digest = digests[index];\n\t\t\treturn digest === undefined ? undefined : room.anchorAt(plan.document, index, digest);\n\t\t};\n\t\tlet gap = (spot: edit.Spot): Wire.Gap | undefined => {\n\t\t\tlet at = anchor(spot.index);\n\t\t\treturn at && { at, side: spot.side };\n\t\t};\n\n\t\tlet wired: Wire.Change[] = [];\n\t\t// The furthest down the plan this batch reached, of what could be\n\t\t// anchored — where the agent leaves its cursor.\n\t\tlet last: number | undefined;\n\n\t\tfor (let change of found) {\n\t\t\tif (change.kind === \"removed\") {\n\t\t\t\tlet at = gap(change.at);\n\t\t\t\tif (at) {\n\t\t\t\t\twired.push({ kind: \"removed\", at, blocks: change.blocks });\n\t\t\t\t\tlast = change.at.index;\n\t\t\t\t}\n\t\t\t\tcontinue;\n\t\t\t}\n\n\t\t\tlet at = anchor(change.index);\n\t\t\tif (!at) continue;\n\t\t\tif (change.kind === \"added\") {\n\t\t\t\twired.push({ kind: \"added\", at, type: change.type, preview: change.preview });\n\t\t\t\tlast = change.index;\n\t\t\t\tcontinue;\n\t\t\t}\n\n\t\t\t// Both ends or neither: a move shown only where it landed reads as\n\t\t\t// new prose, and shown only where it left reads as a deletion.\n\t\t\tlet from = gap(change.from);\n\t\t\tif (from) {\n\t\t\t\twired.push({ kind: \"moved\", at, from, type: change.type, preview: change.preview });\n\t\t\t\tlast = change.index;\n\t\t\t}\n\t\t}\n\n\t\tif (wired.length === 0) return;\n\t\tbroadcast(server, roomId, {\n\t\t\tkind: \"plan:changes\",\n\t\t\tts: 0,\n\t\t\tepoch: plan.document.epoch,\n\t\t\tchanges: options?.attribution\n\t\t\t\t? wired.map(change => ({ ...change, attribution: options.attribution }))\n\t\t\t\t: wired,\n\t\t});\n\n\t\t// Read off the same pass, deliberately. Working it out separately could\n\t\t// disagree, and then the cursor would point at one block while the\n\t\t// marks described another.\n\t\tif (last !== undefined && options?.cursor !== false) attend(plan, server, roomId, last);\n\t} catch (err) {\n\t\tconsole.error(\"[plan] could not say what the agent changed:\", err);\n\t}\n}\n\n/** Relay the current relationship snapshot to the whole room. */\nexport function anchors(\n\tplan: Plan,\n\tserver: Server,\n\troomId: string,\n): void {\n\tlet prose = Questions.prose(plan);\n\tbroadcast(server, roomId, {\n\t\tkind: \"plan:anchors\",\n\t\tts: 0,\n\t\tepoch: plan.document.epoch,\n\t\twidgets: Questions.anchors(plan),\n\t\tthreads: Comments.anchors(plan),\n\t\tprose,\n\t});\n\t// Other post-commit callers also rebase prose; repeat metadata so none can\n\t// publish a stale collapse state after an edit or rewrite.\n\tfor (let item of prose) {\n\t\ttry {\n\t\t\tQuestions.announce(plan, server, roomId, item.widget);\n\t\t} catch (err) {\n\t\t\tconsole.error(\"[plan] could not announce decided prose metadata:\", err);\n\t\t}\n\t}\n}\n\n/**\n * How often to repeat the agent's cursor.\n *\n * Comfortably inside the thirty seconds after which a peer drops a state it\n * has not heard about. Ours rather than the awareness library's, because a\n * cursor that quietly disappears partway through a long turn is not a failure\n * anybody would think to attribute to a renewal cadence changing underneath.\n */\nconst RENEW_MS = 10_000;\n\n/**\n * The colour of the agent's cursor.\n *\n * A graphite, deliberately outside the palette `packages/editor/src/cursor.ts`\n * hands to people: the agent is not one of them, and a cursor that looked like\n * a colleague's would be read as one. Literal rather than a theme token\n * because Lexical validates it with `CSS.supports` and writes it inline, so a\n * `var()` would resolve against the wrong scope or not at all.\n *\n * Duplicated across the package boundary rather than shared for one string. If\n * the palette there ever grows a slate, this is what it must not collide with.\n */\nconst AGENT_COLOR = \"#475569\";\n\n/**\n * Put the agent's cursor where it just edited.\n *\n * Presence rather than a record: it says the agent is working here now, which\n * is why it is broadcast rather than held, and why it does not wait to be\n * seen the way the marks do. Somebody scrolled elsewhere is not shown it at\n * all — that is what the marks and the chips are for.\n */\nfunction attend(\n\tplan: Plan,\n\tserver: Server,\n\troomId: string,\n\tblock: number,\n): void {\n\t// The last turn may still be counting down to taking the cursor away. It is\n\t// about to be somewhere new, so that removal is no longer the truth.\n\tclearTimeout(plan.chat.lingering);\n\tplan.chat.lingering = undefined;\n\n\tlet position = room.endOf(plan.document, block);\n\tlet update = presence.attend(plan.presence, {\n\t\tname: MENTION.slice(1),\n\t\tcolor: AGENT_COLOR,\n\t\tfocusing: true,\n\t\tagent: true,\n\t\tanchorPos: position,\n\t\tfocusPos: position,\n\t\tawarenessData: {},\n\t});\n\n\tbroadcast(server, roomId, {\n\t\tkind: \"plan:awareness\",\n\t\tts: 0,\n\t\tepoch: plan.document.epoch,\n\t\tupdate: encode(update),\n\t});\n\n\t// Restarted, not stacked: an agent that edits twice in a turn should not\n\t// end up with two intervals repeating its cursor.\n\tclearInterval(plan.attention);\n\tplan.attention = setInterval(() => {\n\t\tlet renewed = presence.renew(plan.presence);\n\t\tif (!renewed) return;\n\t\tbroadcast(server, roomId, {\n\t\t\tkind: \"plan:awareness\",\n\t\t\tts: 0,\n\t\t\tepoch: plan.document.epoch,\n\t\t\tupdate: encode(renewed),\n\t\t});\n\t}, RENEW_MS);\n}\n\n/** Take the agent's cursor down, and stop repeating it. */\nexport function release(plan: Plan, server: Server, roomId: string): void {\n\tclearInterval(plan.attention);\n\tplan.attention = undefined;\n\n\tlet update = presence.release(plan.presence);\n\tif (!update) return;\n\n\tbroadcast(server, roomId, {\n\t\tkind: \"plan:awareness\",\n\t\tts: 0,\n\t\tepoch: plan.document.epoch,\n\t\tupdate: encode(update),\n\t});\n}\n\n/** Relay presence verbatim, and remember it for whoever joins next. */\nexport function awareness(plan: Plan, ws: Socket, msg: Wire.Awareness): void {\n\tif (msg.epoch !== plan.document.epoch) return;\n\tpresence.track(plan.presence, ws, decode(msg.update));\n\trelay(ws, { kind: \"plan:awareness\", ts: 0, epoch: plan.document.epoch, update: msg.update });\n}\n\n/** Clear a departed member's cursors rather than leaving peers to time them out. */\nexport function departed(plan: Plan, ws: Socket): void {\n\tlet update = presence.drop(plan.presence, ws);\n\tif (!update) return;\n\trelay(ws, {\n\t\tkind: \"plan:awareness\",\n\t\tts: 0,\n\t\tepoch: plan.document.epoch,\n\t\tupdate: encode(update),\n\t});\n}\n\n/** Release a successfully opened document that has never been exposed to a room. */\nexport async function abortOpening(plan: Plan): Promise {\n\tif (plan.timer) clearTimeout(plan.timer);\n\tclearInterval(plan.attention);\n\tlet persistence = plan.persistence;\n\tpersistence.closing = true;\n\tif (persistence.checkpointTimer) clearTimeout(persistence.checkpointTimer);\n\tpersistence.checkpointTimer = undefined;\n\ttry {\n\t\tlet results = await Promise.allSettled([Chat.close(plan.chat), plan.flushing]);\n\t\tlet failed = results.filter(result => result.status === \"rejected\");\n\t\tif (failed.length === 1) throw failed[0]!.reason;\n\t\tif (failed.length > 1) {\n\t\t\tthrow new AggregateError(failed.map(result => result.reason), \"document cleanup failed\");\n\t\t}\n\t} finally {\n\t\ttry {\n\t\t\tQuestions.shutdown(plan.questions);\n\t\t} finally {\n\t\t\ttry {\n\t\t\t\tpresence.destroy(plan.presence);\n\t\t\t} finally {\n\t\t\t\tplan.document.doc.destroy();\n\t\t\t}\n\t\t}\n\t}\n}\n\n/** Write anything outstanding and let go. */\nexport async function close(plan: Plan): Promise {\n\tif (plan.timer) clearTimeout(plan.timer);\n\tclearInterval(plan.attention);\n\tlet persistence = plan.persistence;\n\tpersistence.closing = true;\n\tif (persistence.checkpointTimer) clearTimeout(persistence.checkpointTimer);\n\tpersistence.checkpointTimer = undefined;\n\tawait Chat.close(plan.chat);\n\tawait plan.flushing;\n\tawait commitHosted(plan, undefined, `state:${crypto.randomUUID()}`, capture(plan), true);\n\tawait checkpointHosted(plan);\n\tQuestions.shutdown(plan.questions);\n\tpresence.destroy(plan.presence);\n\tplan.document.doc.destroy();\n}\n\n/** Current canonical source. */\nexport function source(plan: Plan): string {\n\treturn room.project(plan.document);\n}\n\nexport async function documentIdentity(plan: Plan): Promise<{ id: string; url: string }> {\n\tlet channel = await plan.persistence.storage.channels.get(plan.id);\n\tif (!channel) throw new Error(\"document is unavailable\");\n\treturn { id: channel.id, url: await documentUrl(channel, plan.persistence.storage.channels) };\n}\n\n/** Size of the Yjs history, for the idle compaction check. */\nexport function size(plan: Plan): number {\n\treturn Y.encodeStateAsUpdate(plan.document.doc).byteLength;\n}\n\nexport function placeResearchReference(plan: Plan, id: string): Promise<\"placed\" | \"deferred\"> {\n\treturn exclusive(plan, async () => {\n\t\tif (implementationActive(plan)) return \"deferred\";\n\t\tlet document = await room.restore(\n\t\t\tplan.document.epoch,\n\t\t\tY.encodeStateAsUpdate(plan.document.doc),\n\t\t\troom.project(plan.document),\n\t\t\t[],\n\t\t);\n\t\tdocument.seq = plan.document.seq;\n\t\ttry {\n\t\t\tlet mutation = room.insertResearch(document, id);\n\t\t\tif (!mutation) return \"placed\";\n\t\t\ttry {\n\t\t\t\tawait publishStaged(plan, plan.server, plan.id, { ...plan, document }, mutation);\n\t\t\t} catch (err) {\n\t\t\t\tif (err instanceof ImplementationActiveError) return \"deferred\";\n\t\t\t\tthrow err;\n\t\t\t}\n\t\t\treturn \"placed\";\n\t\t} finally {\n\t\t\tdocument.doc.destroy();\n\t\t}\n\t});\n}\n","editable",false,"branch","documentNavigation","repo","githubnext/chopin","repoId","7d1a8ac3-35b0-4dea-85d4-06a980f3bada","ownerId","org:89615882","organization",true,"site","nexthub","private","admin","settings","maintainer","triage","starred","stars",0,"watchers","watching","pinned","notify","views",[184,185,186,187,188,189,190,191,192,193,194,195,196],"navigation",{"_56":57,"_141":142,"_143":144,"_145":146,"_147":148,"_149":150,"_151":152},"sources",[],"viewer","meta",{"_133":134,"_135":136,"_137":138,"_44":31,"_139":128,"_140":31,"_50":51,"_52":51},"branches",[75,76,77,78,79,80,81,82,83,84,85,86,87,88,89,90,91,92,93,94,95,96,97,98,99,100,101,102,103,104,105,106,107,108,109,110,111,112,113,114,115,116,117,118,119,120,121,122,123,124,125,126,127,128,129,130,131,132],"tags",[71,72,73,74],"cloneUrl","https://git.nexthub.sh/githubnext/chopin.git","color-controls-slice-1","color-controls-slice-2","color-controls-slice-3","color-controls-slice-4","Krzysztof-Cieslak-patch-1","Krzysztof-Cieslak-patch-2","Maggie/app-preview-toolkit","Maggie/ci-playwright-install-timeout","Maggie/collaborative-visual-decision","Maggie/design-audit","Maggie/generic-visual-decision-runtime","Maggie/impeccable-chopin-metadata","Maggie/jev-chat-product","Maggie/jev-composed-assessment","Maggie/native-document-layouts","Maggie/openui-composition-product","Maggie/openui-composition-trial","Maggie/planner-visual-authoring","Maggie/planner-visual-requests","Maggie/visual-selection-eval","artifacts/pr-419-live-preview-20261009","automation/pr-maintenance-state","automation/use-effect/callout-disabled-picker-8cf9e49ba5c09471","automation/use-effect/collaboration-options-ref-4e3a5a08a879a5ae","automation/use-effect/room-workspace-effective-access-a57d40866c71bed0","chore/bun-1.4.2","color-controls","color-controls-plan","comments","design-pr-assets","design/archived-decisions-badge","design/channel-route-remount","design/chat-decision-entry","design/conversation-analysis-markers","design/decision-evidence","design/settled-line-icon","dev/integration","diagram-step-controls","feat/liveapp-frontend-pilot","feat/local-experiments","feat/local-launcher-tracer","fix/autolink-mdx-crash","github-pills/2-pills","github-pills/3-hover-card","maggie/app-preview-toolkit","maggie/ci-playwright-install-timeout","maggie/collaborative-visual-decision","maggie/design-audit","maggie/generic-visual-decision-runtime","maggie/impeccable-chopin-metadata","maggie/jev-composed-assessment","maggie/native-document-layouts","maggie/openui-composition-product","maggie/openui-composition-trial","maggie/planner-visual-authoring","maggie/planner-visual-requests","maggie/visual-selection-eval","main","plain-text-fence-chrome","tokens/09-copy","wireframes/2-renderer","wireframes/3-planner","description","Let's compose rich plans together","owner","githubnext","section","","head","empty","pins",[167,168,169],"documents",{"_162":163,"_164":165},"code",{"_158":159,"_160":161},"automations",{"_155":156,"_157":-5},"delivery",{"_153":51,"_154":-5},"tag","04zvKQ8QOrDUF8TO1Ovz0Zy8_TT","releases","applications","workflows",3,"deployments","commits",1000,"changes",13,"counts",{"_166":165},"total",11,"issue",{"_170":160,"_172":182,"_15":160,"_174":183,"_54":41,"_176":31,"_177":41,"_178":161},{"_170":179,"_172":180,"_15":179,"_174":181,"_54":41,"_176":31,"_177":41,"_178":51},{"_170":171,"_172":173,"_15":171,"_174":175,"_54":41,"_176":31,"_177":41,"_178":165},"id","issues","name","Issues","filters",{},"saved","pinnable","count","tasks","Tasks",{},"Changes",{},{"_170":160,"_172":182,"_15":160,"_174":227,"_54":41,"_176":31,"_177":41},{"_170":179,"_172":180,"_15":179,"_174":226,"_54":41,"_176":31,"_177":41},{"_170":171,"_172":173,"_15":171,"_174":225,"_54":41,"_176":31,"_177":41},{"_170":222,"_172":223,"_15":222,"_174":224,"_54":31,"_176":31,"_177":41},{"_170":219,"_172":220,"_15":219,"_174":221,"_54":31,"_176":31,"_177":41},{"_170":155,"_172":217,"_15":147,"_174":218,"_54":31,"_176":31,"_177":41},{"_170":157,"_172":214,"_15":215,"_174":216,"_54":31,"_176":31,"_177":41},{"_170":153,"_172":212,"_15":153,"_174":213,"_54":31,"_176":31,"_177":41},{"_170":208,"_172":209,"_15":210,"_174":211,"_54":31,"_176":31,"_177":41},{"_170":143,"_172":206,"_15":143,"_174":207,"_54":31,"_176":31,"_177":31},{"_170":203,"_172":204,"_15":203,"_174":205,"_54":31,"_176":31,"_177":31},{"_170":200,"_172":201,"_15":200,"_174":202,"_54":31,"_176":31,"_177":31},{"_170":197,"_172":198,"_15":197,"_174":199,"_54":31,"_176":31,"_177":31},"decisions","Decisions",{},"changelog","Changelog",{},"roadmap","Roadmap",{},"Documents",{},"activity","Activity","insights",{},"Releases",{},"Deployments","automations/deployments",{},"Workflows",{},"discussions","Discussions",{},"goals","Goals",{},{},{},{},"groups","user","agents",[],"now",1794700183686,"sidebar","expanded"]