docs(rfc): RFC-5 polyglot task protocol — ecosystem citizenship for foreign-language tasks

.llm/runs/claude-harness-profile-rfc-benchmark-shzhgv--polyglot-protocol-rfc/bench/spikes/k1/emit-frames.go

@@ -0,0 +1,84 @@
+// K1 emitter (Go) — interleaves an adversarial log corpus with sentinel-framed protocol
+// frames from concurrent goroutines. Frames are single write() calls <= 4096 bytes (the
+// PIPE_BUF atomicity rule the protocol will mandate); logs are deliberately hostile.
+package main
+
+import (
+	"crypto/sha256"
+	"encoding/hex"
+	"encoding/json"
+	"fmt"
+	"os"
+	"strings"
+	"sync"
+)
+
+const SENTINEL = "\x00NSF\x00"
+
+func main() {
+	nLogs := 10000
+	nFrames := 200
+	var wg sync.WaitGroup
+	var mu sync.Mutex // logs use a mutex only sometimes — half the writers are unsynchronized
+
+	// frame writer goroutine: unsynchronized single-write frames (atomicity via PIPE_BUF)
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		for i := 0; i < nFrames; i++ {
+			payload := strings.Repeat("x", 64*(i%16))
+			h := sha256.Sum256([]byte(payload))
+			frame := map[string]any{
+				"v": 1, "t": "progress", "seq": i,
+				"payload": payload, "sha": hex.EncodeToString(h[:8]),
+			}
+			b, _ := json.Marshal(frame)
+			line := SENTINEL + string(b) + "\n"
+			if len(line) > 4096 {
+				panic("frame exceeds PIPE_BUF budget")
+			}
+			os.Stdout.WriteString(line) // one write syscall
+		}
+	}()
+
+	// hostile log writers
+	for w := 0; w < 4; w++ {
+		wg.Add(1)
+		go func(w int) {
+			defer wg.Done()
+			for i := 0; i < nLogs/4; i++ {
+				var line string
+				switch i % 8 {
+				case 0:
+					line = fmt.Sprintf("worker %d: plain log line %d\n", w, i)
+				case 1:
+					line = fmt.Sprintf("{\"looks\":\"like json\",\"seq\":%d,\"success\":true}\n", i) // D-3 trap
+				case 2:
+					line = SENTINEL + "this is not valid json {{{\n" // sentinel-lookalike
+				case 3:
+					line = fmt.Sprintf("mid-line %s{\"t\":\"result\"} embedded sentinel %d\n", SENTINEL, i)
+				case 4:
+					line = strings.Repeat("A", 1024*1024) + "\n" // 1MB line
+				case 5:
+					line = string([]byte{0xff, 0xfe, 0x80, 0x81}) + " binary-ish\n" // invalid UTF-8
+				case 6:
+					line = "\r\n" // CRLF/empty
+				default:
+					line = fmt.Sprintf("{\"t\":\"progress\",\"seq\":%d}\n", i) // frame-shaped, NO sentinel
+				}
+				if i%2 == 0 {
+					mu.Lock()
+					os.Stdout.WriteString(line)
+					mu.Unlock()
+				} else {
+					os.Stdout.WriteString(line) // unsynchronized: big lines may shred
+				}
+			}
+		}(w)
+	}
+	wg.Wait()
+	// terminal result frame
+	res := map[string]any{"v": 1, "t": "result", "seq": nFrames, "outcome": "ok", "framesEmitted": nFrames}
+	b, _ := json.Marshal(res)
+	os.Stdout.WriteString(SENTINEL + string(b) + "\n")
+}

.llm/runs/claude-harness-profile-rfc-benchmark-shzhgv--polyglot-protocol-rfc/bench/spikes/k1/emit_frames.py

