Stop reporting a silent workflow as a completed task

n8n answers 200 with an empty body when a node throws before the Respond node
runs — a refused token exchange is the common way in. The adapter read that as
a completed task carrying "(the workflow returned nothing)", which tells the
caller the task succeeded: the one thing certainly false.

Off-contract replies get the same treatment. They are still returned raw so
drift is visible, but a body that is not the contract cannot be read as
agreement.

Also removes a backlog section that described the events as unbuilt and said an
event should carry sub and azp from the caller's token — the design the user
overruled. Anyone reading it would have built the wrong consumer.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Stéphane Tailland
2026-08-31 14:39:03 +02:00
parent 78456c7902
commit 6eb77d862f
5 changed files with 85 additions and 58 deletions
+30 -37
View File
@@ -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
+40 -14
View File
@@ -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:
+15 -7
View File
@@ -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: