feat: durable pursuit observer plane

docs/api/index.md

@@ -7782,6 +7782,12 @@ Idempotency-Key header (e.g. the runId) — safe retries + upsert.
 
 > **id**: `string`
 
+##### pursuitId?
+
+> `optional` **pursuitId?**: `string`
+
+Stable identity for the long-lived objective. One pursuit may contain many runs.
+
 ##### runId
 
 > **runId**: `string`
@@ -7860,6 +7866,12 @@ Idempotency-Key header (e.g. the runId) — safe retries + upsert.
 
 > **id**: `string`
 
+##### pursuitId?
+
+> `optional` **pursuitId?**: `string`
+
+Stable identity for the long-lived objective. One pursuit may contain many runs.
+
 ##### runId
 
 > **runId**: `string`
@@ -11861,6 +11873,11 @@ Runtime hook contracts. Hooks are execution-scoped observers, not part of an
 `AgentProfile`: profiles stay portable agent recipes; hooks attach to the
 loop or product harness that is running the profile.
 
+A `pursuitId` is deliberately orthogonal to `runId`: a pursuit can span many
+resumed/retried/forked runs while every event remains attributable to the
+durable objective that caused it. The observer plane is outside the agent
+environment and must never be required for agent correctness.
+
 ***
 
 ### RuntimeHookTarget

docs/api/primitive-catalog.md

@@ -370,22 +370,30 @@ Import from `@tangle-network/agent-runtime/conversation` — 54 exports.
 
 ### Product chat turns — edge-safe streaming, persistence, and stable execution IDs
 
-Import from `@tangle-network/agent-runtime/durable` — 11 exports.
+Import from `@tangle-network/agent-runtime/durable` — 28 exports.
 
 | Symbol | Kind | Summary |
 |---|---|---|
+| `createFileObserverHooks` | function | Build the canonical durable observer hook in one call. |
 | `deriveExecutionId` | function | Derive a stable execution id from the run identity. |
 | `discoverDurableSupervisionRun` | function | Discover the stable identities recorded by Runtime's durable supervision |
 | `handleChatTurn` | function | Run one chat turn. Returns immediately with a `ReadableStream` body; |
+| `observerRecordDigest` | function | Compute the canonical SHA-256 digest for an unsigned observer record. |
+| `projectPursuit` | function | Fold one append-only execution journal into a deterministic operator projection. |
+| `supervisePursuit` | function | One-call durable pursuit execution over the canonical `supervise()` kernel. |
+| `verifyObserverRecords` | function | Verify identity, monotonic sequence, payload shape, and the complete digest chain. |
+| `FileObserverJournal` | class | Durable, append-only third-person history for one concrete Runtime execution. |
+| `SupervisePursuitError` | class | A failed Runtime execution whose complete third-person projection was retained. |
 | `ChatStreamEvent` | interface | The NDJSON line protocol every product chat client already speaks. |
 | `ChatTurnHooks` | interface | Product callbacks invoked while one chat turn runs. |
 | `ChatTurnIdentity` | interface | Identity of a chat turn. `tenantId` is the workspace id for workspace- |
 | `ChatTurnProducer` | interface | The live side of a turn returned by the product's `produce` hook. |
 | `ChatTurnResult` | interface | HTTP response values returned for one chat turn. |
 | `DurableSupervisionDiscovery` | interface | Identities discoverable from one `supervise({ runDir })` directory without |
+| `ObserverRecord` | interface | One immutable record in the observer plane. `sequence` is journal order, not |
 | `RunChatTurnInput` | interface | Inputs for one streamed product chat turn. |
 
-**Undocumented supporting types** (add a TSDoc line at the declaration to earn a table row): `DurableCoordinationStreamIdentity`.
+**Undocumented supporting types** (add a TSDoc line at the declaration to earn a table row): `DurableCoordinationStreamIdentity`, `ObserverJournal`, `PursuitNodeProjection`, `PursuitProjection`, `PursuitRunProjection`, `SupervisedPursuitResult`, `SupervisePursuitOptions`, `ObserverRecordKind`, `PursuitNodeStatus`, `PursuitRunStatus`.
 
 ### Bounded tool calls for browser and edge runtimes
 

