simplificaiton

README.md

@@ -26,8 +26,8 @@ namespace is `http://mu.semte.ch/vocabularies/reporting/` (`rep:`).
 
 The run: check → seed (`SELECT DISTINCT ?s`, paged) → columns (per group, per
 chunk of subjects) → assemble (dedup + `rep:collect`) → CSV → register file
-+ report resource. The job is a `cogs:Job` + `task:Task` we write ourselves,
-with `mu-call-scope-id` set so jobs-controller ignores it.
++ report resource. One run per question; the loop stops after `run_report`
+returns.
 
 ## One identity: mu-authorization's
 
@@ -52,7 +52,8 @@ POST /assistant/conversations/:id/turns      { content }           → 202 { id
 service (profiles, lookups, spec, validate, repair, run), and the answer in
 Dutch comes back with a trace of the tool calls. The CSV lands in
 `data/files/`. `history` carries the earlier turns for follow-ups; the
-conversation lives on the caller's side, the service stores nothing.
+conversation lives on the caller's side, the service stores nothing. One
+question makes one report; the loop stops after `run_report`.
 
 `/assistant` is the chat. The turn module in
 `src/chat/` is generic — it imports nothing from `src/llm/` or `src/runner/`
@@ -67,15 +68,13 @@ that way. Failure is a message, never a silent loader.
 `/ask` needs `LLM_BASE_URL` set, and answers 503 without it. A full-org report
 run takes minutes.
 
-The job is a resource: read it at `/jobs/:id`, or watch the dashboard.
-
 ### The internal tools
 
-The loop drives eight tool handlers (`src/llm/tools.js`): `list_profiles`,
-`describe_profile`, `validate_spec`, `lookup_values`, `preview_queries`,
-`run_report`, `report_status`, `export_spec`. They are machinery, not an
+The loop drives four tool handlers (`src/llm/tools.js`): `list_profiles`,
+`describe_profile`, `validate_spec`, `run_report`. They are machinery, not an
 external surface — the tool trace in the answer is the only thing a caller
-sees of them.
+sees of them. All LLM-facing text (system prompt, profile menu, tool
+descriptions, validator errors) is English; the answer the user sees is Dutch.
 
 ## Environment
 

app.js

@@ -1,7 +1,7 @@
-import { app } from 'mu';
+import { app, errorHandler } from 'mu';
 import express from 'express';
 import { loadProfiles } from './src/runner/profile.js';
-import { buildSession, stripStoreError } from './src/db.js';
+import { sessionQuery, sessionUpdate, stripStoreError } from './src/db.js';
 import { ask } from './src/llm/loop.js';
 import { mountChat } from './src/chat/index.js';
 import { reportAssistant } from './src/report-assistant.js';
@@ -41,7 +41,7 @@ app.post('/ask', express.json(), async (req, res) => {
       ...(Array.isArray(history) ? history : []),
       { role: 'user', content: question },
     ];
-    const session = await buildSession();
+    const session = { query: sessionQuery, update: sessionUpdate };
     const { text, trace } = await ask(turns, profiles, session);
     res.json({ answer: text, toolCalls: trace });
   } catch (e) {
@@ -49,3 +49,5 @@ app.post('/ask', express.json(), async (req, res) => {
     res.status(502).json({ error: stripStoreError(e.message) });
   }
 });
+
+app.use(errorHandler);

src/db.js

@@ -1,20 +1,30 @@
-import { query as muQuery, update as muUpdate,
-         sparqlEscapeUri, sparqlEscapeString, sparqlEscapeInt,
-         sparqlEscapeDateTime, sparqlEscapeDate, sparqlEscapeBool } from 'mu';
+// Thin re-export of mu's SPARQL helpers so the runner files share one import
+// site. mu's query/update attach the caller's session from the request
+// context, and mu-authorization decides graphs and visibility; the service
+// never touches an auth header.
+import {
+  query as muQuery,
+  update as muUpdate,
+  sparqlEscapeUri,
+  sparqlEscapeString,
+  sparqlEscapeInt,
+  sparqlEscapeDateTime,
+  sparqlEscapeDate,
+  sparqlEscapeBool,
+} from 'mu';
 
-export { sparqlEscapeUri, sparqlEscapeString, sparqlEscapeInt,
-         sparqlEscapeDateTime, sparqlEscapeDate, sparqlEscapeBool };
-
-export async function sessionQuery(sparql)  { return muQuery(sparql); }
-export async function sessionUpdate(sparql) { return muUpdate(sparql); }
+export const sessionQuery = muQuery;
+export const sessionUpdate = muUpdate;
+export {
+  sparqlEscapeUri,
+  sparqlEscapeString,
+  sparqlEscapeInt,
+  sparqlEscapeDateTime,
+  sparqlEscapeDate,
+  sparqlEscapeBool,
+};
 
 // The store's error text may name graphs or internals we do not pass on.
 export function stripStoreError(text) {
   return String(text).split('\n')[0].slice(0, 300);
 }
-
-// The request context lives for the whole run, also past the 202 a chat turn
-// answers with, so every write stays the caller's.
-export async function buildSession() {
-  return { query: sessionQuery, update: sessionUpdate };
-}

src/llm/describe.js

@@ -32,7 +32,7 @@ async function loadCodeList(profile, f, key) {
   try {
     values = await queryCodeList(f.class);
   } catch (e) {
-    // the read failed; report nothing, the LLM will use lookup_values
+    // the read failed; report nothing, the LLM will use rep:anyOf with words
     values = [];
   }
   cache.set(key, { values, at: Date.now() });
@@ -54,35 +54,36 @@ async function queryCodeList(type) {
 
 export function describeProfile(profile, codeLists = {}) {
   const lines = [];
-  lines.push('Een spec is Turtle, precies in deze vorm (kopieer deze prefixes):');
+  lines.push('A spec is Turtle, in exactly this shape (copy these prefixes):');
   lines.push('@prefix rep:  <http://mu.semte.ch/vocabularies/reporting/> .');
   lines.push('@prefix sh:   <http://www.w3.org/ns/shacl#> .');
   lines.push('@prefix rdfs: <http://www.w3.org/2000/01/rdf-schema#> .');
   lines.push('@prefix dct:  <http://purl.org/dc/terms/> .');
   lines.push('');
-  lines.push('<http://data.lblod.info/id/report-specs/<uw-id>> a rep:ReportSpec , sh:NodeShape ;');
+  lines.push('<http://data.lblod.info/id/report-specs/<your-id>> a rep:ReportSpec , sh:NodeShape ;');
   lines.push('  dct:title "..." ;');
   lines.push(`  rep:profile <${profile.uri}> ;`);
-  lines.push('  sh:targetClass <de rdf:type uit de lijst hieronder> ;');
-  lines.push('  sh:property [ ... ] ;   # filters, optioneel');
+  lines.push('  sh:targetClass <the rdf:type from the list below> ;');
+  lines.push('  sh:property [ ... ] ;   # filters, optional');
   lines.push('  rep:columns ( [ sh:path ( ... ) ; rdfs:label "..." ] ... ) .');
-  lines.push('Elke kolom eindigt op een waardeveld. rep:self als sh:path geeft de eigen URI.');
+  lines.push('Every column ends on a value field. Use rep:self as sh:path for the subject URI.');
   lines.push('');
-  lines.push('sh:targetClass is de volledige URI tussen haakjes achter de entiteit hieronder.');
-  lines.push('Gebruik in sh:path de predicate-URI die achter elke veldnaam staat, niet de veldnaam zelf.');
-  lines.push('Een omgekeerde hop (het veld staat "(omgekeerd)" achter) schrijf je in de lijst als [ sh:inversePath <predicate> ] .');
-  lines.push('Een datumfilter is sh:minInclusive of sh:maxInclusive met "jjjj-mm-dd"^^xsd:dateTime');
-  lines.push('(kopieer ook het xsd-prefix hierboven niet: voeg @prefix xsd: <http://www.w3.org/2001/XMLSchema#> toe).');
+  lines.push('sh:targetClass is the full URI in brackets after the entity below.');
+  lines.push('In sh:path, use the predicate URI shown after each field name, not the field name itself.');
+  lines.push('An inverse hop (the field shows "(inverse)") goes in the list as [ sh:inversePath <predicate> ] .');
+  lines.push('When two entities share one sh:targetClass, pick one with rep:entity <the URI after the entity>.');
+  lines.push('A date filter is sh:minInclusive or sh:maxInclusive with "yyyy-mm-dd"^^xsd:dateTime');
+  lines.push('(the xsd prefix above is not copied: add @prefix xsd: <http://www.w3.org/2001/XMLSchema#>).');
   lines.push('');
   for (const shape of profile.shapes) {
-    lines.push(`${shape.label || shape.targetClass} — sh:targetClass <${shape.targetClass}>`);
+    lines.push(`${shape.label || shape.targetClass} — sh:targetClass <${shape.targetClass}> ; rep:entity <${shape.uri}>`);
     for (const f of shape.fields) {
       const key = `${shape.uri}|${f.path}|${f.inverse ? 1 : 0}`;
       const kind = f.datatype ? datatypeName(f.datatype)
         : f.class ? `→ ${conceptLabel(f.class, codeLists[key])}`
         : f.node ? `→ ${linkedLabel(profile, f.node)}`
         : '?';
-      const dir = f.inverse ? ' (omgekeerd)' : '';
+      const dir = f.inverse ? ' (inverse)' : '';
       const pathShown = typeof f.path === 'string' ? f.path : '?';
       lines.push(`  ${f.name || pathShown}  sh:path <${pathShown}>${dir}  ${kind}`);
     }

src/llm/lookup.js

@@ -1,110 +0,0 @@
-// Search a code list too long to inline: regex on str(?label), so
-// language-tagged labels match. Returns the matches, the total, and an
-// `exact` flag when a label equals the term (case aside).
-
-import { sessionQuery, sparqlEscapeString } from '../db.js';
-import { fieldsOf, entityForClass } from '../runner/profile.js';
-
-const LIMIT = 25;
-
-// field is "shapeLabel.fieldLabel" as the LLM writes it, or a predicate URI.
-// We resolve it to the field's sh:class to know what type to search.
-export async function lookupValues(profile, term, fieldSpec) {
-  const type = resolveClass(profile, fieldSpec);
-  if (!type) return { total: 0, exact: null, matches: [], error: `field "${fieldSpec}" not found in profile "${profile.title}"` };
-
-  // Search within the class first. A common word ("gemeente") also sits in
-  // hundreds of unrelated labels; an unscoped scan drowns the code.
-  let rows = await labelQuery(term, type);
-  // Code lists sometimes type their entries differently than the profile
-  // says (MAR codes are ext:supervision/Nomenclature, the profile says
-  // skos:Concept). Fall back to a scan over everything.
-  if (!rows.length) rows = await labelQuery(term, null);
-  // Order: an exact label (case aside) first, then the rest as the store
-  // ordered them. "Gent" must surface before "AGB Erfgoed Gent".
-  const lower = term.toLowerCase();
-  const exactRow = rows.find(b => b.label.toLowerCase() === lower);
-  const rest = rows.filter(b => b !== exactRow);
-  const ordered = exactRow ? [exactRow, ...rest] : rest;
-  let matches = ordered.slice(0, LIMIT);
-  if (!matches.length) {
-    const byCode = await codeQuery(profile, type, term);
-    matches = byCode;
-  }
-  const exact = matches.find(m => m.label.toLowerCase() === term.toLowerCase());
-  return {
-    total: matches.length >= LIMIT ? `${LIMIT}+` : matches.length,
-    exact: exact ? exact.uri : null,
-    matches,
-  };
-}
-
-async function labelQuery(term, type) {
-  const q = `SELECT DISTINCT ?uri ?label WHERE {
-    ?uri a ${type ? `<${type}>` : '?anyType'} ;
-         <http://www.w3.org/2004/02/skos/core#prefLabel> ?label .
-    FILTER(REGEX(str(?label), ${sparqlEscapeString(escapeRegex(term))}, "i"))
-  } ORDER BY ?label LIMIT ${LIMIT * 8}`;
-  const r = await sessionQuery(q);
-  return r.results.bindings.map(b => ({ label: b.label.value, uri: b.uri.value }));
-}
-
-// MAR codes: the label says "MAR7300 - …" but the searchable code sits on
-// ext:supervision/nomenclatureCode. One extra query, only when the label
-// search came up empty.
-async function codeQuery(profile, type, term) {
-  const q = `SELECT ?uri ?label WHERE {
-    ?uri <http://mu.semte.ch/vocabularies/ext/supervision/nomenclatureCode> ?code ;
-         <http://www.w3.org/2004/02/skos/core#prefLabel> ?label .
-    FILTER(CONTAINS(LCASE(str(?code)), LCASE(${sparqlEscapeString(term)})))
-  } ORDER BY ?label LIMIT ${LIMIT}`;
-  try {
-    const r = await sessionQuery(q);
-    return r.results.bindings.map(b => ({ label: b.label.value, uri: b.uri.value }));
-  } catch {
-    return [];
-  }
-}
-
-function resolveClass(profile, fieldSpec) {
-  if (fieldSpec.includes('/')) return null;
-  // "shapeLabel.fieldLabel" → find the field carrying sh:class
-  const dot = fieldSpec.lastIndexOf('.');
-  if (dot < 0) {
-    // bare predicate URI
-    for (const shape of profile.shapes) {
-      for (const f of shape.fields) {
-        if (f.path === fieldSpec && f.class) return f.class;
-      }
-    }
-    return null;
-  }
-  const shapeLabel = fieldSpec.slice(0, dot);
-  const fieldLabel = fieldSpec.slice(dot + 1);
-  const shape = profile.shapes.find(s => s.label === shapeLabel);
-  if (!shape) return null;
-  const f = shape.fields.find(x => x.name === fieldLabel);
-  // sh:node marks a link, not a code list; the searchable type is the linked
-  // shape's sh:targetClass ("melding.bestuurseenheid" → besluit:Bestuurseenheid).
-  if (f?.class) return f.class;
-  if (f?.node) {
-    const linked = profile.shapes.find(s => s.uri === f.node);
-    return linked?.targetClass || null;
-  }
-  // The model sometimes writes shape.field on a linked shape
-  // ("bestuurseenheid.naam"); resolve it to that shape's target class.
-  for (const s of profile.shapes) {
-    if (s.label === shapeLabel || shortType(s.targetClass) === shapeLabel) {
-      return s.targetClass;
-    }
-  }
-  return null;
-}
-
-function shortType(cls) {
-  return cls.split('#').pop() || cls.split('/').pop();
-}
-
-function escapeRegex(s) {
-  return String(s).replace(/[.*+?^${}()|[\]\\]/g, '\\$&');
-}

src/llm/loop.js

@@ -4,7 +4,7 @@
 
 import { buildTools } from './tools.js';
 
-const MAX_REPAIR_ROUNDS = Number(process.env.MAX_REPAIR_ROUNDS || 3);
+const MAX_ROUNDS = Number(process.env.MAX_ROUNDS || 12);
 const LLM_BASE_URL = process.env.LLM_BASE_URL;
 const LLM_MODEL = process.env.LLM_MODEL;
 const LLM_API_KEY = process.env.LLM_API_KEY;
@@ -17,21 +17,22 @@ The profile lists what can be asked. Compose paths by chaining fields.
 
 Rules:
 - A column must end on a value, not on a link. Chain one more hop.
+- When two entities share one sh:targetClass, the spec also needs rep:entity
+  with that entity's URI; describe_profile shows it after each entity.
 - Filters go in sh:property. Columns go in rep:columns, in the order you want.
 - To match a topic rather than a value, list the Dutch words and compounds you
   would expect and put them all in rep:anyOf.
-- If a lookup returns nothing, or more than one plausible match, ask the user
-  in Dutch, offering labels. Never guess and never show a URI.
 - You cannot count, sort, take the first N, or compare two subjects. Say so.
 - The report's title is the dct:title you write in the spec. There is no other
   title input.
-- One conversation may ask for several reports, one run_report per report.
+- One question makes one report. Write one spec, validate it, run it, then stop.
+  Do not write a second report for the same question.
 
 Always prefix the spec with @prefix lines. Use the prefixes the profile declares.
 Write the spec as one Turtle block. Then call validate_spec. Fix what it says,
-up to three rounds. Then call run_report.
+up to three rounds. Then call run_report. After run_report returns, stop.
 
-Earlier turns are context only. Answer the last one.`;
+Reply to the user in Dutch. Earlier turns are context only. Answer the last one.`;
 
 // ask(questionOrMessages, profiles, session) → { text, trace }
 // A string is one question; an array is the conversation so far,
@@ -46,7 +47,7 @@ export async function ask(questionOrMessages, profiles, session) {
     : questionOrMessages.filter(m => m.role !== 'system');
   const messages = [{ role: 'system', content: SYSTEM_PROMPT }, ...turns];
 
-  for (let round = 0; round < 12; round++) {
+  for (let round = 0; round < MAX_ROUNDS; round++) {
     const reply = await chat(messages);
     messages.push(reply.message);
     const calls = reply.message.tool_calls || [];
@@ -123,15 +124,6 @@ function parseArgs(s) {
   try { return JSON.parse(s); } catch { return {}; }
 }
 
-// messagesFor(question) — a fresh conversation: system prompt plus the user
-// question.
-export function messagesFor(question) {
-  return [
-    { role: 'system', content: SYSTEM_PROMPT },
-    { role: 'user', content: question },
-  ];
-}
-
 // The tool definitions sent to the model.
 import { toolSchemas } from './schemas.js';
 function toolDefs() {

src/llm/schemas.js

@@ -1,4 +1,4 @@
-// The JSON schemas of the eight tools. The descriptions are what the LLM
+// The JSON schemas of the four tools. The descriptions are what the LLM
 // reads to decide which tool to call.
 
 export function toolSchemas() {
@@ -26,54 +26,14 @@ export function toolSchemas() {
         required: ['spec'],
       },
     },
-    {
-      name: 'lookup_values',
-      description: 'Search a code list (too long to inline) for a term. Returns matches, a total, and an exact flag when a label equals the term.',
-      inputSchema: {
-        type: 'object',
-        properties: {
-          profile_id: { type: 'string' },
-          field: { type: 'string', description: 'The field as "entityLabel.fieldLabel", e.g. "melding.type dossier"' },
-          term: { type: 'string', description: 'The Dutch word or name to search for' },
-        },
-        required: ['profile_id', 'field', 'term'],
-      },
-    },
-    {
-      name: 'preview_queries',
-      description: 'Show the SPARQL queries the runner would run for a valid spec, as text. Lets you check before running.',
-      inputSchema: {
-        type: 'object',
-        properties: { spec: { type: 'string' } },
-        required: ['spec'],
-      },
-    },
     {
       name: 'run_report',
-      description: 'Run a valid spec and return the report URI, the row count and the file URI. Runs as the caller, so it sees only what they may see. Waits as long as the run keeps answering; gives up after RUN_TIMEOUT seconds of silence. The report is named by the dct:title in the spec.',
+      description: 'Run a valid spec and return the report URI, the row count and the file URI. Runs as the caller, so it sees only what they may see. Waits as long as the run keeps answering; gives up after RUN_TIMEOUT seconds of silence. The report is named by the dct:title in the spec. Call this once per question, after validate_spec returns ok.',
       inputSchema: {
         type: 'object',
         properties: { spec: { type: 'string' } },
         required: ['spec'],
       },
     },
-    {
-      name: 'report_status',
-      description: 'Read the status of a job by its URI, and the file once it succeeds.',
-      inputSchema: {
-        type: 'object',
-        properties: { job_uri: { type: 'string' } },
-        required: ['job_uri'],
-      },
-    },
-    {
-      name: 'export_spec',
-      description: 'Return the Turtle spec that produced a report, by report URI.',
-      inputSchema: {
-        type: 'object',
-        properties: { report_uri: { type: 'string' } },
-        required: ['report_uri'],
-      },
-    },
   ];
 }

src/llm/tools.js

@@ -1,15 +1,13 @@
-// The eight tools the LLM calls. They return a job URI, a status, a count, a
-// link, column names or an error we wrote — never a cell value. The shape
-// every tool returns is { content: [{ type: 'text', text }] }, which the loop
-// passes straight back to the model.
+// The tools the LLM calls: list_profiles, describe_profile, validate_spec,
+// run_report. They return a report URI, a count, or an error we wrote —
+// never a cell value. The shape every tool returns is
+// { content: [{ type: 'text', text }] }, which the loop passes straight back
+// to the model.
 
 import { checkSpec } from '../runner/check.js';
 import { parseSpec } from '../runner/spec.js';
-import { seedPageQuery } from '../runner/seed.js';
 import { describeProfile, codeListValues } from './describe.js';
 import { run, slug } from '../runner/run.js';
-import { lookupValues } from './lookup.js';
-import { readSpec } from '../runner/csv.js';
 
 const MAX_PATH_DEPTH = Number(process.env.MAX_PATH_DEPTH || 8);
 const RUN_TIMEOUT = Number(process.env.RUN_TIMEOUT || 60) * 1000;
@@ -42,24 +40,6 @@ export function buildTools(profiles, session) {
     return text(errors.length ? errors[0] : 'ok');
   }
 
-  async function lookup_values({ profile_id, field, term }) {
-    const p = findProfile(profile_id);
-    if (!p) return text(`no profile <${profile_id}>.`);
-    const res = await lookupValues(p, term, field);
-    return text(JSON.stringify(res));
-  }
-
-  async function preview_queries({ spec }) {
-    let parsed;
-    try { parsed = parseSpec(spec); }
-    catch (e) { return text(`the spec did not parse: ${e.message}`); }
-    const p = findProfile(parsed.profileUri);
-    if (!p) return text(`no profile <${parsed.profileUri}>.`);
-    const errors = checkSpec(parsed, p, MAX_PATH_DEPTH, profiles);
-    if (errors.length) return text(`the spec is not valid: ${errors[0]}`);
-    return text(buildQueriesText(parsed, p));
-  }
-
   async function list_profiles() {
     return text(profileList().map(x => `${x.title}  <${x.id}>`).join('\n'));
   }
@@ -77,7 +57,7 @@ export function buildTools(profiles, session) {
       const name = parsed.title || 'report';
       await session.onReportStart?.({ title: name, fileName: `${slug(name)}.csv` });
       const waiter = newRunWaiter(RUN_TIMEOUT);
-      const running = run(session.query, session.update, parsed, p, name, null, { spec }, waiter.tick);
+      const running = run(session.query, session.update, parsed, p, name, { spec }, waiter.tick);
       // The end callback also fires after the wait gave up: the run is not
       // cancelled, the file arrives when it arrives. The .catch keeps a late
       // failure from crashing the process.
@@ -91,43 +71,11 @@ export function buildTools(profiles, session) {
     }
   }
 
-  async function report_status({ job_uri }) {
-    const r = await session.query(
-      `SELECT ?status WHERE { <${job_uri}> <http://www.w3.org/ns/adms#status> ?status } LIMIT 1`);
-    const b = r.results.bindings[0];
-    if (!b) return text(`no job <${job_uri}>.`);
-    const status = b.status.value.split('/').pop();
-    let link = '';
-    if (status === 'success') {
-      const f = await session.query(
-        `SELECT ?file WHERE { <${job_uri}> <http://redpencil.data.gift/vocabularies/tasks/resultsContainer> ?c . ?c <http://redpencil.data.gift/vocabularies/tasks/hasFile> ?file } LIMIT 1`);
-      const fb = f.results.bindings[0];
-      if (fb) link = ` file <${fb.file.value}>.`;
-    }
-    return text(`status: ${status}.${link}`);
-  }
-
-  async function export_spec({ report_uri }) {
-    const r = await session.query(
-      `SELECT ?specFile WHERE { <${report_uri}> <http://mu.semte.ch/vocabularies/ext/specFile> ?specFile } LIMIT 1`);
-    const b = r.results.bindings[0];
-    if (!b) return text(`no spec stored on <${report_uri}>.`);
-    try {
-      return text(readSpec(b.specFile.value));
-    } catch {
-      return text(`the spec file ${b.specFile.value} is not on disk anymore.`);
-    }
-  }
-
   const handlers = new Map([
     ['list_profiles', list_profiles],
     ['describe_profile', describe_profile],
     ['validate_spec', validate_spec],
-    ['lookup_values', lookup_values],
-    ['preview_queries', preview_queries],
     ['run_report', run_report],
-    ['report_status', report_status],
-    ['export_spec', export_spec],
   ]);
 
   return { handlers };
@@ -135,11 +83,6 @@ export function buildTools(profiles, session) {
 
 function text(s) { return { content: [{ type: 'text', text: s }] }; }
 
-function buildQueriesText(spec, profile) {
-  const seed = seedPageQuery(spec, Number(process.env.SEED_PAGE_SIZE || 5000), 0);
-  return ['-- seed query', seed, ''].join('\n');
-}
-
 // Waits for a report run while it makes progress. tick restarts the silence
 // clock; idleMs without a single tick rejects. The run is never cancelled:
 // finish's handlers only pass its own settlement on, so a run that answers

src/runner/check.js

@@ -1,4 +1,4 @@
-import { entityForClass, fieldsOf } from './profile.js';
+import { fieldsOf } from './profile.js';
 import { NUMERIC_DATATYPES } from './assemble.js';
 
 const SH = 'http://www.w3.org/ns/shacl#';
@@ -27,21 +27,46 @@ function profileNamed(spec, knownProfiles) {
   return null;
 }
 
-// The target class is in the profile
-function targetClass(spec, profile) {
+// The spec names an entity we have: by rep:entity, or by a targetClass only
+// one entity carries. A class two entities share is refused with both names,
+// so the model can add rep:entity instead of guessing.
+function startEntity(spec, profile) {
   if (!spec.targetClass) {
     return `no sh:targetClass. In "${profile.title}" you can list: ${shapeList(profile)}.`;
   }
-  if (!entityForClass(profile, spec.targetClass)) {
+  if (spec.entity) {
+    const s = profile.shapes.find(x => x.uri === spec.entity);
+    if (!s) {
+      return `no entity <${spec.entity}> in "${profile.title}". It has: ${shapeList(profile)}.`;
+    }
+    if (s.targetClass !== spec.targetClass) {
+      return `rep:entity <${spec.entity}> is a <${s.targetClass}>, but sh:targetClass says <${spec.targetClass}>. Make them agree.`;
+    }
+    return null;
+  }
+  const matches = profile.shapes.filter(s => s.targetClass === spec.targetClass);
+  if (matches.length === 1) return null;
+  if (!matches.length) {
     return `"${profile.title}" cannot list <${spec.targetClass}>. It has: ${shapeList(profile)}.`;
   }
-  return null;
+  const names = matches.map(m => `${m.label || m.uri} <${m.uri}>`).join(', ');
+  return `<${spec.targetClass}> is the class of ${matches.length} entities: ${names}. Add rep:entity <...> to say which one.`;
 }
 
 function shapeList(profile) {
   return profile.shapes.map(s => s.label || s.targetClass).join(', ');
 }
 
+// The shape a spec starts from, after startEntity has approved it: rep:entity
+// wins, otherwise the one shape carrying the targetClass.
+function startShapeOf(spec, profile) {
+  if (spec.entity) {
+    return profile.shapes.find(s => s.uri === spec.entity)?.uri || null;
+  }
+  const matches = profile.shapes.filter(s => s.targetClass === spec.targetClass);
+  return matches.length === 1 ? matches[0].uri : null;
+}
+
 // Columns exist
 function columnsPresent(spec) {
   if (!spec.columns.length) return 'no columns. Add at least one rep:columns entry.';
@@ -70,7 +95,7 @@ function columnLabelsUnique(spec) {
 
 // Every path resolves, columns end on a value, depth capped
 function pathsResolve(spec, profile, maxPathDepth) {
-  const startShapeUri = entityForClass(profile, spec.targetClass);
+  const startShapeUri = startShapeOf(spec, profile);
   for (const col of spec.columns) {
     const err = walk(profile, startShapeUri, col.path, maxPathDepth,
       `column "${col.label}"`, true);
@@ -86,7 +111,7 @@ function pathsResolve(spec, profile, maxPathDepth) {
 
 // sh:min/sh:max only on numbers and dates
 function collectMinMaxTyped(spec, profile) {
-  const startShapeUri = entityForClass(profile, spec.targetClass);
+  const startShapeUri = startShapeOf(spec, profile);
   for (const col of spec.columns) {
     if (col.collect !== SH + 'min' && col.collect !== SH + 'max') continue;
     const name = col.collect === SH + 'min' ? 'sh:min' : 'sh:max';
@@ -176,11 +201,11 @@ function anyOfLimits(spec) {
     const c = filter.constraints;
     if (!c.anyOf) continue;
     if (c.anyOf.length > ANYOF_MAX_TERMS) {
-      return `rep:anyOf has ${c.anyOf.length} terms. The limit is ${ANYOF_MAX_TERMS}. Use sh:in with exact URIs instead (find them with lookup_values).`;
+      return `rep:anyOf has ${c.anyOf.length} terms. The limit is ${ANYOF_MAX_TERMS}. Narrow the words, or split the filter into two sh:property blocks.`;
     }
     for (const term of c.anyOf) {
       if (term.type === 'uri') {
-        return 'a rep:anyOf term is a URI. rep:anyOf takes Dutch words, not URIs. Use sh:in with the exact URI instead.';
+        return 'a rep:anyOf term is a URI. rep:anyOf takes Dutch words, not URIs.';
       }
       if (term.value && term.value.length > ANYOF_MAX_TERM_LENGTH) {
         return `a rep:anyOf term is ${term.value.length} characters. The limit is ${ANYOF_MAX_TERM_LENGTH}.`;
@@ -232,7 +257,7 @@ function parsesAsDate(value, withTime) {
 }
 
 const CHECKS = [
-  ['no sh:targetClass or unknown', targetClass],
+  ['no entity or unknown class', startEntity],
   ['no columns', columnsPresent],
   ['column without rdfs:label', columnLabel],
   ['two columns with one label', columnLabelsUnique],

src/runner/columns.js

@@ -7,7 +7,7 @@ import { sparqlEscapeUri } from '../db.js';
 
 const SUBJECT_CHUNK_SIZE = Math.max(1, Number(process.env.SUBJECT_CHUNK_SIZE) || 100);
 
-export async function fetchColumns(sessionQuery, subjects, spec, values = new Map(), onChunk = null, onProgress = null) {
+export async function fetchColumns(sessionQuery, subjects, spec, values = new Map(), onProgress = null) {
   const groups = groupColumns(spec.columns);
   // The subject list is complete and the chunk size is fixed, so the number
   // of batch queries is known before the first one flies. onProgress fires
@@ -22,7 +22,6 @@ export async function fetchColumns(sessionQuery, subjects, spec, values = new Ma
       done += 1;
       onProgress?.({ done, total });
     }
-    if (onChunk) await onChunk(group);
   }
   return values;
 }

src/runner/csv.js

@@ -1,4 +1,4 @@
-import { mkdirSync, writeFileSync, readFileSync } from 'fs';
+import { mkdirSync, writeFileSync } from 'fs';
 
 // Rows to a CSV file. CSV_SEPARATOR is ';' by default. A cell is quoted
 // only when it holds the separator, a quote or a newline; a quote inside a
@@ -33,8 +33,3 @@ export function writeSpec(csvName, spec) {
   writeFileSync(`${SHARE_DIR}/${name}`, spec, 'utf8');
   return `share://${name}`;
 }
-
-export function readSpec(pointer) {
-  if (!pointer.startsWith('share://')) throw new Error(`not a share:// pointer: ${pointer}`);
-  return readFileSync(`${SHARE_DIR}/${pointer.slice('share://'.length)}`, 'utf8');
-}

src/runner/job.js

@@ -1,152 +0,0 @@
-import { randomUUID } from 'crypto';
-import { sparqlEscapeUri, sparqlEscapeString, sparqlEscapeDateTime } from '../db.js';
-
-// cogs:Job + task:Task, written by us; no other service is involved. Every
-// write carries mu-call-scope-id (REPORT_SCOPE) so the delta notifier lets
-// jobs-controller ignore our statuses — it throws on unknown operations.
-
-const REPORT_SCOPE = 'http://redpencil.data.gift/id/concept/muScope/report-generation';
-const JOB_BASE = process.env.JOB_URI_BASE || 'http://redpencil.data.gift/id/job/';
-const TASK_BASE = process.env.TASK_URI_BASE || 'http://redpencil.data.gift/id/task/';
-const STATUS = {
-  scheduled: 'http://redpencil.data.gift/id/concept/JobStatus/scheduled',
-  busy: 'http://redpencil.data.gift/id/concept/JobStatus/busy',
-  success: 'http://redpencil.data.gift/id/concept/JobStatus/success',
-  failed: 'http://redpencil.data.gift/id/concept/JobStatus/failed',
-};
-const JOB_OPERATION = 'http://lblod.data.gift/id/jobs/concept/JobOperation/reportGeneration';
-const TASK_OPERATION = 'http://lblod.data.gift/id/jobs/concept/TaskOperation/reportGeneration';
-
-const COGS_JOB = 'http://vocab.deri.ie/cogs#Job';
-const TASK_TASK = 'http://redpencil.data.gift/vocabularies/tasks/Task';
-const DATA_CONTAINER = 'http://www.semanticdesktop.org/ontologies/2007/03/22/nfo#DataContainer';
-
-const MU_UUID = 'http://mu.semte.ch/vocabularies/core/uuid';
-const ADMS_STATUS = 'http://www.w3.org/ns/adms#status';
-const DCT = 'http://purl.org/dc/terms/';
-const TASK = 'http://redpencil.data.gift/vocabularies/tasks/';
-const OSLC_MESSAGE = 'http://open-services.net/ns/core#message';
-const OSLC_ERROR_URI = 'http://open-services.net/ns/core#Error';
-
-// Adds the scope header to every job write.
-function withScope(updateFn) {
-  return async sparql => updateFn(sparql, { 'mu-call-scope-id': REPORT_SCOPE });
-}
-
-// Create job + task + result container, all at status scheduled.
-// Returns { jobUri, taskUri, containerUri }.
-export async function createJob(updateFn, title, question, creator) {
-  const upd = withScope(updateFn);
-  const jobId = randomUUID();
-  const taskId = randomUUID();
-  const containerId = randomUUID();
-  const jobUri = `${JOB_BASE}${jobId}`;
-  const taskUri = `${TASK_BASE}${taskId}`;
-  const containerUri = `http://redpencil.data.gift/id/dataContainers/${containerId}`;
-  const now = sparqlEscapeDateTime(new Date());
-
-  const creatorTriple = creator
-    ? `${sparqlEscapeUri(jobUri)} ${sparqlEscapeUri(DCT + 'creator')} ${sparqlEscapeUri(creator)} .`
-    : '';
-  const questionTriple = question
-    ? `${sparqlEscapeUri(jobUri)} ${sparqlEscapeUri('http://mu.semte.ch/vocabularies/ext/question')} ${sparqlEscapeString(question)} .`
-    : '';
-
-  await upd(`
-    INSERT DATA {
-      ${sparqlEscapeUri(jobUri)} a ${sparqlEscapeUri(COGS_JOB)} ;
-        ${sparqlEscapeUri(MU_UUID)} ${sparqlEscapeString(jobId)} ;
-        ${sparqlEscapeUri(DCT + 'title')} ${sparqlEscapeString(title)} ;
-        ${sparqlEscapeUri(ADMS_STATUS)} ${sparqlEscapeUri(STATUS.scheduled)} ;
-        ${sparqlEscapeUri(TASK + 'operation')} ${sparqlEscapeUri(JOB_OPERATION)} ;
-        ${sparqlEscapeUri(DCT + 'created')} ${now} ;
-        ${sparqlEscapeUri(DCT + 'modified')} ${now} .
-      ${creatorTriple}
-      ${questionTriple}
-      ${sparqlEscapeUri(taskUri)} a ${sparqlEscapeUri(TASK_TASK)} ;
-        ${sparqlEscapeUri(MU_UUID)} ${sparqlEscapeString(taskId)} ;
-        ${sparqlEscapeUri(DCT + 'isPartOf')} ${sparqlEscapeUri(jobUri)} ;
-        ${sparqlEscapeUri(ADMS_STATUS)} ${sparqlEscapeUri(STATUS.scheduled)} ;
-        ${sparqlEscapeUri(TASK + 'operation')} ${sparqlEscapeUri(TASK_OPERATION)} ;
-        ${sparqlEscapeUri(TASK + 'index')} ${sparqlEscapeString('0')} ;
-        ${sparqlEscapeUri(DCT + 'created')} ${now} ;
-        ${sparqlEscapeUri(DCT + 'modified')} ${now} ;
-        ${sparqlEscapeUri(TASK + 'resultsContainer')} ${sparqlEscapeUri(containerUri)} .
-      ${sparqlEscapeUri(containerUri)} a ${sparqlEscapeUri(DATA_CONTAINER)} ;
-        ${sparqlEscapeUri(MU_UUID)} ${sparqlEscapeString(containerId)} .
-    }`);
-
-  return { jobUri, taskUri, containerUri };
-}
-
-export async function setStatus(updateFn, uri, status) {
-  const s = STATUS[status];
-  if (!s) throw new Error(`unknown job status "${status}"`);
-  const now = sparqlEscapeDateTime(new Date());
-  await withScope(updateFn)(`
-    DELETE {
-      ${sparqlEscapeUri(uri)} ${sparqlEscapeUri(ADMS_STATUS)} ?old ;
-        ${sparqlEscapeUri(DCT + 'modified')} ?modified .
-    }
-    INSERT {
-      ${sparqlEscapeUri(uri)} ${sparqlEscapeUri(ADMS_STATUS)} ${sparqlEscapeUri(s)} ;
-        ${sparqlEscapeUri(DCT + 'modified')} ${now} .
-    }
-    WHERE {
-      ${sparqlEscapeUri(uri)} ${sparqlEscapeUri(ADMS_STATUS)} ?old .
-      OPTIONAL { ${sparqlEscapeUri(uri)} ${sparqlEscapeUri(DCT + 'modified')} ?modified . }
-    }`);
-}
-
-export async function touch(updateFn, taskUri) {
-  const now = sparqlEscapeDateTime(new Date());
-  await withScope(updateFn)(`
-    DELETE {
-      ${sparqlEscapeUri(taskUri)} ${sparqlEscapeUri(DCT + 'modified')} ?modified .
-    }
-    INSERT {
-      ${sparqlEscapeUri(taskUri)} ${sparqlEscapeUri(DCT + 'modified')} ${now} .
-    }
-    WHERE {
-      OPTIONAL { ${sparqlEscapeUri(taskUri)} ${sparqlEscapeUri(DCT + 'modified')} ?modified . }
-    }`);
-}
-
-// Link the finished CSV to the task's result container. The container is
-// created with the job; the file URI is the logical file from registerFile.
-export async function attachResult(updateFn, taskUri, fileUri) {
-  await withScope(updateFn)(`
-    INSERT {
-      ?container ${sparqlEscapeUri(TASK + 'hasFile')} ${sparqlEscapeUri(fileUri)} .
-    }
-    WHERE {
-      ${sparqlEscapeUri(taskUri)} ${sparqlEscapeUri(TASK + 'resultsContainer')} ?container .
-    }`);
-}
-
-// An oslc:Error with the message, job and task both failed.
-export async function fail(updateFn, jobUri, taskUri, message) {
-  const errorId = randomUUID();
-  const errorUri = `http://redpencil.data.gift/id/jobs/error/${errorId}`;
-  const now = sparqlEscapeDateTime(new Date());
-  await withScope(updateFn)(`
-    DELETE {
-      ${sparqlEscapeUri(jobUri)} ${sparqlEscapeUri(ADMS_STATUS)} ?js .
-      ${sparqlEscapeUri(taskUri)} ${sparqlEscapeUri(ADMS_STATUS)} ?ts .
-    }
-    INSERT DATA {
-      ${sparqlEscapeUri(errorUri)} a ${sparqlEscapeUri(OSLC_ERROR_URI)} ;
-        ${sparqlEscapeUri(MU_UUID)} ${sparqlEscapeString(errorId)} ;
-        ${sparqlEscapeUri(OSLC_MESSAGE)} ${sparqlEscapeString(message)} ;
-        ${sparqlEscapeUri(DCT + 'created')} ${now} .
-      ${sparqlEscapeUri(taskUri)} ${sparqlEscapeUri(TASK + 'error')} ${sparqlEscapeUri(errorUri)} ;
-        ${sparqlEscapeUri(ADMS_STATUS)} ${sparqlEscapeUri(STATUS.failed)} ;
-        ${sparqlEscapeUri(DCT + 'modified')} ${now} .
-      ${sparqlEscapeUri(jobUri)} ${sparqlEscapeUri(ADMS_STATUS)} ${sparqlEscapeUri(STATUS.failed)} ;
-        ${sparqlEscapeUri(DCT + 'modified')} ${now} .
-    }
-    WHERE {
-      ${sparqlEscapeUri(jobUri)} ${sparqlEscapeUri(ADMS_STATUS)} ?js .
-      ${sparqlEscapeUri(taskUri)} ${sparqlEscapeUri(ADMS_STATUS)} ?ts .
-    }`);
-}

src/runner/profile.js

@@ -6,6 +6,7 @@ import { readFileSync } from 'fs';
 const SH = 'http://www.w3.org/ns/shacl#';
 const RDF = 'http://www.w3.org/1999/02/22-rdf-syntax-ns#';
 const OWL_ONTOLOGY = 'http://www.w3.org/2002/07/owl#Ontology';
+const REP = 'http://mu.semte.ch/vocabularies/reporting/';
 
 export async function loadProfiles(dir) {
   const fs = await import('fs');
@@ -52,7 +53,8 @@ export async function loadProfile(file) {
     const targetClass = store.getQuads(shape, SH + 'targetClass', null)[0]?.object.value || null;
     const label = store.getQuads(shape, 'http://www.w3.org/2000/01/rdf-schema#label', null)[0]?.object.value || null;
     const fields = store.getQuads(shape, SH + 'property', null).map(q => parseField(store, q.object));
-    return { uri: shape.value, targetClass, label, fields };
+    const discriminator = parseDiscriminator(store, shape);
+    return { uri: shape.value, targetClass, label, fields, discriminator };
   });
 
   return { uri, title, prefixes, shapes: shapeList, store };
@@ -82,11 +84,46 @@ function parseField(store, p) {
   };
 }
 
+// rep:discriminator tells shapes that share a targetClass apart. sh:minCount 1
+// becomes a plain triple in the seed, sh:maxCount 0 a FILTER NOT EXISTS.
+function parseDiscriminator(store, shape) {
+  const node = store.getQuads(shape, REP + 'discriminator', null)[0]?.object;
+  if (!node) return null;
+  let path = null, inverse = false;
+  const pathQuad = store.getQuads(node, SH + 'path', null)[0]?.object;
+  if (pathQuad) {
+    if (pathQuad.termType === 'BlankNode') {
+      const inv = store.getQuads(pathQuad, SH + 'inversePath', null)[0]?.object;
+      if (inv) { path = inv.value; inverse = true; }
+    } else {
+      path = pathQuad.value;
+    }
+  }
+  const one = (pred) => store.getQuads(node, pred, null)[0]?.object?.value || null;
+  const minCount = one(SH + 'minCount');
+  const maxCount = one(SH + 'maxCount');
+  if (!path || (minCount === null && maxCount === null)) return null;
+  return {
+    path, inverse,
+    minCount: minCount !== null ? Number(minCount) : null,
+    maxCount: maxCount !== null ? Number(maxCount) : null,
+  };
+}
+
 export function entityForClass(profile, cls) {
   const shape = profile.shapes.find(s => s.targetClass === cls);
   return shape ? shape.uri : null;
 }
 
+// The shape a spec starts from. rep:entity wins; a targetClass only resolves
+// when exactly one shape carries it, so a class two entities share is a
+// validator error, not a silent first match.
+export function startShape(profile, spec) {
+  if (spec.entity) return shape(profile, spec.entity);
+  const matches = profile.shapes.filter(s => s.targetClass === spec.targetClass);
+  return matches.length === 1 ? matches[0] : null;
+}
+
 export function shape(profile, shapeUri) {
   return profile.shapes.find(s => s.uri === shapeUri) || null;
 }

src/runner/run.js

@@ -3,22 +3,21 @@ import { fetchColumns } from './columns.js';
 import { assemble } from './assemble.js';
 import { rowsToCsv, writeCsv, writeSpec } from './csv.js';
 import { registerFile, registerReport } from './report.js';
-import { touch } from './job.js';
+import { startShape } from './profile.js';
 
 // Ties the steps together: seed → columns → assemble → CSV → register.
-// Returns { reportUri, fileUri, rowCount }. The task is touched once per
-// column group so the dashboard shows movement. onProgress fires on every
+// Returns { reportUri, fileUri, rowCount }. onProgress fires on every
 // answered query with what the phase knows — { subjects } per seed page,
 // { done, total } per column batch — so a waiter can tell a live run from
 // a stalled one.
 
-export async function run(query, update, parsed, profile, title, taskUri, extra = {}, onProgress = null) {
-  const subjects = await seed(query, parsed,
+export async function run(query, update, parsed, profile, title, extra = {}, onProgress = null) {
+  const shape = startShape(profile, parsed);
+  const subjects = await seed(query, parsed, shape,
     onProgress ? (n) => onProgress({ subjects: n }) : null);
 
   const values = new Map();
-  await fetchColumns(query, subjects, parsed, values,
-    taskUri ? () => touch(update, taskUri) : null, onProgress);
+  await fetchColumns(query, subjects, parsed, values, null, onProgress);
 
   const rows = assemble(subjects, values, parsed);
   const fileName = `${slug(title)}.csv`;

src/runner/seed.js

@@ -6,11 +6,11 @@ import { sparqlEscapeUri, sparqlEscapeString, sparqlEscapeDateTime, sparqlEscape
 const ROW_LIMIT = Number(process.env.ROW_LIMIT || 200000);
 const PAGE_SIZE = Number(process.env.SEED_PAGE_SIZE || 5000);
 
-export async function seed(sessionQuery, spec, onPage = null) {
+export async function seed(sessionQuery, spec, shape = null, onPage = null) {
   const subjects = [];
   let offset = 0;
   for (;;) {
-    const query = seedPageQuery(spec, PAGE_SIZE, offset);
+    const query = seedPageQuery(spec, PAGE_SIZE, offset, shape);
     const result = await sessionQuery(query);
     const rows = result.results.bindings.map(b => b.s.value);
     subjects.push(...rows);
@@ -24,9 +24,11 @@ export async function seed(sessionQuery, spec, onPage = null) {
   return subjects;
 }
 
-export function seedPageQuery(spec, limit, offset) {
+export function seedPageQuery(spec, limit, offset, shape = null) {
   const parts = [];
-  parts.push(`?s a ${sparqlEscapeUri(spec.targetClass)} .`);
+  const targetClass = spec.targetClass || shape?.targetClass;
+  parts.push(`?s a ${sparqlEscapeUri(targetClass)} .`);
+  if (shape?.discriminator) parts.push(discriminatorPattern(shape.discriminator));
   spec.filters.forEach((filter, fi) => {
     parts.push(hopsPattern(filter.path, `f${fi}`, '?s'));
     parts.push(constraintsPattern(filter.constraints, `f${fi}`));
@@ -37,6 +39,16 @@ export function seedPageQuery(spec, limit, offset) {
 LIMIT ${limit} OFFSET ${offset}`;
 }
 
+// A shape that shares its class with another says here what sets its
+// subjects apart: sh:minCount 1 is a plain triple, sh:maxCount 0 a NOT EXISTS.
+function discriminatorPattern(d) {
+  const p = sparqlEscapeUri(d.path);
+  const triple = d.inverse ? `?disc ${p} ?s .` : `?s ${p} ?disc .`;
+  if (d.minCount !== null && d.minCount >= 1) return triple;
+  if (d.maxCount === 0) return `FILTER NOT EXISTS { ${triple} }`;
+  return null;
+}
+
 // Writes out every hop as its own triple pattern.
 // Forward:  ?s <p1> ?f0_1 . ?f0_1 <p2> ?f0_v
 // Inverse:  ?s ^<p1> → ?f0_1 <p1> ?s

src/runner/spec.js

@@ -19,6 +19,7 @@ export function parseSpec(turtle) {
   const one = (s, pred) => store.getQuads(s, pred, null)[0]?.object || null;
 
   const targetClass = one(specNode, SH + 'targetClass')?.value || null;
+  const entity = one(specNode, REP + 'entity')?.value || null;
   const profileUri = one(specNode, REP + 'profile')?.value || null;
   const title = one(specNode, 'http://purl.org/dc/terms/title')?.value || null;
 
@@ -28,7 +29,7 @@ export function parseSpec(turtle) {
   const columnsQuad = one(specNode, REP + 'columns');
   const columns = columnsQuad ? parseColumnList(store, columnsQuad) : [];
 
-  return { uri: specNode.value, title, profileUri, targetClass, filters, columns };
+  return { uri: specNode.value, title, profileUri, targetClass, entity, filters, columns };
 }
 
 function parseFilter(store, p, { withLabel = false } = {}) {