Dkg Importer

DKG Importer Skill

This skill is the agent-readable manual for bulk imports against a DKG V10 node. If you are about to write more than a few thousand triples in one logical operation — a code graph, a Markdown corpus, a GitHub issue archive, a domain-specific dataset — read this first. It documents the contract every existing in-tree importer follows, so the graphs you produce join naturally with graphs other agents and the scanners produce.

For the general node API surface (auth, contextGraphs, SWM/VM publish, SPARQL) see packages/cli/skills/dkg-node/SKILL.md. This skill sits one layer above: it assumes you already know how to call dkg_knowledge_asset_* and focuses on how to call them at scale, repeatedly, without losing data on restart and without fragmenting the graph against parallel producers.

1. The chunking contract (read first)

The daemon's /api/knowledge-assets create + /api/knowledge-assets/<name>/{wm/write,swm/share} loop is the chunked-write API. There is no /api/import/bulk and there will not be one (see ADR 0002 for the rejected-alternative analysis). To push a large graph you call the loop many times, with each call staying under fixed budgets.

These constants are conservative: a 5,000-quad N-Quads payload serialises at roughly 1.0-1.5 MB, well under the daemon's 10 MB MAX_BODY_BYTES cap. Going larger gives no throughput win and risks a 413 on URIs that serialise on the heavy end.

1.1 Known daemon caps and the exact error strings they produce

These are the three hard caps you will hit if you push past the constants above, with the verbatim error text the daemon emits. The HTTP body caps live in packages/cli/src/daemon/http-utils.ts; the gossip cap lives in packages/core/src/constants.ts.

The two /swm/share caps are independent: the 256 KB body cap is on the request you send (URI count × URI length); the 4 MiB gossip cap is on the assertion that ends up in SWM (triples × N-Quads length). It is possible to hit the gossip cap with a single-URI entities array, if the assertion under that root is large enough. entities: "all" triggers it most often because it asks the daemon to gossip every root in one message; for any assertion above ~12k typical-sized triples, expect to split.

In practice this means a robust importer needs two independent halve-and- retry paths on /swm/share: one for 413 (shrink the entities array) and one for 500 (shrink the per-root scope or switch from "all" to explicit batches of N ≤ 400 root URIs). See §5 Error handling for the recipes.

Self-tune from /api/status. Future versions of the daemon advertise their current per-call limits at /api/status under an importLimits block. If present, use those values — they reflect any operator-side tuning. If absent (older daemon), use the constants above verbatim.

2. The write loop

For each logical slice of triples (one slice ≈ one source artefact: one file, one PR, one document, one record group):

POST /api/knowledge-assets   { name, subGraphName, contextGraphId }
POST /api/knowledge-assets/<name>/wm/write   { quads: [...] }   ── one or more times
POST /api/knowledge-assets/<name>/swm/share { entities: [...] }

Reference implementation — see scripts/lib/dkg-daemon.mjs for DkgClient. writeAssertion auto-chunks at a conservative 500-triple default (override via the second-argument batchSize); promote does not chunk — split the entities array yourself before calling it for big imports.

TypeScript sketch

import { DkgClient } from './scripts/lib/dkg-daemon.mjs';

const client = new DkgClient({ token: process.env.DKG_TOKEN });
await client.ensureProject({ id: 'my-corpus', name: 'My Corpus' });
await client.ensureSubGraph(client.cgId, 'code');

async function ensureAssertion(client, body) {
  try {
    await client.request('POST', '/api/knowledge-assets', body);
  } catch (err) {
    if (err.status === 400 && /already exists/i.test(JSON.stringify(err.body ?? err.message))) {
      return;
    }
    throw err;
  }
}

for (const partition of partitions) {                       // one source artefact
  const triples = generateTriples(partition);               // ≤ tens of thousands typical
  const assertionName = `import-${partition.slug}`;
  await ensureAssertion(client, {
    contextGraphId: client.cgId,
    name: assertionName,
    subGraphName: 'code',
  });
  await client.writeAssertion({                             // auto-chunks at 500 quads
    contextGraphId: client.cgId,
    assertionName,
    subGraphName: 'code',
    triples,
  }, { batchSize: 5000 });                                  // bump if your triples are small
  const entities = rootUrisFor(partition);
  for (let i = 0; i < entities.length; i += 400) {          // chunk promote ourselves
    await client.promote({
      contextGraphId: client.cgId,
      assertionName,
      subGraphName: 'code',
      entities: entities.slice(i, i + 400),
    });
  }
}

Python sketch

import os
import requests

