mirror of
https://github.com/mvschwarz/openrig-studio.git
synced 2026-11-07 05:56:21 -10:00
Drive ops are addressed to a surface and acknowledged on apply
review.r2 CRITICAL + PM plan-lock (qitem e2ba1531 / audit C2): one global /api/drive slot delivered every op to every kept-alive surface — including the play/export/delete side effects the contract itself names — because an op could not say which surface it was for. PM ruled per-surface DELIVERY (structural, not global-channel-plus-filter) plus a settle-ack, as one revision. Plan-lock GO on all three questions; drive.md redline rides this candidate for PM review. PROPERTY 1 — per-surface delivery, structural. Each surface has its own slot, keyed by the id it supplies about itself; a surface polls GET /api/drive?surface= <its own id> and reads only its own slot. A mis-addressed op lands in a slot the wrong surface never polls, so it is undeliverable there — nothing to filter, no adopter responsible for filtering correctly. Fail-first: the old global model delivers A's op to kept-alive B and C; the new control (drive A, keep B/C alive, assert delete/export/play never fire on B/C) catches exactly that. BREAKING (Q3, PM-endorsed): an unaddressed POST or GET is refused 400 naming the required surface field. A warned unaddressed default still fans out, so the default that caused the defect is removed, not softened. Every in-repo consumer (drive.test harness, the per-route positive control) migrated in this commit; listed BREAKING in contract-meta with a changelog entry. PROPERTY 2 — settle-ack (Q2, advisory/exposed per PM). POST returns on RECORD (recorded: true) — record and apply kept explicitly apart, a near-synonym pair. The surface acks the highest generation it has APPLIED via ?applied=; the runtime exposes appliedGen, which advances only when apply() has RESOLVED (the harder half, named for applied not received). An agent that sequences awaits appliedGen >= its posted gen; one that does not stays fast. The gap is observable: a control blocks apply() and asserts appliedGen < postedGen during the window, catch-up after. driveSurface exposes applied (resolved) and accepted (offered) separately. drive.md redline marks every changed (stable) section for PM; contract-meta example gains drive.surfaces; the per-route parity guard caught that omission. 359 -> 362 tests, 362/362 green.
This commit is contained in:
1 parent
37c20d9ded
commit
c70a78d97f
6 files changed
+280
-63
No files matched your search
+62
-33
@@ -821,36 +821,51 @@ function persistAnnotations() {
|
||||
fs.writeFileSync(ANNOTATIONS_FILE, JSON.stringify({ scopes }, null, 2));
|
||||
}
|
||||
|
||||
let driveOp = null;
|
||||
let driveSeq = 0;
|
||||
// HOW MANY TIMES A SURFACE HAS ACTUALLY POLLED. Not a declaration — an
|
||||
// observation. A runtime that has served ZERO drive reads since boot has nobody
|
||||
// listening, and it is the only party that can know that.
|
||||
// PER-SURFACE DELIVERY (review.r2 CRITICAL, PM ruling): one global op slot fanned
|
||||
// every op out to every kept-alive surface — including play/export/delete side
|
||||
// effects — because "which surface is open" was not expressible. Now each surface
|
||||
// has its OWN slot, keyed by the surface id, and a surface only ever reads its own
|
||||
// slot: a mis-addressed op lands in a slot the wrong surface never polls, so it is
|
||||
// STRUCTURALLY undeliverable there — not delivered-then-filtered (which would make
|
||||
// every adopter responsible for filtering correctly).
|
||||
//
|
||||
// The defect this exists for shipped and was found on a real box: an agent POSTed
|
||||
// an op to a hand-built surface that had never adopted the helper, got
|
||||
// { ok: true, gen: 2 }, and nothing happened. The agent did everything right. A
|
||||
// runtime is not allowed to report success for a delivery it cannot make.
|
||||
let driveReads = 0;
|
||||
const driveMarker = () => `${BOOT_ID}.${driveSeq}`;
|
||||
// `gen` stays global and strictly increasing across all surfaces; each slot holds
|
||||
// the latest op addressed to IT plus that surface's applied progress.
|
||||
const driveSlots = new Map(); // surfaceId -> { op, appliedGen, reads }
|
||||
let driveSeq = 0;
|
||||
let driveReads = 0; // total polls across all surfaces (contract report)
|
||||
const driveSlot = (surface) => {
|
||||
let slot = driveSlots.get(surface);
|
||||
if (!slot) { slot = { op: null, appliedGen: 0, reads: 0 }; driveSlots.set(surface, slot); }
|
||||
return slot;
|
||||
};
|
||||
// Per-surface marker: a surface watches its OWN slot's latest generation.
|
||||
const driveMarker = (slot) => `${BOOT_ID}.${slot.op ? slot.op.gen : 0}`;
|
||||
|
||||
function writeDrive(op) {
|
||||
// POST RETURNS ON RECORD, NOT APPLY — record and apply are a near-synonym pair
|
||||
// the contract keeps explicitly apart. writeDrive records the op in the target
|
||||
// surface's slot and returns `recorded: true`. Whether the surface has APPLIED it
|
||||
// is a separate question answered by `appliedGen` on GET, which advances only when
|
||||
// the surface acks an apply that RESOLVED (the harder half — applied, not seen).
|
||||
function writeDrive(surface, op) {
|
||||
// ADDRESSING IS MANDATORY (BREAKING, PM ruling): an unaddressed op is refused
|
||||
// rather than fanned out. A warned unaddressed default still fans out, so the
|
||||
// default that caused the defect is removed, not softened.
|
||||
if (typeof surface !== "string" || !surface.trim()) {
|
||||
return { ok: false, error: "a drive op must name its target surface — send { surface, ...intent }; " +
|
||||
"an unaddressed op would reach every kept-alive surface, which is the fan-out this refuses" };
|
||||
}
|
||||
if (!op || typeof op !== "object" || Array.isArray(op)) {
|
||||
return { ok: false, error: "a drive op must be a JSON object carrying intent for the surface" };
|
||||
}
|
||||
driveSeq += 1;
|
||||
driveOp = { ...op, gen: driveSeq, at: new Date().toISOString() };
|
||||
// ACCEPTED IS NOT DELIVERED, AND THE CALLER FINDS OUT HERE RATHER THAN LATER.
|
||||
// `listening` is false when nothing has ever polled this verb: the op is
|
||||
// recorded and no surface will act on it. Reported at the moment of the mistake,
|
||||
// in the response the caller is already reading — a doc line only fires once
|
||||
// somebody goes looking, which is after they have concluded the platform is
|
||||
// broken.
|
||||
//
|
||||
// STILL NOT CLAIMED, and drive.md says so: listening does not mean HONOURED. A
|
||||
// surface can poll, receive, and do nothing. That is the refusal shape, and it
|
||||
// stays unspecified until a real application needs one.
|
||||
return { ok: true, marker: driveMarker(), op: driveOp, listening: driveReads > 0 };
|
||||
const slot = driveSlot(surface);
|
||||
slot.op = { ...op, gen: driveSeq, at: new Date().toISOString() };
|
||||
// ACCEPTED IS NOT DELIVERED, AND THE CALLER FINDS OUT HERE. `listening` is
|
||||
// false when THIS surface has never polled: the op is recorded and that surface
|
||||
// will not act on it. Reported in the response the caller is already reading.
|
||||
return { ok: true, recorded: true, surface, marker: driveMarker(slot), op: slot.op,
|
||||
gen: slot.op.gen, appliedGen: slot.appliedGen, listening: slot.reads > 0 };
|
||||
}
|
||||
|
||||
const readBody = (req) => new Promise((resolve) => {
|
||||
@@ -1091,19 +1106,33 @@ const API_ROUTES = [
|
||||
{ verb: "/api/drive", handler: async (request, req, res) => {
|
||||
if (req.method === "POST") {
|
||||
const raw = await readBody(req);
|
||||
let op;
|
||||
try { op = JSON.parse(raw || "null"); }
|
||||
let body;
|
||||
try { body = JSON.parse(raw || "null"); }
|
||||
catch (e) { return sendJson(res, 400, { ok: false, error: `drive op is not valid JSON: ${e.message}` }); }
|
||||
const result = writeDrive(op);
|
||||
// `surface` is the ADDRESS envelope, reserved like `gen`/`at`; the rest is
|
||||
// opaque intent for that surface.
|
||||
const { surface, ...op } = (body && typeof body === "object" && !Array.isArray(body)) ? body : { surface: null };
|
||||
const result = writeDrive(surface, op);
|
||||
return sendJson(res, result.ok ? 200 : 400, result);
|
||||
}
|
||||
// The surface polls this. Same `?since=` shape as the change signal and
|
||||
// focus, so a surface that already watches one is watching this with the
|
||||
// primitive it has rather than a third mechanism of its own.
|
||||
// The surface polls its OWN slot, addressed by ?surface=<id>. Same `?since=`
|
||||
// shape as the change signal and focus. `?applied=<gen>` is the settle-ack:
|
||||
// the surface reports the highest generation it has APPLIED, and the runtime
|
||||
// exposes it as `appliedGen` so a poster can await appliedGen >= its own gen
|
||||
// before the next op. POST returns on RECORD; this is how apply is observed.
|
||||
const surface = request.searchParams.get("surface");
|
||||
if (typeof surface !== "string" || !surface.trim()) {
|
||||
return sendJson(res, 400, { ok: false, error: "GET /api/drive must name ?surface=<id> — ops are per-surface" });
|
||||
}
|
||||
const slot = driveSlot(surface);
|
||||
slot.reads += 1;
|
||||
driveReads += 1;
|
||||
const applied = Number(request.searchParams.get("applied"));
|
||||
if (Number.isInteger(applied) && applied > slot.appliedGen) slot.appliedGen = applied;
|
||||
const since = request.searchParams.get("since");
|
||||
const marker = driveMarker();
|
||||
return sendJson(res, 200, { ok: true, changed: since === null || since !== marker, marker, op: driveOp });
|
||||
const marker = driveMarker(slot);
|
||||
return sendJson(res, 200, { ok: true, surface, changed: since === null || since !== marker,
|
||||
marker, op: slot.op, appliedGen: slot.appliedGen });
|
||||
} },
|
||||
{ verb: "/api/capture-target", handler: async (request, req, res) => {
|
||||
if (req.method === "POST") {
|
||||
@@ -1167,7 +1196,7 @@ const API_ROUTES = [
|
||||
capabilities: CAPABILITIES,
|
||||
// Inspectable without posting: an agent can ask whether anything is
|
||||
// listening before it drives, rather than learning from a no-op.
|
||||
drive: { listening: driveReads > 0, reads: driveReads, ops: driveSeq },
|
||||
drive: { listening: driveReads > 0, reads: driveReads, ops: driveSeq, surfaces: driveSlots.size },
|
||||
// WHICH KIND OF `by` THIS RUNTIME PRODUCES. focus.md tells a consumer not
|
||||
// to build a trust decision on `by` without knowing which kind of runtime
|
||||
// made it — and until now there was no way to ask. The header NAME is
|
||||
|
||||
+25
-4
@@ -392,6 +392,7 @@ export function deferWhileDirty(isDirty, apply) {
|
||||
// are pure view state and which genuinely want a fresh page to catch up.
|
||||
export function driveSurface({
|
||||
apply,
|
||||
surface,
|
||||
fetchImpl = fetch,
|
||||
endpoint = "/api/drive",
|
||||
intervalMs = 350,
|
||||
@@ -399,9 +400,17 @@ export function driveSurface({
|
||||
onError = () => {},
|
||||
} = {}) {
|
||||
if (typeof apply !== "function") throw new TypeError("driveSurface needs an apply(op) function");
|
||||
// PER-SURFACE DELIVERY: a surface receives ops from its OWN slot, addressed by
|
||||
// the id it supplies about itself. A backgrounded surface polling its own slot
|
||||
// is exactly why a foreground op never reaches it — there is no unaddressed
|
||||
// channel left to poll.
|
||||
if (typeof surface !== "string" || !surface.trim()) {
|
||||
throw new TypeError("driveSurface needs the `surface` id it receives ops for — ops are per-surface");
|
||||
}
|
||||
|
||||
let marker = null; // last marker seen, so `?since=` can be honest
|
||||
let seen = 0; // highest generation already applied
|
||||
let seen = 0; // highest generation ACCEPTED (dedup), not yet proven applied
|
||||
let appliedGen = 0; // highest generation whose apply() RESOLVED — the ack we report
|
||||
let next = null; // newest op that arrived while one was being applied
|
||||
let baselined = false; // has the first poll established what counts as "already there"?
|
||||
let flying = false;
|
||||
@@ -415,6 +424,10 @@ export function driveSurface({
|
||||
// Re-read `next` AFTER the await, not before: anything that arrived
|
||||
// during this application supersedes what we were about to do next.
|
||||
await apply(op);
|
||||
// APPLIED, not merely accepted: the ack advances only once apply() has
|
||||
// RESOLVED, so appliedGen names the harder half. A poster reading
|
||||
// appliedGen < its gen is seeing a real posted-but-not-yet-applied window.
|
||||
appliedGen = op.gen;
|
||||
op = next;
|
||||
next = null;
|
||||
}
|
||||
@@ -436,8 +449,11 @@ export function driveSurface({
|
||||
async function poll() {
|
||||
if (stopped) return;
|
||||
try {
|
||||
const url = marker === null ? endpoint : `${endpoint}?since=${encodeURIComponent(marker)}`;
|
||||
const r = await (await fetchImpl(url, { cache: "no-store" })).json();
|
||||
// Address our own slot, and ack the highest generation we have APPLIED so
|
||||
// the runtime can expose appliedGen to a poster awaiting settle.
|
||||
const params = new URLSearchParams({ surface, applied: String(appliedGen) });
|
||||
if (marker !== null) params.set("since", marker);
|
||||
const r = await (await fetchImpl(`${endpoint}?${params}`, { cache: "no-store" })).json();
|
||||
if (r && r.ok) {
|
||||
marker = r.marker ?? marker;
|
||||
// The baseline is taken from the FIRST answer, before anything is applied.
|
||||
@@ -460,7 +476,12 @@ export function driveSurface({
|
||||
return {
|
||||
offer,
|
||||
stop() { stopped = true; if (timer) clearTimeout(timer); },
|
||||
get applied() { return seen; },
|
||||
// `applied` is the SETTLE-ACK value: the highest generation whose apply()
|
||||
// resolved. `accepted` is the dedup high-water (highest offered/baselined) —
|
||||
// it moves ahead of `applied` while an application is in flight, which is the
|
||||
// observable posted-but-not-yet-applied window.
|
||||
get applied() { return appliedGen; },
|
||||
get accepted() { return seen; },
|
||||
};
|
||||
}
|
||||
|
||||
|
||||
@@ -55,7 +55,7 @@ Two schemas sit beside them: `surface-row.schema.json` for a manifest row, and
|
||||
"capabilities": ["contract.meta", "observe.factory-state", "stream.events", "files.read",
|
||||
"shell.protocol", "focus.read", "focus.write", "drive.read", "drive.write",
|
||||
"annotations.read", "annotations.write", "capture.target"],
|
||||
"drive": { "listening": false, "reads": 0, "ops": 0 },
|
||||
"drive": { "listening": false, "reads": 0, "ops": 0, "surfaces": 0 },
|
||||
"focus": { "attribution": "caller-declared", "identityHeader": null },
|
||||
"annotations": { "persistence": "memory", "scopes": 0, "writes": 0 },
|
||||
"capture": { "target": null, "roots": ["feedback"], "consumedBy": null },
|
||||
@@ -179,6 +179,16 @@ Every route, field, and message type in this contract carries one of:
|
||||
Additive fields and new routes are non-breaking. Breaking changes move
|
||||
`contractVersion` and are named in a changelog entry.
|
||||
|
||||
### Changelog
|
||||
|
||||
- **2026-08-08 — drive per-surface delivery (BREAKING).** `POST /api/drive` now
|
||||
requires a `surface` field and refuses an unaddressed op with `400`; `GET
|
||||
/api/drive` requires `?surface=<id>`. Previously one global channel delivered
|
||||
every op to every polling surface, including `play`/`export`/`delete` side
|
||||
effects on backgrounded-but-alive surfaces. Additive in the same revision (not
|
||||
breaking): the `surface` and `appliedGen` response fields, and the settle-ack
|
||||
`?applied=` poll parameter. See `drive.md` (redline pending PM review).
|
||||
|
||||
## How a surface should decide compatibility
|
||||
|
||||
Check two things at boot, in this order:
|
||||
|
||||
+66
-17
@@ -18,20 +18,35 @@ application built on this SDK, and it is the reason that application can be
|
||||
watched rather than merely instructed. What was missing is the SDK offering it, so
|
||||
that two applications do not invent two incompatible versions of the same thing.
|
||||
|
||||
## The op (stable)
|
||||
> **⚠️ REDLINE — REVISION 2026-08-08 (per-surface delivery + settle-ack). PM
|
||||
> review pending.** Sections marked **(stable — REVISED)** below change a
|
||||
> previously-stable contract, and one change is **BREAKING** (an unaddressed
|
||||
> `POST` is now refused). Every changed section is marked so PM reviews the diff,
|
||||
> not the memory of it. Reason: one global channel delivered every op to every
|
||||
> kept-alive surface — including `play`/`export`/`delete` — because the op could
|
||||
> not name which surface it was for. The revision makes delivery per-surface
|
||||
> (structurally undeliverable to the wrong surface) and adds a settle-ack.
|
||||
|
||||
An op carries **intent**, and the runtime attaches two fields:
|
||||
## The op (stable — REVISED: `surface` address added)
|
||||
|
||||
An op carries **intent**, and the runtime attaches two fields. The **agent**
|
||||
supplies one required envelope field, `surface`, naming which surface the op is
|
||||
for:
|
||||
|
||||
```json
|
||||
{ "gen": 7, "at": "<ISO timestamp>", "...": "whatever this surface's intent IS" }
|
||||
{ "surface": "<surface id>", "gen": 7, "at": "<ISO timestamp>", "...": "whatever this surface's intent IS" }
|
||||
```
|
||||
|
||||
| field | stability | meaning |
|
||||
|---|---|---|
|
||||
| `gen` | stable | server-assigned generation, strictly increasing |
|
||||
| `surface` | **stable — NEW** | the target surface id, supplied by the agent; **required** |
|
||||
| `gen` | stable | server-assigned generation, strictly increasing across all surfaces |
|
||||
| `at` | stable | ISO timestamp the runtime recorded the op |
|
||||
| everything else | **the surface's** | intent, interpreted by the surface alone |
|
||||
|
||||
`surface` is a reserved envelope field like `gen`/`at`: it addresses the op and
|
||||
is not part of the opaque intent.
|
||||
|
||||
### The op is opaque to the runtime, and is never a list of DOM operations (stable)
|
||||
|
||||
The runtime never interprets an op. Two consequences, both deliberate.
|
||||
@@ -46,22 +61,48 @@ and would force every drivable surface to freeze its DOM as an interface it neve
|
||||
agreed to publish. This is the same decision, for the same reason, as `selection`
|
||||
being typed by its surface in `focus.md`.
|
||||
|
||||
## The channel (stable)
|
||||
## The channel (stable — REVISED: per-surface addressing + settle-ack)
|
||||
|
||||
**`POST /api/drive`** — an agent posts an op. The runtime assigns `gen` and
|
||||
answers `{ ok, marker, op }`.
|
||||
**`POST /api/drive`** — an agent posts an op **addressed to a surface**:
|
||||
`{ surface, ...intent }`. The runtime records it in that surface's slot, assigns
|
||||
`gen`, and **returns on RECORD** — the response is
|
||||
`{ ok, recorded: true, surface, gen, marker, op, appliedGen, listening }`.
|
||||
|
||||
**`GET /api/drive`** — the surface polls. It answers the change-signal shape:
|
||||
**Record and apply are kept explicitly apart.** `recorded: true` means the op is
|
||||
stored in the surface's slot. It does **not** mean the surface has applied it —
|
||||
that is `appliedGen` (below). A `POST` never blocks on a surface applying;
|
||||
coupling the agent to surface liveness would let a dead or backgrounded-then-killed
|
||||
surface hang every driver.
|
||||
|
||||
**⚠️ BREAKING — an unaddressed `POST` is refused with `400`.** Previously an op
|
||||
had no `surface` and was delivered to whoever polled; that is the fan-out this
|
||||
revision removes. A warned unaddressed default would still fan out, so the default
|
||||
is removed, not softened. The refusal names the required `surface` field.
|
||||
|
||||
**`GET /api/drive?surface=<id>`** — the surface polls **its own slot**. It answers
|
||||
the change-signal shape plus the settle-ack:
|
||||
|
||||
```json
|
||||
{ "ok": true, "changed": true, "marker": "<opaque>", "op": { "...": "the latest op" } }
|
||||
{ "ok": true, "surface": "<id>", "changed": true, "marker": "<opaque>",
|
||||
"op": { "...": "the latest op for THIS surface" }, "appliedGen": 6 }
|
||||
```
|
||||
|
||||
`?since=<marker>` is honoured exactly as `change-signal.md` describes. A surface
|
||||
already watching the change signal or focus is watching this **with the primitive
|
||||
it already has**, rather than a third polling mechanism of its own.
|
||||
A `GET` without `?surface=` is refused with `400`: ops are per-surface, and an
|
||||
unaddressed poll could hand a surface another surface's ops.
|
||||
|
||||
A runtime nobody has driven answers `op: null`. That is not an error.
|
||||
**Settle-ack.** A surface reports the highest generation it has **applied** via
|
||||
`?applied=<gen>` on its poll; the runtime exposes it as `appliedGen`. It advances
|
||||
only when the surface's application has **resolved** — `appliedGen` names the
|
||||
harder half (applied, not received). An agent that needs to sequence ops awaits
|
||||
`appliedGen >= <the gen its POST returned>` before the next op; one that does not
|
||||
care stays fast. A posted-but-not-yet-applied op is observable as
|
||||
`appliedGen < gen`.
|
||||
|
||||
`?since=<marker>` is honoured exactly as `change-signal.md` describes, per surface.
|
||||
A surface already watching the change signal or focus is watching this **with the
|
||||
primitive it already has**.
|
||||
|
||||
A surface nobody has driven answers `op: null`. That is not an error.
|
||||
|
||||
### Latest intent wins; superseded ops are DROPPED (stable)
|
||||
|
||||
@@ -108,12 +149,13 @@ overlapping applications interleave their writes and leave the surface in a stat
|
||||
**neither op described**. An op arriving mid-flight is held, and when the flight
|
||||
lands only the newest held op runs — so serialising never reintroduces a queue.
|
||||
|
||||
## Adopting it (stable)
|
||||
## Adopting it (stable — REVISED: `surface` required)
|
||||
|
||||
```js
|
||||
import { driveSurface } from "/signal.js";
|
||||
|
||||
const drive = driveSurface({
|
||||
surface: mySurfaceId, // which surface this receives ops for — required
|
||||
apply: async (op) => {
|
||||
if (op.show !== undefined) await showTake(op.show);
|
||||
if (op.say) narrate(op.say); // let the agent say WHY, where the surface talks
|
||||
@@ -121,9 +163,13 @@ const drive = driveSurface({
|
||||
});
|
||||
```
|
||||
|
||||
`apply` is awaited, so returning a promise is how a surface tells the helper that
|
||||
it is still working. `offer(op)` hands an op in directly for a surface driven by
|
||||
something other than the verb, and `stop()` ends the polling.
|
||||
`surface` is the id this surface receives ops for; the helper polls that slot and
|
||||
never sees another surface's ops. `apply` is awaited, so returning a promise is how
|
||||
a surface tells the helper that it is still working — and the settle-ack only
|
||||
advances once that promise resolves. `offer(op)` hands an op in directly for a
|
||||
surface driven by something other than the verb; `applied` reads the highest
|
||||
generation whose `apply` has resolved, `accepted` the highest offered; `stop()`
|
||||
ends the polling.
|
||||
|
||||
**Narrate the change.** An op may carry a line of text the surface shows where it
|
||||
normally talks to the user. A watcher then sees a narrated change rather than
|
||||
@@ -145,6 +191,9 @@ feels driven and one that feels haunted.
|
||||
| specified | shipped |
|
||||
|---|---|
|
||||
| `POST` / `GET /api/drive`, opaque ops, server-assigned `gen` | **yes**, in the reference runtime |
|
||||
| **per-surface delivery: `{ surface, ...intent }` on `POST`, `?surface=` on `GET`; an op reaches only its addressed surface** | **yes** (REVISED), with a control that drives A while B and C are kept alive and asserts `delete`/`export`/`play` never fire on B or C |
|
||||
| **an unaddressed `POST` or `GET` refused `400`** (BREAKING) | **yes** (REVISED), with a control |
|
||||
| **settle-ack: `POST` returns on RECORD; `appliedGen` advances only when `apply` resolves; the record-to-apply gap is observable** | **yes** (REVISED), with a control asserting `appliedGen < gen` during a blocked application and catch-up after |
|
||||
| the namespace advertised at `GET /api/contract` -> `capabilities` (`drive.read`, `drive.write`) | **yes** |
|
||||
| the change-signal `?since=` shape on the read | **yes** |
|
||||
| `driveSurface` dropping superseded ops | **yes**, with a control that plants a queue and fails |
|
||||
|
||||
+115
-7
@@ -43,12 +43,22 @@ async function start() {
|
||||
await new Promise((r) => setTimeout(r, 100));
|
||||
}
|
||||
}
|
||||
// Ops are per-surface now: POST carries { surface, ...intent }, GET addresses
|
||||
// ?surface=<id>. The single-surface tests use a fixed default id "s"; the
|
||||
// per-surface control drives distinct ids.
|
||||
const post = (body) => fetch(`${base}/api/drive`,
|
||||
{ method: "POST", headers: { "content-type": "application/json" }, body: JSON.stringify(body) }).then((r) => r.json());
|
||||
const get = (surface, params = {}) => {
|
||||
const q = new URLSearchParams({ surface, ...params });
|
||||
return fetch(`${base}/api/drive?${q}`, { cache: "no-store" }).then((r) => r.json());
|
||||
};
|
||||
return {
|
||||
dir, base,
|
||||
drive: async (op) => (await fetch(`${base}/api/drive`,
|
||||
{ method: "POST", headers: { "content-type": "application/json" }, body: JSON.stringify(op) })).json(),
|
||||
read: async (since) => (await fetch(`${base}/api/drive${since === undefined ? "" : `?since=${encodeURIComponent(since)}`}`,
|
||||
{ cache: "no-store" })).json(),
|
||||
drive: async (op, surface = "s") => post({ surface, ...op }),
|
||||
driveTo: async (surface, op) => post({ surface, ...op }),
|
||||
postRaw: async (body) => post(body),
|
||||
read: async (since, surface = "s") => get(surface, since === undefined ? {} : { since }),
|
||||
readAt: async (surface, params = {}) => get(surface, params),
|
||||
stop: () => proc.kill(),
|
||||
};
|
||||
}
|
||||
@@ -113,9 +123,11 @@ test("a malformed op is refused by name rather than silently accepted", async (t
|
||||
const s = await start();
|
||||
t.after(() => s.stop());
|
||||
|
||||
const arr = await s.drive(["not", "an", "object"]);
|
||||
// A non-object body cannot carry a surface address, so it is refused by name
|
||||
// at the addressing gate (op-must-be-object is now defensive behind it).
|
||||
const arr = await s.postRaw(["not", "an", "object"]);
|
||||
assert.equal(arr.ok, false);
|
||||
assert.match(arr.error, /JSON object/);
|
||||
assert.match(arr.error, /surface|JSON object/);
|
||||
});
|
||||
|
||||
// ------------------------------------------------- the surface half
|
||||
@@ -130,6 +142,7 @@ test("SUPERSEDED OPS ARE DROPPED, not replayed — the whole reason for a counte
|
||||
const gate = new Promise((r) => { release = r; });
|
||||
|
||||
const d = driveSurface({
|
||||
surface: "s",
|
||||
apply: async (op) => { applied.push(op.gen); if (op.gen === 1) await gate; },
|
||||
fetchImpl: async () => ({ json: async () => ({ ok: true, changed: false, marker: "x", op: null }) }),
|
||||
});
|
||||
@@ -148,6 +161,7 @@ test("SUPERSEDED OPS ARE DROPPED, not replayed — the whole reason for a counte
|
||||
test("an OLDER op arriving late is ignored — order is by generation, not arrival", async (t) => {
|
||||
const applied = [];
|
||||
const d = driveSurface({
|
||||
surface: "s",
|
||||
apply: (op) => { applied.push(op.gen); },
|
||||
fetchImpl: async () => ({ json: async () => ({ ok: true, changed: false, marker: "x", op: null }) }),
|
||||
});
|
||||
@@ -168,6 +182,7 @@ test("two applications never overlap — a mid-flight op waits rather than inter
|
||||
// than by trusting the ordering of the results.
|
||||
let inFlight = 0, maxConcurrent = 0;
|
||||
const d = driveSurface({
|
||||
surface: "s",
|
||||
apply: async () => {
|
||||
inFlight += 1;
|
||||
maxConcurrent = Math.max(maxConcurrent, inFlight);
|
||||
@@ -208,6 +223,7 @@ test("an op posted BEFORE the surface loaded is not replayed", async (t) => {
|
||||
// `play`, `export` or `delete`, firing again in every new tab.
|
||||
const applied = [];
|
||||
const d = driveSurface({
|
||||
surface: "s",
|
||||
apply: (op) => { applied.push(op.gen); },
|
||||
intervalMs: 5,
|
||||
// The runtime keeps the last op forever, so a fresh surface always sees one.
|
||||
@@ -217,7 +233,7 @@ test("an op posted BEFORE the surface loaded is not replayed", async (t) => {
|
||||
|
||||
await new Promise((r) => setTimeout(r, 40));
|
||||
assert.deepEqual(applied, [], `an op that predates the surface was applied: ${applied.join(",")}`);
|
||||
assert.equal(d.applied, 9, "the baseline was not adopted, so a later op could not be told from this one");
|
||||
assert.equal(d.accepted, 9, "the baseline was not adopted, so a later op could not be told from this one");
|
||||
});
|
||||
|
||||
test("...but an op posted AFTER it loaded still applies — the baseline is not a mute button", async (t) => {
|
||||
@@ -227,6 +243,7 @@ test("...but an op posted AFTER it loaded still applies — the baseline is not
|
||||
const applied = [];
|
||||
let current = { gen: 9, select: "old" };
|
||||
const d = driveSurface({
|
||||
surface: "s",
|
||||
apply: (op) => { applied.push(op.select); },
|
||||
intervalMs: 5,
|
||||
fetchImpl: async () => ({ json: async () => ({ ok: true, changed: true, marker: "m", op: current }) }),
|
||||
@@ -244,6 +261,7 @@ test("...but an op posted AFTER it loaded still applies — the baseline is not
|
||||
test("resumeLatestOnLoad opts back IN to catching up, for surfaces that want it", async (t) => {
|
||||
const applied = [];
|
||||
const d = driveSurface({
|
||||
surface: "s",
|
||||
apply: (op) => { applied.push(op.select); },
|
||||
intervalMs: 5,
|
||||
resumeLatestOnLoad: true,
|
||||
@@ -260,6 +278,7 @@ test("the poll survives the runtime going away", async (t) => {
|
||||
// reload to become drivable again — the exact state this primitive removes.
|
||||
let calls = 0;
|
||||
const d = driveSurface({
|
||||
surface: "s",
|
||||
apply: () => {},
|
||||
intervalMs: 5,
|
||||
fetchImpl: async () => { calls += 1; throw new Error("runtime is down"); },
|
||||
@@ -279,6 +298,7 @@ test("end to end: an agent POSTs and an adopting surface applies it", async (t)
|
||||
|
||||
const applied = [];
|
||||
const d = driveSurface({
|
||||
surface: "s",
|
||||
apply: (op) => { applied.push(op); },
|
||||
endpoint: `${s.base}/api/drive`,
|
||||
intervalMs: 20,
|
||||
@@ -302,6 +322,94 @@ test("end to end: an agent POSTs and an adopting surface applies it", async (t)
|
||||
assert.equal(applied[0].show, "take-3");
|
||||
});
|
||||
|
||||
// ---------------------------------------- per-surface delivery + settle-ack
|
||||
|
||||
test("PER-SURFACE DELIVERY: an op addressed to A never reaches kept-alive B or C", async (t) => {
|
||||
// review.r2 CRITICAL / PM's control shape. The shell keeps backgrounded
|
||||
// surfaces ALIVE and polling; one global channel fanned every op to all of
|
||||
// them — including the play/export/delete side-effect classes the contract
|
||||
// names. The fixture MUST keep B and C alive: that is the exact condition, and
|
||||
// a fixture that closed them would prove nothing about the defect.
|
||||
const s = await start();
|
||||
t.after(() => s.stop());
|
||||
|
||||
const appliedA = [], appliedB = [], appliedC = [];
|
||||
const mk = (surface, sink) => driveSurface({
|
||||
surface, apply: (op) => { sink.push(op); }, endpoint: `${s.base}/api/drive`, intervalMs: 15,
|
||||
});
|
||||
const A = mk("A", appliedA), B = mk("B", appliedB), C = mk("C", appliedC);
|
||||
t.after(() => { A.stop(); B.stop(); C.stop(); });
|
||||
|
||||
await new Promise((r) => setTimeout(r, 80)); // all three take their baselines
|
||||
|
||||
// Space the ops past a poll interval so each is applied before the next —
|
||||
// otherwise the generation counter correctly supersedes them down to the last
|
||||
// (that behaviour has its own test above; this control is about ADDRESSING).
|
||||
for (const op of [{ delete: "take-3" }, { export: "reel" }, { play: true }]) {
|
||||
await s.driveTo("A", op);
|
||||
await new Promise((r) => setTimeout(r, 60));
|
||||
}
|
||||
await new Promise((r) => setTimeout(r, 120)); // well past several poll intervals
|
||||
|
||||
assert.equal(appliedA.length, 3, "A did not receive its own ops — per-surface delivery dropped them");
|
||||
assert.deepEqual(appliedB, [],
|
||||
"a kept-alive B received an op addressed to A — delete/export/play fired on the wrong surface, the exact defect");
|
||||
assert.deepEqual(appliedC, [], "a kept-alive C received an op addressed to A");
|
||||
});
|
||||
|
||||
test("ADDRESSING IS MANDATORY: an unaddressed op is refused, so the fan-out default cannot return", async (t) => {
|
||||
// Q3 breaking-now: a warned unaddressed default still fans out. POST and GET
|
||||
// both refuse an unaddressed op by name.
|
||||
const s = await start();
|
||||
t.after(() => s.stop());
|
||||
|
||||
const posted = await s.postRaw({ delete: "everything" }); // no surface
|
||||
assert.equal(posted.ok, false, "an unaddressed op was accepted — the fan-out default is back");
|
||||
assert.match(posted.error, /surface/);
|
||||
|
||||
const polled = await fetch(`${s.base}/api/drive`, { cache: "no-store" }).then((x) => x.json());
|
||||
assert.equal(polled.ok, false, "an unaddressed poll was answered — a surface could receive another's ops");
|
||||
assert.match(polled.error, /surface/);
|
||||
});
|
||||
|
||||
test("SETTLE-ACK: appliedGen exposes a real posted-but-not-yet-applied window", async (t) => {
|
||||
// Q2 condition 3: the record-to-apply gap must be OBSERVABLE via appliedGen <
|
||||
// postedGen, or the ack is decoration. POST returns on RECORD; appliedGen only
|
||||
// advances when the surface's apply() has RESOLVED (the harder half).
|
||||
const s = await start();
|
||||
t.after(() => s.stop());
|
||||
|
||||
let release;
|
||||
const gate = new Promise((r) => { release = r; });
|
||||
const d = driveSurface({
|
||||
surface: "A", endpoint: `${s.base}/api/drive`, intervalMs: 10,
|
||||
apply: async () => { await gate; }, // hold the application open
|
||||
});
|
||||
t.after(() => d.stop());
|
||||
await new Promise((r) => setTimeout(r, 50)); // baseline
|
||||
|
||||
const posted = await s.driveTo("A", { play: true });
|
||||
assert.equal(posted.recorded, true, "POST did not report it RETURNS ON RECORD");
|
||||
const postedGen = posted.gen;
|
||||
|
||||
// apply() is blocked, so appliedGen must lag postedGen — the observable gap
|
||||
await new Promise((r) => setTimeout(r, 60));
|
||||
const mid = await s.readAt("A");
|
||||
assert.ok(mid.appliedGen < postedGen,
|
||||
`no observable gap: appliedGen ${mid.appliedGen} is not < postedGen ${postedGen} — the ack is decoration`);
|
||||
|
||||
release(); // let it settle; appliedGen must catch up
|
||||
const deadline = Date.now() + 3000;
|
||||
let settled;
|
||||
do {
|
||||
settled = await s.readAt("A");
|
||||
if (settled.appliedGen >= postedGen) break;
|
||||
await new Promise((r) => setTimeout(r, 20));
|
||||
} while (Date.now() < deadline);
|
||||
assert.ok(settled.appliedGen >= postedGen,
|
||||
"appliedGen never reached postedGen after apply resolved — settle is unobservable");
|
||||
});
|
||||
|
||||
test("positive control: this suite is capable of failing", () => {
|
||||
assert.throws(() => assert.equal(1, 2));
|
||||
});
|
||||
@@ -120,7 +120,7 @@ test("EVERY declared route answers AS ITSELF — the per-route positive control
|
||||
"/api/files/search": (s) => has(s.body, "hits"),
|
||||
};
|
||||
// Probe query per route: enough to reach the route's own answer shape.
|
||||
const QUERY = { "/api/files/search": "?q=zz" };
|
||||
const QUERY = { "/api/files/search": "?q=zz", "/api/drive": "?surface=probe" };
|
||||
|
||||
const port = 9700 + Math.floor(Math.random() * 250);
|
||||
const proc = spawn(process.execPath,
|
||||
|
||||
Reference in new issue
Block a user