🏛
Government: Inter-Agency Incident Response
An emergency response system where causal tracing connects 911 dispatch, first responder field operations, and interagency command coordination into a single auditable timeline.
★ Important
After-action reviews require a causal timeline, not a timestamp log. PyRapide's causal graph answers "why was this unit dispatched?", "what caused the escalation?", and "was the backup request fulfilled?" with machine-verifiable precision.
Architecture Overview
- DispatchCenter receives 911 calls, dispatches units, and tracks incident status.
- FieldUnit covers first responders reporting en route, on scene, situation assessments, and backup requests.
- CommandCenter handles interagency coordination, escalation decisions, resource allocation, and after-action reports.
Interfaces
incident_interfaces.py python
1from pyrapide import interface, action, module, when
2from pyrapide import architecture, connect, Pattern, Engine
3from pyrapide import must_match, never
4import asyncio
5
6# ── Interfaces ──────────────────────────────────────────
7
8@interface
9class DispatchCenter:
10 """911/emergency dispatch center."""
11 @action
12 async def incident_reported(self, incident_id: str, type: str,
13 location: str, priority: str) -> None: ...
14 @action
15 async def units_dispatched(self, incident_id: str,
16 unit_ids: list[str]) -> None: ...
17 @action
18 async def incident_updated(self, incident_id: str,
19 status: str, details: str) -> None: ...
20
21@interface
22class FieldUnit:
23 """First responder field unit (police, fire, EMS)."""
24 @action
25 async def en_route(self, unit_id: str, incident_id: str) -> None: ...
26 @action
27 async def on_scene(self, unit_id: str, incident_id: str) -> None: ...
28 @action
29 async def situation_report(self, unit_id: str, incident_id: str,
30 assessment: str) -> None: ...
31 @action
32 async def requesting_backup(self, unit_id: str, incident_id: str,
33 resource_type: str) -> None: ...
34 @action
35 async def incident_resolved(self, unit_id: str,
36 incident_id: str) -> None: ...
37
38@interface
39class CommandCenter:
40 """Interagency command and coordination center."""
41 @action
42 async def escalation(self, incident_id: str, level: str,
43 agencies: list[str]) -> None: ...
44 @action
45 async def resource_allocated(self, incident_id: str,
46 resource_type: str, source: str) -> None: ...
47 @action
48 async def after_action_report(self, incident_id: str,
49 timeline: list[dict]) -> None: ...
Module Logic
incident_modules.py python
1# ── Modules ─────────────────────────────────────────────
2
3@module(implements=CommandCenter)
4class IncidentCommand:
5 @when("FieldUnit.requesting_backup")
6 async def handle_backup_request(self, event):
7 await self.resource_allocated(
8 incident_id=event.data["incident_id"],
9 resource_type=event.data["resource_type"],
10 source="mutual_aid"
11 )
12
13 @when("FieldUnit.situation_report")
14 async def evaluate_escalation(self, event):
15 if "mass_casualty" in event.data["assessment"].lower():
16 await self.escalation(
17 incident_id=event.data["incident_id"],
18 level="level_3",
19 agencies=["fire", "ems", "police", "national_guard"]
20 )
21
22@module(implements=FieldUnit)
23class FirstResponder:
24 @when("DispatchCenter.units_dispatched")
25 async def respond_to_dispatch(self, event):
26 for unit_id in event.data["unit_ids"]:
27 await self.en_route(
28 unit_id=unit_id,
29 incident_id=event.data["incident_id"]
30 )
Architecture and Constraints
The constraints encode accountability and standard operating procedures:
- incident_must_dispatch: every reported incident must result in units being dispatched.
- dispatch_must_respond: dispatched units must acknowledge and go en route.
- on_scene_must_report: units on scene must file a situation report.
- backup_must_be_fulfilled: every backup request must result in resource allocation.
- incident_must_have_aar: every incident must produce an after-action report.
- no_resolve_without_sitrep: an incident must never be resolved without a situation report.
incident_architecture.py python
1# ── Architecture ────────────────────────────────────────
2
3@architecture
4class IncidentResponseSystem:
5 dispatch: DispatchCenter
6 field: FieldUnit
7 command: CommandCenter
8
9 def connections(self):
10 return [
11 # Dispatch sends to field units
12 connect(Pattern.match("DispatchCenter.units_dispatched"), "field"),
13 # Field reports go to both dispatch and command
14 connect(Pattern.match("FieldUnit.situation_report"), "dispatch"),
15 connect(Pattern.match("FieldUnit.situation_report"), "command"),
16 connect(Pattern.match("FieldUnit.requesting_backup"), "command"),
17 # Command decisions go to dispatch for execution
18 connect(Pattern.match("CommandCenter.*"), "dispatch"),
19 ]
20
21 def constraints(self):
22 return [
23 # Every incident must result in units being dispatched
24 must_match(
25 trigger="DispatchCenter.incident_reported",
26 response="DispatchCenter.units_dispatched",
27 name="incident_must_dispatch"
28 ),
29
30 # Every dispatched unit must report en_route
31 must_match(
32 trigger="DispatchCenter.units_dispatched",
33 response="FieldUnit.en_route",
34 name="dispatch_must_respond"
35 ),
36
37 # Every unit on scene must file a situation report
38 must_match(
39 trigger="FieldUnit.on_scene",
40 response="FieldUnit.situation_report",
41 name="on_scene_must_report"
42 ),
43
44 # A backup request must result in resource allocation
45 must_match(
46 trigger="FieldUnit.requesting_backup",
47 response="CommandCenter.resource_allocated",
48 name="backup_must_be_fulfilled"
49 ),
50
51 # Every incident must eventually have an after-action report
52 must_match(
53 trigger="DispatchCenter.incident_reported",
54 response="CommandCenter.after_action_report",
55 name="incident_must_have_aar"
56 ),
57
58 # An incident must never be resolved without
59 # a situation report first
60 never(
61 pattern=("FieldUnit.en_route",
62 "FieldUnit.incident_resolved"),
63 unless="FieldUnit.situation_report",
64 name="no_resolve_without_sitrep"
65 ),
66 ]
Execution and After-Action Review
incident_analysis.py python
1# ── Execute and Generate After-Action Report ───────────
2
3async def main():
4 engine = Engine()
5 computation = await engine.run(IncidentResponseSystem)
6
7 from pyrapide import (
8 critical_path, root_causes, backward_slice,
9 forward_slice, causal_distance, to_mermaid
10 )
11
12 # Generate the critical path for incident response time
13 path = critical_path(computation)
14 print("Critical response path:")
15 for i, event in enumerate(path):
16 print(f" {i+1}. {event.name} ({event.timestamp})")
17
18 # For any escalation, trace back to the root incident
19 escalations = [e for e in computation.events
20 if e.name == "CommandCenter.escalation"]
21 for esc in escalations:
22 causes = root_causes(computation, esc)
23 chain = backward_slice(computation, esc)
24 print(f"\nEscalation to {esc.data['level']}:")
25 print(f" Root incident: {causes[0].data}")
26 print(f" Causal chain: {len(chain)} events")
27
28 # Export the full causal graph for the after-action review
29 mermaid = to_mermaid(computation)
30 with open("incident_timeline.md", "w") as f:
31 f.write(f"```mermaid\n{mermaid}\n```")
32
33asyncio.run(main())
💡 Tip
Export the causal graph with
to_mermaid to generate visual timelines for after-action review boards. The graph shows exactly which events caused which escalations and resource allocations.