= Introduction =

== Audience and scope ==

This exercise is intended for EDURange developers learning how to write event processors with the message bus.

The packaged <code>mbus</code> and <code>aostore</code> modules are treated as read-only for this exercise. All new code belongs under a demonstration directory such as:

<pre lang="text">
workspace/
mbus/
aostore/
demos/
__init__.py
event_processors/
__init__.py
</pre>

The exercise uses two problem families.

* '''1. Shell and TTY event processing'''
** Send and receive messages through the bus.
** Capture and inspect message history.
** Represent high-level shell command events.
** Recreate the regex-based milestone role in a simplified form.
** Mock up TTY-derived observations such as keystroke timing.
* '''2. Graph event processing'''
** Represent graph edge relationships as source events.
** Derive direct path facts from edge relation events.
** Add feedback by deriving longer paths from known paths.
** Simulate nondeterministic processor scheduling while keeping source input fixed.

The shell examples draw from EDURange's institutional knowledge. The graph examples provide a compact computer science model for safe feedback loops.

== Running the examples ==

Adjust paths to match the local checkout layout if necessary. Start an interpreter from the workspace root with project packages available.

<pre lang="bash">
PYTHONPATH="$PWD/mbus:$PWD/aostore:$PWD" python3.14
</pre>

When module files are changed, restart the interpreter unless the group is deliberately practicing <code>importlib.reload()</code>.

== Create the demonstration package ==

Before the first durable file edit, create the demonstration package directories:

<pre lang="bash">
mkdir -p demos/event_processors
touch demos/__init__.py
touch demos/event_processors/__init__.py
</pre>

== Important note about capture records ==

The current demonstration works close to the low-level capture layer. As a result, it exposes <code>MessageSymbolRegistration</code> records while building the temporary history helper.

That is not the intended ordinary user experience. In a more complete release, higher-level capture and history infrastructure should handle symbol registration bookkeeping on behalf of processor authors. In this exercise, symbol registration records are visible because the demonstration history helper reconstructs readable messages from compact captured records directly.

== Suggested group session flow ==

; Session 1 - Bus and capture basics

* '''Phase 1:'''
** Signal over the bus by hand.
** Inspect received messages.
** Inspect low-level captured records.
* '''Phase 2:'''
** Discuss why symbol registration appears in this temporary demo.
** Build the capture history helper.

Estimated time: 1-3 hours (1-2 hours work, 0.5-1 hours setup)

; Session 2 - Shell command and regex milestones

* '''Phase 3:'''
** Represent a shell command as a dictionary.
* '''Phase 4:'''
** Replace it with <code>ShellCommandEvent</code>.
* '''Phase 5:'''
** Manually evaluate regex milestone expressions.
* '''Phase 6:'''
** Implement <code>RegexMilestoneProcessor</code>.
** Inspect source and analysis events in history.

Estimated time: 1.5-3 hours (1.5-2.5 hours work, 0-0.5 hours setup)

; Session 3 - TTY-derived observations

* '''Phase 7:'''
** Create raw TTY read, write, and line events.
** Show why regex milestones need reconstructed shell command events.
* '''Phase 8:'''
** Implement keystroke timing.
** Discuss shell command reconstruction as the next derived-event problem (to be continued in Session 6 onward).

Estimated time: 1-2.5 hours (1-2 hours work, 0-0.5 hours setup)

; Session 4 - Graph-derived facts

* '''Phase 9:'''
** Define graph edges and paths.
* '''Phase 10:'''
** Add the demo-only nonblocking receive helper.
* '''Phase 11:'''
** Implement direct path derivation.
* '''Phase 12:'''
** Explore the extension rule by hand.
* '''Phase 13:'''
** Add path extension feedback.
** Discuss duplicate prevention.

Estimated time: 1.5-3.5 hours (1.5-3 hours work, 0-0.5 hours setup)

; Session 5 - Nondeterministic processor scheduling

* '''Phase 14:'''
** Add the randomized processor runner.
* '''Phase 15:'''
** Keep source events fixed.
** Vary processor scheduling by seed.
** Compare final path facts across runs.

Estimated time: 1.5-3.5 hours (1.5-3 hours work, 0-0.5 hours setup)

; Session 6 - Follow-up refinements

* '''Phase 16:'''
** Rebuild a graph index from history.
* '''Phase 17:'''
** Implement shell command reconstruction.
** Replace local Python payload assumptions with explicit payload projection formats when the bus format layer is ready for the exercise.

Estimated time: 1.5-4.5 hours (1.5-4 hours work, 0-0.5 hours setup)

= Event Processor Demonstration Plan =

== Phase 1: Signal over the bus by hand ==

; Goal

Send one message through the direct bus and inspect what the receiver sees.

; Discussion points

* A bus is useful only when something emits and something receives.
* Public message setup uses readable strings: topic, producer, and message type.
* The direct broker delivers local Python payloads in this demonstration.
* Capture must be attached before delivery in the current prototype.

; Interpreter exploration

Import the bus and capture interfaces:

<pre lang="python">
from dataclasses import dataclass, field

from mbus.broker.direct import DirectMessageBus
from mbus.capture.sink import CaptureSink
from mbus.message.records import CapturedMessage
from mbus.message.symbols import MessageSymbolRegistration
</pre>

Create a temporary capture sink in the interpreter. This sink stores whatever the bus sends to the capture layer:

<pre lang="python">
CapturedRecord = MessageSymbolRegistration | CapturedMessage

@dataclass
class ListCaptureSink(CaptureSink):
records: list[CapturedRecord] = field(default_factory=list)

def capture_symbol_registration(
self,
registration: MessageSymbolRegistration,
) -> None:
self.records.append(registration)

def capture(self, message: CapturedMessage) -> None:
self.records.append(message)
</pre>

Create the bus and attach capture:

<pre lang="python">
bus = DirectMessageBus()
sink = ListCaptureSink()
bus.set_capture_sink(sink)
</pre>

Inspect the empty capture sink:

<pre lang="python">
sink
sink.records
len(sink.records)
</pre>

Expected output shape:

<pre lang="text">
ListCaptureSink(records=[])
[]
0
</pre>

Create a receiver. This describes the messages the receiver wants to hear:

<pre lang="python">
receiver = bus.subscribe(
msg_topic="demo.signals",
msg_producer="demo-source",
msg_type="ping",
)
</pre>

Inspect the receiver and confirm that capture is still empty:

<pre lang="python">
type(receiver)
vars(receiver)
sink.records
</pre>

Expected output shape:

<pre lang="text">
<class 'mbus.broker.direct._BrokerReceiver'>
{'_queue': ...}
[]
</pre>

Create an emitter. Registering the emitter gives the bus enough information to bind readable names to compact internal IDs:

<pre lang="python">
emitter = bus.register_emitter(
msg_topic="demo.signals",
msg_producer="demo-source",
default_msg_type="ping",
)
</pre>

Inspect the captured registration records:

<pre lang="python">
sink.records
[type(record).__name__ for record in sink.records]

for record in sink.records:
print(record)
</pre>

Expected output shape:

<pre lang="text">
['MessageSymbolRegistration', 'MessageSymbolRegistration', 'MessageSymbolRegistration']
MessageSymbolRegistration(... symbol='demo.signals' ...)
MessageSymbolRegistration(... symbol='ping' ...)
MessageSymbolRegistration(... symbol='demo-source' ...)
</pre>

Emit one message:

<pre lang="python">
emitter.emit({"text": "hello bus"})
</pre>

Inspect capture before receiving:

