felix.ruizdearcaute/natural-language-report-service
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

more semantic.works

Browse Source
This commit is contained in:
felix
2026-09-14 15:34:29 +02:00
parent 120a3d259c
commit 789a676425
18 changed files with 307 additions and 410 deletions
Download Patch File Download Diff File Expand all files Collapse all files

33
src/chat/index.js
View File

@@ -5,10 +5,10 @@
// answer(turn) is the only thing the service writes. See the plan, §3, for
// the turn object. This file knows nothing about what the answer is.
import express from 'express';
import { query as muQuery, update as muUpdate } from 'mu';
import {
readConversation, readHistory, findAssistant,
writeMessage, setDocumentUrl, dropDocument,
groupsQuery, groupsUpdate,
} from './store.js';
const TEXT = {
@@ -23,25 +23,22 @@ export function mountChat(app, { path = '/assistant', answer }) {
const content = String(req.body?.content ?? '').trim();
if (!content) return res.status(400).json({ error: 'content is required' });
// 1. the access check: can this session read the conversation?
// 1. the access check: the query carries the caller's session, and the
// auth layer answers with what they may read
let conversation;
try {
conversation = await readConversation(req.params.id, req.get('mu-session-id'));
conversation = await readConversation(req.params.id);
} catch (e) {
console.error('[chat] conversation read failed:', e);
return res.status(500).json({ error: e.message });
}
if (!conversation) return res.status(404).json({ error: 'no such conversation' });
// 2. everything from here on carries the captured groups
const query = sparql => groupsQuery(sparql, conversation.groups);
const update = sparql => groupsUpdate(sparql, conversation.groups);
let assistant, history, userMessage;
try {
assistant = process.env.CHAT_ASSISTANT_URI || await findAssistant(query);
history = await readHistory(query, conversation.uri, assistant);
userMessage = await writeMessage(update, {
assistant = process.env.CHAT_ASSISTANT_URI || await findAssistant();
history = await readHistory(conversation.uri, assistant);
userMessage = await writeMessage({
conversationUri: conversation.uri,
content,
maker: conversation.creator,
@@ -52,22 +49,24 @@ export function mountChat(app, { path = '/assistant', answer }) {
return res.status(500).json({ error: e.message });
}
// 3. answer now; the assistant's message comes when it comes
// 2. answer now; the assistant's message comes when it comes. The
// template's helpers keep attaching this request's session for the
// whole continuation, so the writes below stay the caller's;
// mu-authorization is in charge of what lands where.
res.status(202).json({ id: userMessage.id });
const turn = {
conversation: { uri: conversation.uri, id: conversation.id, title: conversation.title },
history,
content,
query,
update,
groups: conversation.groups,
query: muQuery,
update: muUpdate,
finished: false,
say: (text, attachments = []) => writeMessage(update, {
say: (text, attachments = []) => writeMessage({
conversationUri: conversation.uri, content: text, maker: assistant, attachments,
}),
setUrl: (documentUri, url) => setDocumentUrl(update, documentUri, url, conversation.uri),
drop: (documentUri) => dropDocument(update, documentUri),
setUrl: (documentUri, url) => setDocumentUrl(documentUri, url, conversation.uri),
drop: (documentUri) => dropDocument(documentUri),
};
try {

98
src/chat/store.js
View File

@@ -2,23 +2,11 @@
// service it lives in. Full URIs everywhere: the auth layer rejects a prefixed
// name with no PREFIX line with an opaque 500.
import { randomUUID } from 'crypto';
import { sparqlEscapeUri, sparqlEscapeString, sparqlEscapeDateTime } from 'mu';
import { query as muQuery, update as muUpdate,
sparqlEscapeUri, sparqlEscapeString, sparqlEscapeDateTime } from 'mu';
const ENDPOINT = process.env.MU_SPARQL_ENDPOINT;
const HISTORY_LIMIT = Number(process.env.CHAT_HISTORY_LIMIT || 20);
// The mu template logs mu.query/mu.update the same way (helpers/mu/sparql.js);
// these fetch paths bypass it, so log here under the same switches.
const LOG_QUERIES = process.env.LOG_SPARQL_QUERIES != undefined
? /^(true|1)$/i.test(process.env.LOG_SPARQL_QUERIES)
: /^(true|1)$/i.test(process.env.LOG_SPARQL_ALL || '');
const LOG_UPDATES = process.env.LOG_SPARQL_UPDATES != undefined
? /^(true|1)$/i.test(process.env.LOG_SPARQL_UPDATES)
: /^(true|1)$/i.test(process.env.LOG_SPARQL_ALL || '');
function logSparql(kind, sparql) {
console.log(`[sparql] ${kind}:\n${sparql}`);
}
const T = {
type: 'http://www.w3.org/1999/02/22-rdf-syntax-ns#type',
uuid: 'http://mu.semte.ch/vocabularies/core/uuid',
@@ -48,77 +36,29 @@ const s = sparqlEscapeString;
const dt = (value) => sparqlEscapeDateTime(value)
.replace('xsd:dateTime', '<http://www.w3.org/2001/XMLSchema#dateTime>');
// --- the two ways to talk to the auth layer ---------------------------------
// With the caller's session, inside the request. Returns the JSON result and
// the allowed-groups header, which is the identity we carry past the 202.
// A bare fetch rather than mu.query: the header is on the response, and this
// way we do not depend on what the template's helper chooses to return.
export async function sessionQuery(sparql, sessionId) {
if (LOG_QUERIES) logSparql('query', sparql);
const res = await fetch(ENDPOINT, {
method: 'POST',
headers: {
'Content-Type': 'application/x-www-form-urlencoded',
Accept: 'application/sparql-results+json',
'mu-session-id': sessionId,
},
body: new URLSearchParams({ query: sparql }).toString(),
});
if (!res.ok) throw new Error(`query failed (${res.status}): ${firstLine(await res.text())}`);
return { json: await res.json(), groups: res.headers.get('mu-auth-allowed-groups') };
}
// With the captured groups, after the 202.
export async function groupsQuery(sparql, groups) {
if (LOG_QUERIES) logSparql('query', sparql);
const res = await fetch(ENDPOINT, {
method: 'POST',
headers: {
'Content-Type': 'application/x-www-form-urlencoded',
Accept: 'application/sparql-results+json',
'mu-auth-allowed-groups': groups,
},
body: new URLSearchParams({ query: sparql }).toString(),
});
if (!res.ok) throw new Error(`query failed (${res.status}): ${firstLine(await res.text())}`);
return res.json();
}
export async function groupsUpdate(sparql, groups) {
if (LOG_UPDATES) logSparql('update', sparql);
const res = await fetch(ENDPOINT, {
method: 'POST',
headers: { 'Content-Type': 'application/sparql-update', 'mu-auth-allowed-groups': groups },
body: sparql,
});
if (!res.ok) throw new Error(`update failed (${res.status}): ${firstLine(await res.text())}`);
}
function firstLine(text) { return String(text).split('\n')[0].slice(0, 300); }
const bindings = (json) => json.results.bindings;
// --- reads --------------------------------------------------------------------
// The conversation by uuid, as the caller sees it. null when not readable.
// Also returns the allowed groups from the same round trip.
export async function readConversation(id, sessionId) {
const { json, groups } = await sessionQuery(`
// The conversation by uuid, as the caller sees it (the query carries the
// caller's session; the auth layer answers with what they may read).
// null when not readable.
export async function readConversation(id) {
const json = await muQuery(`
SELECT ?conversation ?title ?creator WHERE {
?conversation ${u(T.type)} ${u(T.thread)} ;
${u(T.uuid)} ${s(id)} .
OPTIONAL { ?conversation ${u(T.title)} ?title . }
OPTIONAL { ?conversation ${u(T.maker)} ?creator . }
} LIMIT 1`, sessionId);
} LIMIT 1`);
const b = bindings(json)[0];
if (!b) return null;
if (!groups) throw new Error('no mu-auth-allowed-groups header on the conversation read; cannot carry the identity past the 202');
return { uri: b.conversation.value, id, title: b.title?.value, creator: b.creator?.value, groups };
return { uri: b.conversation.value, id, title: b.title?.value, creator: b.creator?.value };
}
// The last HISTORY_LIMIT messages, oldest first, as { role, content }.
export async function readHistory(queryFn, conversationUri, assistantUri) {
const json = await queryFn(`
export async function readHistory(conversationUri, assistantUri) {
const json = await muQuery(`
SELECT ?content ?maker ?created WHERE {
?message ${u(T.hasContainer)} ${u(conversationUri)} ;
${u(T.content)} ?content ;
@@ -132,8 +72,8 @@ export async function readHistory(queryFn, conversationUri, assistantUri) {
}
// The one assistant the caller can read. Throws unless there is exactly one.
export async function findAssistant(queryFn) {
const json = await queryFn(`
export async function findAssistant() {
const json = await muQuery(`
SELECT ?agent WHERE { ?agent ${u(T.type)} ${u(T.softwareAgent)} . } LIMIT 2`);
const rows = bindings(json);
if (rows.length !== 1) {
@@ -147,7 +87,7 @@ export async function findAssistant(queryFn) {
// One message, both types, its attachments, and the conversation touched, in
// one update. `title` is only written when given (the first question names
// an unnamed conversation). Returns { uri, id, documents: [{ uri, id }] }.
export async function writeMessage(updateFn, { conversationUri, content, maker, attachments = [], title }) {
export async function writeMessage({ conversationUri, content, maker, attachments = [], title }) {
const id = randomUUID();
const uri = `${MESSAGE_BASE}${id}`;
const now = dt(new Date());
@@ -171,7 +111,7 @@ export async function writeMessage(updateFn, { conversationUri, content, maker,
const conversationPairs = [`${u(T.lastActivity)} ${now}`];
if (title) conversationPairs.push(`${u(T.title)} ${s(title)}`);
await updateFn(`
await muUpdate(`
DELETE {
${u(conversationUri)} ${u(T.lastActivity)} ?old .
}
@@ -195,9 +135,9 @@ export async function writeMessage(updateFn, { conversationUri, content, maker,
}
// The file is ready.
export async function setDocumentUrl(updateFn, documentUri, url, conversationUri) {
export async function setDocumentUrl(documentUri, url, conversationUri) {
const now = dt(new Date());
await updateFn(`
await muUpdate(`
DELETE {
${u(documentUri)} ${u(T.url)} ?oldUrl .
${u(conversationUri)} ${u(T.lastActivity)} ?old .
@@ -213,8 +153,8 @@ export async function setDocumentUrl(updateFn, documentUri, url, conversationUri
}
// The file failed. The card disappears; the caller says why in a message.
export async function dropDocument(updateFn, documentUri) {
await updateFn(`
export async function dropDocument(documentUri) {
await muUpdate(`
DELETE {
?message ${u(T.attachment)} ${u(documentUri)} .
${u(documentUri)} ?p ?o .

128
src/db.js
View File

@@ -5,129 +5,23 @@ import { query as muQuery, update as muUpdate,
export { sparqlEscapeUri, sparqlEscapeString, sparqlEscapeInt,
sparqlEscapeDateTime, sparqlEscapeDate, sparqlEscapeBool };
// The caller's session. Everything that touches report data.
// mu attaches mu-session-id / mu-auth-allowed-groups from the incoming
// request itself; we never set headers here.
// All SPARQL goes through the template's own helpers, like
// ipdc-bookmarks-service. They attach the caller's session
// (mu-session-id, mu-call-id) from the request context, and
// mu-authorization decides graphs and visibility. The service never
// touches an auth header itself.
export async function sessionQuery(sparql) { return muQuery(sparql); }
export async function sessionUpdate(sparql) { return muUpdate(sparql); }
// The fetch paths below bypass the template's query helpers, so their
// logging (LOG_SPARQL_ALL / LOG_SPARQL_QUERIES / LOG_SPARQL_UPDATES, the
// same switches helpers/mu/sparql.js reads) is mirrored here.
const LOG_QUERIES = process.env.LOG_SPARQL_QUERIES != undefined
? /^(true|1)$/i.test(process.env.LOG_SPARQL_QUERIES)
: /^(true|1)$/i.test(process.env.LOG_SPARQL_ALL || '');
const LOG_UPDATES = process.env.LOG_SPARQL_UPDATES != undefined
? /^(true|1)$/i.test(process.env.LOG_SPARQL_UPDATES)
: /^(true|1)$/i.test(process.env.LOG_SPARQL_ALL || '');
function logSparql(kind, sparql) {
console.log(`[sparql] ${kind}:\n${sparql}`);
}
// No session. Only what the stack makes public. Everything shown to the LLM.
// Bare HTTP POST: muQuery always attaches the request's session, which would
// widen or narrow the answer depending on the caller.
export async function publicQuery(sparql) {
if (LOG_QUERIES) logSparql('query', sparql);
const res = await fetch(`${process.env.MU_SPARQL_ENDPOINT}`, {
method: 'POST',
headers: {
'Content-Type': 'application/x-www-form-urlencoded',
'Accept': 'application/sparql-results+json',
},
body: new URLSearchParams({ query: sparql }).toString(),
});
if (!res.ok) {
const text = await res.text();
throw new Error(`public query failed (${res.status}): ${stripStoreError(text)}`);
}
return res.json();
}
// A bare update, the same trick as publicQuery. muUpdate would attach the
// request's session; job writes after the 202 must not depend on it.
export async function plainUpdate(sparql, extraHeaders = {}) {
if (LOG_UPDATES) logSparql('update', sparql);
const res = await fetch(`${process.env.MU_SPARQL_ENDPOINT}`, {
method: 'POST',
headers: {
'Content-Type': 'application/sparql-update',
...extraHeaders,
},
body: sparql,
});
if (!res.ok) {
const text = await res.text();
throw new Error(`update failed (${res.status}): ${stripStoreError(text)}`);
}
}
// Background half of a run. Call captureGroups() once inside the request,
// then send the captured header on every query after the 202: the template
// stops attaching the session once the request is answered, and the auth
// layer would silently answer from the public graph.
export async function captureGroups(sessionQueryFn) {
const probe = await sessionQueryFn('SELECT ?s WHERE { ?s ?p ?o } LIMIT 1');
const header = probe.headers?.['mu-auth-allowed-groups'];
if (!header) {
throw new Error('no mu-auth-allowed-groups header on the first response; cannot carry the identity past the 202');
}
return header;
}
export async function groupsQuery(sparql, allowedGroups) {
if (LOG_QUERIES) logSparql('query', sparql);
const res = await fetch(`${process.env.MU_SPARQL_ENDPOINT}`, {
method: 'POST',
headers: {
'Content-Type': 'application/x-www-form-urlencoded',
'Accept': 'application/sparql-results+json',
'mu-auth-allowed-groups': allowedGroups,
},
body: new URLSearchParams({ query: sparql }).toString(),
});
if (!res.ok) {
const text = await res.text();
throw new Error(`query failed (${res.status}): ${stripStoreError(text)}`);
}
return res.json();
}
export async function groupsUpdate(sparql, allowedGroups, extraHeaders = {}) {
if (LOG_UPDATES) logSparql('update', sparql);
const res = await fetch(`${process.env.MU_SPARQL_ENDPOINT}`, {
method: 'POST',
headers: {
'Content-Type': 'application/sparql-update',
'mu-auth-allowed-groups': allowedGroups,
...extraHeaders,
},
body: sparql,
});
if (!res.ok) {
const text = await res.text();
throw new Error(`update failed (${res.status}): ${stripStoreError(text)}`);
}
return res.json();
}
// 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 per-request session bundle the MCP tools use. Built here (next to the
// other mu-touching code) so llm/ never imports from 'mu' or sessionQuery
// (CI checks 1 and 2). Read tools use the caller's session through muQuery;
// run_report captures the groups once and runs in the background with them.
// The per-request bundle the tools use: the template's query and update.
// The request context lives for the whole run, also past the 202 the chat
// turn answers with, so mu-authorization keeps rewriting with the caller's
// session. We add nothing to it.
export async function buildSession() {
const query = async (sparql) => muQuery(sparql);
let groupsCache = null;
return {
query,
get groups() {
if (!groupsCache) groupsCache = captureGroups(query);
return groupsCache;
},
};
}
return { query: sessionQuery, update: sessionUpdate };
}

12
src/llm/describe.js
View File

@@ -1,11 +1,11 @@
// Turn a shapes graph into the flat list an LLM reads well (plan §7.3). Not
// Turtle: a plain text menu of entities, their fields and the code lists.
// Code lists under INLINE_VALUES_MAX are inlined at boot from the public
// graph (publicQuery) and refreshed every VALUES_TTL, so the LLM never has
// to search the short ones.
// Code lists under INLINE_VALUES_MAX are inlined at request time
// (sessionQuery; mu-authorization decides what is visible) and refreshed
// every VALUES_TTL, so the LLM never has to search the short ones.
import { fieldsOf, entityForClass } from '../runner/profile.js';
import { publicQuery } from '../db.js';
import { sessionQuery } from '../db.js';
const INLINE_VALUES_MAX = Number(process.env.INLINE_VALUES_MAX || 50);
const VALUES_TTL = Number(process.env.VALUES_TTL || 3600) * 1000;
@@ -34,7 +34,7 @@ async function loadCodeList(profile, f, key) {
try {
values = await queryCodeList(f.class);
} catch (e) {
// public graph read failed; report nothing, the LLM will use lookup_values
// the read failed; report nothing, the LLM will use lookup_values
values = [];
}
cache.set(key, { values, at: Date.now() });
@@ -46,7 +46,7 @@ async function queryCodeList(type) {
?uri a <${type}> .
OPTIONAL { ?uri <http://www.w3.org/2004/02/skos/core#prefLabel> ?label . }
} ORDER BY ?label LIMIT ${INLINE_VALUES_MAX + 1}`;
const r = await publicQuery(q);
const r = await sessionQuery(q);
const bindings = r.results.bindings;
return bindings.slice(0, INLINE_VALUES_MAX).map(b => ({
uri: b.uri.value,

15
src/llm/lookup.js
View File

@@ -1,9 +1,10 @@
// lookup_values: search a code list too long to inline. One publicQuery,
// regex on str(?label) so language-tagged labels still match. Returns the
// matches, the total, and an `exact` flag when a label equals the term (case
// aside) so the LLM can settle disambiguation on its own.
// lookup_values: search a code list too long to inline. One query through
// the template helper, regex on str(?label) so language-tagged labels still
// match. Returns the matches, the total, and an `exact` flag when a label
// equals the term (case aside) so the LLM can settle disambiguation on its
// own.
import { publicQuery, sparqlEscapeString } from '../db.js';
import { sessionQuery, sparqlEscapeString } from '../db.js';
import { fieldsOf, entityForClass } from '../runner/profile.js';
const LIMIT = 25;
@@ -18,7 +19,7 @@ export async function lookupValues(profile, term, fieldSpec) {
?uri <http://www.w3.org/2004/02/skos/core#prefLabel> ?label ; a ?type .
FILTER(REGEX(str(?label), ${sparqlEscapeString(escapeRegex(term))}, "i"))
} ORDER BY ?label LIMIT ${LIMIT * 8}`;
const r = await publicQuery(q);
const r = await sessionQuery(q);
const bindings = r.results.bindings;
// The profile's sh:class is the type the data should carry, but code lists
// sometimes type their entries differently (MAR codes are
@@ -59,7 +60,7 @@ async function codeQuery(profile, type, term) {
FILTER(CONTAINS(LCASE(str(?code)), LCASE(${sparqlEscapeString(term)})))
} ORDER BY ?label LIMIT ${LIMIT}`;
try {
const r = await publicQuery(q);
const r = await sessionQuery(q);
return r.results.bindings.map(b => ({ label: b.label.value, uri: b.uri.value }));
} catch {
return [];

19
src/llm/tools.js
View File

@@ -1,9 +1,7 @@
// The eight MCP tools the LLM calls. Each maps to a runner/ function or a
// pure profile/db read. Tools that touch session data (run_report,
// report_status, export_spec) live here and call into runner/; they never
// import sessionQuery themselves (CI check 2). What they return is always a
// job URI, a status, a count, a link, column names or an error we wrote,
// never a cell value (CI check 3, the §8 guarantee).
// profile/db read. What they return is always a job URI, a status, a count,
// a link, column names or an error we wrote, never a cell value (the §8
// guarantee).
//
// The shape every tool returns is { content: [{ type: 'text', text }] } so the
// loop can pass it straight back to the model.
@@ -19,10 +17,10 @@ const MAX_PATH_DEPTH = Number(process.env.MAX_PATH_DEPTH || 8);
const RUN_TIMEOUT = Number(process.env.RUN_TIMEOUT || 60) * 1000;
// buildTools(profiles, session) → { tools: Tool[], handlers: Map<name, fn> }
// `session` is the request's { query, update, groups, capture } bundle the app
// builds once per /mcp request. run_report captures the identity and runs in
// the background exactly like POST /reports does, but waits up to
// RUN_TIMEOUT so the LLM gets the result in one turn when it is quick.
// `session` is the request's { query, update } bundle: the template's own
// helpers, which attach the caller's session from the request context.
// run_report starts the run and waits up to RUN_TIMEOUT so the LLM gets the
// result in one turn when it is quick.
export function buildTools(profiles, session) {
const profileList = () => [...profiles.values()].map(p => ({ id: p.uri, title: p.title }));
@@ -77,11 +75,10 @@ export function buildTools(profiles, session) {
const errors = checkSpec(parsed, p, MAX_PATH_DEPTH, profiles);
if (errors.length) return text(`the spec is not valid: ${errors[0]}`);
try {
const g = await session.groups;
// The spec's own dct:title names the report; there is no other input.
const name = parsed.title || 'report';
await session.onReportStart?.({ title: name, fileName: `${slug(name)}.csv` });
const running = run(g, parsed, p, name);
const running = run(session.query, session.update, parsed, p, name);
// The end callback fires even when withTimeout below has given up: the
// run is not cancelled, and the file arrives when it arrives. The
// rejection handler also keeps a late failure from crashing Node 20.

2
src/report-assistant.js
View File

@@ -19,7 +19,7 @@ export function reportAssistant(whenProfiles) {
const session = {
query: turn.query,
groups: turn.groups,
update: turn.update,
async onReportStart({ fileName }) {
const said = await turn.say(TEXT.making, [{ name: fileName, mediaType: 'text/csv' }]);
pendingDocument = said.documents[0].uri;

26
src/runner/run.js
View File

@@ -4,29 +4,27 @@ import { assemble } from './assemble.js';
import { rowsToCsv, writeCsv } from './csv.js';
import { registerFile, registerReport } from './report.js';
import { touch } from './job.js';
import { groupsQuery, groupsUpdate } from '../db.js';
// Ties the six steps together. Runs with the captured groups (see
// captureGroups), returns { reportUri, fileUri, rowCount }. The task is
// touched once per column group so the dashboard shows movement.
// Ties the six steps together. `query` and `update` are the template's own
// helpers: they attach the caller's session from the request context, which
// lives for the whole run (also past the 202), so mu-authorization keeps
// deciding graphs and visibility. Returns { reportUri, fileUri, rowCount }.
// The task is touched once per column group so the dashboard shows movement.
export async function run(allowedGroups, parsed, profile, title, taskUri, extra = {}) {
const queryFn = sparql => groupsQuery(sparql, allowedGroups);
const updateFn = (sparql, extra) => groupsUpdate(sparql, allowedGroups, extra);
const subjects = await seed(queryFn, parsed);
export async function run(query, update, parsed, profile, title, taskUri, extra = {}) {
const subjects = await seed(query, parsed);
const values = new Map();
await fetchColumns(queryFn, subjects, parsed, values,
taskUri ? () => touch(updateFn, taskUri) : null);
await fetchColumns(query, subjects, parsed, values,
taskUri ? () => touch(update, taskUri) : null);
const rows = assemble(subjects, values, parsed);
const fileName = `${slug(title)}.csv`;
const csv = rowsToCsv(rows);
const filePath = writeCsv(fileName, csv);
const fileUri = await registerFile((sparql) => updateFn(sparql), fileName, filePath);
const reportUri = await registerReport((sparql) => updateFn(sparql), title, fileUri, extra);
const fileUri = await registerFile(update, fileName, filePath);
const reportUri = await registerReport(update, title, fileUri, extra);
return { reportUri, fileUri, rowCount: rows.length - 1, filePath };
}
@@ -38,4 +36,4 @@ export function slug(title) {
.replace(/[^a-z0-9]+/g, '-')
.replace(/^-+|-+$/g, '')
.slice(0, 80) || 'report';
}
}