PORT = int(os.environ.get('DKG_PORT', '9200'))
TOKEN_PATH = os.path.expanduser('~/.dkg/auth.token')
with open(TOKEN_PATH) as f:
    token = f.read().strip()
H = {'Authorization': f'Bearer {token}', 'Content-Type': 'application/json'}
BASE = f'http://localhost:{PORT}/api'
CHUNK = 5000
ROOT_CHUNK = 400

def ensure_assertion(cg, name, sg):
    res = requests.post(f'{BASE}/knowledge-assets',
                        headers=H, json={'contextGraphId': cg, 'name': name, 'subGraphName': sg})
    if res.status_code == 400 and 'already exists' in res.text.lower():
        return
    res.raise_for_status()

def write_assertion(cg, name, sg, triples, entities):
    ensure_assertion(cg, name, sg)
    for i in range(0, len(triples), CHUNK):
        requests.post(f'{BASE}/knowledge-assets/{name}/wm/write',
                      headers=H, json={'contextGraphId': cg, 'subGraphName': sg,
                                        'quads': triples[i:i+CHUNK]}).raise_for_status()
    for i in range(0, len(entities), ROOT_CHUNK):
        requests.post(f'{BASE}/knowledge-assets/{name}/swm/share',
                      headers=H, json={'contextGraphId': cg, 'subGraphName': sg,
                                        'entities': entities[i:i+ROOT_CHUNK]}).raise_for_status()

3. Resumability via the import manifest

A 10,000-partition import that fails on partition 7,453 must not start over from partition 1. The pattern is a small RDF manifest assertion the importer maintains as it works. The reference implementation lives in scripts/lib/manifest.mjs.

Setup

import { createImportManifest, markPartitionStatus, loadImportManifest, pendingPartitions }
  from './scripts/lib/manifest.mjs';

const importId = 'my-corpus-2026-01-15';
const partitions = enumerateSourceArtefacts().map((p) => p.key);   // strings, one per slice
await createImportManifest({
  client, importId, partitions, subGraphName: 'meta',
});

createImportManifest writes a single urn:dkg:import:<id> assertion to the meta sub-graph listing every partition with initialStatus = "pending". Manifests follow the chunking contract automatically: the import root and every partition URI are promoted to SWM in chunks of ROOT_CHUNK ≤ 400 so a peer node (or this node after a restart) can read the manifest back from SWM to resume — promoting only the root would leave partition triples in WM only.

Per-partition lifecycle

for (const part of partitions) {
  await markPartitionStatus({
    client, importId, partitionKey: part, status: 'in_progress', subGraphName: 'meta',
  });
  try {
    await importOne(part);
    await markPartitionStatus({ client, importId, partitionKey: part, status: 'done', subGraphName: 'meta' });
  } catch (err) {
    await markPartitionStatus({ client, importId, partitionKey: part, status: 'failed', subGraphName: 'meta' });
    throw err;       // or continue, depending on policy
  }
}

Status events are append-only — each markPartitionStatus call writes a fresh StatusEvent triple with a timestamp and promotes BOTH the partition root and the new event root to SWM so peers (or a resume on a different node) see the progress, not just the local WM. (The promote root list is [partIri, evIri] because the new partIri imp:statusEvent evIri edge has subject partIri; promoting only evIri would leave that edge in WM and a peer-side loadImportManifest would never observe the new status.) loadImportManifest resolves the "current" status as the latest event per partition using a standard "max row" SPARQL pattern (FILTER NOT EXISTS against any later event), which avoids the classic SAMPLE+MAX decorrelation foot-gun. This append-only pattern also avoids needing SPARQL DELETE/INSERT and gives you a complete audit trail for free.

Resume

const { partitions: state } = await loadImportManifest({ client, importId, subGraphName: 'meta' });
const pending = pendingPartitions(state);
for (const part of pending) {
  await importOne(part.key);   // pick up where we left off
}

4. Canonical URIs (look-before-mint)

If your import is producing nodes that other producers also produce — files, packages, GitHub PRs, etc. — reuse their URIs, don't fork a new namespace.

Canonical patterns (ADR 0003):

urn:dkg:code:package:<pkgName>                  Package (workspace name)
urn:dkg:code:file:<pkgName>/<relPath>           Source file (relPath ≡ path inside the package)
urn:dkg:github:repo:<owner>/<name>              GitHub repo node
urn:dkg:github:pr:<owner>/<name>/<num>          GitHub PR
urn:dkg:github:issue:<owner>/<name>/<num>       GitHub issue
urn:dkg:import:<id>                             Your own import manifest

Encoding rule: every path segment is encodeURIComponent'd. A file with spaces, @, +, parens, etc. would otherwise produce an IRI Oxigraph rejects with Invalid IRI code point.