<pre lang="python">
len(sink.records)
[type(record).__name__ for record in sink.records]
sink.records[-1]
</pre>

Expected output shape:

<pre lang="text">
4
['MessageSymbolRegistration', 'MessageSymbolRegistration', 'MessageSymbolRegistration', 'CapturedMessage']
CapturedMessage(... payload={'text': 'hello bus'} ...)
</pre>

Receive the message:

<pre lang="python">
message = receiver.receive()

message
message.msg_topic
message.msg_producer
message.msg_type
message.payload
</pre>

Expected output shape:

<pre lang="text">
ReceivedMessage(...)
'demo.signals'
'demo-source'
'ping'
{'text': 'hello bus'}
</pre>

Compare the received message with the captured record:

<pre lang="python">
captured = sink.records[-1]

captured
captured.msg_topic_id
captured.msg_type_id
captured.msg_producer_id
captured.payload
</pre>

Expected output shape:

<pre lang="text">
CapturedMessage(...)
TopicID(...)
MessageTypeID(...)
ProducerID(...)
{'text': 'hello bus'}
</pre>

; What this phase shows

The receiver gets the readable message view. The capture sink sees compact IDs and local payload data. The readable names are still available through symbol registration records, but manually reconstructing them is awkward.

That awkwardness motivates the next phase.

== Phase 2: Build a temporary capture history helper ==

; Goal

Create a small demonstration helper that turns captured records back into readable history messages.

; Discussion points

* Capture records are useful, but raw captured messages use compact IDs.
* A history helper can rebuild readable message history from registration records.
* This is demonstration infrastructure, not the final persistence/query API.
* The helper writes captured records into an in-memory <code>aostore</code> sequential unit.

; Interpreter exploration before the file edit

List only the captured messages:

<pre lang="python">
captured_messages = []

for record in sink.records:
if isinstance(record, CapturedMessage):
captured_messages.append(record)

captured_messages
</pre>

Expected output shape:

<pre lang="text">
[CapturedMessage(... payload={'text': 'hello bus'} ...)]
</pre>

Inspect the first captured message:

<pre lang="python">
captured_message = captured_messages[0]

captured_message.msg_topic_id
captured_message.msg_type_id
captured_message.msg_producer_id
</pre>

Expected output shape:

<pre lang="text">
TopicID(...)
MessageTypeID(...)
ProducerID(...)
</pre>

Manually reconstruct a topic name:

<pre lang="python">
topic_names = {}

for record in sink.records:
if isinstance(record, MessageSymbolRegistration):
if record.symbol_kind.name == "TOPIC":
topic_names[record.symbol_id] = record.symbol

topic_names
topic_names[captured_message.msg_topic_id]
</pre>

Expected output shape:

<pre lang="text">
{TopicID(...): 'demo.signals'}
'demo.signals'
</pre>

This works, but repeating it by hand would distract from writing event processors.

; File edit

Create:

<pre lang="text">
demos/event_processors/capture_history.py
</pre>

<pre lang="python">
#!/usr/bin/env python3
# demos/event_processors/capture_history.py

"""Capture and history helpers for event processor demonstrations."""

from collections.abc import Iterable
from dataclasses import dataclass, field

from aostore.sequentialunit.listsequentialunit import ListSequentialUnit
from mbus.capture.sink import CaptureSink
from mbus.message.records import CapturedMessage
from mbus.message.symbols import (
MessageSymbolKind,
MessageSymbolRegistration,
MessageTypeID,
ProducerID,
TopicID,
)

CapturedRecord = MessageSymbolRegistration | CapturedMessage

@dataclass
class AOStoreCaptureSink(CaptureSink):
records: ListSequentialUnit[CapturedRecord] = field(
default_factory=ListSequentialUnit
)

def capture_symbol_registration(
self,
registration: MessageSymbolRegistration,
) -> None:
self.records.append(registration)

def capture(self, message: CapturedMessage) -> None:
self.records.append(message)

def all_records(self) -> tuple[CapturedRecord, ...]:
records = self.records.sequential_read(0, len(self.records))
result = tuple(records)
return result

@dataclass(frozen=True, kw_only=True)
class HistoryMessage:
msg_topic: str
msg_type: str
msg_producer: str
payload: object
bus_sequence: int
topic_sequence: int

class CaptureHistory:
_records: tuple[CapturedRecord, ...]
_topics: dict[TopicID, str]
_types: dict[MessageTypeID, str]
_producers: dict[ProducerID, str]

def __init__(self, records: Iterable[CapturedRecord]) -> None:
self._records = tuple(records)
self._topics = {}
self._types = {}
self._producers = {}
self._index_symbols()

def _index_symbols(self) -> None:
for record in self._records:
if isinstance(record, MessageSymbolRegistration):
self._record_symbol(record)

def _record_symbol(self, record: MessageSymbolRegistration) -> None:
if record.symbol_kind is MessageSymbolKind.TOPIC:
self._topics[record.symbol_id] = record.symbol
elif record.symbol_kind is MessageSymbolKind.MSG_TYPE:
self._types[record.symbol_id] = record.symbol
elif record.symbol_kind is MessageSymbolKind.PRODUCER:
self._producers[record.symbol_id] = record.symbol

def messages(self) -> list[HistoryMessage]:
messages = []
for record in self._records:
if isinstance(record, CapturedMessage):
message = self._decode_message(record)
messages.append(message)
return messages

def by_topic(self, msg_topic: str) -> list[HistoryMessage]:
messages = []
for message in self.messages():
if message.msg_topic == msg_topic:
messages.append(message)
return messages

def since(self, bus_sequence: int) -> list[HistoryMessage]:
messages = []
for message in self.messages():
if message.bus_sequence >= bus_sequence:
messages.append(message)
return messages

def _decode_message(self, message: CapturedMessage) -> HistoryMessage:
history_message = HistoryMessage(
msg_topic=self._topics[message.msg_topic_id],
msg_type=self._types[message.msg_type_id],
msg_producer=self._producers[message.msg_producer_id],
payload=message.payload,
bus_sequence=message.bus_sequence,
topic_sequence=message.topic_sequence,
)
return history_message
</pre>

; Interpreter check

Restart the interpreter, then run:

<pre lang="python">
from mbus.broker.direct import DirectMessageBus
from demos.event_processors.capture_history import (
AOStoreCaptureSink,
CaptureHistory,
)

bus = DirectMessageBus()
sink = AOStoreCaptureSink()
bus.set_capture_sink(sink)

receiver = bus.subscribe(
msg_topic="demo.signals",
msg_producer="demo-source",
msg_type="ping",
)

emitter = bus.register_emitter(
msg_topic="demo.signals",
msg_producer="demo-source",
default_msg_type="ping",
)

emitter.emit({"text": "history helper"})
message = receiver.receive()

history = CaptureHistory(sink.all_records())
history.messages()
</pre>

Expected output shape:

<pre lang="text">
[HistoryMessage(msg_topic='demo.signals', msg_type='ping', msg_producer='demo-source', payload={'text': 'history helper'}, ...)]
</pre>

Inspect the decoded history:

<pre lang="python">
history_message = history.messages()[0]

history_message.msg_topic
history_message.msg_type
history_message.msg_producer
history_message.payload
</pre>

Expected output:

<pre lang="text">
'demo.signals'
'ping'
'demo-source'
{'text': 'history helper'}
</pre>

== Phase 3: Explore shell command data by hand ==

; Goal

Identify the high-level event shape needed by a regex milestone processor.

; Discussion points

