// runtime-v2 multi-lane ipc protocol types and sequence cursor algebra
// mirrors klbr-ipc/src/v2.rs and klbr-ipc/src/cursor.rs 1:1
// distinct lanes: scratchpad, cell (diagnostics), speech (local.sent), input, lifecycle
export type Lane = "scratchpad" | "cell" | "speech" | "input" | "lifecycle";
// scratchpad lane
export interface ScratchpadPayload {
text: string;
reasoning?: string;
execution_id?: string | null;
delta?: boolean | string | null;
kind?: "thought" | "prose" | "tool_call_started" | "tool_call_proposed" | "accumulated";
thoughts?: string;
prose?: string;
tool_call?: {
id: string;
name: string;
arguments: string;
};
step?: number;
tokens?: number;
duration_ms?: number;
active?: boolean;
}
// cell execution lane
export type CellState = "ok" | "error" | "yielded" | "interrupted";
export type CellStreamKind = "stdout" | "stderr" | "result";
export interface CellStreamOutput {
stream: CellStreamKind;
text: string;
timestamp?: number;
}
export type CellPayload =
| {
type: "started";
execution_id: string;
generation: string;
revision: string;
code: string;
}
| {
type: "completed";
execution_id: string;
state: CellState;
output: CellStreamOutput[];
truncated_bytes?: number;
error?: string | null;
};
// provenance metadata - origin is descriptive provenance, never permission/trust rank
// aligned 1:1 with Rust OriginKind and Provenance (klbr-runtime/src/work.rs)
export type OriginKind = "human" | "agent" | "checker" | "system" | "external";
export type ProvenanceOrigin = OriginKind;
export interface ProvenanceMetadata {
origin_kind: OriginKind;
origin: OriginKind;
reporter_id: string;
reporter: string;
source_refs?: string[];
observed_ms: number;
}
export function normalizeProvenance(raw: unknown): ProvenanceMetadata | undefined {
if (!raw || typeof raw !== "object") return undefined;
const p = raw as Record<string, unknown>;
const origin = (p.origin_kind ?? p.origin ?? "agent") as OriginKind;
const reporter = String(p.reporter_id ?? p.reporter ?? origin);
const observed_ms =
typeof p.observed_ms === "number" && Number.isFinite(p.observed_ms)
? p.observed_ms
: Date.now();
const source_refs = Array.isArray(p.source_refs) ? p.source_refs.map(String) : [];
return {
origin_kind: origin,
origin,
reporter_id: reporter,
reporter,
source_refs,
observed_ms,
};
}
// capability health status & diagnostic observations
export type CapabilityHealthStatus =
| "healthy"
| "degraded"
| "unavailable"
| "unknown"
| "stale";
export interface CapabilityHealthObservation {
id: string;
name: string;
scope: string;
status: CapabilityHealthStatus;
target?: string;
target_id?: string;
evidence?: string;
impact?: string;
observed_ms: number;
valid_until_ms?: number;
origin: OriginKind;
origin_kind?: OriginKind;
reporter?: string;
reporter_id?: string;
provenance?: ProvenanceMetadata;
}
export function normalizeHealthObservation(raw: unknown): CapabilityHealthObservation {
if (!raw || typeof raw !== "object") {
throw new Error("malformed health observation: must be an object");
}
const h = raw as Record<string, unknown>;
if (typeof h.id !== "string" || !h.id.trim()) {
throw new Error("malformed health observation: missing or empty id");
}
const prov = h.provenance ? normalizeProvenance(h.provenance) : undefined;
const origin = (h.origin_kind ?? h.origin ?? prov?.origin_kind ?? "system") as OriginKind;
const reporter = h.reporter_id ?? h.reporter ?? prov?.reporter_id;
const statusStr =
typeof h.status === "object" && h.status && "status" in (h.status as any)
? (h.status as any).status
: h.status;
const target =
typeof h.target_id === "string"
? h.target_id
: typeof h.target === "string"
? h.target
: undefined;
return {
id: h.id,
name: typeof h.name === "string" ? h.name : h.id,
scope: typeof h.scope === "string" ? h.scope : "main",
status: (statusStr || "unknown") as CapabilityHealthStatus,
target,
target_id: target,
evidence: typeof h.evidence === "string" ? h.evidence : undefined,
impact: typeof h.impact === "string" ? h.impact : undefined,
observed_ms: typeof h.observed_ms === "number" ? h.observed_ms : Date.now(),
valid_until_ms: typeof h.valid_until_ms === "number" ? h.valid_until_ms : undefined,
origin,
origin_kind: origin,
reporter: reporter ? String(reporter) : undefined,
reporter_id: reporter ? String(reporter) : undefined,
provenance: prov,
};
}
// derive effective capability health: healthy checkups expire into stale
export function resolveCapabilityHealth(
obs: CapabilityHealthObservation,
nowMs: number = Date.now()
): CapabilityHealthStatus {
if (obs.status === "healthy" && obs.valid_until_ms !== undefined && nowMs > obs.valid_until_ms) {
return "stale";
}
return obs.status;
}
// work item diagnostic representation - human & agent work share identical scheduling projection
export type WorkItemStatus =
| "pending"
| "selected"
| "deferred"
| "executing"
| "completed"
| "failed"
| "cancelled"
| "claimed";
export interface WorkItemDiagnostic {
id: string;
title: string;
purpose: string;
status: WorkItemStatus;
target?: string;
target_id?: string;
origin: OriginKind;
origin_kind?: OriginKind;
reporter?: string;
reporter_id?: string;
provenance?: ProvenanceMetadata;
selection_reason?: string;
deferred_reason?: string;
dependencies?: string[];
created_ms: number;
age_ms?: number;
}
export function normalizeWorkItem(raw: unknown): WorkItemDiagnostic {
if (!raw || typeof raw !== "object") {
throw new Error("malformed work item: must be an object");
}
const w = raw as Record<string, unknown>;
if (typeof w.id !== "string" || !w.id.trim()) {
throw new Error("malformed work item: missing or empty id");
}
const prov = w.provenance ? normalizeProvenance(w.provenance) : undefined;
const origin = (w.origin_kind ?? w.origin ?? prov?.origin_kind ?? "agent") as OriginKind;
const reporter = w.reporter_id ?? w.reporter ?? prov?.reporter_id;
const statusStr =
typeof w.status === "object" && w.status && "status" in (w.status as any)
? (w.status as any).status
: w.status;
const purpose = typeof w.purpose === "string" ? w.purpose : (typeof w.title === "string" ? w.title : "");
const title = typeof w.title === "string" ? w.title : (purpose || w.id);
const target =
typeof w.target_id === "string"
? w.target_id
: typeof w.target === "string"
? w.target
: undefined;
return {
id: w.id,
title,
purpose,
status: (statusStr || "pending") as WorkItemStatus,
target,
target_id: target,
origin,
origin_kind: origin,
reporter: reporter ? String(reporter) : undefined,
reporter_id: reporter ? String(reporter) : undefined,
provenance: prov,
selection_reason: typeof w.selection_reason === "string" ? w.selection_reason : undefined,
deferred_reason: typeof w.deferred_reason === "string" ? w.deferred_reason : undefined,
dependencies: Array.isArray(w.dependencies) ? w.dependencies.map(String) : undefined,
created_ms: typeof w.created_ms === "number" ? w.created_ms : Date.now(),
age_ms: typeof w.age_ms === "number" ? w.age_ms : undefined,
};
}
export interface CandidateRevisionDiagnostic {
candidate_id: string;
work_item_id?: string;
base_revision: string;
target_revision: string;
status: "draft" | "testing" | "pinned" | "active" | "rolled_back" | string;
epoch?: number;
test_receipts?: Array<{
suite_id: string;
passed: boolean;
duration_ms: number;
}>;
}
export function normalizeCandidate(raw: unknown): CandidateRevisionDiagnostic | null {
if (!raw || typeof raw !== "object") return null;
const c = raw as Record<string, unknown>;
if (typeof c.candidate_id !== "string" || !c.candidate_id.trim()) {
throw new Error("malformed candidate: missing or empty candidate_id");
}
const work_item_id =
typeof c.work_item_id === "string"
? c.work_item_id
: typeof c.item_id === "string"
? c.item_id
: undefined;
return {
candidate_id: c.candidate_id,
work_item_id,
base_revision: typeof c.base_revision === "string" ? c.base_revision : "",
target_revision: typeof c.target_revision === "string" ? c.target_revision : "",
status: typeof c.status === "string" ? c.status : "draft",
epoch: typeof c.epoch === "number" ? c.epoch : undefined,
test_receipts: Array.isArray(c.test_receipts) ? c.test_receipts : undefined,
};
}
// speech / external effect lane
export type SpeechPayload =
| {
type: "sent";
effect_id: string;
text: string;
event_id: number;
status: string;
provenance?: ProvenanceMetadata;
}
| {
type: "held";
effect_id: string;
text: string;
status: string;
reason: string;
provenance?: ProvenanceMetadata;
}
| {
type: "failed";
effect_id: string;
text: string;
status: string;
reason?: string;
error?: string;
provenance?: ProvenanceMetadata;
};
// input lane
export interface InputPayload {
source: string;
text: string;
author?: string | null;
author_id?: string | null;
client_id?: string | null;
provenance?: ProvenanceMetadata;
}
export interface OrientationDiagnostic {
candidate_id?: string;
work_item_id?: string;
frontier_id?: string;
parent_id?: string | null;
action?: string;
actor?: string;
status?: string;
staged_ms?: number;
adopted_ms?: number;
carrier?: unknown;
}
export function normalizeOrientation(raw: unknown): OrientationDiagnostic | null {
if (!raw || typeof raw !== "object") return null;
const o = raw as Record<string, unknown>;
const candId =
typeof o.candidate_id === "string"
? o.candidate_id
: typeof (o.candidate as any)?.candidate_id === "string"
? (o.candidate as any).candidate_id
: undefined;
const workId =
typeof o.work_item_id === "string"
? o.work_item_id
: typeof o.item_id === "string"
? o.item_id
: undefined;
const parentId =
typeof o.parent_id === "string"
? o.parent_id
: typeof o.expected_parent === "string"
? o.expected_parent
: null;
return {
candidate_id: candId,
work_item_id: workId,
frontier_id: typeof o.frontier_id === "string" ? o.frontier_id : undefined,
parent_id: parentId,
action: typeof o.action === "string" ? o.action : undefined,
actor:
typeof o.actor === "string"
? o.actor
: typeof o.adoption_actor === "string"
? o.adoption_actor
: undefined,
status: typeof o.status === "string" ? o.status : undefined,
staged_ms: typeof o.staged_ms === "number" ? o.staged_ms : undefined,
adopted_ms: typeof o.adopted_ms === "number" ? o.adopted_ms : undefined,
carrier: o.candidate_carrier ?? o.carrier,
};
}
// lifecycle lane
export type LinkedDiagnostic = Record<string, unknown> | null;
/** Extract production-linked state without inventing candidate/health records. */
export function extractLinkedDiagnostics(raw: unknown): {
target?: LinkedDiagnostic;
context?: LinkedDiagnostic;
output?: LinkedDiagnostic;
posture?: LinkedDiagnostic;
adoption?: LinkedDiagnostic;
} {
if (!raw || typeof raw !== "object") return {};
const data = raw as Record<string, unknown>;
const result: Record<string, LinkedDiagnostic> = {};
for (const key of ["target", "context", "output", "posture", "adoption"]) {
if (key in data && data[key] && typeof data[key] === "object") {
result[key] = data[key] as Record<string, unknown>;
}
}
return result;
}
export interface DiagnosticData {
health?: CapabilityHealthObservation[];
work_items?: WorkItemDiagnostic[];
candidate?: CandidateRevisionDiagnostic | null;
orientation?: OrientationDiagnostic | null;
}
export interface LifecyclePayload {
kind: string;
data?: DiagnosticData | Record<string, unknown> | unknown;
step?: number;
seconds?: number;
reason?: string;
details?: string;
health?: CapabilityHealthObservation[];
work_items?: WorkItemDiagnostic[];
candidate?: CandidateRevisionDiagnostic | null;
orientation?: OrientationDiagnostic | null;
provenance?: ProvenanceMetadata;
}
export interface ExtractedDiagnostics {
health?: CapabilityHealthObservation[];
work_items?: WorkItemDiagnostic[];
candidate?: CandidateRevisionDiagnostic | null;
orientation?: OrientationDiagnostic | null;
}
export function extractDiagnostics(lp: LifecyclePayload): ExtractedDiagnostics {
const dataObj = (lp.data && typeof lp.data === "object" ? lp.data : null) as Record<string, unknown> | null;
const result: ExtractedDiagnostics = {};
// Reject double-nested data.data patterns (gate malformation)
if (dataObj && "data" in dataObj && typeof dataObj.data === "object" && dataObj.data !== null) {
const inner = dataObj.data as Record<string, unknown>;
if ("health" in inner || "work_items" in inner || "candidate" in inner) {
throw new Error(
`malformed lifecycle payload: diagnostics are double-nested as data.data.* - ` +
`emit flat {kind, data: {health, work_items, candidate}} instead`
);
}
}
// 1. Health
const rawHealth = dataObj?.health ?? lp.health;
if (Array.isArray(rawHealth)) {
result.health = rawHealth.map(normalizeHealthObservation);
} else if (lp.kind === "capability.health" && dataObj) {
if (Array.isArray(dataObj.health)) {
result.health = dataObj.health.map(normalizeHealthObservation);
} else if (typeof dataObj.id === "string") {
result.health = [normalizeHealthObservation(dataObj)];
}
}
// 2. Work items
const rawWork = dataObj?.work_items ?? lp.work_items;
if (Array.isArray(rawWork)) {
result.work_items = rawWork.map(normalizeWorkItem);
} else if (lp.kind === "work.updated" && dataObj) {
if (Array.isArray(dataObj.work_items)) {
result.work_items = dataObj.work_items.map(normalizeWorkItem);
} else if (typeof dataObj.id === "string") {
result.work_items = [normalizeWorkItem(dataObj)];
}
}
// 3. Candidate
const rawCandidate = dataObj && "candidate" in dataObj ? dataObj.candidate : lp.candidate;
if (rawCandidate !== undefined) {
result.candidate = rawCandidate ? normalizeCandidate(rawCandidate) : null;
} else if (lp.kind === "candidate.updated" && dataObj) {
if ("candidate" in dataObj) {
result.candidate = dataObj.candidate ? normalizeCandidate(dataObj.candidate) : null;
} else if (typeof dataObj.candidate_id === "string") {
result.candidate = normalizeCandidate(dataObj);
}
}
// 4. Orientation
const rawOrientation = dataObj && "orientation" in dataObj ? dataObj.orientation : lp.orientation;
if (rawOrientation !== undefined) {
result.orientation = rawOrientation ? normalizeOrientation(rawOrientation) : null;
} else if (
(lp.kind.startsWith("orientation.") || lp.kind.startsWith("frontier.")) &&
dataObj
) {
result.orientation = normalizeOrientation(dataObj);
}
return result;
}
// union of lane payloads
export type LanePayload =
| ScratchpadPayload
| CellPayload
| SpeechPayload
| InputPayload
| LifecyclePayload;
// sequence-stamped event record with discriminated lane payloads
export type V2EventRecord =
| { seq: number; session_id: string; lane: "scratchpad"; payload: ScratchpadPayload; created_ms: number }
| { seq: number; session_id: string; lane: "cell"; payload: CellPayload; created_ms: number }
| { seq: number; session_id: string; lane: "speech"; payload: SpeechPayload; created_ms: number }
| { seq: number; session_id: string; lane: "input"; payload: InputPayload; created_ms: number }
| { seq: number; session_id: string; lane: "lifecycle"; payload: LifecyclePayload; created_ms: number };
// client -> daemon v2 messages
export type V2ClientMsg =
| { type: "subscribe"; session_id: string; after_seq?: number | null; limit?: number | null }
| { type: "history"; session_id: string; before_seq: number; limit: number }
| { type: "send_input"; session_id: string; source: string; text: string; client_request_id?: string | null }
| { type: "interrupt"; session_id: string; execution_id?: string | null }
| { type: "reset_kernel"; session_id: string }
| { type: "compact"; session_id: string }
| { type: "stop"; session_id: string }
| { type: "drain"; session_id: string }
| { type: "activate"; session_id: string; expected_epoch: number; release_path: string }
| { type: "session_status"; session_id: string }
| { type: "session_list" }
| { type: "ack"; session_id: string; seq: number }
| { type: "ping"; timestamp?: number | null };
// daemon -> client v2 messages
export type V2ServerMsg =
| {
type: "sync_started";
session_id: string;
from_seq: number;
to_seq: number;
/** The oldest record the daemon can still serve; absent = it cannot page. */
earliest_seq?: number | null;
/** The record the session's active context fold drained through, when it has one. */
fold_seq?: number | null;
}
| { type: "event"; record: V2EventRecord }
| { type: "sync_completed"; session_id: string; latest_seq: number; earliest_seq?: number | null }
| { type: "live_chunk"; session_id: string; lane: Lane; execution_id?: string | null; stream?: string | null; delta: string }
| { type: "ack"; session_id: string; seq: number; client_request_id?: string | null; duplicate?: boolean | null }
| { type: "pong"; timestamp?: number | null }
| { type: "operator_ack"; action: string; session_id?: string | null; success: boolean; message?: string | null }
| { type: "session_status"; session_id: string; state: unknown }
| { type: "session_list"; sessions: Array<[string, unknown]> }
| { type: "error"; code: string; message: string };
// Sequence cursor algebra
export type CursorAction =
| { action: "accept"; seq: number }
| { action: "duplicate"; seq: number }
| { action: "gap"; expected: number; received: number };
export class CursorTracker {
private _lastSeq: number;
private _sparse: boolean;
constructor(initialSeq: number = 0, sparse: boolean = true) {
this._lastSeq = initialSeq;
this._sparse = sparse;
}
get lastSeq(): number {
return this._lastSeq;
}
get expectedSeq(): number {
return this._lastSeq + 1;
}
reset(seq: number): void {
this._lastSeq = seq;
}
process(incomingSeq: number): CursorAction {
if (this._sparse) {
if (incomingSeq > this._lastSeq) {
this._lastSeq = incomingSeq;
return { action: "accept", seq: incomingSeq };
} else {
return { action: "duplicate", seq: incomingSeq };
}
}
if (incomingSeq === this._lastSeq + 1) {
this._lastSeq = incomingSeq;
return { action: "accept", seq: incomingSeq };
} else if (incomingSeq <= this._lastSeq) {
return { action: "duplicate", seq: incomingSeq };
} else {
return {
action: "gap",
expected: this._lastSeq + 1,
received: incomingSeq,
};
}
}
processContiguous(incomingSeq: number): CursorAction {
if (incomingSeq === this._lastSeq + 1) {
this._lastSeq = incomingSeq;
return { action: "accept", seq: incomingSeq };
} else if (incomingSeq <= this._lastSeq) {
return { action: "duplicate", seq: incomingSeq };
} else {
return {
action: "gap",
expected: this._lastSeq + 1,
received: incomingSeq,
};
}
}
}
export type ResumptionAction<T> =
| { action: "ready"; items: T[] }
| { action: "duplicate"; seq: number }
| { action: "gap"; expected: number; received: number };
export class ResumptionWindow<T> {
private tracker: CursorTracker;
private buffer: Map<number, T>;
private sparse: boolean;
constructor(initialSeq: number = 0, sparse: boolean = false) {
this.tracker = new CursorTracker(initialSeq, sparse);
this.buffer = new Map();
this.sparse = sparse;
}
static sparse<T>(initialSeq: number = 0): ResumptionWindow<T> {
return new ResumptionWindow<T>(initialSeq, true);
}
get lastSeq(): number {
return this.tracker.lastSeq;
}
get expectedSeq(): number {
return this.tracker.expectedSeq;
}
get bufferedCount(): number {
return this.buffer.size;
}
reset(seq: number): void {
this.tracker.reset(seq);
this.buffer.clear();
}
process(seq: number, item: T): ResumptionAction<T> {
if (seq <= this.tracker.lastSeq || this.buffer.has(seq)) {
return { action: "duplicate", seq };
}
if (this.sparse) {
this.tracker.process(seq);
return { action: "ready", items: [item] };
}
if (seq === this.tracker.expectedSeq) {
this.tracker.processContiguous(seq);
const ready: T[] = [item];
while (this.buffer.has(this.tracker.expectedSeq)) {
const nextSeq = this.tracker.expectedSeq;
const nextItem = this.buffer.get(nextSeq)!;
this.buffer.delete(nextSeq);
this.tracker.processContiguous(nextSeq);
ready.push(nextItem);
}
return { action: "ready", items: ready };
} else {
const expected = this.tracker.expectedSeq;
this.buffer.set(seq, item);
return { action: "gap", expected, received: seq };
}
}
drainWatermark(watermark: number): T[] {
const ready: T[] = [];
const keys = Array.from(this.buffer.keys()).sort((a, b) => a - b);
for (const seq of keys) {
if (seq <= watermark) {
const item = this.buffer.get(seq)!;
this.buffer.delete(seq);
if (seq > this.tracker.lastSeq) {
this.tracker.reset(seq);
}
ready.push(item);
} else {
break;
}
}
return ready;
}
missingGaps(): Array<[number, number]> {
const gaps: Array<[number, number]> = [];
let curr = this.tracker.expectedSeq;
const sortedKeys = Array.from(this.buffer.keys()).sort((a, b) => a - b);
for (const seq of sortedKeys) {
if (seq > curr) {
gaps.push([curr, seq - 1]);
}
curr = seq + 1;
}
return gaps;
}
}
export function processEvent(
window: ResumptionWindow<V2EventRecord>,
record: V2EventRecord
): ResumptionAction<V2EventRecord> {
return window.process(record.seq, record);
}
// Compatibility types and helper functions for klbr-web consumers
/** The window request and the history request, and what the daemon reports about its own history.
* `subscribe` with `after_seq: 0` and a `limit` asks for the newest window it serves — at most `limit`
* records, opening after the session's own context fold — rather than for the whole session. The daemon
* reports the range it chose, how far back it can still serve, and the record its fold drained through in
* `sync_started`; a daemon that knows nothing about windows ignores the limit and replays everything.
* `history` asks for the records behind a loaded window, for a reader who pressed a button, and is
* answered with the same replay shape as a subscribe.
*
* A range and the floor under it are announced together, in `sync_started`, and not only when the
* replay finishes: a console that learns there is history behind its window only at the end would grow
* a control at the top of a transcript the reader is already reading, and move the page under them. */
export type ClientMsgV2 = V2ClientMsg;
export type ServerMsgV2 =
| V2ServerMsg
| { type: "event"; seq: number; session_id: string; lane: Lane; payload: LanePayload; created_ms: number };
export type LaneEventRecord = V2EventRecord | {
seq: number;
session_id: string;
lane: Lane;
payload: any;
created_ms: number;
};
export type SpeechStatus = "confirmed" | "held" | "failed" | "rejected";
export type CellOutputStream = CellStreamKind;
export type CellExecutionStatus = CellState | "running";
export interface SpeechReceipt {
effect_id: string;
status: SpeechStatus | string;
event_id?: number | null;
reason?: string | null;
text: string;
author?: string;
source?: string;
provenance?: ProvenanceMetadata;
}
export function validateSpeechStatus(
rawStatus: unknown,
rawType?: unknown
): { status: "confirmed" | "held" | "failed"; reason?: string } {
// ONLY explicitly confirmed status becomes confirmed
if (rawStatus === "confirmed" && (rawType === "sent" || rawType === undefined)) {
return { status: "confirmed" };
}
if (rawStatus === "failed" || rawType === "failed") {
return { status: "failed", reason: "delivery failed" };
}
if (rawStatus === "held" || rawType === "held") {
return { status: "held", reason: "held by supervisor policy" };
}
// Unknown or missing statuses must NEVER default to 'confirmed'
// Visibly classify as held / error
const statusStr = typeof rawStatus === "string" ? rawStatus : "unknown";
return { status: "held", reason: `unvalidated delivery status: ${statusStr}` };
}
export type CellOutputRecord = CellStreamOutput;
export interface CellExecutionPayload {
type?: "started" | "completed";
execution_id?: string;
command?: string;
code?: string;
state?: CellState;
stream?: CellStreamKind;
text?: string;
output?: CellStreamOutput[];
outputs?: CellOutputRecord[];
status?: CellExecutionStatus;
truncated?: boolean;
truncated_bytes?: number;
exit_code?: number;
duration_ms?: number;
error?: string | null;
}
const VALID_LANES = new Set<string>(["scratchpad", "cell", "speech", "input", "lifecycle"]);
const VALID_HEALTH_STATUSES = new Set(["healthy", "degraded", "unavailable", "unknown", "stale"]);
const VALID_ORIGINS = new Set(["human", "agent", "checker", "system", "external"]);
const VALID_WORK_STATUSES = new Set([
"pending",
"selected",
"deferred",
"executing",
"completed",
"failed",
"cancelled",
"claimed",
]);
export function validateLifecyclePayload(p: Record<string, unknown>): void {
const dataObj = (p.data && typeof p.data === "object" ? p.data : null) as Record<string, unknown> | null;
const health = dataObj?.health ?? p.health;
if (health !== undefined) {
if (!Array.isArray(health)) {
throw new Error("malformed lifecycle payload: health must be an array");
}
for (const h of health) {
if (!h || typeof h !== "object") {
throw new Error("malformed lifecycle payload: health observation must be an object");
}
if (typeof (h as any).id !== "string" || !(h as any).id.trim()) {
throw new Error("malformed lifecycle payload: health observation id must be a non-empty string");
}
const st =
typeof (h as any).status === "object" && (h as any).status && "status" in (h as any).status
? (h as any).status.status
: (h as any).status;
if (typeof st !== "string" || !VALID_HEALTH_STATUSES.has(st)) {
throw new Error(`malformed lifecycle payload: invalid health status: ${st}`);
}
const orig =
(h as any).origin_kind ??
(h as any).origin ??
(h as any).provenance?.origin_kind ??
(h as any).provenance?.origin;
if (orig !== undefined && (!VALID_ORIGINS.has(orig) || typeof orig !== "string")) {
throw new Error(`malformed lifecycle payload: invalid health origin: ${orig}`);
}
}
}
const workItems = dataObj?.work_items ?? p.work_items;
if (workItems !== undefined) {
if (!Array.isArray(workItems)) {
throw new Error("malformed lifecycle payload: work_items must be an array");
}
for (const w of workItems) {
if (!w || typeof w !== "object") {
throw new Error("malformed lifecycle payload: work item must be an object");
}
if (typeof (w as any).id !== "string" || !(w as any).id.trim()) {
throw new Error("malformed lifecycle payload: work item id must be a non-empty string");
}
const st =
typeof (w as any).status === "object" && (w as any).status && "status" in (w as any).status
? (w as any).status.status
: (w as any).status;
if (typeof st !== "string" || !VALID_WORK_STATUSES.has(st)) {
throw new Error(`malformed lifecycle payload: invalid work item status: ${st}`);
}
const orig =
(w as any).origin_kind ??
(w as any).origin ??
(w as any).provenance?.origin_kind ??
(w as any).provenance?.origin;
if (orig !== undefined && (!VALID_ORIGINS.has(orig) || typeof orig !== "string")) {
throw new Error(`malformed lifecycle payload: invalid work item origin: ${orig}`);
}
}
}
const candidate = dataObj && "candidate" in dataObj ? dataObj.candidate : p.candidate;
if (candidate !== undefined && candidate !== null) {
if (typeof candidate !== "object") {
throw new Error("malformed lifecycle payload: candidate must be an object");
}
const c = candidate as Record<string, unknown>;
if (typeof c.candidate_id !== "string" || !c.candidate_id.trim()) {
throw new Error("malformed lifecycle payload: candidate candidate_id must be a non-empty string");
}
if (typeof c.base_revision !== "string") {
throw new Error("malformed lifecycle payload: candidate base_revision must be a string");
}
if (typeof c.target_revision !== "string") {
throw new Error("malformed lifecycle payload: candidate target_revision must be a string");
}
if (typeof c.status !== "string") {
throw new Error("malformed lifecycle payload: candidate status must be a string");
}
}
}
export function validateEventRecord(record: unknown): V2EventRecord {
if (!record || typeof record !== "object") {
throw new Error("malformed event record: record must be an object");
}
const r = record as Record<string, unknown>;
if (typeof r.seq !== "number" || !Number.isFinite(r.seq) || r.seq < 0) {
throw new Error(`malformed event record: invalid seq: ${r.seq}`);
}
if (typeof r.session_id !== "string" || r.session_id.trim().length === 0) {
throw new Error("malformed event record: invalid session_id");
}
if (typeof r.lane !== "string" || !VALID_LANES.has(r.lane)) {
throw new Error(`malformed event record: invalid lane: ${r.lane}`);
}
if (typeof r.created_ms !== "number" || !Number.isFinite(r.created_ms) || r.created_ms < 0) {
throw new Error(`malformed event record: invalid created_ms: ${r.created_ms}`);
}
if (!r.payload || typeof r.payload !== "object") {
throw new Error("malformed event record: payload must be a non-null object");
}
const p = r.payload as Record<string, unknown>;
switch (r.lane) {
case "scratchpad": {
if (
typeof p.text !== "string" &&
typeof p.thoughts !== "string" &&
typeof p.prose !== "string" &&
!p.tool_call
) {
throw new Error("malformed scratchpad payload");
}
// Reject diagnostics on scratchpad lane (gate malformation)
if ("health" in p || "work_items" in p || "candidate" in p) {
throw new Error("malformed scratchpad payload: diagnostics must not enter scratchpad lane");
}
// Reject stringified JSON diagnostics in text field
if (typeof p.text === "string") {
try {
const parsed = JSON.parse(p.text);
if (parsed && typeof parsed === "object" &&
(("type" in parsed && (parsed.type === "capability.checked" || parsed.type === "work.completed")) ||
("health" in parsed || "work_items" in parsed || "candidate" in parsed))) {
throw new Error(
"malformed scratchpad payload: diagnostics JSON detected in text field - " +
"emit as lifecycle lane event instead"
);
}
} catch (e) {
if (e instanceof SyntaxError) {
// Not JSON, that's fine for scratchpad
} else {
throw e;
}
}
}
break;
}
case "cell": {
if (
p.type !== "started" &&
p.type !== "completed" &&
!p.execution_id &&
!p.state &&
!p.status &&
!p.code &&
!p.command
) {
throw new Error("malformed cell payload");
}
break;
}
case "speech": {
if (typeof p.text !== "string" && typeof p.reason !== "string") {
throw new Error("malformed speech payload");
}
if ("health" in p || "work_items" in p || "candidate" in p) {
throw new Error("malformed speech payload: diagnostics must not enter speech lane");
}
break;
}
case "input": {
if (typeof p.text !== "string") {
throw new Error("malformed input payload");
}
break;
}
case "lifecycle": {
if (typeof p.kind !== "string" || !p.kind.trim()) {
throw new Error("malformed lifecycle payload: kind must be a non-empty string");
}
// Reject double-nested patterns (e.g. hooks.activated wrapping candidate.activated)
const dataObj = p.data && typeof p.data === "object" ? p.data as Record<string, unknown> : null;
if (dataObj && "kind" in dataObj && "data" in dataObj) {
const innerData = dataObj.data;
if (innerData && typeof innerData === "object" &&
("candidate" in (innerData as Record<string, unknown>) ||
"health" in (innerData as Record<string, unknown>) ||
"work_items" in (innerData as Record<string, unknown>))) {
throw new Error(
`malformed lifecycle payload: nested kind/data structure detected (${p.kind} -> ${dataObj.kind}) - ` +
`emit flat {kind, data: {health, work_items, candidate}} instead`
);
}
}
validateLifecyclePayload(p);
break;
}
}
return r as unknown as V2EventRecord;
}
// Explicit protocol version planes for the v2 client vocabulary. Missing
// fields mean the current version (backward compatible); a PRESENT field
// with an unknown value is a typed refusal, never a silent accept.
export const V2_PROTOCOL_VERSION = 2;
// Distinct version plane for the remote relay framing (worker relay
// handshake), mirroring klbr_runtime::protocol::RELAY_PROTOCOL_VERSION.
export const RELAY_PROTOCOL_VERSION = 1;
export function checkRelayProtocolVersion(value: unknown): void {
if (typeof value !== "number" || value !== RELAY_PROTOCOL_VERSION) {
throw new Error(
`unknown relay protocol version: ${JSON.stringify(value)}; this client agrees with relay version ${RELAY_PROTOCOL_VERSION}`
);
}
}
export function checkProtocolVersion(value: unknown): void {
if (value === undefined) {
return; // frames without a version field are current-version frames
}
if (typeof value !== "number" || !Number.isInteger(value) || value <= 0) {
throw new Error(`malformed protocol version: ${JSON.stringify(value)}`);
}
if (value !== V2_PROTOCOL_VERSION) {
throw new Error(
`unknown server protocol version: ${value}; this client speaks version ${V2_PROTOCOL_VERSION}`
);
}
}
export function parseServerMsgV2(raw: string): ServerMsgV2 {
const parsed = JSON.parse(raw);
if (!parsed || typeof parsed !== "object" || !("type" in parsed)) {
throw new Error("malformed message: missing type");
}
checkProtocolVersion(parsed.protocol_version);
switch (parsed.type) {
case "sync_started": {
if (
typeof parsed.session_id !== "string" ||
typeof parsed.from_seq !== "number" ||
typeof parsed.to_seq !== "number"
) {
throw new Error("malformed sync_started message");
}
break;
}
case "sync_completed": {
if (
typeof parsed.session_id !== "string" ||
typeof parsed.latest_seq !== "number"
) {
throw new Error("malformed sync_completed message");
}
break;
}
case "event": {
if (parsed.record) {
validateEventRecord(parsed.record);
} else {
validateEventRecord(parsed);
}
break;
}
case "live_chunk": {
if (
typeof parsed.session_id !== "string" ||
!VALID_LANES.has(parsed.lane) ||
typeof parsed.delta !== "string"
) {
throw new Error("malformed live_chunk message");
}
break;
}
case "ack": {
if (
typeof parsed.session_id !== "string" ||
typeof parsed.seq !== "number"
) {
throw new Error("malformed ack message");
}
break;
}
case "pong": {
break;
}
case "operator_ack": {
if (
typeof parsed.action !== "string" ||
typeof parsed.success !== "boolean" ||
(parsed.session_id != null && typeof parsed.session_id !== "string") ||
(parsed.message != null && typeof parsed.message !== "string")
) {
throw new Error("malformed operator_ack message");
}
break;
}
case "session_status": {
if (typeof parsed.session_id !== "string" || !("state" in parsed)) {
throw new Error("malformed session_status message");
}
break;
}
case "session_list": {
if (!Array.isArray(parsed.sessions)) {
throw new Error("malformed session_list message");
}
break;
}
case "error": {
if (
typeof parsed.code !== "string" ||
typeof parsed.message !== "string"
) {
throw new Error("malformed error message");
}
break;
}
default:
throw new Error(`unknown server message type: ${parsed.type}`);
}
return parsed as ServerMsgV2;
}
export function extractLaneRecord(msg: ServerMsgV2 & { type: "event" }): LaneEventRecord {
if ("record" in msg && msg.record && typeof msg.record === "object") {
return msg.record;
}
return {
seq: (msg as any).seq ?? 0,
session_id: (msg as any).session_id ?? "",
lane: (msg as any).lane,
payload: (msg as any).payload ?? {},
created_ms: (msg as any).created_ms ?? Date.now(),
};
}
export interface InputOutcome {
status: "queued" | "running" | "replied" | "held" | "failed" | "no_reply";
reason?: string;
error?: string;
details?: string;
retryable?: boolean;
}
export interface DiagnosticItem {
id: string;
seq?: number;
created_ms: number;
kind: string;
title: string;
summary?: string;
data: unknown;
}
export function separateThoughtsAndProse(sp: ScratchpadPayload): { thoughts: string; prose: string } {
// Prefer explicit sp.reasoning as provider reasoning
if (sp.reasoning !== undefined && sp.reasoning !== null) {
const thoughts = sp.reasoning.trim();
const prose = (sp.text || sp.prose || (sp as any).content || "").trim();
return { thoughts, prose };
}
let thoughts = (sp.thoughts || "").trim();
let prose = (sp.prose || "").trim();
// If already separated
if (thoughts && prose) {
if (thoughts === prose) {
if (sp.kind === "thought") prose = "";
else thoughts = "";
}
return { thoughts, prose };
}
const raw = (sp.text || (sp as any).content || "").trim();
if (!raw) {
return { thoughts, prose };
}
// Detect <think>...</think> or <thought>...</thought>
const match = raw.match(/<(?:think|thought)>([\s\S]*?)<\/(?:think|thought)>/i);
if (match) {
const extractedThoughts = match[1].trim();
const extractedProse = raw.replace(/<(?:think|thought)>[\s\S]*?<\/(?:think|thought)>/gi, "").trim();
return {
thoughts: thoughts || extractedThoughts,
prose: prose || extractedProse,
};
}
if (sp.kind === "thought" || (sp as any).stream === "thought" || (sp as any).stream === "think") {
return {
thoughts: thoughts || raw,
prose,
};
}
if (sp.kind === "prose" || (sp as any).stream === "prose") {
return {
thoughts,
prose: prose || raw,
};
}
if (thoughts) {
if (raw !== thoughts) {
prose = raw;
}
return { thoughts, prose };
}
if (prose) {
if (raw !== prose) {
thoughts = raw;
}
return { thoughts, prose };
}
if ((sp as any).thoughts && typeof (sp as any).thoughts === "string") {
thoughts = (sp as any).thoughts.trim();
}
if ((sp as any).content && typeof (sp as any).content === "string") {
prose = (sp as any).content.trim();
}
if (!thoughts && !prose) {
prose = raw;
}
return { thoughts, prose };
}
export function isFailurePayload(payload: unknown): { isFailure: boolean; reason?: string } {
if (!payload) return { isFailure: false };
if (typeof payload === "string") {
const trimmed = payload.trim();
if (trimmed.startsWith("tool_result: error") || trimmed.startsWith("error:")) {
return { isFailure: true, reason: trimmed };
}
if (trimmed.startsWith("{") && trimmed.endsWith("}")) {
try {
const obj = JSON.parse(trimmed);
return isFailurePayload(obj);
} catch {
return { isFailure: false };
}
}
return { isFailure: false };
}
if (typeof payload === "object") {
const p = payload as Record<string, unknown>;
if (p.status === "error" || p.status === "failed") {
const reason = p.error || p.reason || p.content || `status: ${p.status}`;
return { isFailure: true, reason: typeof reason === "string" ? reason : JSON.stringify(reason) };
}
if (p.type === "tool_result" && (p.status === "error" || p.error)) {
const reason = p.error || p.content || "tool execution error";
return { isFailure: true, reason: typeof reason === "string" ? reason : JSON.stringify(reason) };
}
if (p.error && typeof p.error === "string") {
return { isFailure: true, reason: p.error };
}
}
return { isFailure: false };
}
export function isDiagnosticPayload(payload: unknown): {
isDiagnostic: boolean;
kind?: string;
data?: unknown;
} {
if (!payload) return { isDiagnostic: false };
let obj: any = payload;
if (typeof payload === "string") {
const trimmed = payload.trim();
if (trimmed.startsWith("{") && trimmed.endsWith("}")) {
try {
obj = JSON.parse(trimmed);
} catch {
return { isDiagnostic: false };
}
} else {
return { isDiagnostic: false };
}
}
if (typeof obj !== "object" || obj === null) return { isDiagnostic: false };
if (obj.type === "context_receipt" || obj.kind === "context.receipt" || obj.context_receipt !== undefined) {
return { isDiagnostic: true, kind: "context.receipt", data: obj.context_receipt || obj };
}
if (obj.type === "request_usage" || obj.kind === "model.usage" || obj.request_sha256 !== undefined) {
return { isDiagnostic: true, kind: "model.usage", data: obj };
}
if (obj.kind === "kernel.reset" || obj.type === "kernel.reset") {
return { isDiagnostic: true, kind: "kernel.reset", data: obj };
}
if (obj.kind === "runtime.slice" || obj.type === "runtime.slice") {
return { isDiagnostic: true, kind: "runtime.slice", data: obj };
}
return { isDiagnostic: false };
}