refactor(storage): own structured mutation commit effects

packages/agent/src/dkg-agent-base.ts

@@ -110,7 +110,7 @@ import {
   pickNetworkTunables,
   isSparqlUpdateOperation,
 } from '@origintrail-official/dkg-core';
-import { GraphManager, PrivateContentStore, SystemRecordLaneForwarderV1, createTripleStore, isExternalBackend, structuredMutationMightMutate, structuredMutationTouchedGraphs, type TripleStore, type TripleStoreConfig, type Quad, type LargeLiteralStorageConfig, type QueryOptions } from '@origintrail-official/dkg-storage';
+import { GraphManager, PrivateContentStore, SystemRecordLaneForwarderV1, captureStructuredMutationEffects, createTripleStore, isExternalBackend, type TripleStore, type TripleStoreConfig, type Quad, type LargeLiteralStorageConfig, type QueryOptions } from '@origintrail-official/dkg-storage';
 import { emptyRpcUsageWindow, EVMChainAdapter, NoChainAdapter, enrichEvmError, buildKnowledgeAssetUal, type EVMAdapterConfig, type ChainAdapter, type CreateContextGraphParams, type CreateOnChainContextGraphParams, type CreateOnChainContextGraphResult, type TxResult, type V10PublishingConvictionAccountInfo, type RpcUsageWindow } from '@origintrail-official/dkg-chain';
 import {
   DKGPublisher, PublishHandler, SharedMemoryHandler, UpdateHandler, ChainEventPoller, AccessHandler, AccessClient,
@@ -578,14 +578,13 @@ export function createListContextGraphsCacheInvalidatingStore(
       : undefined,
     structuredMutation: innerStore.structuredMutation
       ? (mutation, options) => {
-          // Capture scope before the first await so caller-side mutation cannot
-          // redirect cache invalidation after the backend has committed.
-          const targetGraphs = [...structuredMutationTouchedGraphs(mutation)];
-          const mightMutate = structuredMutationMightMutate(mutation);
+          const effects = captureStructuredMutationEffects(mutation);
           return invalidateAfterMutation(
             () => innerStore.structuredMutation!(mutation, options),
-            () => mightMutate,
-            () => targetGraphs.forEach((graph) => markProjectionDirty?.(undefined, graph)),
+            () => effects !== undefined,
+            () => effects?.touchedGraphs.forEach(
+              (graph) => markProjectionDirty?.(undefined, graph),
+            ),
           );
         }
       : undefined,

packages/agent/test/replace-subject-agent-wrapper.test.ts

@@ -15,7 +15,7 @@
 import { mkdtemp, rm } from 'node:fs/promises';
 import { tmpdir } from 'node:os';
 import { join } from 'node:path';
-import { afterEach, describe, expect, it } from 'vitest';
+import { afterEach, describe, expect, it, vi } from 'vitest';
 import {
   CHANGELOG_GRAPH,
   OxigraphStore,
@@ -150,4 +150,76 @@ describe('#1863 replaceSubject through the agent store wrapper', () => {
     proj.markDirtyForGraph('urn:dkg:publisher:control-plane');
     expect(entries.has('urn:dkg:publisher:control-plane')).toBe(false);
   });
+
+  it('invalidates structured mutation effects after success without decoding them in Agent', async () => {
+    let release!: () => void;
+    const inFlight = new Promise<void>((resolve) => { release = resolve; });
+    const options = { source: 'agent.test.structured-mutation-effects' };
+    let inner!: TripleStore;
+    const structuredMutation = vi.fn(function (
+      this: TripleStore,
+      _mutation: unknown,
+      receivedOptions: unknown,
+    ) {
+      expect(this).toBe(inner);
+      expect(receivedOptions).toBe(options);
+      return inFlight;
+    });
+    inner = { structuredMutation } as unknown as TripleStore;
+    const invalidate = vi.fn();
+    const markProjectionDirty = vi.fn();
+    const store = createListContextGraphsCacheInvalidatingStore(
+      inner,
+      invalidate,
+      markProjectionDirty,
+    );
+    const mutation = {
+      kind: 'copy-subject-projection' as const,
+      input: {
+        sourceGraphUris: ['urn:test:source'],
+        targetGraphUri: 'urn:test:target',
+        roots: ['urn:test:root'],
+        descendantSuffix: '/',
+        excludedPredicates: [],
+      },
+    };
+
+    const pending = store.structuredMutation!(mutation, options);
+    mutation.input.targetGraphUri = 'urn:test:redirected';
+    expect(invalidate).not.toHaveBeenCalled();
+    release();
+    await pending;
+
+    expect(invalidate).toHaveBeenCalledOnce();
+    expect(markProjectionDirty).toHaveBeenCalledOnce();
+    expect(markProjectionDirty).toHaveBeenCalledWith(undefined, 'urn:test:target');
+  });
+
+  it('does not invalidate structured mutation failures or structural no-ops', async () => {
+    const invalidate = vi.fn();
+    const markProjectionDirty = vi.fn();
+    const inner = {
+      structuredMutation: vi.fn(async () => undefined),
+    } as unknown as TripleStore;
+    const store = createListContextGraphsCacheInvalidatingStore(
+      inner,
+      invalidate,
+      markProjectionDirty,
+    );
+
+    await store.structuredMutation!({
+      kind: 'delete-subjects',
+      input: { graphUri: 'urn:test:target', subjects: [] },
+    });
+    expect(invalidate).not.toHaveBeenCalled();
+    expect(markProjectionDirty).not.toHaveBeenCalled();
+
+    inner.structuredMutation = vi.fn(async () => { throw new Error('commit failed'); });
+    await expect(store.structuredMutation!({
+      kind: 'delete-subjects',
+      input: { graphUri: 'urn:test:target', subjects: ['urn:test:subject'] },
+    })).rejects.toThrow('commit failed');
+    expect(invalidate).not.toHaveBeenCalled();
+    expect(markProjectionDirty).not.toHaveBeenCalled();
+  });
 });

packages/storage/src/adapters/blazegraph.ts

@@ -34,8 +34,8 @@
 } from '../atomic-graph-replace.js';
 import {
   buildStructuredMutationUpdate,
+  captureStructuredMutationEffects,
   normalizeStructuredMutation,
-  structuredMutationMightMutate,
 } from '../bounded-structured-mutation.js';
 import { quadToNQuad } from '../bounded-rdf.js';
 import { readResponseTextBounded } from '../http-response-limit.js';