* The old milestone strategy operated on command-level records.
* A command-level record includes context such as host and working directory.
* It also includes shell input and shell output.
* Raw TTY events do not directly have this shape; reconstruction comes later.

; Interpreter exploration

Represent a shell command as a dictionary:

<pre lang="python">
command = {
"session_id": "session-1",
"command_index": 0,
"host": "alpha",
"cwd": "/home/student",
"input_text": "ls -la\n",
"output_text": "total 12\n-rw-r--r-- 1 student student 0 notes.txt\n",
"started_at_ns": 1000,
"ended_at_ns": 2000,
}
</pre>

Inspect the fields used by regex milestone checks:

<pre lang="python">
command["host"]
command["cwd"]
command["input_text"]
command["output_text"]
</pre>

Expected output:

<pre lang="text">
'alpha'
'/home/student'
'ls -la\n'
'total 12\n-rw-r--r-- 1 student student 0 notes.txt\n'
</pre>

Try an accidental wrong key:

<pre lang="python">
command["working_directory"]
</pre>

Expected output:

<pre lang="text">
KeyError: 'working_directory'
</pre>

; What this phase shows

The dictionary contains the information, but it does not document the shape of the event. A typo is discovered only when the field is used. A named event class makes the processor code easier to read and easier to explain.

== Phase 4: Persist the event record classes ==

; Goal

Create reusable payload classes for the TTY, shell command, and regex milestone examples.

; File edit

Create:

<pre lang="text">
demos/event_processors/events.py
</pre>

<pre lang="python">
#!/usr/bin/env python3
# demos/event_processors/events.py

"""Payload records for event processor demonstrations."""

from dataclasses import dataclass
from typing import Literal

@dataclass(frozen=True, kw_only=True)
class TTYReadEvent:
session_id: str
source_index: int
timestamp_ns: int
data: bytes

@dataclass(frozen=True, kw_only=True)
class TTYWriteEvent:
session_id: str
source_index: int
timestamp_ns: int
data: bytes

@dataclass(frozen=True, kw_only=True)
class LineDisciplineEvent:
session_id: str
source_index: int
timestamp_ns: int
line: str

@dataclass(frozen=True, kw_only=True)
class KeystrokeTimingEvent:
session_id: str
source_index: int
timestamp_ns: int
previous_timestamp_ns: int | None
delta_ns: int | None
text: str

@dataclass(frozen=True, kw_only=True)
class ShellCommandEvent:
session_id: str
command_index: int
host: str
cwd: str
input_text: str
output_text: str
started_at_ns: int
ended_at_ns: int

RegexMilestoneField = Literal[
"host",
"cwd",
"input_text",
"output_text",
]

@dataclass(frozen=True, kw_only=True)
class RegexMilestoneExpression:
name: str
field: RegexMilestoneField
pattern: str

@dataclass(frozen=True, kw_only=True)
class RegexMilestoneResult:
expression_name: str
field: RegexMilestoneField
matched: bool
match_text: str | None = None

@dataclass(frozen=True, kw_only=True)
class RegexMilestoneAnalysisEvent:
session_id: str
command_index: int
results: tuple[RegexMilestoneResult, ...]
matched_count: int
expression_count: int
</pre>

; Interpreter check

Restart the interpreter, then run:

<pre lang="python">
from demos.event_processors.events import ShellCommandEvent

command = ShellCommandEvent(
session_id="session-1",
command_index=0,
host="alpha",
cwd="/home/student",
input_text="ls -la\n",
output_text="total 12\n-rw-r--r-- 1 student student 0 notes.txt\n",
started_at_ns=1000,
ended_at_ns=2000,
)

command
command.host
command.cwd
command.input_text
command.output_text
</pre>

Expected output shape:

<pre lang="text">
ShellCommandEvent(session_id='session-1', command_index=0, ...)
'alpha'
'/home/student'
'ls -la\n'
'total 12\n-rw-r--r-- 1 student student 0 notes.txt\n'
</pre>

; What this phase shows

The event shape is now explicit. Other code can depend on a named <code>ShellCommandEvent</code> instead of assuming that a loose dictionary has the expected fields.

== Phase 5: Evaluate regex milestones manually ==

; Goal

Recreate the core milestone check in a small, transparent form.

; Discussion points

* A regex milestone expression names one check against one shell command field.
* The field may be command input, command output, working directory, or host context.
* The result records whether the expression matched and, when available, the matched text.
* Manual evaluation is useful for learning the shape of the problem before writing a processor.

; Interpreter exploration

Import regular expressions and the payload classes:

<pre lang="python">
import re

from demos.event_processors.events import (
RegexMilestoneExpression,
ShellCommandEvent,
)
</pre>

Create one command event:

<pre lang="python">
command = ShellCommandEvent(
session_id="session-1",
command_index=0,
host="alpha",
cwd="/home/student",
input_text="ls -la\n",
output_text="total 12\n-rw-r--r-- 1 student student 0 notes.txt\n",
started_at_ns=1000,
ended_at_ns=2000,
)
</pre>

Create three milestone expressions:

<pre lang="python">
expressions = (
RegexMilestoneExpression(
name="used-ls",
field="input_text",
pattern=r"\bls\b",
),
RegexMilestoneExpression(
name="saw-notes",
field="output_text",
pattern=r"notes\.txt",
),
RegexMilestoneExpression(
name="home-directory",
field="cwd",
pattern=r"^/home/student$",
),
)
</pre>

Evaluate one expression:

<pre lang="python">
expression = expressions[0]
text = getattr(command, expression.field)
match = re.search(expression.pattern, text)

expression.name
expression.field
text
match.group(0)
</pre>

Expected output:

<pre lang="text">
'used-ls'
'input_text'
'ls -la\n'
'ls'
</pre>

Evaluate all expressions:

<pre lang="python">
manual_results = []

for expression in expressions:
text = getattr(command, expression.field)
match = re.search(expression.pattern, text)
manual_results.append((expression.name, match is not None))

manual_results
</pre>

Expected output:

<pre lang="text">
[('used-ls', True), ('saw-notes', True), ('home-directory', True)]
</pre>

; What this phase shows

Manual evaluation is simple for one command, but the loop is already event processor logic. The next phase preserves the loop in a reusable processor.

== Phase 6: Persist the RegexMilestone processor ==

; Goal

Create a processor that receives shell command events and emits regex milestone analysis events.

; File edit

Create:

<pre lang="text">
demos/event_processors/regex_milestones.py
</pre>

<pre lang="python">
#!/usr/bin/env python3
# demos/event_processors/regex_milestones.py

"""Regex milestone processor for shell command events."""

import re

from mbus.broker.direct import DirectMessageBus
from mbus.broker.endpoints import Emitter, Receiver

from demos.event_processors.events import (
RegexMilestoneAnalysisEvent,
RegexMilestoneExpression,
RegexMilestoneResult,
ShellCommandEvent,
)

class RegexMilestoneProcessor:
_receiver: Receiver
_emitter: Emitter
_expressions: tuple[RegexMilestoneExpression, ...]

def __init__(
self,
bus: DirectMessageBus,
expressions: tuple[RegexMilestoneExpression, ...],
) -> None:
self._receiver = bus.subscribe(
msg_topic="demo.shell.command",
msg_producer="shell-reconstructor",
msg_type="shell-command",
)
self._emitter = bus.register_emitter(
msg_topic="demo.shell.regex_milestone",
msg_producer="regex-milestone-processor",
default_msg_type="regex-milestone-analysis",
)
self._expressions = expressions

def process_one(self) -> RegexMilestoneAnalysisEvent:
message = self._receiver.receive()
command = message.payload

