diff --git a/CLAUDE.md b/CLAUDE.md index a6d2108..e64395c 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -118,56 +118,49 @@ Recouvrement's dunning is the candidate, and it is a human-validated step already. The `classic-models-mandate` client is left in the realm, unused, for that case. -## Backlog +## Events over Kafka — built, and one rule about them -**Events over Kafka, emitted by `classic-models-api`.** The target for the -reactive half. The estate emits no domain event today — the API has no signals -and no outbound webhook, and there is no bus in the cluster — which is what made -a nightly sweep look like the only option. +Built 2026-08-31. `classic-models-api` emits domain events through a +transactional outbox; a relay publishes them; `approvisionnement-stock-watch` +consumes `product.stock_changed` and is the reactive half this repo had been +waiting on. The contract is `classic-models-api/docs/DOMAIN_EVENTS.md` — read it +before writing a consumer. -The emission belongs **in the API**, not in a database CDC and not in a gateway -plugin. Only the API knows which change is a business event rather than a row -update, and only it holds the identity that made it — `sub` and `azp` from the -caller's token, which the event has to carry if a downstream agent is to act -under the same authority as the person who triggered it. +**An event carries no actor, and that is deliberate.** -Two facts about that codebase shape the work: +An earlier draft of this file said the opposite — that the event had to carry +`sub` and `azp` from the caller's token so a downstream agent could act under +the same authority. The user overruled it: -- Every write goes through one `BaseModelViewSet` in - `api/v1/classicmodels/views.py`. One hook there covers products, orders, - payments and the rest — Django signals scattered across models would be the - harder version of the same thing. -- **The models are `managed = False`.** The tables are the pre-existing - classicmodels schema and Django owns no migrations for them. A transactional - outbox means introducing the estate's first managed table beside unmanaged - ones — which works, and is worth deciding deliberately rather than - discovering. +> *"Nous gérons ici des événements, pas des 'commandes'. L'agent s'exécute avec +> sa propre identité machine, on se fout de savoir qui est à l'origine de +> l'émission de l'événement."* -An outbox is the part that matters: emitting from a viewset after `save()` drops -the event if the process dies between the two, and an agent that never hears -about a stock movement is worse than one that hears late. +The reasoning, kept because it generalises: an event is a **fact**, not a +command. A consumer acts under its own identity. An actor field would have had +no consumer and one obvious misuse — anyone able to write to the topic could name +any actor and have an agent act as them. A signed token is proof; a JSON field +is not. Causality travels as `traceparent`, in the system that exists for it. -Until this exists, detection stays unbuilt rather than being approximated by a -periodic scan: a batch that pretends to be reactive is worse than an absent -feature, because it looks finished. - -The periodic ordering run does not wait on any of it. +The emission still belongs **in the API** rather than in a database CDC or a +gateway plugin: only the API knows which change is a business event rather than +a row update. ## Two controls that become mandatory here -They were deferrable with a human in the loop. They are not, once an agent -loops on its own. +Both **built** 2026-08-31, before the first agent looped unattended. -- **Read volume caps per `(azp, sub)`**, enforced on Kong. An agent reads a - thousand records in seconds where a human clicks ten; a scope authorises the - read but says nothing about volume. +- **Read volume caps per `(azp, sub)`** — a scope authorises a read and says + nothing about volume; an agent reads a thousand records where a human clicks + ten. - **Authorisation on tool-call parameters**, not just the tool name. `write` authorises writing, not "this €40k transfer". -An autonomous agent hitting a rule that requires approval must **fail cleanly -and escalate**, never decide. And any approval screen must be rendered from the -structured tool-call parameters — never from a sentence the model produced, or a -prompt injection writes what the human validates. +The rules that stay true whatever is built next: an autonomous agent hitting a +rule that requires approval must **fail cleanly and escalate**, never decide. +And any approval screen must be rendered from the structured tool-call +parameters — never from a sentence the model produced, or a prompt injection +writes what the human validates. ## Before writing workflows diff --git a/adapter/src/a2a_n8n_adapter/__pycache__/server.cpython-312.pyc b/adapter/src/a2a_n8n_adapter/__pycache__/server.cpython-312.pyc index 44b643c..3502eb4 100644 Binary files a/adapter/src/a2a_n8n_adapter/__pycache__/server.cpython-312.pyc and b/adapter/src/a2a_n8n_adapter/__pycache__/server.cpython-312.pyc differ diff --git a/adapter/src/a2a_n8n_adapter/server.py b/adapter/src/a2a_n8n_adapter/server.py index ceb774c..c1c1418 100644 --- a/adapter/src/a2a_n8n_adapter/server.py +++ b/adapter/src/a2a_n8n_adapter/server.py @@ -27,6 +27,7 @@ from __future__ import annotations import contextlib import logging +from enum import Enum, auto from collections.abc import AsyncIterator import httpx @@ -152,12 +153,14 @@ class WorkflowExecutor(AgentExecutor): ) return - answer, escalate = _read(response) - if escalate: - # An autonomous agent that meets a rule needing approval must fail - # cleanly and escalate, never decide. The caller sees a task that - # did not complete, which is the honest shape for "a human has to - # look at this". + answer, outcome = _read(response) + if outcome is not Outcome.ANSWERED: + # Three different failures, one honest shape. An agent that met a + # rule needing approval must escalate rather than decide; a workflow + # that answered nothing did not answer; a workflow that answered off + # contract cannot be read as agreement. `completed` on any of them + # tells the caller the task succeeded, which is the one thing that is + # certainly false. await updater.failed(message=updater.new_agent_message(parts=[Part(text=answer)])) return @@ -167,23 +170,46 @@ class WorkflowExecutor(AgentExecutor): await TaskUpdater(event_queue, context.task_id, context.context_id).cancel() -def _read(response: httpx.Response) -> tuple[str, bool]: - """What the workflow said, and whether it is escalating. +class Outcome(Enum): + """What came back, which is not the same question as whether HTTP worked.""" - A workflow answers with `{"answer": "...", "escalate": false}`. Anything else - is returned as raw text rather than guessed at — a workflow that has drifted - from the contract should be visible, not silently reinterpreted. + #: The contract was honoured and the workflow is standing behind its answer. + ANSWERED = auto() + #: The workflow answered, and is handing the decision to a human. + ESCALATED = auto() + #: HTTP 200 with an empty body. n8n does that when a node throws before the + #: Respond node runs — a failed token exchange is the common way in. It is a + #: failure that looks exactly like a success from the status code alone. + SILENT = auto() + #: A body that is not the contract. Returned raw rather than guessed at: a + #: workflow that has drifted should be visible, not reinterpreted. + OFF_CONTRACT = auto() + + +def _read(response: httpx.Response) -> tuple[str, Outcome]: + """What the workflow said, and what kind of thing it said. + + A workflow answers with `{"answer": "...", "escalate": false}`. """ + body = response.text.strip() + if not body: + return ( + "The workflow produced no answer. It most often means a node threw " + "before the reply was built — a refused token exchange, usually.", + Outcome.SILENT, + ) + try: payload = response.json() except ValueError: - return response.text.strip() or "(the workflow returned nothing)", False + return body, Outcome.OFF_CONTRACT if isinstance(payload, list) and payload: # n8n returns an array when the workflow's last node emits items. payload = payload[0] if isinstance(payload, dict) and "answer" in payload: - return str(payload["answer"]), bool(payload.get("escalate", False)) - return response.text.strip() or "(the workflow returned nothing)", False + answer = str(payload["answer"]) + return answer, Outcome.ESCALATED if payload.get("escalate", False) else Outcome.ANSWERED + return body, Outcome.OFF_CONTRACT def _subject_token(context: RequestContext) -> str | None: diff --git a/adapter/tests/__pycache__/test_adapter.cpython-312-pytest-9.1.1.pyc b/adapter/tests/__pycache__/test_adapter.cpython-312-pytest-9.1.1.pyc index a9816cb..9319c00 100644 Binary files a/adapter/tests/__pycache__/test_adapter.cpython-312-pytest-9.1.1.pyc and b/adapter/tests/__pycache__/test_adapter.cpython-312-pytest-9.1.1.pyc differ diff --git a/adapter/tests/test_adapter.py b/adapter/tests/test_adapter.py index 5e88d6a..1708893 100644 --- a/adapter/tests/test_adapter.py +++ b/adapter/tests/test_adapter.py @@ -11,7 +11,7 @@ import httpx import pytest from a2a_n8n_adapter.config import Settings -from a2a_n8n_adapter.server import _read, build_agent_card +from a2a_n8n_adapter.server import Outcome, _read, build_agent_card def settings(**over) -> Settings: @@ -35,27 +35,35 @@ def response(payload, status=200) -> httpx.Response: class TestReadingTheWorkflow: def test_the_contracted_shape(self) -> None: - assert _read(response({"answer": "yes", "escalate": False})) == ("yes", False) + assert _read(response({"answer": "yes", "escalate": False})) == ("yes", Outcome.ANSWERED) def test_n8n_wraps_its_last_node_in_an_array(self) -> None: - assert _read(response([{"answer": "yes"}])) == ("yes", False) + assert _read(response([{"answer": "yes"}])) == ("yes", Outcome.ANSWERED) def test_escalation_is_carried_not_swallowed(self) -> None: # An autonomous agent meeting a rule that needs approval must escalate. # Losing this flag would turn "a human must look" into "done". assert _read(response({"answer": "needs approval", "escalate": True})) == ( "needs approval", - True, + Outcome.ESCALATED, ) def test_an_off_contract_reply_is_returned_raw(self) -> None: # Visible rather than reinterpreted: a workflow that has drifted should # look wrong, not be quietly guessed at. - text, escalate = _read(response("plain text")) - assert text == "plain text" and escalate is False + # Off contract is not agreement. Returned raw so the drift is visible, + # but never as a completed answer. + text, outcome = _read(response("plain text")) + assert text == "plain text" and outcome is Outcome.OFF_CONTRACT def test_an_empty_reply_says_so(self) -> None: - assert _read(response("")) == ("(the workflow returned nothing)", False) + # HTTP 200 with an empty body is what n8n returns when a node throws + # before the reply is built — a refused token exchange, usually. Reading + # it as a completed answer told the caller the task succeeded, which was + # the one thing certainly false. + text, outcome = _read(response("")) + assert outcome is Outcome.SILENT + assert "no answer" in text class TestAgentCard: