90 lines
No EOL
3.4 KiB
Markdown
90 lines
No EOL
3.4 KiB
Markdown
# Message Pump — End-to-End Flow (v2.0)
|
|
|
|
The AgentServer message pump processes individual messages through a single, linear, attack-resistant pipeline. The outer dispatcher runs a continuous async loop, draining per-thread message buffers (queues) until empty — enabling persistent, branched reasoning without artificial limits.
|
|
|
|
```mermaid
|
|
flowchart TD
|
|
subgraph MessagePump
|
|
subgraph Init
|
|
start([Start])
|
|
raw[/Optional<br>Raw Bytes/]
|
|
wrapstart["Wrap<br><start>{...}</start>"]
|
|
end
|
|
enq1([QUEUE 1])
|
|
rawwaiting{Raw<br>Msg<br>Waiting?}
|
|
waitRaw([Wait])
|
|
subgraph Process
|
|
extract["Extract<br>Tree"]
|
|
split["Split<br>Tree"]
|
|
subgraph foreach [For Each Message]
|
|
getmsg[Get Msg]
|
|
badTo{<To> Missing?}
|
|
endnoto([Discard])
|
|
addfrom["Add .from"]
|
|
repair[Repair + C14N]
|
|
validate[Validate]
|
|
invalidMsg{Bad<br>Message?}
|
|
badmsg([Discard])
|
|
more{More?}
|
|
end
|
|
enqueue([QUEUE 2])
|
|
xmlWaiting{XML<br>Waiting?}
|
|
waitXml([Wait])
|
|
subgraph Async
|
|
lookup[Lookup Listener]
|
|
route[Route]
|
|
wait[await Response]
|
|
wrap["Wrap<br><from>{...}</from>"]
|
|
end
|
|
end
|
|
end
|
|
|
|
start --> raw --> wrapstart --> enq1 --> rawwaiting
|
|
rawwaiting --> |NO| waitRaw
|
|
rawwaiting ---> |YES| extract
|
|
extract --> split --> foreach
|
|
getmsg --> badTo
|
|
badTo --> |YES| endnoto
|
|
badTo --> |NO| addfrom --> repair --> validate --> invalidMsg
|
|
invalidMsg ---> |NO| more --> |Yes| getmsg
|
|
invalidMsg --> |YES| badmsg
|
|
more --> |NO| enqueue
|
|
enqueue --> xmlWaiting
|
|
xmlWaiting --> |NO| waitXml
|
|
xmlWaiting ---> |YES| lookup --> route --> wait --> wrap --> enq1
|
|
|
|
```
|
|
## Detailed Stages (Per-Message)
|
|
|
|
1. **Ingress/Enqueue**: Raw bytes → repair → preliminary tree → enqueue to target thread buffer.
|
|
|
|
2. **Dispatcher Loop**: Single async non-blocking loop selects next message from per-thread queues (breadth-first default for fairness).
|
|
|
|
3. **Processing**:
|
|
- Full repair + C14N.
|
|
- Envelope validation.
|
|
- Routing decision:
|
|
- **Meta Branch** (`https://xml-pipeline.org/ns/meta/v1` namespace): Handled directly by privileged core handler (no listener lookup or XSD validation needed).
|
|
- Purpose: Introspection and reserved organism primitives.
|
|
- Examples:
|
|
- `request-schema`, `request-example`, `request-prompt`, `list-capabilities` (returns XSD bytes, example XML, prompt fragment, or capability list).
|
|
- Thread primitives like `spawn-thread`, `clear-context`.
|
|
- Privileged: Controlled via YAML `meta` flags (e.g., `allow_schema_requests: "admin"` or "none"). Remote queries optional.
|
|
- Why separate: Faster, safer (no user listener involved), topology privacy preserved.
|
|
- Capability namespace → normal listener route (XSD validation + deserialization).
|
|
|
|
- Typed handler call → raw bytes.
|
|
|
|
4. **Response Handling**:
|
|
- Dummy wrap → extract multi-payloads.
|
|
- Each enqueued as new message(s) in appropriate thread buffer(s).
|
|
|
|
5. **Egress**: Dequeue → C14N/sign → send.
|
|
|
|
## Key Properties
|
|
- Continuous looping until all thread buffers empty — natural iteration/subthreading without one-shot constraints.
|
|
- Multi-payload enqueues enable parallel branches/thoughts.
|
|
- Scheduling balances deep dives vs fair exploration.
|
|
- Attack-resistant at every step.
|
|
|
|
XML in → queued → processed → multi-out → re-queued. Loops forever if needed. Safely. Permanently. |