From 3cd4fea600b82f68f8cc840ed3db0ee193344550 Mon Sep 17 00:00:00 2001
From: jochen <jschoubben@gmail.com>
Date: Wed, 11 Nov 2026 13:08:17 +0100
Subject: [PATCH] T-3: the finding writer in src/views (TypeScript twin),
 transport handed in
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Builds: design 47 G §383, §384, §390, §439, §461, §485, §503

Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ
---
 src/views/index.ts  |   1 +
 src/views/writer.ts | 345 ++++++++++++++++++++++++++++++++++++++++++++
 test/sdk.test.ts    |  93 ++++++++++++
 3 files changed, 439 insertions(+)
 create mode 100644 src/views/writer.ts

diff --git a/src/views/index.ts b/src/views/index.ts
index a489b23b..5496130f 100644
--- a/src/views/index.ts
+++ b/src/views/index.ts
@@ -563,3 +563,4 @@ export function statusTools(
     },
   ];
 }
+export * from "./writer.js";
diff --git a/src/views/writer.ts b/src/views/writer.ts
new file mode 100644
index 00000000..ba0efb90
--- /dev/null
+++ b/src/views/writer.ts
@@ -0,0 +1,345 @@
+// The finding writer, TypeScript side (design 47 G §461, §485, §503): the same acts as the Go `views` writer, with the
+// same bodies (§439) and rules, for the observers written in TypeScript. It keeps an observer's findings in its own state
+// `findings` and says them as events (§383, §384). It names no seat and no module: the observer hands in the subjects of
+// `finding-written`, `finding-cleared`, `findings-stated` and `findings-asked`, which the seat's generated contract makes.
+// Its acts, `write`, `clear` and a statement, run one at a time in the order they are asked, on one chain of promises.
+
+import type { State } from "../state/index.js";
+
+/** How long an event the bus did not acknowledge is sent again before it is logged as an error (§383). */
+export const PUBLISH_RETRY_MS = 60 * 1000;
+/** How often a failing `observe` is called again until a set is said (§485). */
+export const OBSERVE_RETRY_MS = 60 * 1000;
+
+/** What a finding is about. */
+export interface FindingSubject {
+  scope: string;
+  id: string;
+  machine?: string;
+  also?: string[];
+}
+
+/** One observation, as it was said. */
+export interface FindingEvidence {
+  at: string;
+  said: string;
+}
+
+/** One answer the operator may give from a notification. */
+export interface FindingAction {
+  label: string;
+  verb: string;
+  machine?: string;
+  arguments?: Record<string, string>;
+  level?: "acknowledge" | "approve";
+}
+
+/**
+ * One finding as an observer says it: the layout whole of §439's list. It is the body of `finding-written`, of each entry
+ * of `findings-stated`, and the value of its key `<scope>.<id>.<kind>` in the state `findings`.
+ */
+export interface FindingBody {
+  kind: string;
+  scope: string;
+  id: string;
+  at?: string;
+  since?: string;
+  severity: "urgent" | "warning";
+  summary: string;
+  headline?: string;
+  explanation?: string;
+  subject: FindingSubject;
+  evidence?: FindingEvidence[];
+  source: string;
+  observer?: string;
+  owner?: string;
+  "waiting-on"?: string;
+  actions?: FindingAction[];
+  needs?: string;
+  resolver?: string;
+  ref?: string;
+  run?: string;
+}
+
+/** The body of `finding-cleared`. */
+export interface FindingCleared {
+  kind: string;
+  scope: string;
+  id: string;
+  at: string;
+  why: string;
+}
+
+/** A key the observer cannot judge yet since its start: its `since` and its bound (a duration such as `5m`). */
+export interface PendingKey {
+  key: string;
+  since: string;
+  bound: string;
+}
+
+/** The body of `findings-stated`. */
+export interface FindingsStated {
+  at: string;
+  asked?: string;
+  findings: FindingBody[];
+  pending: PendingKey[];
+}
+
+/** The body of `findings-asked`. */
+export interface FindingsAsked {
+  ask: string;
+  at: string;
+  by: string;
+  module?: string;
+}
+
+/** What the observer's own truth says now. */
+export interface Observed {
+  findings: FindingBody[];
+  pending: PendingKey[];
+}
+
+/** The key of a finding in the state `findings`: `<scope>.<id>.<kind>`. */
+export function findingKey(scope: string, id: string, kind: string): string {
+  return `${scope}.${id}.${kind}`;
+}
+
+export interface WriterOptions {
+  /** The observer's module name: it fills the `observer` of a finding and tells an ask for another module. */
+  module: string;
+  /** The observer's own state `findings`. */
+  state: State<FindingBody>;
+  /** The subjects of the four events. */
+  subjects: { written: string; cleared: string; stated: string; asked: string };
+  /** The findings the observer observes now and its pending keys. It is called inside the writer's one chain: it must not call `write` or `clear`. */
+  observe(): Promise<Observed>;
+  /** Say one message and resolve when the bus acknowledged it; `id` is the bus's de-duplication id. */
+  publish(subject: string, body: unknown, id: string): Promise<void>;
+  /** Make a consumer of `subject` that delivers only the messages published after it is made; resolves to its stop. */
+  listen(subject: string, handle: (body: unknown) => void): Promise<() => void>;
+  /** Put a set larger than one bus message in the observer's sent store; resolves to what to publish instead. */
+  send?(label: string, body: unknown): Promise<unknown>;
+  log?(line: string): void;
+  now?(): Date;
+  /** Pause for `ms`; resolve false when the writer closed. */
+  wait?(ms: number): Promise<boolean>;
+}
+
+/** The finding writer of an observer. */
+export class Writer {
+  private readonly o: WriterOptions;
+  private chain: Promise<unknown> = Promise.resolve();
+  private seq = 0;
+  private stop?: () => void;
+  private asks: string[] = [];
+  private busy = false;
+  private closed = false;
+  private readonly closing: Promise<void>;
+  private close_!: () => void;
+
+  constructor(o: WriterOptions) {
+    this.o = o;
+    this.closing = new Promise((r) => (this.close_ = r));
+  }
+
+  private log(line: string): void {
+    this.o.log?.(line);
+  }
+  private now(): Date {
+    return this.o.now ? this.o.now() : new Date();
+  }
+  private stamp(d: Date): string {
+    return d.toISOString();
+  }
+  private async pause(ms: number): Promise<boolean> {
+    if (this.o.wait) return this.o.wait(ms);
+    return new Promise((resolve) => {
+      const t = setTimeout(() => resolve(!this.closed), ms);
+      void this.closing.then(() => {
+        clearTimeout(t);
+        resolve(false);
+      });
+    });
+  }
+
+  /** Run `act` after every act asked before it has ended. */
+  private act<T>(act: () => Promise<T>): Promise<T> {
+    const run = this.chain.then(act, act);
+    this.chain = run.catch(() => undefined);
+    return run;
+  }
+
+  /**
+   * Make the consumer of `findings-asked`, then state the findings now; later each ask for this module (or for none) is
+   * answered by a statement. Resolves once the consumer exists; the start statement, retried every minute until `observe`
+   * answers, runs on.
+   */
+  async start(): Promise<void> {
+    await this.listen();
+    this.request("");
+  }
+
+  /** The bus connection was made again: make the consumer again, from the newest ask, and state the findings again (§485). */
+  async reconnected(): Promise<void> {
+    if (this.closed) return;
+    try {
+      await this.listen();
+    } catch (e) {
+      this.log(`error: the finding writer could not make its consumer of ${this.o.subjects.asked} again: ${e}`);
+    }
+    this.request("");
+  }
+
+  /** End the consumer and the statements. */
+  close(): void {
+    this.closed = true;
+    this.stop?.();
+    this.stop = undefined;
+    this.close_();
+  }
+
+  private async listen(): Promise<void> {
+    const stop = await this.o.listen(this.o.subjects.asked, (body) => this.heard(body));
+    this.stop?.();
+    this.stop = stop;
+  }
+
+  private heard(body: unknown): void {
+    const a = body as FindingsAsked | null;
+    if (!a || typeof a.ask !== "string") {
+      this.log("an ask for findings could not be read");
+      return;
+    }
+    if (a.module && a.module !== this.o.module) return;
+    this.request(a.ask);
+  }
+
+  private request(ask: string): void {
+    this.asks.push(ask);
+    if (this.busy) return;
+    this.busy = true;
+    void this.drain();
+  }
+
+  private async drain(): Promise<void> {
+    while (this.asks.length > 0 && !this.closed) {
+      const asked = this.asks[this.asks.length - 1];
+      this.asks = [];
+      for (;;) {
+        try {
+          await this.state(asked);
+          break;
+        } catch (e) {
+          this.log(`the findings could not be stated, no set said, trying again in ${OBSERVE_RETRY_MS / 1000}s: ${e}`);
+          if (!(await this.pause(OBSERVE_RETRY_MS))) {
+            this.busy = false;
+            return;
+          }
+        }
+      }
+    }
+    this.busy = false;
+  }
+
+  /** Keep the finding in its key and say `finding-written` only when what it says changed; the key's `since` stays. */
+  write(f: FindingBody): Promise<void> {
+    return this.act(async () => {
+      const [full, old] = await this.complete(f);
+      await this.o.state.put(key(full), full);
+      if (old && unsaid(old) === unsaid(full)) return;
+      await this.say(this.o.subjects.written, "finding-written", key(full), full);
+    });
+  }
+
+  /** Delete the key and say `finding-cleared`, also of a key that does not exist, so a keeper holding it clears it. */
+  clear(scope: string, id: string, kind: string, why: string): Promise<void> {
+    return this.act(async () => {
+      const k = findingKey(scope, id, kind);
+      await this.o.state.delete(k);
+      const body: FindingCleared = { kind, scope, id, at: this.stamp(this.now()), why };
+      await this.say(this.o.subjects.cleared, "finding-cleared", k, body);
+    });
+  }
+
+  /** Judge every finding anew, write each key, delete each key neither stated nor pending, and say `findings-stated`. */
+  state(asked = ""): Promise<void> {
+    return this.act(async () => {
+      const obs = await this.o.observe();
+      const keep = new Set<string>();
+      const body: FindingsStated = { at: this.stamp(this.now()), ...(asked ? { asked } : {}), findings: [], pending: [] };
+      for (const f of obs.findings) {
+        const [full] = await this.complete(f);
+        await this.o.state.put(key(full), full);
+        keep.add(key(full));
+        body.findings.push(full);
+      }
+      for (const p of obs.pending) {
+        keep.add(p.key);
+        body.pending.push(p);
+      }
+      for (const k of await this.o.state.keys()) {
+        if (!keep.has(k)) await this.o.state.delete(k);
+      }
+      let out: unknown = body;
+      if (this.o.send) {
+        try {
+          out = await this.o.send(`findings-stated.${this.o.module}`, body);
+        } catch (e) {
+          this.log(`error: the findings-stated of ${this.o.module} could not be sent, the next statement covers it: ${e}`);
+          return;
+        }
+      }
+      await this.say(this.o.subjects.stated, "findings-stated", this.o.module, out);
+    });
+  }
+
+  private async complete(f: FindingBody): Promise<[FindingBody, FindingBody | null]> {
+    if (!f.kind || !f.scope || !f.id) throw new Error(`a finding needs its kind, scope and id: ${f.scope}.${f.id}.${f.kind}`);
+    const e = await this.o.state.get(key(f));
+    const old = e ? e.value : null;
+    const at = f.at ?? this.stamp(this.now());
+    const since = old?.since ?? f.since ?? at;
+    return [{ ...f, observer: f.observer || this.o.module, at, since }, old];
+  }
+
+  /** Publish awaiting the acknowledgement, again for a minute when unacknowledged, then log the error naming what it said. */
+  private async say(subject: string, event: string, what: string, body: unknown): Promise<void> {
+    const id = `${this.o.module}-${event}-${this.now().getTime()}-${++this.seq}`;
+    const start = this.now().getTime();
+    for (let delay = 1000; ; delay = Math.min(delay * 2, 10000)) {
+      try {
+        await this.o.publish(subject, body, id);
+        return;
+      } catch (e) {
+        if (this.now().getTime() - start >= PUBLISH_RETRY_MS || this.closed || !(await this.pause(delay))) {
+          this.log(`error: ${event} for ${what} was not acknowledged by the bus, the next statement covers it: ${e}`);
+          return;
+        }
+      }
+    }
+  }
+}
+
+function key(f: FindingBody): string {
+  return findingKey(f.scope, f.id, f.kind);
+}
+
+/** A finding as compared to tell whether what it says changed: not the time it was last seen, nor what saw it. */
+function unsaid(f: FindingBody): string {
+  const { at: _at, run: _run, ...rest } = f;
+  return canonical(rest);
+}
+
+/** JSON with every object's keys in order, so two equal findings compare equal whatever order they were built in. */
+function canonical(v: unknown): string {
+  if (Array.isArray(v)) return `[${v.map(canonical).join(",")}]`;
+  if (v && typeof v === "object") {
+    const o = v as Record<string, unknown>;
+    return `{${Object.keys(o)
+      .filter((k) => o[k] !== undefined)
+      .sort()
+      .map((k) => `${JSON.stringify(k)}:${canonical(o[k])}`)
+      .join(",")}}`;
+  }
+  return JSON.stringify(v);
+}
diff --git a/test/sdk.test.ts b/test/sdk.test.ts
index cd71f496..8f59871e 100644
--- a/test/sdk.test.ts
+++ b/test/sdk.test.ts
@@ -833,3 +833,96 @@ test("a first view follows its owner's events, asks again on a gap and by hand,
   assert.equal(wRaised.length, 1, "a view held before its bound raises nothing");
   assert.equal(wCleared.length, 1, "nothing raised, nothing cleared");
 });