@@ -0,0 +1,60 @@
+# K1 emitter (python3) — same adversarial contract as the Go emitter, via threads.
+# Frames: single os.write() <= 4096 bytes (PIPE_BUF atomicity). Logs: hostile.
+import hashlib
+import json
+import os
+import sys
+import threading
+
+SENTINEL = b"\x00NSF\x00"
+N_LOGS = 10000
+N_FRAMES = 200
+mu = threading.Lock()
+fd = sys.stdout.fileno()
+
+
+def frames() -> None:
+    for i in range(N_FRAMES):
+        payload = "x" * (64 * (i % 16))
+        sha = hashlib.sha256(payload.encode()).hexdigest()[:16]
+        frame = {"v": 1, "t": "progress", "seq": i, "payload": payload, "sha": sha}
+        line = SENTINEL + json.dumps(frame, separators=(",", ":")).encode() + b"\n"
+        assert len(line) <= 4096, "frame exceeds PIPE_BUF budget"
+        os.write(fd, line)  # one write syscall
+
+
+def logs(w: int) -> None:
+    for i in range(N_LOGS // 4):
+        k = i % 8
+        if k == 0:
+            line = f"worker {w}: plain log line {i}\n".encode()
+        elif k == 1:
+            line = json.dumps({"looks": "like json", "seq": i, "success": True}).encode() + b"\n"
+        elif k == 2:
+            line = SENTINEL + b"this is not valid json {{{\n"
+        elif k == 3:
+            line = b"mid-line " + SENTINEL + b'{"t":"result"} embedded %d\n' % i
+        elif k == 4:
+            line = b"A" * (1024 * 1024) + b"\n"
+        elif k == 5:
+            line = bytes([0xFF, 0xFE, 0x80, 0x81]) + b" binary-ish\n"
+        elif k == 6:
+            line = b"\r\n"
+        else:
+            line = json.dumps({"t": "progress", "seq": i}).encode() + b"\n"
+        if i % 2 == 0:
+            with mu:
+                os.write(fd, line)
+        else:
+            os.write(fd, line)
+
+
+threads = [threading.Thread(target=frames)] + [
+    threading.Thread(target=logs, args=(w,)) for w in range(4)
+]
+for t in threads:
+    t.start()
+for t in threads:
+    t.join()
+res = {"v": 1, "t": "result", "seq": N_FRAMES, "outcome": "ok", "framesEmitted": N_FRAMES}
+os.write(fd, SENTINEL + json.dumps(res, separators=(",", ":")).encode() + b"\n")

.llm/runs/claude-harness-profile-rfc-benchmark-shzhgv--polyglot-protocol-rfc/bench/spikes/k1/run-k1.ts

@@ -0,0 +1,174 @@
+/**
+ * K1 — sentinel-NDJSON stdout frame demux under adversarial logs (plan L8/K1).
+ *
+ * Demux algorithm v2 (the spec-relevant reference): STREAMING SENTINEL-SCAN, not line-anchored.
+ * Rationale (measured in demux-v1, kept in results as k1-v1 rows): a frame write is atomic
+ * (single write <= PIPE_BUF), but it may land INSIDE another writer's not-yet-terminated
+ * >PIPE_BUF log line — line-anchored parsing then sees the sentinel mid-line and drops the
+ * frame (python3 lost 8-44/200 frames per rep). Sentinel-scan recovers frames wherever they
+ * are embedded; surrounding log bytes stay log content.
+ *
+ * Criteria: 200 progress frames + 1 result frame per emitter, exactly once, arrival-ordered,
+ * sha-verified; frame-shaped log lines (no sentinel) never hijacked; sentinel+invalid-JSON
+ * counted as malformed-diagnostic log, never a frame; no crash; throughput recorded.
+ *
+ *   deno run --allow-read --allow-write --allow-run --allow-env bench/spikes/k1/run-k1.ts
+ */
+
+const RUN_DIR = new URL('../../..', import.meta.url).pathname;
+const K1 = `${RUN_DIR}bench/spikes/k1`;
+const SENTINEL = new Uint8Array([0x00, 0x4e, 0x53, 0x46, 0x00]); // \x00NSF\x00
+const FRAME_MAX = 4200;
+const REPS = 5;
+
+type Row = Record<string, unknown>;
+const out: Row[] = [];
+const emit = (o: Row) => out.push(o);
+emit({ kind: 'meta', bench: 'k1-frame-transport', demux: 'sentinel-scan-v2', startedAt: new Date().toISOString(), reps: REPS });
+
+class Demux {
+  frames: Array<Record<string, unknown>> = [];
+  logLines = 0;
+  malformedSentinel = 0;
+  frameShapedLogs = 0;
+  private mode: 'log' | 'frame' = 'log';
+  private sentMatch = 0; // sentinel bytes matched so far (log mode)
+  private lineHead: number[] = []; // first 64 bytes of current log line
+  private lineHasContent = false;
+  private frameBuf: number[] = [];
+  private dec = new TextDecoder('utf-8', { fatal: false });
+
+  private logByte(b: number) {
+    if (b === 0x0a) {
+      if (this.lineHasContent || this.lineHead.length > 0 || true) this.endLogLine();
+      return;
+    }
+    this.lineHasContent = true;
+    if (this.lineHead.length < 64) this.lineHead.push(b);
+  }
+
+  private endLogLine() {
+    const head = this.dec.decode(new Uint8Array(this.lineHead));
+    if (head.startsWith('{"t":')) this.frameShapedLogs++;
+    this.logLines++;
+    this.lineHead = [];
+    this.lineHasContent = false;
+  }
+
+  push(chunk: Uint8Array) {
+    for (let i = 0; i < chunk.length; i++) {
+      const b = chunk[i];
+      if (this.mode === 'log') {
+        if (b === SENTINEL[this.sentMatch]) {
+          this.sentMatch++;
+          if (this.sentMatch === SENTINEL.length) {
+            this.sentMatch = 0;
+            this.mode = 'frame';
+            this.frameBuf = [];
+          }
+        } else {
+          // flush withheld partial-sentinel bytes into the log line, then this byte
+          for (let k = 0; k < this.sentMatch; k++) this.logByte(SENTINEL[k]);
+          this.sentMatch = 0;
+          if (b === SENTINEL[0]) { this.sentMatch = 1; } else this.logByte(b);
+        }
+      } else {
+        if (b === 0x0a) {
+          const body = this.dec.decode(new Uint8Array(this.frameBuf));
+          let ok = false;
+          try {
+            const obj = JSON.parse(body);
+            if (obj && typeof obj === 'object' && !Array.isArray(obj) && typeof obj.t === 'string') {
+              this.frames.push(obj);
+              ok = true;
+            }
+          } catch { /* fallthrough */ }
+          if (!ok) {
+            this.malformedSentinel++;
+            for (const fb of this.frameBuf) this.logByte(fb);
+            this.logByte(0x0a);
+          }
+          this.mode = 'log';
+          this.frameBuf = [];
+        } else {
+          this.frameBuf.push(b);
+          if (this.frameBuf.length > FRAME_MAX) {
+            this.malformedSentinel++;
+            for (const fb of this.frameBuf) this.logByte(fb);
+            this.mode = 'log';
+            this.frameBuf = [];
+          }
+        }
+      }
+    }
+  }
+
+  finish() {
+    if (this.mode === 'frame') { this.malformedSentinel++; this.mode = 'log'; }
+    if (this.lineHasContent) this.endLogLine();
+  }
+}
+
+async function run(cmd: string[], label: string, rep: number) {
+  const t0 = performance.now();
+  const proc = new Deno.Command(cmd[0], {
+    args: cmd.slice(1),
+    stdout: 'piped',
+    stderr: 'null',
+  }).spawn();
+  const d = new Demux();
+  let bytes = 0;
+  const reader = proc.stdout.getReader();
+  while (true) {
+    const { done, value } = await reader.read();
+    if (done) break;
+    bytes += value.length;
+    d.push(value);
+  }
+  d.finish();
+  const status = await proc.status;
+  const wallMs = performance.now() - t0;
+
+  const progress = d.frames.filter((f) => f.t === 'progress');
+  const results = d.frames.filter((f) => f.t === 'result');
+  const seqs = progress.map((f) => f.seq as number).sort((a, b) => a - b);
+  const seqExact = seqs.length === 200 && seqs.every((v, i) => v === i);
+  const arrival = progress.map((f) => f.seq as number);
+  const inOrder = arrival.every((v, i) => i === 0 || v > arrival[i - 1]);
+  let shaOk = true;
+  for (const f of progress) {
+    const digest = await crypto.subtle.digest('SHA-256', new TextEncoder().encode(f.payload as string));
+    const hex = [...new Uint8Array(digest)].map((b) => b.toString(16).padStart(2, '0')).join('');
+    if (!hex.startsWith(f.sha as string)) { shaOk = false; break; }
+  }
+  const pass = seqExact && inOrder && shaOk && results.length === 1;
+  emit({
+    kind: 'demux', label, rep, pass, wallMs: +wallMs.toFixed(1), exitCode: status.code,
+    mbProcessed: +(bytes / 1048576).toFixed(1), framesRecovered: progress.length,
+    resultFrames: results.length, seqExact, inOrder, shaOk, logLines: d.logLines,
+    malformedSentinelAsLog: d.malformedSentinel, frameShapedLogsNotHijacked: d.frameShapedLogs,
+    throughputMBs: +(bytes / 1048576 / (wallMs / 1000)).toFixed(1),
+  });
+  return pass;
+}
+
+await new Deno.Command('go', { args: ['build', '-o', `${K1}/bin/emitter-go`, `${K1}/emit-frames.go`], env: { ...Deno.env.toObject() }, cwd: K1 }).output();
+
+let allPass = true;
+for (let rep = 0; rep < REPS; rep++) {
+  allPass = (await run([`${K1}/bin/emitter-go`], 'go', rep)) && allPass;
+  allPass = (await run(['python3', `${K1}/emit_frames.py`], 'python3', rep)) && allPass;
+}
+
+emit({
+  kind: 'fd3-feasibility',
+  extraFdSupport: false,
+  note: 'Deno.Command exposes only stdin/stdout/stderr; no API passes additional inherited fds. The pre-registered fd-3 fallback branch is unavailable on the Deno host; socket transports (K3) are the alternative channel.',
+});
+emit({
+  kind: 'v1-lesson',
+  note: 'Line-anchored demux (v1) lost 8-44/200 python3 frames per rep: atomic frames embedded inside another writer\'s unterminated >PIPE_BUF log line. Spec rule derived: demux MUST sentinel-scan the byte stream, not split lines first; frame span (sentinel..newline) integrity is guaranteed by single-write <= PIPE_BUF atomicity.',
+});
+emit({ kind: 'summary', allPass, finishedAt: new Date().toISOString() });
+await Deno.writeTextFile(`${RUN_DIR}results/raw/k1.jsonl`, out.map((o) => JSON.stringify(o)).join('\n') + '\n');
+console.log(`K1 ${allPass ? 'PASS' : 'FAIL'} -> results/raw/k1.jsonl`);