fix(daemon): composite-operation membership scope — one batch per request for the boundToWorkflow lens and composeMission

Corrective gate on the operation-scoped membership index (75245ed6):
two composite operations run list() plus indexer.get() per slice, and
with per-operation batch clearing every uncached get built its own
2-scan batch — 2+2N membership scans (82 at 40 slices, guard-reproduced:
route lens 3888.7ms at 353 slices with one queue row; ReviewGatherer.
composeMission is the same shape on the flagship Review path).

Add SliceIndexer.withMembershipBatch(fn): a synchronous, depth-counted
composite-operation scope. The outermost open never reuses a prior
operation's batch (the index is lazy — first membership access inside
the scope builds from then-current queue state; request-boundary
freshness, no snapshot claim); nested scopes share the outermost batch;
the outermost exit always clears via finally (exception-safe); and
standalone list()/get() keep their unchanged one-operation semantics
(their clears are now conditional on scope depth). The slices GET
handler wraps its whole request (list + lens get loop + mission
sidecar) in one scope; composeMission wraps its list + per-slice
gather projection. composeSliceWithContext is deliberately unwrapped
(its second get is detailCache-absorbed — no O(N) batching).

Load contracts pinned by test: one lens request and one mission compose
each execute a constant number of membership scans (both were 82 at 40
slices; the Review counter counts the membership query family by named
shape — the exact 204-read inventory, including the 122 pre-existing
attention/agent/hasActiveQitem projections, is documented in-test).
Scope API pins: in-scope sharing, two-scope stale-generation freshness,
nested reuse, exception-safe cleanup. Composite benchmarks at 353
slices + 8000x3KB queue: route-shaped 194.0ms, Review-shaped 276.9ms.

qitem-ccf87c0d amended gate (corrective on c8f85802 + 75245ed6)

Co-Authored-By: Claude Fable 5 <[email protected]>
This commit is contained in:
v-openrig-buildandClaude Fable 5 committed 2026-07-20 05:05:57 +00:00
1 parent 75245ed69b
commit 5c95a8ee6f
6 files changed
+353 -45

No files matched your search

