From 06d0ab5e2de9776bb13f54648d15effabdc367a5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?St=C3=A9phane=20Tailland?= Date: Mon, 31 Aug 2026 07:46:24 +0200 Subject: [PATCH] Wake Approvisionnement on a stock event instead of a clock MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The reactive half of the agent. It consumes product.stock_changed from the catalog topic and reports only a crossing of the threshold — every further order line on an already-low reference would otherwise raise the same alarm, and an agent that repeats itself is one a human learns to ignore. A stock-out is treated as its own fact even when the reference was already low. The detection lands on a watchlist readable over a webhook, because a proposal that reached only an execution log would be lost between two nightly runs. Co-Authored-By: Claude Opus 5 --- CLAUDE.md | 23 ++++ deploy/README.md | 0 workflows/README.md | 72 ++++++++++++ workflows/approvisionnement-stock-watch.json | 115 +++++++++++++++++++ 4 files changed, 210 insertions(+) create mode 100644 deploy/README.md create mode 100644 workflows/approvisionnement-stock-watch.json diff --git a/CLAUDE.md b/CLAUDE.md index 8326245..a6d2108 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -176,3 +176,26 @@ endpoints and the known blockers. Deliverables in English. Export workflows as JSON and commit them; do not rely on the n8n instance as the source of truth. + +## The Kafka credential is created by hand + +`kafka-classic-models` in n8n holds the SCRAM password of the +`classic-models-agent-approvisionnement` KafkaUser and the Strimzi cluster CA. +It is not in this repo and must not be: n8n encrypts stored credentials with +`N8N_ENCRYPTION_KEY`, and the committed workflow JSON references it by id only. + +To recreate it after a rebuild: + +``` +kubectl -n kafka get secret classic-models-agent-approvisionnement \ + -o jsonpath='{.data.password}' | base64 -d +kubectl -n kafka get secret kafka-cluster-ca-cert \ + -o jsonpath='{.data.ca\.crt}' | base64 -d +``` + +then POST to `/rest/credentials` with type `kafka`, `ssl: true`, the CA in the +`ca` field, `saslMechanism: scram-sha-512`, brokers +`kafka-kafka-bootstrap.kafka.svc.cluster.local:9093`. The workflow's +`credentials.kafka.id` then needs the new id. + +The KafkaUser itself lives in `infra-kafka/kafka-users.yaml`. diff --git a/deploy/README.md b/deploy/README.md new file mode 100644 index 0000000..e69de29 diff --git a/workflows/README.md b/workflows/README.md index 12197e0..664ec1f 100644 --- a/workflows/README.md +++ b/workflows/README.md @@ -147,3 +147,75 @@ looks exactly like one computed from all of it, so the difference has to be in the output. A catalog that outgrows one page needs a paging loop **and** the read-volume cap this repo already lists as mandatory for autonomous agents. + +## approvisionnement-stock-watch + +The reactive half, and the one the design had been waiting on. It is woken by +`product.stock_changed` on `classic-models.catalog` rather than by a clock, so a +reference that runs low at 14:32 is noticed at 14:32 instead of being reported +at 06:00 the next morning. + +Verified end to end against the live catalog: + +``` +S12_1099 68 → 38 detected: "fell from 68 to 38, below the threshold of 50" + 38 → 30 not detected — already below, figures refreshed instead + 30 → 0 detected: "is out of stock. Nothing can be promised." + 0 → 68 not detected, entry refreshed (a correction, not demand) +POST {"clear":["S12_1099"]} → watching: 0 +``` + +### Only a crossing is news + +`before >= threshold && after < threshold`. Without that test, every further +order line on an already-low reference raises the same alarm again, and an agent +that repeats itself is one a human learns to ignore. + +A stock-out is treated as its own fact even when the reference was already below +the threshold: *low* and *nothing left to promise* are different problems. + +A movement that changes nothing newsworthy still **refreshes** an entry already +being watched. Otherwise the watchlist would show the figure from the moment of +detection, which is the number least likely to still be true. + +`reason` carries `order:` for demand and `manual-adjustment` for someone +correcting the books. Both can justify a purchase order; only demand says +anything about the future, so the distinction is recorded rather than filtered +on. + +### The detection has somewhere to go + +`$getWorkflowStaticData('global')`, keyed by reference, readable over +`GET /webhook/stock-watchlist` and cleared with +`POST {"clear": ["S24_2000"]}`. The purchase *proposal* still has nowhere to be +written — the domain has no purchase order and no supplier entity — but a +detection that only reached an execution log would be lost between two nightly +runs, which would make the reactive half pointless. + +The obvious next consumer is `check-availability`: "can we promise X?" is a +better answer when it knows X crossed the threshold at 14:32. + +### Kafka details worth keeping + +- **The offset moves only on success** (`resolveOffset: onCompletion`, + `allowedStatuses: ["success"]`). A failed execution replays its event rather + than losing it — for an autonomous agent a detection dropped in silence is + worse than one made twice. +- **`fromBeginning: false`.** Otherwise every restart replays the whole + retention window and re-detects crossings dealt with days ago. +- Most messages on the topic are not `stock_changed`, and the workflow says so + explicitly and succeeds. A workflow that threw on an uninteresting event would + replay it forever. +- The consumer has its **own** SCRAM account, + `classic-models-agent-approvisionnement`: Read on that one topic and that one + consumer-group prefix. It cannot read sales or payments, and it cannot write — + so a compromised workflow can raise no event another agent would act on. + +### Two traps + +- **n8n's Kafka credential has a `ca` field.** The cluster CA goes there, so + there was no need to touch `NODE_EXTRA_CA_CERTS` or to disable verification. + Not obvious from the node's UI, which shows it only once `ssl` is on. +- **A webhook declaring several methods has one output per method**, in the + order declared. Wiring only the first silently drops every POST: the execution + reports **success** and n8n answers `No item to return was found`. diff --git a/workflows/approvisionnement-stock-watch.json b/workflows/approvisionnement-stock-watch.json new file mode 100644 index 0000000..a61647e --- /dev/null +++ b/workflows/approvisionnement-stock-watch.json @@ -0,0 +1,115 @@ +{ + "name": "Approvisionnement \u2014 stock watch", + "nodes": [ + { + "parameters": { + "topic": "classic-models.catalog", + "groupId": "approvisionnement-stock-watch", + "resolveOffset": "onCompletion", + "allowedStatuses": [ + "success" + ], + "options": { + "jsonParseMessage": true, + "onlyMessage": true, + "fromBeginning": false + } + }, + "type": "n8n-nodes-base.kafkaTrigger", + "typeVersion": 2, + "position": [ + 0, + 0 + ], + "id": "kafka-trigger-stock-watch", + "name": "Catalog events", + "credentials": { + "kafka": { + "id": "8qNZXLzvdx5rWQ0V", + "name": "kafka-classic-models" + } + } + }, + { + "parameters": { + "jsCode": "// A stock movement, judged the moment it happens.\n//\n// This is the reactive half of Approvisionnement. The 06:00 run proposes\n// purchase orders on a buying cycle, which is genuinely periodic; noticing that\n// a reference is about to run out is not, and a nightly sweep would report at\n// 06:00 what was true at 14:32 the previous afternoon.\n//\n// What it reacts to is a fact the API published, not a row it went looking for.\nconst THRESHOLD = 50;\nconst COVER = 3; // a proposal covers three times the threshold\nconst STOCK_CHANGED = 'com.sttlab.classicmodels.product.stock_changed.v1';\n\nconst event = $json ?? {};\n\n// The topic carries every catalog event, so most messages are not ours. Saying\n// so explicitly keeps the offset moving: a workflow that threw on an\n// uninteresting event would replay it forever.\nif (event.type !== STOCK_CHANGED) {\n return [{ json: { considered: false, type: event.type ?? 'unreadable' } }];\n}\n\nconst data = event.data ?? {};\nconst before = Number(data.before?.quantityinstock);\nconst after = Number(data.after?.quantityinstock);\nif (!Number.isFinite(before) || !Number.isFinite(after)) {\n return [{ json: {\n considered: false,\n escalate: true,\n reason: `${event.subject}: the movement carried no readable quantities.`,\n }}];\n}\n\n// Only a *crossing* is news. Without this, every further order line on an\n// already-low reference raises the same alarm again, and an agent that repeats\n// itself is one a human learns to ignore.\nconst crossed = before >= THRESHOLD && after < THRESHOLD;\n// A stock-out is a different fact from a low stock, even when the reference was\n// already below the threshold \u2014 nothing can be promised at all.\nconst emptied = after === 0 && before > 0;\n\n// `order:` means demand; anything else is someone correcting the books. Both\n// justify a purchase order, but only demand is a signal about the future.\nconst fromDemand = String(data.reason ?? '').startsWith('order:');\n\nconst store = $getWorkflowStaticData('global');\nstore.watchlist = store.watchlist ?? {};\n\nif (!crossed && !emptied) {\n // Not news, but if this reference is already being watched its figures have\n // just moved. Refresh them rather than leaving a human reading the number it\n // held at the moment of detection, which is the number least likely to still\n // be true.\n const watched = store.watchlist[event.subject];\n if (watched) {\n watched.on_hand = after;\n watched.propose_order = THRESHOLD * COVER - after;\n watched.last_movement_at = event.time ?? new Date().toISOString();\n }\n return [{ json: {\n considered: true,\n detected: false,\n watching: Boolean(watched),\n productcode: event.subject,\n on_hand: after,\n }}];\n}\n\nconst state = { detected: true, escalate: true,\n productcode: event.subject,\n on_hand: after,\n was: before,\n threshold: THRESHOLD,\n stock_out: after === 0,\n driver: fromDemand ? 'demand' : 'correction',\n movement: data.reason ?? null,\n propose_order: THRESHOLD * COVER - after,\n observed_at: event.time ?? new Date().toISOString(),\n // Kept so this detection can be joined to the request that caused it. It is\n // the only link back \u2014 the event deliberately names no actor.\n traceparent: event.traceparent ?? null,\n};\n\nstate.reason = state.stock_out\n ? `${state.productcode} is out of stock (was ${before}). Nothing can be promised.`\n : `${state.productcode} fell from ${before} to ${after}, below the threshold of ${THRESHOLD}.`;\n\n// The watchlist is the destination, and it is a real one: the proposal still\n// has nowhere to be written \u2014 the domain has no purchase order and no supplier\n// entity \u2014 so a detection that only logged would be lost between two nightly\n// runs. Here it survives, keyed by reference so a reference cannot appear twice.\nstore.watchlist[state.productcode] = state;\n\nreturn [{ json: state }];\n" + }, + "type": "n8n-nodes-base.code", + "typeVersion": 2, + "position": [ + 220, + 0 + ], + "id": "detect-crossing", + "name": "Judge the movement" + }, + { + "parameters": { + "httpMethod": [ + "GET", + "POST" + ], + "path": "stock-watchlist", + "responseMode": "lastNode", + "options": {}, + "multipleMethods": true + }, + "type": "n8n-nodes-base.webhook", + "typeVersion": 2, + "position": [ + 0, + 200 + ], + "id": "watchlist-read", + "name": "Watchlist", + "webhookId": "stock-watchlist-get" + }, + { + "parameters": { + "jsCode": "// What has been detected and not yet dealt with.\n//\n// A detection that goes only into an execution log is one nobody reads. This\n// makes the watchlist readable \u2014 by the periodic run, which can start from what\n// was already noticed rather than rediscovering it, and by a human.\n//\n// GET /webhook/stock-watchlist \u2192 the current list\n// POST /webhook/stock-watchlist {\"clear\": [\"S24_2000\"]} \u2192 drop entries\nconst store = $getWorkflowStaticData('global');\nstore.watchlist = store.watchlist ?? {};\n\n// GET carries no body; the node still runs, and clear is then empty.\nconst body = $json.body ?? {};\nconst clear = Array.isArray(body.clear) ? body.clear : [];\nfor (const code of clear) delete store.watchlist[code];\n\nconst entries = Object.values(store.watchlist);\nreturn [{ json: {\n watching: entries.length,\n cleared: clear,\n // Worst first: an out-of-stock reference is not the same problem as one that\n // merely crossed the threshold.\n entries: entries.sort((a, b) => a.on_hand - b.on_hand),\n}}];\n" + }, + "type": "n8n-nodes-base.code", + "typeVersion": 2, + "position": [ + 220, + 200 + ], + "id": "watchlist-render", + "name": "Render the watchlist" + } + ], + "connections": { + "Catalog events": { + "main": [ + [ + { + "node": "Judge the movement", + "type": "main", + "index": 0 + } + ] + ] + }, + "Watchlist": { + "main": [ + [ + { + "node": "Render the watchlist", + "type": "main", + "index": 0 + } + ], + [ + { + "node": "Render the watchlist", + "type": "main", + "index": 0 + } + ] + ] + } + }, + "settings": { + "executionOrder": "v1" + } +}