if not isinstance(command, ShellCommandEvent):
raise TypeError(
f"expected ShellCommandEvent, got {type(command).__name__}"
)

analysis = analyze_regex_milestones(command, self._expressions)
self._emitter.emit(analysis)
return analysis

def analyze_regex_milestones(
command: ShellCommandEvent,
expressions: tuple[RegexMilestoneExpression, ...],
) -> RegexMilestoneAnalysisEvent:
results = []
for expression in expressions:
result = evaluate_regex_milestone(command, expression)
results.append(result)

matched_count = sum(result.matched for result in results)
analysis = RegexMilestoneAnalysisEvent(
session_id=command.session_id,
command_index=command.command_index,
results=tuple(results),
matched_count=matched_count,
expression_count=len(expressions),
)
return analysis

def evaluate_regex_milestone(
command: ShellCommandEvent,
expression: RegexMilestoneExpression,
) -> RegexMilestoneResult:
text = getattr(command, expression.field)
match = re.search(expression.pattern, text)
match_text = None

if match is not None:
match_text = match.group(0)

result = RegexMilestoneResult(
expression_name=expression.name,
field=expression.field,
matched=match is not None,
match_text=match_text,
)
return result
</pre>

; Interpreter check

Restart the interpreter, then set up the bus:

<pre lang="python">
from mbus.broker.direct import DirectMessageBus
from demos.event_processors.capture_history import (
AOStoreCaptureSink,
CaptureHistory,
)
from demos.event_processors.events import (
RegexMilestoneExpression,
ShellCommandEvent,
)
from demos.event_processors.regex_milestones import RegexMilestoneProcessor

bus = DirectMessageBus()
sink = AOStoreCaptureSink()
bus.set_capture_sink(sink)
</pre>

Subscribe to the analysis topic before running the processor:

<pre lang="python">
analysis_receiver = bus.subscribe(
msg_topic="demo.shell.regex_milestone",
msg_producer="regex-milestone-processor",
msg_type="regex-milestone-analysis",
)
</pre>

Create the shell command emitter and processor:

<pre lang="python">
command_emitter = bus.register_emitter(
msg_topic="demo.shell.command",
msg_producer="shell-reconstructor",
default_msg_type="shell-command",
)

expressions = (
RegexMilestoneExpression(
name="used-ls",
field="input_text",
pattern=r"\bls\b",
),
RegexMilestoneExpression(
name="saw-notes",
field="output_text",
pattern=r"notes\.txt",
),
RegexMilestoneExpression(
name="home-directory",
field="cwd",
pattern=r"^/home/student$",
),
)

processor = RegexMilestoneProcessor(bus, expressions)
</pre>

Emit a shell command event:

<pre lang="python">
command = ShellCommandEvent(
session_id="session-1",
command_index=0,
host="alpha",
cwd="/home/student",
input_text="ls -la\n",
output_text="total 12\n-rw-r--r-- 1 student student 0 notes.txt\n",
started_at_ns=1000,
ended_at_ns=2000,
)

command_emitter.emit(command)
</pre>

Process the command:

<pre lang="python">
analysis = processor.process_one()
analysis
analysis.results
analysis.matched_count
analysis.expression_count
</pre>

Expected output shape:

<pre lang="text">
RegexMilestoneAnalysisEvent(... matched_count=3, expression_count=3)
(RegexMilestoneResult(...), RegexMilestoneResult(...), RegexMilestoneResult(...))
3
3
</pre>

Receive the derived analysis event:

<pre lang="python">
analysis_message = analysis_receiver.receive()

analysis_message.msg_topic
analysis_message.msg_type
analysis_message.msg_producer
analysis_message.payload
</pre>

Expected output shape:

<pre lang="text">
'demo.shell.regex_milestone'
'regex-milestone-analysis'
'regex-milestone-processor'
RegexMilestoneAnalysisEvent(... matched_count=3, expression_count=3)
</pre>

Inspect history:

<pre lang="python">
history = CaptureHistory(sink.all_records())
messages = history.messages()

for message in messages:
print(message.bus_sequence, message.msg_topic, message.msg_type)
</pre>

Expected output shape:

<pre lang="text">
0 demo.shell.command shell-command
1 demo.shell.regex_milestone regex-milestone-analysis
</pre>

; What this phase shows

The regex milestone processor consumes one event and emits another. The derived analysis event is available to any later subscriber. History contains both the source command event and the derived analysis event.

== Phase 7: Explore why shell reconstruction is separate ==

; Goal

Show that raw TTY observations are not the same thing as high-level shell command events.

; Discussion points

* Regex milestones should inspect high-level command events.
* TTY reads, TTY writes, and line-discipline lines are lower-level observations.
* A shell command event is reconstructed by correlating those observations.
* Keeping reconstruction separate prevents the regex milestone processor from becoming a large TTY parser.

; Interpreter exploration

Create a few low-level observations:

<pre lang="python">
from demos.event_processors.events import (
LineDisciplineEvent,
TTYReadEvent,
TTYWriteEvent,
)

read_1 = TTYReadEvent(
session_id="session-1",
source_index=0,
timestamp_ns=1000,
data=b"l",
)

read_2 = TTYReadEvent(
session_id="session-1",
source_index=1,
timestamp_ns=1100,
data=b"s",
)

line = LineDisciplineEvent(
session_id="session-1",
source_index=2,
timestamp_ns=1200,
line="ls\n",
)

write = TTYWriteEvent(
session_id="session-1",
source_index=3,
timestamp_ns=1600,
data=b"notes.txt\n",
)
</pre>

Inspect the raw pieces:

<pre lang="python">
read_1.data
read_2.data
line.line
write.data
</pre>

Expected output:

<pre lang="text">
b'l'
b's'
'ls\n'
b'notes.txt\n'
</pre>

Assemble command fields manually:

<pre lang="python">
input_text = line.line
output_text = write.data.decode("utf-8", errors="replace")

input_text
output_text
</pre>

Expected output:

<pre lang="text">
'ls\n'
'notes.txt\n'
</pre>

; What this phase shows

The useful command event is derived from several lower-level observations. A later shell reconstructor can subscribe to line-discipline events, query recent reads/writes/timing observations, and emit <code>ShellCommandEvent</code> records.

== Phase 8: Add keystroke timing as a smaller TTY-derived processor ==

; Goal

Create a simple stateful processor before attempting full shell reconstruction.

; Discussion points

* Timing requires comparing a TTY read with the previous read in the same session.
* A derived timing event makes the result explicit.
* Other processors can consume timing events instead of recalculating deltas.

; Interpreter exploration before the file edit

Calculate timing manually from the two read events:

<pre lang="python">
read_events = [read_1, read_2]
previous_timestamp = None
timing_rows = []

for event in read_events:
delta = None
if previous_timestamp is not None:
delta = event.timestamp_ns - previous_timestamp
timing_rows.append((event.source_index, previous_timestamp, delta))
previous_timestamp = event.timestamp_ns

timing_rows
</pre>

Expected output:

<pre lang="text">
[(0, None, None), (1, 1000, 100)]
</pre>

; File edit

Create:

<pre lang="text">
demos/event_processors/keystroke_timing.py
</pre>

<pre lang="python">
#!/usr/bin/env python3
# demos/event_processors/keystroke_timing.py

"""Keystroke timing processor for TTY read events."""

from mbus.broker.direct import DirectMessageBus
from mbus.broker.endpoints import Emitter, Receiver

from demos.event_processors.events import (
KeystrokeTimingEvent,
TTYReadEvent,
)