docs/api/runtime.md

@@ -14905,6 +14905,7 @@ caller that owns the code registers it here once and names it from data thereaft
 
 #### Extended by
 
+- [`SupervisePursuitOptions`](durable.md#supervisepursuitoptions)
 - [`SuperviseTestOptions`](testing.md#supervisetestoptions)
 
 #### Properties

src/durable/index.ts

@@ -12,6 +12,12 @@
  *     persist and pass as both execution and turn identity on dispatch.
  *   - `discoverDurableSupervisionRun`: inspect a durable supervision directory
  *     without already knowing the root/run identities written inside it.
+ *   - `FileObserverJournal`: tamper-evident, append-only third-person history
+ *     for one concrete Runtime execution.
+ *   - `projectPursuit`: a rebuildable operator read model over that history;
+ *     it owns no execution or coordination semantics.
+ *   - `supervisePursuit`: one-call adapter over canonical `supervise()` that
+ *     gives each isolated run a stable cross-run pursuit identity.
  */
 
 export type {
@@ -24,6 +30,29 @@ export type {
 } from './chat-engine'
 export { handleChatTurn } from './chat-engine'
 export { deriveExecutionId } from './execution-handle'
+export {
+  createFileObserverHooks,
+  FileObserverJournal,
+  type ObserverJournal,
+  type ObserverRecord,
+  type ObserverRecordKind,
+  observerRecordDigest,
+  verifyObserverRecords,
+} from './observer-journal'
+export {
+  type PursuitNodeProjection,
+  type PursuitNodeStatus,
+  type PursuitProjection,
+  type PursuitRunProjection,
+  type PursuitRunStatus,
+  projectPursuit,
+} from './observer-projection'
+export {
+  type SupervisedPursuitResult,
+  SupervisePursuitError,
+  type SupervisePursuitOptions,
+  supervisePursuit,
+} from './supervise-pursuit'
 export {
   type DurableCoordinationStreamIdentity,
   type DurableSupervisionDiscovery,

src/durable/observer-journal.ts

@@ -0,0 +1,277 @@
+import { createHash } from 'node:crypto'
+import { readFile } from 'node:fs/promises'
+import { resolve } from 'node:path'
+import {
+  type RuntimeDecisionPoint,
+  type RuntimeHookEvent,
+  type RuntimeHooks,
+  withPursuitContext,
+} from '../runtime-hooks'
+import { parseCommittedJsonLines, prepareJsonlAppend, writeAllBytes } from './jsonl-file'
+
+export type ObserverRecordKind = 'event' | 'decision'
+
+/**
+ * One immutable record in the observer plane. `sequence` is journal order, not
+ * execution order; causal/runtime order remains available on the underlying event.
+ * `previousDigest` + `digest` make deletion, reordering, or mutation detectable.
+ */
+export interface ObserverRecord {
+  readonly schemaVersion: 1
+  readonly pursuitId: string
+  readonly sequence: number
+  readonly kind: ObserverRecordKind
+  readonly observedAt: number
+  readonly previousDigest?: string
+  readonly event?: RuntimeHookEvent
+  readonly decision?: RuntimeDecisionPoint
+  readonly digest: string
+}
+
+export interface ObserverJournal {
+  appendEvent(event: RuntimeHookEvent): Promise<ObserverRecord>
+  appendDecision(point: RuntimeDecisionPoint): Promise<ObserverRecord>
+  read(): Promise<readonly ObserverRecord[]>
+  hooks(): RuntimeHooks
+}
+
+type UnsignedObserverRecord = Omit<ObserverRecord, 'digest'>
+
+/**
+ * Durable, append-only third-person history for one concrete Runtime execution.
+ * It consumes Runtime's existing hook stream and does not participate in execution
+ * decisions. A broken observer therefore cannot change what an agent is allowed to do.
+ *
+ * The write discipline deliberately matches `FileSpawnJournal`: serialized appends,
+ * torn-tail recovery, short-write handling, and fsync before acknowledgement. One
+ * execution owns one journal file; higher-level pursuit aggregation joins isolated
+ * journals by `pursuitId` instead of making independent processes share a write head.
+ */
+export class FileObserverJournal implements ObserverJournal {
+  readonly path: string
+  readonly pursuitId: string
+  private tail: Promise<void> = Promise.resolve()
+  private initialized = false
+  private sequence = 0
+  private previousDigest: string | undefined
+  private appendFailure: Error | undefined
+
+  constructor(path: string, pursuitId: string) {
+    const stableId = pursuitId.trim()
+    if (stableId.length === 0) {
+      throw new TypeError('FileObserverJournal: pursuitId must be non-empty')
+    }
+    this.path = resolve(path)
+    this.pursuitId = stableId
+  }
+
+  hooks(): RuntimeHooks {
+    return withPursuitContext(this.pursuitId, {
+      onEvent: (event) => this.appendEvent(event).then(() => undefined),
+      onDecisionPoint: (point) => this.appendDecision(point).then(() => undefined),
+    })
+  }
+
+  appendEvent(event: RuntimeHookEvent): Promise<ObserverRecord> {
+    return this.enqueue('event', event)
+  }
+
+  appendDecision(point: RuntimeDecisionPoint): Promise<ObserverRecord> {
+    return this.enqueue('decision', point)
+  }
+
+  async read(): Promise<readonly ObserverRecord[]> {
+    await this.tail
+    this.assertComplete()
+    let text: string
+    try {
+      text = await readFile(this.path, 'utf8')
+    } catch (error) {
+      if (isNoEnt(error)) return []
+      throw error
+    }
+    return verifyObserverRecords(
+      parseCommittedJsonLines<ObserverRecord>(text, this.path),
+      this.pursuitId,
+    )
+  }
+
+  private enqueue(
+    kind: ObserverRecordKind,
+    value: RuntimeHookEvent | RuntimeDecisionPoint,
+  ): Promise<ObserverRecord> {
+    if (value.pursuitId !== this.pursuitId) {
+      return Promise.reject(
+        new Error(
+          `FileObserverJournal: ${kind} pursuitId ${String(value.pursuitId)} does not match ${this.pursuitId}`,
+        ),
+      )
+    }
+
+    let result: ObserverRecord | undefined
+    const operation = this.tail.then(async () => {
+      this.assertComplete()
+      try {
+        await this.initialize()
+      } catch (error) {
+        this.appendFailure ??= toError(error)
+        throw error
+      }
+
+      const unsigned: UnsignedObserverRecord = {
+        schemaVersion: 1,
+        pursuitId: this.pursuitId,
+        sequence: this.sequence + 1,
+        kind,
+        observedAt: Date.now(),
+        ...(this.previousDigest ? { previousDigest: this.previousDigest } : {}),
+        ...(kind === 'event'
+          ? { event: value as RuntimeHookEvent }
+          : { decision: value as RuntimeDecisionPoint }),
+      }
+      const record: ObserverRecord = Object.freeze({
+        ...unsigned,
+        digest: observerRecordDigest(unsigned),
+      })
+      try {
+        await this.writeRecord(record)
+      } catch (error) {
+        this.appendFailure ??= toError(error)
+        throw error
+      }
+      this.sequence = record.sequence
+      this.previousDigest = record.digest
+      result = record
+    })
+    this.tail = operation.then(
+      () => undefined,
+      () => undefined,
+    )
+    return operation.then(() => {
+      if (!result) throw new Error('FileObserverJournal: append completed without a record')
+      return result
+    })
+  }
+
+  private async initialize(): Promise<void> {
+    if (this.initialized) return
+    const records = await this.readExistingUnsafe()
+    const verified = verifyObserverRecords(records, this.pursuitId)
+    const tail = verified.at(-1)
+    this.sequence = tail?.sequence ?? 0
+    this.previousDigest = tail?.digest
+    // Only latch after recovery + verification succeed. A transient read error or
+    // corruption must never leave an instance pretending it initialized cleanly.
+    this.initialized = true
+  }
+
+  private async readExistingUnsafe(): Promise<ObserverRecord[]> {
+    let text: string
+    try {
+      text = await readFile(this.path, 'utf8')
+    } catch (error) {
+      if (isNoEnt(error)) return []
+      throw error
+    }
+    return parseCommittedJsonLines<ObserverRecord>(text, this.path)
+  }
+
+  private async writeRecord(record: ObserverRecord): Promise<void> {
+    const fs = await import('node:fs/promises')
+    const path = await import('node:path')
+    await fs.mkdir(path.dirname(this.path), { recursive: true })
+    const needsSeparator = await prepareJsonlAppend(this.path)
+    const handle = await fs.open(this.path, 'a')
+    try {
+      await writeAllBytes(handle, `${needsSeparator ? '\n' : ''}${JSON.stringify(record)}\n`)
+      await handle.sync()
+    } finally {
+      await handle.close()
+    }
+  }
+
+  private assertComplete(): void {
+    if (!this.appendFailure) return
+    throw new Error(
+      'FileObserverJournal: a prior durable append failed; observer completeness is unknown',
+      { cause: this.appendFailure },
+    )
+  }
+}
+
+/** Verify identity, monotonic sequence, payload shape, and the complete digest chain. */
+export function verifyObserverRecords(
+  records: readonly ObserverRecord[],
+  pursuitId?: string,
+): readonly ObserverRecord[] {
+  let previousDigest: string | undefined
+  let expectedSequence = 1
+  for (const record of records) {
+    if (record.schemaVersion !== 1) throw new Error('observer journal: unsupported schemaVersion')
+    if (pursuitId !== undefined && record.pursuitId !== pursuitId) {
+      throw new Error(`observer journal: pursuit identity mismatch at sequence ${record.sequence}`)
+    }
+    if (record.sequence !== expectedSequence) {
+      throw new Error(
+        `observer journal: non-contiguous sequence ${record.sequence}; expected ${expectedSequence}`,
+      )
+    }
+    if (record.previousDigest !== previousDigest) {
+      throw new Error(`observer journal: digest-chain break at sequence ${record.sequence}`)
+    }
+    if ((record.kind === 'event') === (record.event === undefined)) {
+      throw new Error(`observer journal: invalid event payload at sequence ${record.sequence}`)
+    }
+    if ((record.kind === 'decision') === (record.decision === undefined)) {
+      throw new Error(`observer journal: invalid decision payload at sequence ${record.sequence}`)
+    }
+    if (record.event !== undefined && record.event.pursuitId !== record.pursuitId) {
+      throw new Error(
+        `observer journal: nested event pursuit mismatch at sequence ${record.sequence}`,
+      )
+    }
+    if (record.decision !== undefined && record.decision.pursuitId !== record.pursuitId) {
+      throw new Error(
+        `observer journal: nested decision pursuit mismatch at sequence ${record.sequence}`,
+      )
+    }
+    const { digest, ...unsigned } = record
+    const expected = observerRecordDigest(unsigned)
+    if (digest !== expected) {
+      throw new Error(`observer journal: digest mismatch at sequence ${record.sequence}`)
+    }
+    previousDigest = digest
+    expectedSequence += 1
+  }
+  return Object.freeze([...records])
+}
+
+/** Compute the canonical SHA-256 digest for an unsigned observer record. */
+export function observerRecordDigest(record: Omit<ObserverRecord, 'digest'>): string {
+  return createHash('sha256').update(JSON.stringify(record)).digest('hex')
+}
+
+/** Build the canonical durable observer hook in one call. */
+export function createFileObserverHooks(
+  path: string,
+  pursuitId: string,
+): {
+  readonly journal: FileObserverJournal
+  readonly hooks: RuntimeHooks
+} {
+  const journal = new FileObserverJournal(path, pursuitId)
+  return { journal, hooks: journal.hooks() }
+}
+
+function isNoEnt(error: unknown): boolean {
+  return (
+    typeof error === 'object' &&
+    error !== null &&
+    'code' in error &&
+    (error as { code?: unknown }).code === 'ENOENT'
+  )
+}
+
+function toError(error: unknown): Error {
+  return error instanceof Error ? error : new Error(String(error))
+}