@@ -417,14 +417,15 @@
     options?: QueryOptions,
   ): Promise<void> {
     const normalized = normalizeStructuredMutation(mutation);
+    const effects = captureStructuredMutationEffects(normalized);
     if (normalized.kind === 'replace-subject-predicates') {
       assertQuadLiteralsMutf8Safe([...normalized.input.replacementQuads], {
         maxBytes: JAVA_WRITE_UTF_MAX_BYTES,
         label: 'BlazegraphStore.structuredMutation',
       });
     }
     const update = buildStructuredMutationUpdate(normalized);
-    if (!update || !structuredMutationMightMutate(normalized)) return;
+    if (!update || !effects) return;
     await this.sparqlUpdate(
       update,
       { ...options, source: options?.source ?? 'blazegraph.structuredMutation' },
@@ -544,7 +545,7 @@
 
   async listGraphs(options?: TripleStoreQueryOptions): Promise<string[]> {
     const r = await this.query(
       'SELECT DISTINCT ?g WHERE { GRAPH ?g { ?s ?p ?o } }',
       options,
     );
     if (r.type !== 'bindings') return [];
@@ -560,7 +561,7 @@
   async countQuads(graphUri?: string, options?: QueryOptions): Promise<number> {
     const sparql = graphUri
       ? `SELECT (COUNT(*) AS ?c) WHERE { GRAPH <${escapeUri(graphUri)}> { ?s ?p ?o } }`
       : `SELECT (COUNT(*) AS ?c) WHERE { { ?s ?p ?o } UNION { GRAPH ?g { ?s ?p ?o } } }`;
     const r = await this.query(sparql, {
       ...options,
       source: options?.source ?? 'blazegraph.countQuads',

packages/storage/src/adapters/oxigraph-worker.ts

@@ -13,9 +13,8 @@ import type {
 import { registerTripleStoreAdapter } from '../triple-store.js';
 import { GraphWriteGenTracker } from '../graph-write-gen.js';
 import {
+  captureStructuredMutationEffects,
   normalizeStructuredMutation,
-  structuredMutationMightMutate,
-  structuredMutationTouchedGraphs,
 } from '../bounded-structured-mutation.js';
 
 /**
@@ -698,10 +697,9 @@ export class OxigraphWorkerStore implements TripleStore {
     _options?: TripleStoreQueryOptions,
   ): Promise<void> {
     const normalized = normalizeStructuredMutation(mutation);
+    const effects = captureStructuredMutationEffects(normalized);
     await this.call('structuredMutation', normalized);
-    if (structuredMutationMightMutate(normalized)) {
-      this.writeGen.recordGraphWrites(structuredMutationTouchedGraphs(normalized));
-    }
+    if (effects) this.writeGen.recordGraphWrites(effects.touchedGraphs);
   }
   async query(sparql: string, options?: TripleStoreQueryOptions): Promise<QueryResult> {
     return this.callWithTimeout<QueryResult>(this.operationTimeoutMs, options?.signal, 'query', sparql);

packages/storage/src/adapters/oxigraph.ts

@@ -28,9 +28,8 @@ import {
 } from '../atomic-graph-replace.js';
 import {
   buildStructuredMutationUpdate,
+  captureStructuredMutationEffects,
   normalizeStructuredMutation,
-  structuredMutationMightMutate,
-  structuredMutationTouchedGraphs,
 } from '../bounded-structured-mutation.js';
 import { quadsToNQuads } from '../bounded-rdf.js';
 import { assertQuadLiteralsMutf8Safe, JAVA_WRITE_UTF_MAX_BYTES } from '@origintrail-official/dkg-core';
@@ -421,17 +420,18 @@ export class OxigraphStore implements TripleStore {
     _options?: TripleStoreQueryOptions,
   ): Promise<void> {
     const normalized = normalizeStructuredMutation(mutation);
+    const effects = captureStructuredMutationEffects(normalized);
     if (normalized.kind === 'replace-subject-predicates') {
       assertQuadLiteralsMutf8Safe([...normalized.input.replacementQuads], {
         maxBytes: JAVA_WRITE_UTF_MAX_BYTES,
         label: 'OxigraphStore.structuredMutation',
       });
     }
     const update = buildStructuredMutationUpdate(normalized);
-    if (!update || !structuredMutationMightMutate(normalized)) return;
+    if (!update || !effects) return;
     this.store.update(update);
     this.scheduleFlush();
-    this.writeGen.recordGraphWrites(structuredMutationTouchedGraphs(normalized));
+    this.writeGen.recordGraphWrites(effects.touchedGraphs);
   }
 
   async listGraphs(options?: TripleStoreQueryOptions): Promise<string[]> {

packages/storage/src/adapters/sparql-http.ts

@@ -54,10 +54,9 @@
 } from '../atomic-graph-replace.js';
 import {
   buildStructuredMutationUpdate,
+  captureStructuredMutationEffects,
   normalizeStructuredMutation,
   structuredMutationGuardedGraphs,
-  structuredMutationMightMutate,
-  structuredMutationTouchedGraphs,
 } from '../bounded-structured-mutation.js';
 import {
   assertNotReservedInternalGraphV1,
@@ -1299,7 +1298,7 @@
     options?: QueryOptions,
   ): Promise<void> {
     const normalized = normalizeStructuredMutation(mutation);
-    const touchedGraphs = structuredMutationTouchedGraphs(normalized);
+    const effects = captureStructuredMutationEffects(normalized);
     this.assertGenericMutationScope(structuredMutationGuardedGraphs(normalized), 'structuredMutation');
     if (normalized.kind === 'replace-subject-predicates') {
       assertQuadLiteralsMutf8Safe([...normalized.input.replacementQuads], {
@@ -1308,24 +1307,24 @@
       });
     }
     const update = buildStructuredMutationUpdate(normalized);
-    if (!update || !structuredMutationMightMutate(normalized)) return;
+    if (!update || !effects) return;
     try {
       await this.postUpdate(
         update,
         { ...options, source: options?.source ?? 'sparql-http.structuredMutation' },
         'structuredMutation',
-        touchedGraphs,
+        effects.touchedGraphs,
       );
     } catch (error) {
       // A remote endpoint may commit before its response is lost. Fail open for
       // cache coherence: invalidate graph enumeration and advance each affected
       // write generation even though the caller still receives the failure.
       this.invalidateListGraphsCache();
-      this.writeGen.recordGraphWrites(touchedGraphs);
+      this.writeGen.recordGraphWrites(effects.touchedGraphs);
       throw error;
     }
     this.invalidateListGraphsCache();
-    this.writeGen.recordGraphWrites(touchedGraphs);
+    this.writeGen.recordGraphWrites(effects.touchedGraphs);
   }
 
   async query(sparql: string, options?: SparqlHttpQueryOptions): Promise<QueryResult> {
@@ -1522,7 +1521,7 @@
   async countQuads(graphUri?: string, options?: QueryOptions): Promise<number> {
     const sparql = graphUri
       ? `SELECT (COUNT(*) AS ?c) WHERE { GRAPH <${escapeUri(graphUri)}> { ?s ?p ?o } }`
       : `SELECT (COUNT(*) AS ?c) WHERE { { ?s ?p ?o } UNION { GRAPH ?g { ?s ?p ?o } } }`;
     const r = await this.query(sparql, {
       ...options,
       source: options?.source ?? 'sparql-http.countQuads',

packages/storage/src/bounded-structured-mutation.ts

@@ -224,7 +224,7 @@
     # sparql-scan-allow: R3 -- one bounded retention prune; OFFSET <= 100000 and LIMIT <= 10000; never page-walked
     GROUP BY ?subject
     ORDER BY DESC(COALESCE(
       xsd:integer(STR(?latestPrimaryRank)),
       xsd:integer(STR(?latestSecondaryRank)),
       0
     )) DESC(STR(?subject))
@@ -547,7 +547,7 @@
 WHERE {
   VALUES ?sourceGraph { ${sources} }
   VALUES ?root { ${roots} }
   # sparql-scan-allow: R2 -- ?sourceGraph is VALUES-bound to at most 8 validated exact graph IRIs
   GRAPH ?sourceGraph { ?subject ?predicate ?object }
   FILTER(?subject = ?root || STRSTARTS(STR(?subject), CONCAT(STR(?root), ${sparqlString(normalized.descendantSuffix)})))
   ${excluded}
@@ -680,3 +680,17 @@
 export function structuredMutationMightMutate(mutation: StructuredMutation): boolean {
   return mutation.kind !== 'delete-subjects' || mutation.input.subjects.length > 0;
 }
+
+/** Immutable graph-scoped effects captured before a structured mutation is dispatched. */
+export interface StructuredMutationEffects {
+  readonly touchedGraphs: readonly string[];
+}
+
+/** Capture canonical effects without executing or probing a store capability. */
+export function captureStructuredMutationEffects(
+  mutation: StructuredMutation,
+): StructuredMutationEffects | undefined {
+  if (!structuredMutationMightMutate(mutation)) return undefined;
+  const touchedGraphs = Object.freeze([...structuredMutationTouchedGraphs(mutation)]);
+  return Object.freeze({ touchedGraphs });
+}

packages/storage/src/changelog-store.ts

@@ -24,10 +24,9 @@ import {
 } from './store-chain-capability.js';
 import type { SystemRecordLaneControllerV1 } from './system-record-materializer-v1.js';
 import {
+  captureStructuredMutationEffects,
   normalizeStructuredMutation,
   structuredMutationGuardedGraphs,
-  structuredMutationMightMutate,
-  structuredMutationTouchedGraphs,
 } from './bounded-structured-mutation.js';
 
 /**
@@ -460,6 +459,7 @@ export class ChangelogStore implements TripleStore, ChangelogReader {
 
   async structuredMutation(mutation: StructuredMutation, options?: QueryOptions): Promise<void> {
     const normalized = normalizeStructuredMutation(mutation);
+    const effects = captureStructuredMutationEffects(normalized);
     const operation = this.inner.structuredMutation;
     if (!operation) {
       throw new UnsupportedTripleStoreCapabilityError('structuredMutation', 'ChangelogStore');
@@ -477,8 +477,8 @@ export class ChangelogStore implements TripleStore, ChangelogReader {
         }
         throw error;
       }
-      if (structuredMutationMightMutate(normalized)) {
-        await this.markPostMutation(structuredMutationTouchedGraphs(normalized), options);
+      if (effects) {
+        await this.markPostMutation(effects.touchedGraphs, options);
       }
     });
   }

packages/storage/src/graph-set-index-store.ts

@@ -23,9 +23,8 @@ import {
 import { isAtomicGraphReplaceStagingGraph } from './atomic-graph-replace.js';
 import { ManagedOxigraphBackendUnownedError } from './managed-oxigraph-ownership-v1-internal.js';
 import {
+  captureStructuredMutationEffects,
   normalizeStructuredMutation,
-  structuredMutationMightMutate,
-  structuredMutationTouchedGraphs,
 } from './bounded-structured-mutation.js';
 import {
   CACHED_READ_GATE_V1,
@@ -559,6 +558,7 @@ export class GraphSetIndexStore implements TripleStore {
 
   async structuredMutation(mutation: StructuredMutation, options?: QueryOptions): Promise<void> {
     const normalized = normalizeStructuredMutation(mutation);
+    const effects = captureStructuredMutationEffects(normalized);
     const operation = this.inner.structuredMutation;
     if (!operation) {
       throw new UnsupportedTripleStoreCapabilityError('structuredMutation', 'GraphSetIndexStore');
@@ -571,10 +571,10 @@ export class GraphSetIndexStore implements TripleStore {
       }
       throw error;
     }
-    if (!this.enabled || !structuredMutationMightMutate(normalized)) return;
+    if (!this.enabled || !effects) return;
     this.bumpMutation();
     await this.maintainTouchedGraphs(
-      [...structuredMutationTouchedGraphs(normalized)],
+      [...effects.touchedGraphs],
       'structuredMutation',
       options,
     );

packages/storage/src/index.ts

@@ -51,9 +51,11 @@ export {
 } from './atomic-graph-replace.js';
 export {
   BOUNDED_MUTATION_MAX_PRUNE_DELETE,
+  captureStructuredMutationEffects,
   chunkCopySubjectProjectionInput,
   structuredMutationMightMutate,
   structuredMutationTouchedGraphs,
+  type StructuredMutationEffects,
 } from './bounded-structured-mutation.js';
 // System-record V1 (#2052 Stack B2). Default-unused: these modules perform no
 // I/O, scheduling, timer, or per-store lane work until the daemon supervisor