class KeystrokeTimingProcessor:
_receiver: Receiver
_emitter: Emitter
_last_timestamp_by_session: dict[str, int]

def __init__(self, bus: DirectMessageBus) -> None:
self._receiver = bus.subscribe(
msg_topic="demo.tty.read",
msg_producer="tty-source",
msg_type="tty-read",
)
self._emitter = bus.register_emitter(
msg_topic="demo.keystroke.timing",
msg_producer="keystroke-timing",
default_msg_type="keystroke-timing",
)
self._last_timestamp_by_session = {}

def process_one(self) -> KeystrokeTimingEvent:
message = self._receiver.receive()
event = message.payload

if not isinstance(event, TTYReadEvent):
raise TypeError(f"expected TTYReadEvent, got {type(event).__name__}")

previous_timestamp = self._last_timestamp_by_session.get(
event.session_id
)
delta = None

if previous_timestamp is not None:
delta = event.timestamp_ns - previous_timestamp

self._last_timestamp_by_session[event.session_id] = event.timestamp_ns

text = event.data.decode("utf-8", errors="replace")
timing = KeystrokeTimingEvent(
session_id=event.session_id,
source_index=event.source_index,
timestamp_ns=event.timestamp_ns,
previous_timestamp_ns=previous_timestamp,
delta_ns=delta,
text=text,
)
self._emitter.emit(timing)
return timing
</pre>

; Interpreter check

Restart the interpreter, then run:

<pre lang="python">
from mbus.broker.direct import DirectMessageBus
from demos.event_processors.capture_history import AOStoreCaptureSink
from demos.event_processors.events import TTYReadEvent
from demos.event_processors.keystroke_timing import KeystrokeTimingProcessor

bus = DirectMessageBus()
sink = AOStoreCaptureSink()
bus.set_capture_sink(sink)

timing_receiver = bus.subscribe(
msg_topic="demo.keystroke.timing",
msg_producer="keystroke-timing",
msg_type="keystroke-timing",
)

read_emitter = bus.register_emitter(
msg_topic="demo.tty.read",
msg_producer="tty-source",
default_msg_type="tty-read",
)

processor = KeystrokeTimingProcessor(bus)
</pre>

Emit reads and process them:

<pre lang="python">
read_emitter.emit(
TTYReadEvent(
session_id="session-1",
source_index=0,
timestamp_ns=1000,
data=b"l",
)
)

read_emitter.emit(
TTYReadEvent(
session_id="session-1",
source_index=1,
timestamp_ns=1100,
data=b"s",
)
)

processor.process_one()
processor.process_one()
</pre>

Expected output shape:

<pre lang="text">
KeystrokeTimingEvent(... previous_timestamp_ns=None, delta_ns=None, text='l')
KeystrokeTimingEvent(... previous_timestamp_ns=1000, delta_ns=100, text='s')
</pre>

Receive timing events:

<pre lang="python">
timing_receiver.receive().payload
timing_receiver.receive().payload
</pre>

Expected output shape:

<pre lang="text">
KeystrokeTimingEvent(... source_index=0 ...)
KeystrokeTimingEvent(... source_index=1 ...)
</pre>

== Phase 9: Introduce graph events ==

; Goal

Switch to a small graph problem that can demonstrate feedback loops safely.

; Discussion points

* An edge is a source fact.
* A path is a derived fact.
* A direct edge implies a direct path.
* Longer paths can be derived later from known paths and edges.

; File edit

Create:

<pre lang="text">
demos/event_processors/graph_events.py
</pre>

<pre lang="python">
#!/usr/bin/env python3
# demos/event_processors/graph_events.py

"""Graph event payloads for message bus demonstrations."""

from dataclasses import dataclass

@dataclass(frozen=True, kw_only=True)
class EdgeDeclaredEvent:
graph_id: str
source: str
target: str

@dataclass(frozen=True, kw_only=True)
class PathKnownEvent:
graph_id: str
source: str
target: str
hop_count: int

def fact_key(self) -> tuple[str, str, str]:
result = (self.graph_id, self.source, self.target)
return result
</pre>

; Interpreter exploration

<pre lang="python">
from demos.event_processors.graph_events import (
EdgeDeclaredEvent,
PathKnownEvent,
)

edge = EdgeDeclaredEvent(
graph_id="demo-graph",
source="A",
target="B",
)

path = PathKnownEvent(
graph_id=edge.graph_id,
source=edge.source,
target=edge.target,
hop_count=1,
)

edge
path
path.fact_key()
</pre>

Expected output shape:

<pre lang="text">
EdgeDeclaredEvent(graph_id='demo-graph', source='A', target='B')
PathKnownEvent(graph_id='demo-graph', source='A', target='B', hop_count=1)
('demo-graph', 'A', 'B')
</pre>

== Phase 10: Create a demo-only nonblocking receive helper ==

; Goal

Prepare for processor scheduling without editing <code>mbus</code>.

; Discussion points

* <code>Receiver.receive()</code> blocks until a message is available.
* A scheduler needs to try a processor and move on if no message is waiting.
* The current package does not expose a public nonblocking receiver method.
* This helper is local demonstration code. It can be replaced by a real public API later.

; File edit

Create:

<pre lang="text">
demos/event_processors/demo_receive.py
</pre>

<pre lang="python">
#!/usr/bin/env python3
# demos/event_processors/demo_receive.py

"""Demo-only helpers for receiver inspection and scheduling."""

from queue import Empty

from mbus.broker.endpoints import Receiver
from mbus.message.records import ReceivedMessage

class DemoReceiverError(RuntimeError):
pass

def try_receive(receiver: Receiver) -> ReceivedMessage | None:
queue = getattr(receiver, "_queue", None)

if queue is None:
raise DemoReceiverError(
"demo try_receive requires the current direct broker receiver"
)

try:
message = queue.get_nowait()
except Empty:
message = None

return message
</pre>

; Interpreter check

Restart the interpreter, then run:

<pre lang="python">
from mbus.broker.direct import DirectMessageBus
from demos.event_processors.demo_receive import try_receive

bus = DirectMessageBus()

receiver = bus.subscribe(
msg_topic="demo.signals",
msg_producer="demo-source",
msg_type="ping",
)

try_receive(receiver)
</pre>

Expected output:

<pre lang="text">
None
</pre>

Now emit a message:

<pre lang="python">
from demos.event_processors.capture_history import AOStoreCaptureSink

sink = AOStoreCaptureSink()
bus.set_capture_sink(sink)

emitter = bus.register_emitter(
msg_topic="demo.signals",
msg_producer="demo-source",
default_msg_type="ping",
)

emitter.emit({"text": "demo nonblocking receive"})
message = try_receive(receiver)

message
message.payload
</pre>

Expected output shape:

<pre lang="text">
ReceivedMessage(...)
{'text': 'demo nonblocking receive'}
</pre>

== Phase 11: Create the direct path processor ==

; Goal

Implement the graph rule:

<pre lang="text">
edge(A, B) -> path(A, B)
</pre>

; Discussion points

* The source emits edge events.
* The direct path processor emits derived path events.
* Other processors can subscribe to derived path events.
* The processor can be run by hand or by a scheduler.

; File edit

Create:

<pre lang="text">
demos/event_processors/graph_reachability.py
</pre>

<pre lang="python">
#!/usr/bin/env python3
# demos/event_processors/graph_reachability.py

"""Graph reachability processors for message bus demonstrations."""

from mbus.broker.direct import DirectMessageBus
from mbus.broker.endpoints import Emitter, Receiver