@@ -123,6 +123,14 @@ export class ReviewGatherer {
}
composeMission(mission: string): ComposedMissionReview | null {
// qitem-ccf87c0d corrective — one mission compose is ONE composite
// operation: the cold list() plus gatherSlice()->indexer.get() per
// mission slice share ONE membership batch (pre-scope each uncached get
// built its own 2-scan batch: 2+2N membership scans — 82 at 40 slices).
return this.indexer.withMembershipBatch(() => this.composeMissionInBatch(mission));
}
private composeMissionInBatch(mission: string): ComposedMissionReview | null {
const slices = this.indexer.list().filter((s) => s.missionId === mission);
if (slices.length === 0 && !this.missionDirExists(mission)) return null;
const nowIso = this.now();
@@ -261,6 +261,8 @@ export class SliceIndexer {
private detailCache: Map<string, CachedSlice> = new Map();
// qitem-ccf87c0d — batch membership index (see MembershipIndex doc).
private membershipIndex: MembershipIndex | null = null;
// qitem-ccf87c0d corrective — depth of open withMembershipBatch scopes.
private batchDepth = 0;
// VM-005: authored mission-status sidecar cache (same TTL as the listing).
private missionStatusCache: {
statuses: Record<string, { authoredStatus: string | null }>;
@@ -280,6 +282,34 @@ export class SliceIndexer {
return this.sliceRoots().some((root) => this.isDirectory(root));
}
/** qitem-ccf87c0d corrective — the explicit COMPOSITE-OPERATION scope: a
* caller performing list()-plus-per-slice-get() as ONE user-visible
* operation (the slices route's boundToWorkflow lens, ReviewGatherer.
* composeMission) wraps it here so every inner list/get shares ONE
* membership batch instead of building one per uncached get (the 2+2N
* regression). Contract, pinned by test:
* - the OUTERMOST open never reuses a prior operation's batch (drops
* any existing one); the index is LAZY, so the first membership
* access inside the scope builds from then-current queue state —
* request-boundary freshness, no snapshot claim;
* - nested/reentrant scopes share the outermost batch;
* - the outermost exit ALWAYS clears (finally — exception-safe), so
* separate operations never share a generation;
* - standalone list()/get() outside any scope keep their own
* one-operation batch semantics unchanged (their clears are
* conditional on depth === 0).
* Synchronous by design — matches the synchronous list/get call graph. */
withMembershipBatch<T>(fn: () => T): T {
if (this.batchDepth === 0) this.membershipIndex = null;
this.batchDepth++;
try {
return fn();
} finally {
this.batchDepth--;
if (this.batchDepth === 0) this.membershipIndex = null;
}
}
/** Drops both caches. Used by tests + by a future explicit-refresh route. */
invalidate(): void {
this.listingCache = null;
@@ -365,12 +395,13 @@ export class SliceIndexer {
const locations = this.readSliceLocations();
// ONE batch membership index serves this whole rebuild (the <=4 LIKE-
// execution load contract), then drops so later uncached operations
// read current queue state (qitem-18f3300d operation scoping).
// read current queue state (qitem-18f3300d operation scoping) — unless
// an enclosing withMembershipBatch scope owns the batch lifetime.
let entries: SliceListEntry[];
try {
entries = locations.map((location) => this.toListEntry(location));
} finally {
this.membershipIndex = null;
if (this.batchDepth === 0) this.membershipIndex = null;
}
this.listingCache = { entries, expiresAt: now + this.cacheTtlMs };
return entries;
@@ -387,12 +418,13 @@ export class SliceIndexer {
if (!location) return null;
// Uncached get(): its own operation-scoped batch, dropped afterward so
// the next uncached operation sees current queue state (qitem-18f3300d).
// the next uncached operation sees current queue state (qitem-18f3300d)
// — unless an enclosing withMembershipBatch scope owns the batch lifetime.
let record: SliceRecord;
try {
record = this.buildRecord(location);
} finally {
this.membershipIndex = null;
if (this.batchDepth === 0) this.membershipIndex = null;
}
this.detailCache.set(name, { record, expiresAt: now + this.cacheTtlMs });
return record;
+47 -41
View File
@@ -53,49 +53,55 @@ export function slicesRoutes(): Hono {
if (refresh === "1" || refresh === "true") {
deps.indexer.invalidate();
}
const all = deps.indexer.list();
let filtered = filter === "all" ? all : all.filter((s) => s.status === filter);
// Workflows in Spec Library v0: optional lens filter — narrow to
// slices bound to a workflow_instance of <name>:<version>.
const boundToWorkflow = c.req.query("boundToWorkflow");
let boundDiagnostic: { specName: string; specVersion: string; matched: number; total: number } | null = null;
if (boundToWorkflow) {
const colonIdx = boundToWorkflow.lastIndexOf(":");
if (colonIdx === -1) {
return c.json({
error: "boundToWorkflow_invalid",
hint: "Format is boundToWorkflow=<specName>:<specVersion>",
}, 400);
// qitem-ccf87c0d corrective — ONE HTTP request is ONE composite
// operation: the list rebuild, the boundToWorkflow per-slice get loop,
// and the mission sidecar share ONE membership batch (pre-scope, each
// uncached get built its own 2-scan batch: 2+2N total queue scans).
return deps.indexer.withMembershipBatch(() => {
const all = deps.indexer.list();
let filtered = filter === "all" ? all : all.filter((s) => s.status === filter);
// Workflows in Spec Library v0: optional lens filter — narrow to
// slices bound to a workflow_instance of <name>:<version>.
const boundToWorkflow = c.req.query("boundToWorkflow");
let boundDiagnostic: { specName: string; specVersion: string; matched: number; total: number } | null = null;
if (boundToWorkflow) {
const colonIdx = boundToWorkflow.lastIndexOf(":");
if (colonIdx === -1) {
return c.json({
error: "boundToWorkflow_invalid",
hint: "Format is boundToWorkflow=<specName>:<specVersion>",
}, 400);
}
const specName = boundToWorkflow.slice(0, colonIdx);
const specVersion = boundToWorkflow.slice(colonIdx + 1);
const db = deps.indexer.db;
const before = filtered.length;
filtered = filtered.filter((slice) => {
// Re-resolve binding per slice. The indexer's list payload
// doesn't carry workflowName so we do the join here. v0 cost
// is bounded by the slice count + a small SQL per slice (the
// membership batch is shared across the whole request).
const sliceRecord = deps.indexer.get(slice.name);
if (!sliceRecord || sliceRecord.qitemIds.length === 0) return false;
const binding = findSliceWorkflowBinding(db, sliceRecord.qitemIds);
return binding.primary?.workflowName === specName
&& binding.primary?.workflowVersion === specVersion;
});
boundDiagnostic = { specName, specVersion, matched: filtered.length, total: before };
}
const specName = boundToWorkflow.slice(0, colonIdx);
const specVersion = boundToWorkflow.slice(colonIdx + 1);
const db = deps.indexer.db;
const before = filtered.length;
filtered = filtered.filter((slice) => {
// Re-resolve binding per slice. The indexer's list payload
// doesn't carry workflowName so we do the join here. v0 cost
// is bounded by the slice count + a small SQL per slice; with
// ~109 slices this stays sub-100ms in practice.
const sliceRecord = deps.indexer.get(slice.name);
if (!sliceRecord || sliceRecord.qitemIds.length === 0) return false;
const binding = findSliceWorkflowBinding(db, sliceRecord.qitemIds);
return binding.primary?.workflowName === specName
&& binding.primary?.workflowVersion === specVersion;
// Sort by lastActivityAt DESC (most recently touched first); slices
// without activity sort to the end.
filtered.sort(compareByActivityDesc);
return c.json({
slices: filtered,
totalCount: filtered.length,
filter,
boundToWorkflow: boundDiagnostic,
// VM-005 (release-0.4.7): additive authored mission-status sidecar so
// chip surfaces can honor authored-wins precedence without a second
// round-trip. The `slices` array itself is byte-untouched.
missions: deps.indexer.missionAuthoredStatuses(),
});
boundDiagnostic = { specName, specVersion, matched: filtered.length, total: before };
}
// Sort by lastActivityAt DESC (most recently touched first); slices
// without activity sort to the end.
filtered.sort(compareByActivityDesc);
return c.json({
slices: filtered,
totalCount: filtered.length,
filter,
boundToWorkflow: boundDiagnostic,
// VM-005 (release-0.4.7): additive authored mission-status sidecar so
// chip surfaces can honor authored-wins precedence without a second
// round-trip. The `slices` array itself is byte-untouched.
missions: deps.indexer.missionAuthoredStatuses(),
});
});
@@ -498,3 +498,84 @@ describe("GET /api/review/rig (OPR.0.4.4.22)", () => {
expect(res.status).toBe(503);
});
});
// ---------------------------------------------------------------------------
// qitem-ccf87c0d amended gate — composeMission composite-operation load
// contract. On 75245ed6, composeMission runs a cold list() (one 2-scan
// batch) then gatherSlice()->indexer.get() once per uncached mission slice
// (one MORE 2-scan batch each): 2 + 2N MEMBERSHIP scans — 82 at 40 slices
// (guard-reproduced; the exact inventory is 204 total queue reads, of which
// 82 are membership — see the matcher doc below). The contract: one mission
// compose = a CONSTANT number of membership-scan executions.
// ---------------------------------------------------------------------------
describe("qitem-ccf87c0d — composeMission MEMBERSHIP-scan load contract", () => {
/** Count EXECUTIONS (.all/.iterate/.get) of the MEMBERSHIP-SCAN shapes —
* the statement class the batch/scope fix owns, counted by NAMED shape
* (guard audit trail; inventory at 40 slices/1 row: these two shapes are
* 41+41=82 pre-fix, exactly the 2+2N regression; the OTHER 122 queue
* reads are pre-existing Review projections — 82x attention/agent
* projection reads, 40x hasActiveQitem `state IN … AND tags LIKE ?` —
* present on parent 7b19b73e and untouched by the locked fix, so
* counting them would make the constant-scan contract unsatisfiable
* in scope):
* (a) typed membership scan — tags LIKE '%slice:%'
* (b) batch fallback scan — FROM queue_items with NO WHERE clause
* (c) legacy per-slice tiers — WHERE tags LIKE ? ORDER BY / body LIKE ?
* INSERT seeding and PK point lookups (WHERE qitem_id IN) pass through. */
function instrumentMembershipScans(target: Database.Database): () => number {
let n = 0;
const origPrepare = target.prepare.bind(target);
(target as unknown as { prepare: (sql: string) => unknown }).prepare = (sql: string) => {
const stmt = origPrepare(sql) as unknown as Record<string, (...a: unknown[]) => unknown>;
const flat = sql.replace(/\s+/g, " ");
const isMembershipScan = /from queue_items/i.test(flat)
&& !/^\s*insert/i.test(flat)
&& (
/tags LIKE '%slice:%'/i.test(flat)
|| !/ where /i.test(flat)
|| /where tags like \? order by/i.test(flat)
|| /where body like \?/i.test(flat)
);
if (!isMembershipScan) return stmt;
return {
all: (...a: unknown[]) => { n++; return stmt.all!(...a); },
iterate: (...a: unknown[]) => { n++; return stmt.iterate!(...a); },
get: (...a: unknown[]) => { n++; return stmt.get!(...a); },
run: (...a: unknown[]) => stmt.run!(...a),
};
};
return () => n;
}
it("one composeMission over 40 indexed slices executes <= 4 membership queue scans AND composes all 40 ledger entries", () => {
const db = createDb();
migrate(db, [
coreSchema, bindingsSessionsSchema, eventsSchema, streamItemsSchema,
queueItemsSchema, queueTransitionsSchema, missionControlActionsSchema,
queueItemSummarySchema,
]);
db.prepare(`INSERT INTO rigs (id, name) VALUES ('r-load', 'rig')`).run();
// Guard repro shape: exactly one queue row (matches no slice, so no
// per-slice IN lookups fire — the count isolates the scan class).
db.prepare(
`INSERT INTO queue_items (qitem_id, ts_created, ts_updated, source_session, destination_session, state, priority, body)
VALUES ('q-lone', '2026-07-04T00:00:00.000Z', '2026-07-04T00:00:00.000Z', 'a@r', 'b@r', 'done', 'routine', 'fixture')`,
).run();
const ws = makeFixtureWorkspace();
for (let i = 0; i < 40; i++) {
writeFixtureSlice(ws, "load-mission", `ld-${String(i).padStart(2, "0")}-topic`, {});
}
const indexer = new SliceIndexer({ slicesRoot: ws.root, dogfoodEvidenceRoot: null, db });
const scans = instrumentMembershipScans(db);
const gatherer = new ReviewGatherer({ db, indexer, gitRepoPath: null, now: () => NOW });
const composed = gatherer.composeMission("load-mission");
expect(composed).toBeTruthy();
expect(composed!.ledger).toHaveLength(40); // semantic: every slice composed
// Pre-fix on 75245ed6: 2 (cold list batch) + 40x2 (one batch per
// uncached gatherSlice get) = 82 membership scans. Contract: constant.
expect(scans()).toBeLessThanOrEqual(4);
db.close();
fs.rmSync(ws.root, { recursive: true, force: true });
});
});
+118
View File
@@ -941,3 +941,121 @@ describe("qitem-18f3300d — NUL parity + cross-operation membership freshness (
expect(indexer.get("fresh-c-fallback")!.qitemIds).toEqual(["q-c-fallback"]);
});
});
// qitem-ccf87c0d amended gate — the explicit composite-operation scope API
// (withMembershipBatch). RED pre-fix via missing export; pins define the
// contract: outermost-fresh open, in-scope sharing, nested reuse,
// exception-safe close, and unchanged standalone semantics.
describe("qitem-ccf87c0d — withMembershipBatch scope API (RED: API missing)", () => {
let db: Database.Database;
let cleanup: string;
beforeEach(() => {
db = createDb();
migrate(db, [coreSchema, eventsSchema, streamItemsSchema, queueItemsSchema]);
db.prepare(`INSERT INTO rigs (id, name) VALUES ('r-1', 'rig')`).run();
cleanup = fs.mkdtempSync(path.join(os.tmpdir(), "slice-indexer-scope-"));
});
afterEach(() => {
db.close();
fs.rmSync(cleanup, { recursive: true, force: true });
});
type Scoped = SliceIndexer & { withMembershipBatch<T>(fn: () => T): T };
function scoped(indexer: SliceIndexer): Scoped {
const s = indexer as Scoped;
expect(typeof s.withMembershipBatch, "withMembershipBatch must be a public SliceIndexer method").toBe("function");
return s;
}
/** Same membership-scan counter as the route/Review contracts (all
* queue_items scans; PK IN lookups + INSERTs excluded — no gather reads
* exist at the indexer seam so total == membership here). */
function instrumentScans(target: Database.Database): () => number {
let n = 0;
const origPrepare = target.prepare.bind(target);
(target as unknown as { prepare: (sql: string) => unknown }).prepare = (sql: string) => {
const stmt = origPrepare(sql) as unknown as Record<string, (...a: unknown[]) => unknown>;
const isScan = /from\s+queue_items/i.test(sql)
&& !/^\s*insert/i.test(sql)
&& !/where\s+qitem_id\s+in/i.test(sql);
if (!isScan) return stmt;
return {
all: (...a: unknown[]) => { n++; return stmt.all!(...a); },
iterate: (...a: unknown[]) => { n++; return stmt.iterate!(...a); },
get: (...a: unknown[]) => { n++; return stmt.get!(...a); },
run: (...a: unknown[]) => stmt.run!(...a),
};
};
return () => n;
}
function seedSlices(n: number): string {
const root = path.join(cleanup, "missions");
for (let i = 0; i < n; i++) {
writeSlice(path.join(root, "scope-mission", "slices"), `sc-${String(i).padStart(2, "0")}-topic`, {
"README.md": "---\nstatus: active\n---\n",
});
}
return root;
}
it("in-scope list() + N gets share ONE batch: <= 4 total queue scans across the whole scope", () => {
const root = seedSlices(12);
const indexer = new SliceIndexer({ slicesRoot: root, dogfoodEvidenceRoot: null, db });
const s = scoped(indexer);
const scans = instrumentScans(db);
s.withMembershipBatch(() => {
const entries = indexer.list();
for (const e of entries) indexer.get(e.name);
});
expect(scans()).toBeLessThanOrEqual(4);
});
it("REQUEST-BOUNDARY FRESHNESS: scope1 builds a batch via get(A); insert membership for already-known uncached B; scope2 get(B) sees it (scope-open is always fresh)", () => {
const root = seedSlices(2); // sc-00 (A) + sc-01 (B), both known from the start
const indexer = new SliceIndexer({ slicesRoot: root, dogfoodEvidenceRoot: null, db });
const s = scoped(indexer);
s.withMembershipBatch(() => {
expect(indexer.get("sc-00-topic")!.qitemIds).toEqual([]); // scope1 builds the generation
});
insertQitem(db, { qitemId: "q-scope-b", body: "no mention", tags: ["slice:sc-01-topic"] });
s.withMembershipBatch(() => {
// B was already known to scope1's batch — a stale-generation reuse
// would serve []; the fresh scope2 open must see the new row.
expect(indexer.get("sc-01-topic")!.qitemIds).toEqual(["q-scope-b"]);
});
});
it("NESTED scopes share the outermost batch and clear only at outermost exit", () => {
const root = seedSlices(4);
const indexer = new SliceIndexer({ slicesRoot: root, dogfoodEvidenceRoot: null, db });
const s = scoped(indexer);
const scans = instrumentScans(db);
s.withMembershipBatch(() => {
indexer.get("sc-00-topic");
s.withMembershipBatch(() => {
indexer.get("sc-01-topic");
});
// Inner exit must NOT have cleared the shared batch.
indexer.get("sc-02-topic");
});
expect(scans()).toBeLessThanOrEqual(4);
// After outermost exit the batch is gone: a later write is visible.
insertQitem(db, { qitemId: "q-after-scope", body: "no mention", tags: ["slice:sc-03-topic"] });
expect(indexer.get("sc-03-topic")!.qitemIds).toEqual(["q-after-scope"]);
});
it("EXCEPTION inside a scope still clears at exit (finally): the next operation is fresh", () => {
const root = seedSlices(2);
const indexer = new SliceIndexer({ slicesRoot: root, dogfoodEvidenceRoot: null, db });
const s = scoped(indexer);
expect(() => s.withMembershipBatch(() => {
indexer.get("sc-00-topic");
throw new Error("boom");
})).toThrow("boom");
insertQitem(db, { qitemId: "q-post-throw", body: "no mention", tags: ["slice:sc-01-topic"] });
expect(indexer.get("sc-01-topic")!.qitemIds).toEqual(["q-post-throw"]);
});
});
@@ -177,3 +177,66 @@ describe("GET /api/slices?boundToWorkflow=<name>:<version>", () => {
expect(body.boundToWorkflow).toBeNull();
});
});
// ---------------------------------------------------------------------------
// qitem-ccf87c0d amended gate — boundToWorkflow composite-operation load
// contract. The route runs list() then indexer.get() per filtered slice; on
// 75245ed6 each uncached get builds its own 2-scan batch: 82 total queue
// scans at 40 slices (guard-reproduced). Contract: one lens request = a
// CONSTANT number of queue_items scan executions. This path has NO gather
// reads, so the count-EVERYTHING rule applies cleanly (LIKE and non-LIKE;
// INSERTs and PK point lookups WHERE qitem_id IN excluded).
// ---------------------------------------------------------------------------
describe("qitem-ccf87c0d — boundToWorkflow total queue-scan load contract", () => {
function instrumentQueueScans(target: Database.Database): () => number {
let n = 0;
const origPrepare = target.prepare.bind(target);
(target as unknown as { prepare: (sql: string) => unknown }).prepare = (sql: string) => {
const stmt = origPrepare(sql) as unknown as Record<string, (...a: unknown[]) => unknown>;
const isScan = /from\s+queue_items/i.test(sql)
&& !/^\s*insert/i.test(sql)
&& !/where\s+qitem_id\s+in/i.test(sql);
if (!isScan) return stmt;
return {
all: (...a: unknown[]) => { n++; return stmt.all!(...a); },
iterate: (...a: unknown[]) => { n++; return stmt.iterate!(...a); },
get: (...a: unknown[]) => { n++; return stmt.get!(...a); },
run: (...a: unknown[]) => stmt.run!(...a),
};
};
return () => n;
}
it("one GET /api/slices?boundToWorkflow over 40 slices executes <= 4 total queue_items scans, response diagnostic intact", async () => {
const db = createDb();
migrate(db, [
coreSchema, eventsSchema, streamItemsSchema, queueItemsSchema,
queueTransitionsSchema, workflowSpecsSchema, workflowInstancesSchema,
workflowStepTrailsSchema, missionControlActionsSchema,
]);
db.prepare(`INSERT INTO rigs (id, name) VALUES ('r-load', 'rig')`).run();
ensureQitem(db, "q-lone", "matches nothing");
const base = fs.mkdtempSync(path.join(os.tmpdir(), "btw-load-"));
const slicesRoot = path.join(base, "missions");
for (let i = 0; i < 40; i++) {
const dir = path.join(slicesRoot, "load-mission", "slices", `ld-${String(i).padStart(2, "0")}-topic`);
fs.mkdirSync(dir, { recursive: true });
fs.writeFileSync(path.join(dir, "README.md"), `---\nstatus: active\n---\n# t${i}\n`);
}
const indexer = new SliceIndexer({ slicesRoot, dogfoodEvidenceRoot: null, db });
const projector = new SliceDetailProjector({ db, indexer });
const app = buildApp({ indexer, projector });
const scans = instrumentQueueScans(db);
const res = await app.request("/api/slices?boundToWorkflow=openrig-velocity:1.0");
expect(res.status).toBe(200);
const body = (await res.json()) as { boundToWorkflow: { matched: number; total: number }; slices: unknown[] };
expect(body.boundToWorkflow).toEqual({ specName: "openrig-velocity", specVersion: "1.0", matched: 0, total: 40 });
expect(body.slices).toHaveLength(0);
// Pre-fix on 75245ed6: 2 (cold list batch) + 40x2 (uncached get per
// filtered slice) = 82. Contract: constant.
expect(scans()).toBeLessThanOrEqual(4);
db.close();
fs.rmSync(base, { recursive: true, force: true });
});
});