Data Model & Persistence

Every source the pipeline screens — sanctions lists, BHRRC, Phase-2 deep research, and the legacy engine — persists into the same Issue node shape, reconciled additively against whatever already exists for the organisation. This page covers the Neo4j schema, provenance fields, the corrected organisation-status facts, additive-merge reconciliation, concurrency correctness, article materialisation, and the read path. See Issue Research Pipeline for how persistence fits into the pipeline as a whole.

Neo4j schema

An Issue is a standalone node connected to its evidence (Article, via REFERENCES) and to the organisation(s) it is reported against (HAS_ISSUE, an edge carrying which ESG satellite the issue belongs to):

type Issue @node {
  id: ID! @id
  title: String!
  text: String!
  date: DateTime!
  sentiment: Float!
  involvement: IssueInvolvementType!
  source: String
  tier: IssueTier
  slaveryPractice: SlaveryPractice
  articles: [Article!]! @relationship(type: "REFERENCES", direction: OUT)
  organisation: [Organisation!]!
    @relationship(type: "HAS_ISSUE", direction: IN, properties: "IssueProperties")
}

type IssueProperties @relationshipProperties {
  satelliteMode: SatelliteMode!
}

Provenance fields

Three fields distinguish which pipeline authored an issue, and were added additively alongside the existing legacy schema — every field is nullable, so pre-existing issues and pre-existing reads are unaffected:

source != null is also the predicate the reconciliation pool uses to select which existing issues are eligible merge targets — see below.

Organisation status fields

Additive-merge reconciliation (never delete)

PersistIssuesResearch is the single write path every source funnels through — BHRRC and the sanctions profiles call it via PersistBhrrcIssues/directly, and Phase 2 calls it per cell. It never deletes or overwrites an existing Issue; a re-run can only add new Issue nodes or add new article REFERENCES to ones that already exist (ADR-0022).

1. Drop and default. Incoming ConsolidatedIssues with involvement === 'NONE' are filtered out before anything else (the Neo4j enum has no NONE value). source defaults to 'native' when the caller doesn't supply one.
2. Load the pool. The org's existing native-authored issues (source != null) are read — either fetched fresh (loadIssueReconcilePool) or supplied by the caller as a shared per-phase pool (see concurrency).
3. Cluster. clusterIssues (an LLM clustering call built from buildReducePrompt) groups existing ∪ incoming refs into clusters.
4. Decide per issue. planReconcile turns each incoming issue's cluster membership into one of three verbs:

A deterministic year-compatibility guard (yearCompatible) prevents over-merging: two timeframes are compatible unless both carry known years that are disjoint — an unknown/absent year on either side never blocks a merge, but a clear year mismatch (e.g. a 2020 incident vs. a 2024 one) forces create instead of silently folding two different incidents together.

If the clustering call itself throws, reconciliation degrades additively, not silently: every incoming issue is created as new (never dropped), and a reconcile-clustering-failed: … entry is added to the returned errors so the degrade is operator-visible. The worst case under this design is a duplicate issue; a dropped one is never acceptable for a compliance screening pipeline (ADR-0022). A per-article idempotency backstop (IssueRepository.connectArticles) also holds regardless of clustering behaviour — it reads currently-connected article ids, diffs in JS, and connects only what's new in a single MERGE-backed mutation.

Concurrency correctness

Two mechanisms make the additive-merge design safe under the harness's actual concurrency, both described in ADR-0022:

• IssueReconcilePool (helpers/issue-reconcile-pool.ts) is fetched once per phase — not once per persist — via loadIssueReconcilePool, and threaded through every persist that phase enqueues (~2 reads per run, versus ~10 under the original per-persist re-fetch design). Each persist reconciles against pool.rows and calls pool.add() for every issue it creates, so a later persist in the same phase sees an earlier persist's creates without re-reading Neo4j.
• createPersistChain (p-limit(1), helpers/serial-persist-chain.ts) serialises every persist within a phase into one queue: persists run one-at-a-time in enqueue order (so each can safely read-cluster-write against the pool's current state), while each source's own research still runs concurrently and only enqueues once it resolves. This is why Phase 1 surfaces issues source-by-source rather than all at once — sources finish research in parallel, but their persists land through the same serial gate.

The in-memory pool mutation is only safe because of that serialisation — no two persists in a phase ever touch pool.rows concurrently. Across phases the guarantee is different: Phase 2 loads its own fresh pool rather than reusing Phase 1's, because Phase 1 and Phase 2 are separate Inngest steps — a step retry re-runs the entire phase against current DB state (ADR-0022), so reusing another phase's in-memory pool would reconcile against a stale or foreign snapshot instead of what's actually in Neo4j when the step starts.

Article materialisation

MaterialiseArticlesFromChunks turns the native pipeline's own search chunks into Article nodes — there is no Diffbot extraction on the native path. For each chunk it calls CreateArticle with source: ArticleSource.Tavily and extract: false, falling back to the chunk's own title/content rather than re-fetching the page. It returns a url → articleId map, keyed by normalised URL, which the persist step (PersistIssuesResearch) uses to resolve each consolidated issue's source URLs into article ids for the REFERENCES connect. Per-chunk failures are isolated (Promise.all over try/catch per chunk) — one bad URL never drops the rest of the cell's article links.

Read path