from demos.event_processors.demo_receive import try_receive
from demos.event_processors.graph_events import (
EdgeDeclaredEvent,
PathKnownEvent,
)

class DirectPathProcessor:
_receiver: Receiver
_emitter: Emitter
_known_paths: set[tuple[str, str, str]]

def __init__(self, bus: DirectMessageBus) -> None:
self._receiver = bus.subscribe(
msg_topic="demo.graph.edge",
msg_producer="graph-source",
msg_type="edge-declared",
)
self._emitter = bus.register_emitter(
msg_topic="demo.graph.path",
msg_producer="direct-path",
default_msg_type="path-known",
)
self._known_paths = set()

def process_one(self) -> PathKnownEvent | None:
message = self._receiver.receive()
result = self._process_payload(message.payload)
return result

def process_available(self) -> int:
message = try_receive(self._receiver)

if message is None:
work_count = 0
else:
self._process_payload(message.payload)
work_count = 1

return work_count

def _process_payload(self, payload: object) -> PathKnownEvent | None:
if not isinstance(payload, EdgeDeclaredEvent):
raise TypeError(
f"expected EdgeDeclaredEvent, got {type(payload).__name__}"
)

path = PathKnownEvent(
graph_id=payload.graph_id,
source=payload.source,
target=payload.target,
hop_count=1,
)

if path.fact_key() not in self._known_paths:
self._known_paths.add(path.fact_key())
self._emitter.emit(path)
result = path
else:
result = None

return result
</pre>

; Interpreter check

Restart the interpreter, then run:

<pre lang="python">
from mbus.broker.direct import DirectMessageBus
from demos.event_processors.capture_history import AOStoreCaptureSink
from demos.event_processors.graph_events import EdgeDeclaredEvent
from demos.event_processors.graph_reachability import DirectPathProcessor

bus = DirectMessageBus()
sink = AOStoreCaptureSink()
bus.set_capture_sink(sink)

path_receiver = bus.subscribe(
msg_topic="demo.graph.path",
msg_producer="direct-path",
msg_type="path-known",
)

edge_emitter = bus.register_emitter(
msg_topic="demo.graph.edge",
msg_producer="graph-source",
default_msg_type="edge-declared",
)

processor = DirectPathProcessor(bus)
</pre>

Call the processor before any edge exists:

<pre lang="python">
processor.process_available()
</pre>

Expected output:

<pre lang="text">
0
</pre>

Emit one edge:

<pre lang="python">
edge_emitter.emit(
EdgeDeclaredEvent(
graph_id="demo-graph",
source="A",
target="B",
)
)
</pre>

Process the edge:

<pre lang="python">
processor.process_available()
</pre>

Expected output:

<pre lang="text">
1
</pre>

Receive the derived path:

<pre lang="python">
path_message = path_receiver.receive()
path_message.payload
</pre>

Expected output:

<pre lang="text">
PathKnownEvent(graph_id='demo-graph', source='A', target='B', hop_count=1)
</pre>

== Phase 12: Explore the path extension rule by hand ==

; Goal

Understand the feedback rule before implementing it.

; Discussion points

* If <code>A</code> reaches <code>B</code>, and there is an edge from <code>B</code> to <code>C</code>, then <code>A</code> reaches <code>C</code>.
* Derived path facts can lead to more derived path facts.
* Duplicate prevention keeps the feedback loop from repeating the same fact forever.

; Interpreter exploration

<pre lang="python">
from demos.event_processors.graph_events import (
EdgeDeclaredEvent,
PathKnownEvent,
)

edge_bc = EdgeDeclaredEvent(
graph_id="demo-graph",
source="B",
target="C",
)

path_ab = PathKnownEvent(
graph_id="demo-graph",
source="A",
target="B",
hop_count=1,
)
</pre>

Check whether the path and edge connect:

<pre lang="python">
path_ab.target
edge_bc.source
path_ab.target == edge_bc.source
</pre>

Expected output:

<pre lang="text">
'B'
'B'
True
</pre>

Derive a new path:

<pre lang="python">
path_ac = PathKnownEvent(
graph_id="demo-graph",
source=path_ab.source,
target=edge_bc.target,
hop_count=path_ab.hop_count + 1,
)

path_ac
path_ac.fact_key()
</pre>

Expected output:

<pre lang="text">
PathKnownEvent(graph_id='demo-graph', source='A', target='C', hop_count=2)
('demo-graph', 'A', 'C')
</pre>

Manually prevent duplicate path facts:

<pre lang="python">
known_paths = set()
known_paths.add(path_ab.fact_key())

path_ac.fact_key() in known_paths

if path_ac.fact_key() not in known_paths:
known_paths.add(path_ac.fact_key())

known_paths
</pre>

Expected output shape:

<pre lang="text">
False
{('demo-graph', 'A', 'B'), ('demo-graph', 'A', 'C')}
</pre>

== Phase 13: Add the path extension processor ==

; Goal

Implement the feedback rule:

<pre lang="text">
path(A, B) + edge(B, C) -> path(A, C)
</pre>

; Discussion points

* The extension processor listens to edge events and path events.
* It emits new path events when a known path can be extended.
* It also listens to its own output because an extended path may be extendable again.
* This is a controlled feedback loop.

; File edit

Append this class to:

<pre lang="text">
demos/event_processors/graph_reachability.py
</pre>

<pre lang="python">

class PathExtensionProcessor:
_edge_receiver: Receiver
_direct_path_receiver: Receiver
_extension_path_receiver: Receiver
_emitter: Emitter
_edges: set[tuple[str, str, str]]
_paths: dict[tuple[str, str, str], PathKnownEvent]

def __init__(self, bus: DirectMessageBus) -> None:
self._edge_receiver = bus.subscribe(
msg_topic="demo.graph.edge",
msg_producer="graph-source",
msg_type="edge-declared",
)
self._direct_path_receiver = bus.subscribe(
msg_topic="demo.graph.path",
msg_producer="direct-path",
msg_type="path-known",
)
self._extension_path_receiver = bus.subscribe(
msg_topic="demo.graph.path",
msg_producer="path-extension",
msg_type="path-known",
)
self._emitter = bus.register_emitter(
msg_topic="demo.graph.path",
msg_producer="path-extension",
default_msg_type="path-known",
)
self._edges = set()
self._paths = {}

def process_edge_available(self) -> int:
message = try_receive(self._edge_receiver)

if message is None:
work_count = 0
else:
paths = self._process_edge_payload(message.payload)
work_count = 1 + len(paths)

return work_count

def process_direct_path_available(self) -> int:
message = try_receive(self._direct_path_receiver)

if message is None:
work_count = 0
else:
paths = self._process_path_payload(message.payload)
work_count = 1 + len(paths)

return work_count

def process_extension_path_available(self) -> int:
message = try_receive(self._extension_path_receiver)

if message is None:
work_count = 0
else:
paths = self._process_path_payload(message.payload)
work_count = 1 + len(paths)

return work_count

def _process_edge_payload(
self,
payload: object,
) -> list[PathKnownEvent]:
if not isinstance(payload, EdgeDeclaredEvent):
raise TypeError(
f"expected EdgeDeclaredEvent, got {type(payload).__name__}"
)

edge = payload
self._edges.add((edge.graph_id, edge.source, edge.target))
paths = self._derive_from_new_edge(edge)
return paths

def _process_path_payload(
self,
payload: object,
) -> list[PathKnownEvent]:
if not isinstance(payload, PathKnownEvent):
raise TypeError(
f"expected PathKnownEvent, got {type(payload).__name__}"
)