+
+test("the finding writer writes, clears and states findings as the Go writer does", async () => {
+  const { Writer, OBSERVE_RETRY_MS } = await import("../dist/views/index.js");
+  const store = new Map<string, any>();
+  const st: any = {
+    get: async (k: string) => (store.has(k) ? { key: k, value: store.get(k), revision: 1 } : null),
+    put: async (k: string, v: any) => (store.set(k, structuredClone(v)), 1),
+    delete: async (k: string) => void store.delete(k),
+    keys: async () => [...store.keys()].sort(),
+  };
+  let clock = Date.parse("2026-11-11T12:00:00Z");
+  const sent: { subject: string; body: any }[] = [];
+  let refuse = 0;
+  const waits: number[] = [];
+  let ask: ((b: unknown) => void) | undefined;
+  let consumers = 0;
+  let observes = 0;
+  const logs: string[] = [];
+  const w = new Writer({
+    module: "m",
+    state: st,
+    subjects: { written: "w", cleared: "c", stated: "s", asked: "a" },
+    observe: async () => {
+      if (++observes <= 2) throw new Error("not yet");
+      return { findings: [{ kind: "k", scope: "module", id: "a", severity: "warning", summary: "now", source: "p", subject: { scope: "module", id: "a" } }], pending: [{ key: "module.wait.k", since: "2026-11-11T11:00:00Z", bound: "5m" }] };
+    },
+    publish: async (subject, body) => {
+      if (refuse > 0) { refuse--; throw new Error("no ack"); }
+      sent.push({ subject, body });
+    },
+    listen: async (_s, h) => { consumers++; ask = h; return () => {}; },
+    log: (l) => logs.push(l),
+    now: () => new Date(clock),
+    wait: async (ms) => { waits.push(ms); clock += ms; return true; },
+  });
+  const f = { kind: "k", scope: "module", id: "a", severity: "warning" as const, summary: "one", source: "p", subject: { scope: "module", id: "a" } };
+  const first = new Date(clock).toISOString();
+  await w.write(f);
+  clock += 3600_000;
+  await w.write({ ...f, run: "other" }); // nothing it says changed
+  await w.write({ ...f, summary: "two" });
+  assert.deepEqual(sent.map((s) => s.subject), ["w", "w"]);
+  assert.equal(sent[1].body.since, first);
+  assert.equal(sent[1].body.observer, "m");
+  await w.clear("module", "never", "k", "held by the keeper");
+  assert.equal(sent[2].subject, "c");
+  assert.equal(sent[2].body.why, "held by the keeper");
+  // an unacknowledged event is sent again for a minute, then logged naming the finding
+  refuse = 1000;
+  const t0 = clock;
+  await w.write({ ...f, id: "b" });
+  assert.ok(clock - t0 >= 60_000 && clock - t0 < 70_000);
+  assert.ok(logs.some((l) => l.includes("module.b.k") && l.startsWith("error")));
+  refuse = 0;
+  // a start states what is observed now, retrying observe every minute, and deletes the keys not stated or pending
+  store.set("module.old.k", f);
+  waits.length = 0;
+  await w.start();
+  for (let i = 0; i < 50 && sent.length < 4; i++) await new Promise((r) => setImmediate(r));
+  const stated = sent[3];
+  assert.equal(stated.subject, "s");
+  assert.deepEqual(waits.slice(0, 2), [OBSERVE_RETRY_MS, OBSERVE_RETRY_MS]);
+  assert.equal(stated.body.findings.length, 1);
+  assert.equal(stated.body.findings[0].since, first);
+  assert.equal(stated.body.pending[0].key, "module.wait.k");
+  assert.deepEqual([...store.keys()].sort(), ["module.a.k"]);
+  // an ask for another module is not answered, one for this module is, with its id; a reconnect makes the consumer again
+  ask!({ ask: "a1", at: "", by: "keeper", module: "other" });
+  ask!({ ask: "a2", at: "", by: "keeper", module: "m" });
+  for (let i = 0; i < 50 && sent.length < 5; i++) await new Promise((r) => setImmediate(r));
+  assert.equal(sent.length, 5);
+  assert.equal(sent[4].body.asked, "a2");
+  await w.reconnected();
+  for (let i = 0; i < 50 && sent.length < 6; i++) await new Promise((r) => setImmediate(r));
+  assert.equal(consumers, 2);
+  assert.equal(sent.length, 6);
+  w.close();
+  // one act at a time: a write asked during a statement waits, so the statement never deletes the key it made
+  let release!: () => void;
+  const gate = new Promise<void>((r) => (release = r));
+  const w2 = new Writer({
+    module: "m", state: st, subjects: { written: "w", cleared: "c", stated: "s", asked: "a" },
+    observe: async () => { await gate; return { findings: [], pending: [] }; },
+    publish: async () => {}, listen: async () => () => {},
+  });
+  const stating = w2.state();
+  const writing = w2.write({ ...f, id: "during" });
+  for (let i = 0; i < 20; i++) await new Promise((r) => setImmediate(r));
+  assert.equal(store.has("module.during.k"), false);
+  release();
+  await Promise.all([stating, writing]);
+  assert.equal(store.has("module.during.k"), true);
+});