Pre-mint check:

1. Compute the normalised slug for your would-be URI (lowercase → ASCII-fold → strip stopwords → hyphenate → ≤60 chars).
2. Call dkg_memory_search with the unnormalised label.
3. If any hit's normalised slug matches yours, reuse the existing URI — prefer hits in higher layers (VM > SWM > WM).
4. Otherwise mint per the pattern above.

If you're producing the canonical code-graph triples for the workspace's own packages, use the helpers in scripts/lib/ontology.mjs rather than redeclaring class/property IRIs.

5. Error handling

HTTP 413 on /wm/write (MAX_BODY_BYTES = 10 MB)

You exceeded the request-body cap with too many N-Quads. Halve and retry:

try {
  await client.writeOne(slice);
} catch (err) {
  if (err.status !== 413) throw err;
  // Halve the chunk size for the next attempt; exponential backoff is fine.
  await client.writeOne(slice.slice(0, slice.length / 2));
  await client.writeOne(slice.slice(slice.length / 2));
}

If you hit 413 frequently on /wm/write, check /api/status for the daemon's current importLimits and tune your CHUNK constant down. Don't paper over it by bumping retries.

HTTP 413 on /swm/share (SMALL_BODY_BYTES = 256 KB)

This is a different 413 from the /wm/write one: the share route uses a smaller body limit because its requests should be small (just root URIs). You hit it by sending too many URIs in entities — roughly 4,000+ URIs at typical lengths. Recovery is to shrink the entities array, not the underlying assertion:

async function promoteRoots(assertion, roots, batchSize = 400) {
  for (let i = 0; i < roots.length; i += batchSize) {
    const batch = roots.slice(i, i + batchSize);
    try {
      await client.request('POST', `/api/knowledge-assets/${assertion}/swm/share`, {
        contextGraphId: client.cgId, subGraphName, entities: batch,
      });
    } catch (err) {
      if (err.status !== 413) throw err;
      // Smaller batch and retry the same range.
      i -= batchSize;
      batchSize = Math.max(50, Math.floor(batchSize / 2));
    }
  }
}

HTTP 500 on /swm/share with "too large for gossip"

The assertion you're promoting serialises to an encoded payload above 4 MiB. This is independent of how many root URIs you pass — even entities: ["<one-uri>"] can trip this if that one root's transitive triple set is bigger than 4 MiB. The error message is verbatim:

HTTP 500 "Promoted assertion too large for gossip
(XXXX KB, limit 4 MB). Promote fewer entities per call."

Recovery is to shrink the per-promote scope, which means: if you were promoting with entities: "all", switch to an explicit URI batch sized so its transitive triples land under 4 MiB. There is no formula because triple fan-out varies; start around 200-400 roots per call for code graphs and 50-100 for prose corpora with long string literals, then halve on the cap.

async function promoteAllInBatches(assertion, allRoots) {
  let batch = 400;
  for (let i = 0; i < allRoots.length; i += batch) {
    try {
      await promoteRoots(assertion, allRoots.slice(i, i + batch));
    } catch (err) {
      if (!/too large for gossip/.test(err.message)) throw err;
      i -= batch;
      batch = Math.max(50, Math.floor(batch / 2));
    }
  }
}

If both 413 and 500-gossip fire on the same import, you need both recovery loops — they're orthogonal. As of PR #4 the daemon ships an in-process async promote queue (POST /api/knowledge-assets/<name>/swm/share-async) that removes both failure modes by promoting one classifying-and-retrying job at a time in the background. See §6 Async promote queue for the new contract — for a fresh importer, prefer the async path and skip both halve-and-retry loops above.

HTTP 401 / 403

Token problem, not a chunking problem. See dkg-node/SKILL.md §4 "If you get 401 or 403 on a protected route, diagnose in this order" — call GET /api/agent/identity to confirm who the daemon thinks you are.

Connection errors / 5xx

Standard retry with exponential backoff. The daemon does not implement idempotency tokens. wm/write is safe to retry with the same payload (duplicate triples are deduped server-side), and retrying swm/share is safe too. Raw POST /api/knowledge-assets (create) returns HTTP 400 when the assertion already exists; higher-level helpers can normalize that into idempotent success by treating an already exists response as reuse.

Daemon restart mid-import