path = payload
self._paths[path.fact_key()] = path
paths = self._derive_from_new_path(path)
return paths

def _derive_from_new_edge(
self,
edge: EdgeDeclaredEvent,
) -> list[PathKnownEvent]:
paths = []

for known_path in self._paths.values():
if known_path.graph_id != edge.graph_id:
continue
if known_path.target != edge.source:
continue

next_path = PathKnownEvent(
graph_id=edge.graph_id,
source=known_path.source,
target=edge.target,
hop_count=known_path.hop_count + 1,
)

if self._emit_if_new(next_path):
paths.append(next_path)

return paths

def _derive_from_new_path(
self,
path: PathKnownEvent,
) -> list[PathKnownEvent]:
paths = []

for graph_id, edge_source, edge_target in self._edges:
if graph_id != path.graph_id:
continue
if edge_source != path.target:
continue

next_path = PathKnownEvent(
graph_id=path.graph_id,
source=path.source,
target=edge_target,
hop_count=path.hop_count + 1,
)

if self._emit_if_new(next_path):
paths.append(next_path)

return paths

def _emit_if_new(self, path: PathKnownEvent) -> bool:
if path.fact_key() in self._paths:
result = False
else:
self._paths[path.fact_key()] = path
self._emitter.emit(path)
result = True

return result
</pre>

; Interpreter check

Restart the interpreter, then run:

<pre lang="python">
from mbus.broker.direct import DirectMessageBus
from demos.event_processors.capture_history import (
AOStoreCaptureSink,
CaptureHistory,
)
from demos.event_processors.graph_events import EdgeDeclaredEvent
from demos.event_processors.graph_reachability import (
DirectPathProcessor,
PathExtensionProcessor,
)

bus = DirectMessageBus()
sink = AOStoreCaptureSink()
bus.set_capture_sink(sink)

extension_receiver = bus.subscribe(
msg_topic="demo.graph.path",
msg_producer="path-extension",
msg_type="path-known",
)

edge_emitter = bus.register_emitter(
msg_topic="demo.graph.edge",
msg_producer="graph-source",
default_msg_type="edge-declared",
)

direct_processor = DirectPathProcessor(bus)
extension_processor = PathExtensionProcessor(bus)
</pre>

Emit two edges in fixed order:

<pre lang="python">
edge_emitter.emit(
EdgeDeclaredEvent(
graph_id="demo-graph",
source="A",
target="B",
)
)

edge_emitter.emit(
EdgeDeclaredEvent(
graph_id="demo-graph",
source="B",
target="C",
)
)
</pre>

Process the direct paths:

<pre lang="python">
direct_processor.process_available()
direct_processor.process_available()
</pre>

Expected output:

<pre lang="text">
1
1
</pre>

Let the extension processor learn the edges:

<pre lang="python">
extension_processor.process_edge_available()
extension_processor.process_edge_available()
</pre>

Expected output:

<pre lang="text">
1
1
</pre>

Let the extension processor learn the direct paths:

<pre lang="python">
extension_processor.process_direct_path_available()
extension_processor.process_direct_path_available()
</pre>

Expected output shape:

<pre lang="text">
1
2
</pre>

The second call may emit the extended <code>A -> C</code> path after the processor has seen both direct paths and both edges.

Receive the extended path:

<pre lang="python">
message = extension_receiver.receive()
message.payload
</pre>

Expected output:

<pre lang="text">
PathKnownEvent(graph_id='demo-graph', source='A', target='C', hop_count=2)
</pre>

Inspect captured graph path history:

<pre lang="python">
history = CaptureHistory(sink.all_records())
path_messages = history.by_topic("demo.graph.path")

for message in path_messages:
path = message.payload
print(
message.bus_sequence,
message.msg_producer,
path.source,
path.target,
path.hop_count,
)
</pre>

Expected output shape:

<pre lang="text">
... direct-path A B 1
... direct-path B C 1
... path-extension A C 2
</pre>

== Phase 14: Add a randomized processor scheduler ==

; Goal

Simulate concurrency effects without changing source input order.

; Discussion points

* The source will emit the same edge events in the same order every run.
* The scheduler varies which processor gets a chance to run first.
* This simulates uncertainty from processing time, scheduling priority, startup timing, or transmission delay.
* The graph example should converge to the same final path facts even when processor scheduling, and hence the order of intermediate derived processor events, differs.

; File edit

Create:

<pre lang="text">
demos/event_processors/runners.py
</pre>

<pre lang="python">
#!/usr/bin/env python3
# demos/event_processors/runners.py

"""Small processor runners for event processor demonstrations."""

from collections.abc import Callable
from dataclasses import dataclass, field
from random import Random

@dataclass(frozen=True, kw_only=True)
class ProcessorStep:
name: str
run: Callable[[], int]

@dataclass
class RandomProcessorRunner:
steps: tuple[ProcessorStep, ...]
seed: int
max_rounds: int = 30
trace: list[tuple[int, str, int]] = field(default_factory=list)

def run_until_quiet(self) -> list[tuple[int, str, int]]:
rng = Random(self.seed)

for round_index in range(self.max_rounds):
work_count = self._run_round(rng, round_index)

if work_count == 0:
break

result = list(self.trace)
return result

def _run_round(self, rng: Random, round_index: int) -> int:
steps = list(self.steps)
rng.shuffle(steps)
round_work_count = 0

for step in steps:
step_work_count = step.run()
self.trace.append((round_index, step.name, step_work_count))
round_work_count += step_work_count

return round_work_count
</pre>

; Interpreter check

Restart the interpreter, then inspect how a runner can shuffle placeholder steps:

<pre lang="python">
from demos.event_processors.runners import (
ProcessorStep,
RandomProcessorRunner,
)

calls = []

def first_step() -> int:
calls.append("first")
return 0

def second_step() -> int:
calls.append("second")
return 0

runner = RandomProcessorRunner(
steps=(
ProcessorStep(name="first", run=first_step),
ProcessorStep(name="second", run=second_step),
),
seed=1,
)

runner.run_until_quiet()
calls
</pre>

Expected output shape:

<pre lang="text">
[(0, '...', 0), (0, '...', 0)]
['...', '...']
</pre>

== Phase 15: Run the graph example with randomized scheduling ==

; Goal

Run the same graph input under different processor schedules and compare the final path facts.

; Interpreter setup

<pre lang="python">
from mbus.broker.direct import DirectMessageBus
from demos.event_processors.capture_history import (
AOStoreCaptureSink,
CaptureHistory,
)
from demos.event_processors.graph_events import (
EdgeDeclaredEvent,
PathKnownEvent,
)
from demos.event_processors.graph_reachability import (
DirectPathProcessor,
PathExtensionProcessor,
)
from demos.event_processors.runners import (
ProcessorStep,
RandomProcessorRunner,
)
</pre>

Define a helper function in the interpreter:

<pre lang="python">
def run_graph_demo(seed: int) -> tuple[
list[tuple[int, str, int]],
list[tuple[int, str, str, int]],
]:
bus = DirectMessageBus()
sink = AOStoreCaptureSink()
bus.set_capture_sink(sink)

edge_emitter = bus.register_emitter(
msg_topic="demo.graph.edge",
msg_producer="graph-source",
default_msg_type="edge-declared",
)

direct_processor = DirectPathProcessor(bus)
extension_processor = PathExtensionProcessor(bus)

edges = (
EdgeDeclaredEvent(
graph_id="demo-graph",
source="A",
target="B",
),
EdgeDeclaredEvent(
graph_id="demo-graph",
source="B",
target="C",
),
EdgeDeclaredEvent(
graph_id="demo-graph",
source="C",
target="D",
),
)

