[{"_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","c927effefb9d7dfbc695358b24721471a937a3f2","path","apps/server/src/plan/service.ts","oid","6945d1354443ca265ad297698ee71f91d95e9768","size",30911,"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\";\n\nimport { MENTION } from \"@chopin/protocol/address\";\n\nimport * as presence from \"./presence\";\nimport * as room from \"./room\";\nimport * as Chat from \"../chat/service\";\nimport * as Comments from \"../comments/service\";\nimport * as Questions from \"../questions/service\";\nimport { broadcast, relay, reply, tell } from \"../wire\";\n\nimport type { Server } from \"bun\";\nimport type { Plan as Wire, Request } from \"@chopin/protocol\";\nimport type { Socket, SocketData } from \"../wire\";\nimport type { Presence } from \"./presence\";\nimport type { Document } from \"./room\";\nimport type * as edit from \"./edit\";\nimport type { Block } from \"./edit\";\nimport type { JsonValue, Lease, StoredChannel } from \"../storage/model\";\nimport type { StorageAdapter } from \"../storage/port\";\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/** 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\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};\n\nexport type Backend = {\n\tstorage: StorageAdapter;\n\tlease: () => Lease;\n\tfatal: (error: unknown) => void;\n};\n\ntype Persistence = Backend & {\n\tchannelId: string;\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\tepoch: string;\n\tsource: string;\n\tdocument: Uint8Array;\n\tsidecar: JsonValue;\n\tsidecarText: string;\n};\n\nexport type Plan = {\n\tid: string;\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/**\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/**\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\tquestions: Questions.Record[];\n\topenQuestions: Questions.StoredOpen[];\n\tthreads: Comments.Record[];\n\ttranscript: Chat.Chat[\"entries\"];\n};\n\nfunction state(plan: Plan): Sidecar {\n\treturn {\n\t\tversion: 1,\n\t\trevision: plan.revision,\n\t\tdocumentSeq: plan.document.seq,\n\t\tquestions: [...plan.records.values()],\n\t\topenQuestions: Questions.dump(plan.questions),\n\t\tthreads: [...plan.threads.values()],\n\t\ttranscript: plan.chat.entries,\n\t};\n}\n\nfunction jsonState(plan: Plan): { value: JsonValue; text: string } {\n\tlet text = JSON.stringify(state(plan));\n\treturn { value: JSON.parse(text) as JsonValue, text };\n}\n\nfunction capture(plan: Plan): Captured {\n\tlet sidecar = jsonState(plan);\n\treturn {\n\t\tepoch: plan.document.epoch,\n\t\tsource: room.project(plan.document),\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 restoredState(value: JsonValue, pristine: boolean): 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 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 (\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 questions = objects(item.questions, \"question record\");\n\tfor (let question of questions) {\n\t\tif (\n\t\t\t(question.status !== \"open\" && question.status !== \"answered\"\n\t\t\t\t&& question.status !== \"cancelled\")\n\t\t\t|| !question.definition\n\t\t\t|| typeof question.definition !== \"object\"\n\t\t\t|| Array.isArray(question.definition)\n\t\t) throw new Error(\"hosted channel has an invalid question record\");\n\t}\n\tlet openQuestions = objects(item.openQuestions, \"open questionnaire\");\n\tlet openIds = new Set(openQuestions.map(entry => entry.id as string));\n\tif (questions.some(record => (record.status === \"open\") !== openIds.has(record.id as string))) {\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\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}\n\treturn {\n\t\tversion: 1,\n\t\trevision: item.revision,\n\t\tdocumentSeq: item.documentSeq,\n\t\tquestions: questions as never[],\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};\n}\n\nfunction digest(source: string): string {\n\treturn `sha256:${createHash(\"sha256\").update(source).digest(\"hex\")}`;\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): 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 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});\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});\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\tif (plan.document.epoch === captured.epoch) {\n\t\t\tplan.document.checkpoint = new Uint8Array(captured.document);\n\t\t}\n\t\tscheduleCheckpoint(plan);\n\t} catch (err) {\n\t\tdurable.fatal(err);\n\t\tthrow err;\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: digest(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: digest(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\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/** Persist a sidecar-only state transition. */\nexport function persist(plan: Plan): Promise {\n\tlet captured = capture(plan);\n\tlet commit = () => commitHosted(plan, undefined, `state:${crypto.randomUUID()}`, captured);\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/** Sidecar-only commit for a caller already holding `exclusive`. */\nexport function persistExclusive(plan: Plan): Promise {\n\treturn commitHosted(plan, undefined, `state:${crypto.randomUUID()}`, capture(plan));\n}\n\ntype RestoredHosted = {\n\tdocument: Document;\n\tneedsInitialCheckpoint: boolean;\n\tsidecar: Sidecar;\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 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);\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 (digest(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 };\n}\n\n/** Project a closed channel without attaching it to the live room registry. */\nexport async function readStored(\n\tloaded: StoredChannel,\n): Promise<{ source: string; revision: number }> {\n\tlet restored = await restoreHosted(loaded.channel.id, loaded);\n\ttry {\n\t\treturn { source: room.project(restored.document), revision: restored.sidecar.revision };\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 } = await restoreHosted(id, loaded);\n\n\tlet plan: Plan = {\n\t\tid,\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),\n\t\toutlines: new Map(),\n\t\trecords: new Map(sidecar.questions.map(record => [record.id, record])),\n\t\tthreads: new Map(sidecar.threads.map(record => [record.id, record])),\n\t\trevision: sidecar.revision,\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\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 (needsInitialCheckpoint) await checkpointHosted(plan);\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)]);\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\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 (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;\n\tgauge.recent.push(Date.now());\n\n\tplan.queue.push({ ws, rid: msg.rid, id: msg.id, update });\n\tschedule(plan);\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\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(plan.document, batch.map(item => item.update));\n\n\tif (!outcome.ok) {\n\t\tconsole.warn(\"[plan] rejected batch:\", outcome.issues.join(\", \"));\n\n\t\tlet now = Date.now();\n\t\tfor (let item of batch) {\n\t\t\tlet gauge = meter(plan, item.ws);\n\t\t\tgauge.invalid = [...recent(gauge.invalid, INVALID_WINDOW_MS), now];\n\t\t\tif (gauge.invalid.length >= INVALID_LIMIT) {\n\t\t\t\titem.ws.close(ABUSIVE, \"repeated invalid plan updates\");\n\t\t\t}\n\t\t}\n\n\t\tlet rebuilt = await room.rebuild(plan.document);\n\t\tplan.document = rebuilt;\n\t\t// Cursors describe positions in a history that no longer exists. The\n\t\t// agent's is in there too, and the interval repeating it would outlive\n\t\t// the presence it repeats.\n\t\tclearInterval(plan.attention);\n\t\tplan.attention = undefined;\n\t\tpresence.destroy(plan.presence);\n\t\tplan.presence = presence.create();\n\t\t// So do anchors and passages, and unlike a cursor nobody re-announces\n\t\t// them. Without this every highlight in the room stays dark until the\n\t\t// agent happens to edit.\n\t\t//\n\t\t// Guarded because the `plan:reset` below is what tells everyone to\n\t\t// re-open. A throw here would strand the whole room on an epoch that no\n\t\t// longer exists, to save some highlights that are already stale.\n\t\ttry {\n\t\t\tQuestions.rebase(plan);\n\t\t\tComments.rebase(plan);\n\t\t} catch (err) {\n\t\t\tconsole.error(\"[plan] could not carry anchors onto the rebuilt document:\", err);\n\t\t}\n\t\tawait replaceHosted(plan, `epoch:${rebuilt.epoch}`, capture(plan));\n\n\t\tbroadcast(plan.server, plan.id, {\n\t\t\tkind: \"plan:reset\",\n\t\t\tts: 0,\n\t\t\tepoch: rebuilt.epoch,\n\t\t\treason: \"rebuilt\",\n\t\t});\n\t\treturn;\n\t}\n\n\tlet relationshipsChanged = false;\n\tlet before = signature(plan);\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\tplan.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\tawait commitHosted(plan, merged, operationId, capture(plan));\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): Promise {\n\tplan.document.seq++;\n\tplan.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});\n\t} catch (err) {\n\t\tconsole.error(\"[plan] could not broadcast a persisted update:\", err);\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): 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: 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) 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\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});\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/** 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));\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\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","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",1794658588912,"sidebar","expanded"]