WM survives restarts under the Oxigraph persistence contract (the archived incident report characterises the bug fixed in OriginTrail/dkg#636-639). On resume, loadImportManifest gives you the "where was I?" answer; if a particular assertion's WM state is partial, you can either:

• Retry the assertion — treat POST /api/knowledge-assets (create) "already exists" as reuse (or call a helper that does), then re-run wm/write to re-assert the same triples without duplication.
• Discard the partial assertion with POST /api/knowledge-assets/<name>/wm/discard and start over from your last done partition.

HTTP 400 on finalize/publish with Rule 4: rootEntity ... already exists

This is the #1 trap for "real-world" graph importers — Wikidata, schema.org, Graphify-style code graphs, EPCIS event streams, anything where the same subject URI legitimately appears across many logical artefacts. It fires from the daemon's autoPartition step (during finalize: true on create, or as part of /api/knowledge-assets/{name}/vm/publish) and looks like:

HTTP 400 "Rule 4 violation: rootEntity <http://www.wikidata.org/entity/Q2831>
already exists as the root of knowledge collection 17 in context graph 4.
Use POST /api/update to extend the existing knowledge collection."

The rule: every Knowledge Asset (KA) within a context graph has exactly one root entity, and a given subject URI can be the root of at most one KA per CG. Multiple KAs sharing a root would make on-chain ownership / attribution ambiguous, so the contract enforces uniqueness. The error message's /api/update hint is correct if you want to extend the existing KA — but for a bulk import producing many KAs that mention the same entities ("Michael Jackson appears in 500 of my 5,000 album KAs"), updating isn't what you want. You want each KA to have its own unrelated root.

The fix — partition-scoped blank-node rewrite. Before submitting quads for partition N, rewrite every Wikidata / external URI to a partition-scoped blank node and anchor them under a single, unique-per-partition root:

function buildPartitionQuads(partitionIdx, rawQuads, anchorUri) {
  // 1. Mint one partition-scoped anchor — this becomes the KA's sole root.
  //    URI is unique per partition; blank nodes underneath it inherit
  //    partition scope so Q2831 in partition 17 != Q2831 in partition 18
  //    from the contract's perspective.
  const anchor = `<${anchorUri}>`;     // e.g. urn:dkg:miles-stress:partition:17
  const blankFor = new Map();          // subject-URI -> deterministic _:bN
  let bnCounter = 0;
  const blankNodeFor = (uri) => {
    if (!blankFor.has(uri)) {
      // Deterministic skolem-ish label keeps the rewrite repeatable across
      // resume runs without coordinating state.
      blankFor.set(uri, `_:p${partitionIdx}_b${bnCounter++}`);
    }
    return blankFor.get(uri);
  };

  // 2. Rewrite every non-anchor URI in the subject (and object, when an IRI)
  //    position to its partition-scoped blank node.
  //
  // Detect IRIs generically via the RFC 3986 scheme grammar rather than
  // hard-coding a scheme list. Earlier drafts checked only `http` / `urn:`,
  // which silently misses valid RDF IRIs that use other schemes — `did:`,
  // `ipfs:`, `tag:`, `file:`, plain-IRI imports etc. — and lets colliding
  // root entities leak through to keep hitting Rule 4 on subsequent
  // partitions. If your parser exposes `term.termType === 'NamedNode'`,
  // prefer that over the regex.
  const ABS_IRI = /^[A-Za-z][A-Za-z0-9+\-.]*:/;
  const isIri = (s) => ABS_IRI.test(s);
  const out = [];
  for (const { s, p, o } of rawQuads) {
    const subj = isIri(s) ? blankNodeFor(s) : s;
    const obj  = (o.kind === 'iri' && o.value !== anchorUri)
      ? blankNodeFor(o.value)
      : serializeObject(o);
    out.push(`${subj} <${p}> ${obj} .`);
  }

  // 3. Link the anchor to every rewritten root with `<anchor> stress:contains <_:bN>`
  //    so the KA's transitive triple set is reachable from the single root.
  for (const blank of new Set(blankFor.values())) {
    out.push(`${anchor} <urn:dkg:stress:contains> ${blank} .`);
  }
  out.push(`${anchor} a <urn:dkg:stress:Partition> .`);
  return out;
}

The result: each KA has one root (the anchor), every Wikidata URI inside appears only as a blank-node label, and partitions sharing entities don't collide. Battle-tested in scripts/testnet-publish-stress/publish-loop.mjs (Base Sepolia, miles-publish-stress-26may, 5000-partition stress run); see that file for a full reference implementation including pace-control, checkpointing and retry-with-unique-name.

If your data has a natural "real" root that's already unique per artefact (e.g. an EPCIS event ID, a GitHub PR URL, a build ID), use that as the anchor instead of minting a synthetic one — the blank-node rewrite still applies for everything under it.

Both the synchronous per-KA /api/knowledge-assets/{name}/vm/publish and the async promote queue run through autoPartition, so this trap exists on both paths. Fix it at the importer level, before any quads reach the daemon.