for edge in edges:
edge_emitter.emit(edge)

steps = (
ProcessorStep(
name="direct path from edge",
run=direct_processor.process_available,
),
ProcessorStep(
name="extension learns edge",
run=extension_processor.process_edge_available,
),
ProcessorStep(
name="extension learns direct path",
run=extension_processor.process_direct_path_available,
),
ProcessorStep(
name="extension learns extension path",
run=extension_processor.process_extension_path_available,
),
)

runner = RandomProcessorRunner(
steps=steps,
seed=seed,
)
trace = runner.run_until_quiet()

history = CaptureHistory(sink.all_records())
path_facts = []

for message in history.by_topic("demo.graph.path"):
path = message.payload
if isinstance(path, PathKnownEvent):
fact = (
message.bus_sequence,
path.source,
path.target,
path.hop_count,
)
path_facts.append(fact)

result = (trace, path_facts)
return result
</pre>

Run one schedule:

<pre lang="python">
trace_1, paths_1 = run_graph_demo(1)

trace_1
paths_1
</pre>

Expected output shape:

<pre lang="text">
[(0, '...', ...), (0, '...', ...), ...]
[(..., 'A', 'B', 1), (..., 'B', 'C', 1), (..., 'C', 'D', 1), ...]
</pre>

Run another schedule:

<pre lang="python">
trace_2, paths_2 = run_graph_demo(2)

trace_2
paths_2
</pre>

Compare the schedules:

<pre lang="python">
trace_1 == trace_2
</pre>

Expected output:

<pre lang="text">
False
</pre>

Compare the final path facts without bus sequence numbers:

<pre lang="python">
final_paths_1 = {
(source, target, hop_count)
for _, source, target, hop_count in paths_1
}

final_paths_2 = {
(source, target, hop_count)
for _, source, target, hop_count in paths_2
}

final_paths_1
final_paths_2
final_paths_1 == final_paths_2
</pre>

Expected output:

<pre lang="text">
{('A', 'B', 1), ('B', 'C', 1), ('C', 'D', 1), ('A', 'C', 2), ('B', 'D', 2), ('A', 'D', 3)}
{('A', 'B', 1), ('B', 'C', 1), ('C', 'D', 1), ('A', 'C', 2), ('B', 'D', 2), ('A', 'D', 3)}
True
</pre>

Inspect the first schedule:

<pre lang="python">
for round_index, step_name, work_count in trace_1:
print(round_index, step_name, work_count)
</pre>

Inspect the first captured path order:

<pre lang="python">
for bus_sequence, source, target, hop_count in paths_1:
print(bus_sequence, source, target, hop_count)
</pre>

Inspect the second schedule and path order:

<pre lang="python">
for round_index, step_name, work_count in trace_2:
print(round_index, step_name, work_count)

for bus_sequence, source, target, hop_count in paths_2:
print(bus_sequence, source, target, hop_count)
</pre>

; What this phase shows

The source input order did not change. The processor scheduling changed. The captured event order may differ. The final reachability facts should match.

This models an important distributed-systems lesson: event history order and final derived state are related, but they are not the same thing.

== Phase 16: Later refinement: graph index from history ==

; Goal

Introduce restart/recovery thinking only after the basic graph feedback example is familiar.

; Discussion points

* The first graph processors keep local working state.
* Local working state is enough for a short demonstration.
* Longer-lived processors need recovery behavior after restart.
* Since edge and path facts are captured as events, a graph index can later be rebuilt from history.
* This points toward projection-style services without changing the basic processor model.

; Suggested later file

Create later:

<pre lang="text">
demos/event_processors/graph_index.py
</pre>

Suggested responsibilities:

<pre lang="text">
read captured history
collect EdgeDeclaredEvent payloads
collect PathKnownEvent payloads
answer whether a path fact exists
find edges starting at a node
find paths ending at a node
</pre>

This refinement belongs after the team has already run the graph reachability demo successfully.

== Phase 17: Later refinement: shell command reconstruction ==

; Goal

Connect the TTY examples back to the shell command event shape used by regex milestones.

; Discussion points

* <code>ShellCommandEvent</code> is the high-level event consumed by the regex milestone processor.
* <code>TTYReadEvent</code>, <code>TTYWriteEvent</code>, and <code>LineDisciplineEvent</code> are lower-level observations.
* A reconstructor can subscribe to line-discipline events and query recent related observations.
* The reconstructor should emit shell command events; it should not perform regex analysis itself.

; Suggested responsibilities

<pre lang="text">
listen for completed line-discipline events
query history since the previous command boundary for the same session
collect matching reads, writes, and timing events
construct ShellCommandEvent
emit ShellCommandEvent on demo.shell.command
</pre>

This is an appropriate follow-up exercise after the team understands both regex milestone processing and history queries.

== Related ==

=== Top-level topic page ===
[[Message Bus Service]]

2026-04-03T20:42:34Z

Jwgranville: /* The EDURange Project */

Team Charter

2026-03-31T19:24:54Z

Jwgranville:

change me!

See https://www.pmi.org/learning/library/team-charter-development-5128 !

Team Charter

2026-03-31T19:23:51Z

Jwgranville: Created page with "change me!"

change me!

2026-01-24T01:18:05Z

Jwgranville: /* Developer Links */

Welcome to the EDURange Wiki. This is less of a formal documentation as it is a living document.

I think that if it's a choice between losing track of people's contributions and having a messy wiki, I'd much prefer a messy wiki. It's my hope that we can gather all of our notes, thoughts, and loosely related ephemera here, rather than having it scattered across correspondence and various sharing platforms.

We don't have many conventions yet. Please follow the Golden Rule and treat others as you would be treated. Put your name on your changes. Consider posting to the discussion tab or appending to a page rather than overwriting it. Try not to overwrite the personal work of others.

If it's installable/runnable/code of any sort, put it on GitHub as well - preferably under the edurange organization so that our future contributors will have seamless access to it too.

Thanks for your time.

~Joe G

== The EDURange Project ==
[[:Category:Project History and Roadmaps|Project History and Roadmaps]]

== Developer Links ==
!!! [[New Version Planning]] !!!

[[Style Guidelines and Developer Tools]]

[[Reference Materials]]

[[:Category:Demonstrations|Demonstrations]]

[[Design Stories (OLD)]]

[[Interface Design]]

[[Scenario Creation Guide]]

[[Terraform Design]]

[[Hosting Requirements]]

[[Domain Modeling]]

We're still figuring out Media Wiki, so I kept the following...

== Orphaned Default Media Wiki Content ==
<strong>Welcome to the EDURange Wiki</strong>

Consult the [https://www.mediawiki.org/wiki/Special:MyLanguage/Help:Contents User's Guide] for information on using the wiki software.

== Getting started with Media Wiki ==
* [https://www.mediawiki.org/wiki/Special:MyLanguage/Manual:Configuration_settings Configuration settings list]
* [https://www.mediawiki.org/wiki/Special:MyLanguage/Manual:FAQ MediaWiki FAQ]
* [https://lists.wikimedia.org/postorius/lists/mediawiki-announce.lists.wikimedia.org/ MediaWiki release mailing list]
* [https://www.mediawiki.org/wiki/Special:MyLanguage/Localisation#Translation_resources Localise MediaWiki for your language]
* [https://www.mediawiki.org/wiki/Special:MyLanguage/Manual:Combating_spam Learn how to combat spam on your wiki]

New Version Planning

2026-01-24T00:54:52Z

Jwgranville: