CityOS Under SI-Core: A Worked Example Across All Invariants
Community Article Published
February 8, 2026
Draft v0.1 — Non-normative supplement to SI-Core / SI-NOS / CityOS design
This document is non-normative. It shows how one possible CityOS could look when it is run “under” SI-Core / SI-NOS, using the invariants and patterns from the other 60-series notes.
Normative contracts live in the SI-Core / SI-NOS / SIM/SIS / EVAL / ETH specs and in the domain-specific CityOS design docs.
0. Conventions used in this draft (non-normative)
This draft follows the portability conventions used in 069/084+ when an artifact might be exported, hashed, or attested (ETH policies/constraints, appeals & overrides, break-glass procedures, review packets, board minutes, compliance bundles):
created_atis operational time (advisory unless time is attested).as_ofcarries markers only (time claim + optional revocation view markers) and SHOULD declareclock_profile: "si/clock-profile/utc/v1"when exported.trustcarries digests only (trust anchors + optional revocation view digests). Never mix markers intotrust.bindingspins meaning as{id,digest}(meaningful identities must not be digest-only).- Avoid floats in policy-/digest-bound artifacts: prefer scaled integers (
*_bp,*_ppm) and integer micro/milliseconds. - If you hash/attest legal/procedural artifacts, declare canonicalization explicitly:
canonicalization: "si/jcs-strict/v1"andcanonicalization_profile_digest: "sha256:...". digest_rulestrings (when present) are explanatory only; verifiers MUST compute digests using pinned schemas/profiles, not by parsingdigest_rule.
Numeric conventions used in examples:
- For weights and ratios in
[0,1], export as basis points:x_bp = round(x * 10000). - For probabilities in
[0,1], export as basis points:p_bp = round(p * 10000). - If a clause uses a legal threshold with decimals, either:
- export
value_scaled_int+ explicitscale, or - keep the raw numeric payload in a referenced artifact and export only its digest (
*_ref/*_hash) plus display-only summaries.
- export
Internal computation may still use floats; the convention here is about exported/hashed representations (e.g., policy releases, review packets, compliance bundles).
1. Scenario: a flood-aware CityOS
We follow a mid-size city that runs a CityOS with three coupled subsystems:
- Water / flood control — canals, pumps, gates, retention basins.
- Mobility / traffic — main roads, tunnels, bridges, public transit.
- Hospitals / health — ER capacity, ICU beds, ambulance routing.
We walk through four phases:
- Normal day — routine optimization.
- Heavy-rain forecast — pre-emptive reconfiguration.
- Actual flooding — emergency response.
- Recovery — reopening and postmortem.
At each phase, we show how [OBS] / [ID] / [ETH] / [MEM] / [EVAL] / Jump / RML / PoLB / GRP show up structurally.
2. Invariants → CityOS mapping
A non-normative mapping table:
| Invariant | CityOS object | Examples |
|----------|---------------|----------|
| [OBS] Observation | Observation bundles from sensors & semantic compression | Flood risk state, traffic state, hospital load, weather forecast |
| [ID] Identity & principals | Human / org / SI principals; roles & personas | Mayor, CityOps operator, HospitalOps, Resident, Emergency SI agent |
| [ETH] Ethics & constraints | Domain ethics policies | “No road closure that blocks all hospital access”, fairness across districts |
| [MEM] Memory | SIM/SIS, EvalTrace, Genius Library, AuditLog | Flood incident history, per-district risk summaries, experiment logs |
| [EVAL] Evaluation surfaces | Goal / Eval surfaces for city | Safety, access, fairness, economic damage, citizen wellbeing |
| Jump | Atomic decision moves | “Close gate X 30%”, “Re-route buses”, “Trigger SMS alerts in sector 12” |
| RML | Real-world effect layer | Gate actuations, signal changes, notification sends |
| PoLB | Policy Load Balancer | Progressive launch/backoff/degradation & blast-radius control (canary, shadow, waves) |
| GRP | Genius Replay Protocol | Reusing high-value “flood playbooks” from past events |
| SCover / SCI / CAS | Observability & forensics | Coverage across flood/traffic/hospital SIRs, incident counts, replay stability |
CityOS becomes just another SI-Core governed stack:
- each critical operation sits behind a Jump+RML boundary,
- every effectful decision is traceable in SIR / JumpTrace / RMLTrace / EthicsTrace,
- every “clever trick” (Genius) is replayable and auditable, not magic.
3. Phase 0: baseline configuration & roles
Before we look at events, we define roles/personas and goal surfaces.
3.1 Roles & personas
Non-normative examples:
roles:
- id: role:city_ops
capabilities:
- "view_city_risk"
- "approve_gate_profiles"
- "coordinate_cross_domain"
- "approve_high_risk_jump"
- id: role:traffic_ops
capabilities:
- "view_traffic_state"
- "propose_traffic_plan"
- "override_traffic_plans"
- id: role:hospital_ops
capabilities:
- "view_hospital_load"
- "request_ambulance_reroutes"
- id: role:mayor
capabilities:
- "view_citywide_risk"
- "declare_emergency"
- "approve_emergency_mandate"
- "request_postmortem_report"
- id: role:auditor
capabilities:
- "view_eth_traces"
- "view_eval_traces"
- "view_audit_logs"
- "request_replay"
- id: role:resident
capabilities:
- "view_local_risk"
- "receive_alerts"
- id: role:emergency_si_agent
capabilities:
- "propose_emergency_plan"
- "execute_within_preapproved_rml_budget"
- "request_human_approval_if_over_budget"
rml_budget: # illustrative; enforce in ID+ETH before RML
max_rml_level: 2
allowlisted_effects:
- "gate.actuate"
- "road.close"
- "signal.update_pattern"
- "notify.send"
require_human_approval_for:
- "evacuation.order"
- "infrastructure.shutdown"
max_population_share_bp: 1000 # 10% (1000 bp)`
expires_after: "2h"
personas:
- id: persona:city_ops_console
role: role:city_ops
explanation_style: "technical"
metric_projection: "ops_view"
- id: persona:traffic_ops_console
role: role:traffic_ops
explanation_style: "technical"
metric_projection: "ops_view"
- id: persona:hospital_ops_console
role: role:hospital_ops
explanation_style: "technical"
metric_projection: "ops_view"
- id: persona:mayor_dashboard
role: role:mayor
explanation_style: "simple"
metric_projection: "policy_view"
- id: persona:auditor_view
role: role:auditor
explanation_style: "formal"
metric_projection: "audit_view"
- id: persona:resident_app
role: role:resident
explanation_style: "simple"
metric_projection: "resident_view"
- id: persona:emergency_si_console
role: role:emergency_si_agent
explanation_style: "technical"
metric_projection: "ops_view"
3.2 City goal surfaces
A compact sketch of a CityOS GoalSurface for floods:
goal_surface:
id: "city.flood_response/v1"
domain: "city"
# Convention (illustrative):
# - weight is positive and expresses importance
# - direction expresses maximize/minimize
# NOTE: Use weight_bp for contracts/digests; derive float weight = weight_bp/10000.0 inside the runtime.
objectives:
primary:
- { id: "safety", direction: "maximize", weight_bp: 4000 }
- { id: "hospital_access", direction: "maximize", weight_bp: 3000 }
- { id: "economic_damage", direction: "minimize", weight_bp: 2000 }
- { id: "fairness", direction: "maximize", weight_bp: 1000 }
constraints:
hard:
- id: "hospital_access_min"
expr: "min(district.hospital_access_score_bp) >= 8000"
- id: "evac_death_risk_max_ppm"
# 10 ppm = 1e-5 (avoid floats in exported/policy-bound artifacts)
expr: "max(route.estimated_death_risk_ppm) <= 10"
soft:
- id: "economic_damage_soft_cap"
expr: "economic_damage_ratio_bp <= 12000 OR safety_at_risk == true" # ratio in basis points
Later, we’ll see E-Jumps (experiment Jumps) that define EvalSurfaces over this GoalSurface, and G-Jumps that replay previously validated Genius traces.
4. Phase 1: normal day — routine Jumps
On a normal day, CityOS is still making decisions, but PoLB is in conservative mode and Genius usage is minimal.
4.1 Observations ([OBS]) via semantic compression
Sensors → SCE → SIM produce semantic city states:
obs: city_state/v1
flood:
segments:
- id: "canal-12"
risk_score_bp: 1200
level_mm: 1300
trend: "stable"
traffic:
roads:
- id: "artery-n1"
congestion_index_bp: 4000
incidents: 0
hospitals:
- id: "general-hospital"
er_load_bp: 6500
icu_load_bp: 7200
weather:
forecast:
rain_mm_next_6h: 3
heavy_rain_prob_bp: 500
[OBS] invariants:
- all observations come as typed semantic units (from the Semantic Compression article),
- raw streams are accessible through
backing_refsif needed, but not used as primary inputs for Jumps.
4.2 Example Jump: traffic optimization
A low-stakes routine Jump:
jump city.optimize_traffic {
input obs: CityTrafficObservation
input goals: GoalSurface<city.traffic>
input constraints: ETHConstraints
plans = generate_candidate_signal_plans(obs.road_state)
best = argmax(plans, p => goals.evaluate(p, obs))
eth_decision = eth.check(best)
# Decision shape alignment (series):
# ETHDecision: { decision: "ALLOW"|"DENY"|"ESCALATE", reason_code?: str, details?: any }
if eth_decision.decision != "ALLOW":
# DENY -> blocked; ESCALATE -> routed (implementation choice)
return JumpResult(blocked=True, eth_decision=eth_decision)
# NOTE: The Jump remains side-effect-free until RML execution.
effect rml_level=1 {
rml.emit("traffic.apply_signal_plan", args=best)
compensator: rml.emit("traffic.restore_previous_plan")
}
}
- [ETH] is relatively simple here (fairness, max delay bounds).
- [MEM] records the JumpTrace and RMLTrace.
- PoLB for this Jump type might be always-on (no special staging).
Even in “boring mode”, every structural element is already in place; the emergency phases reuse the same plumbing.
5. Phase 2: heavy-rain forecast — pre-emptive E-Jumps & PoLB
A severe weather service issues a forecast:
weather:
forecast:
rain_mm_next_6h: 80
heavy_rain_prob_bp: 8500
high_wind_prob_bp: 6000
The E-layer (experiments) kicks in: we want to test new flood-gate strategies and traffic patterns before the worst hits.
5.1 E-Jump: design a safe experiment
We use the “Evaluation as Goal Surface” ideas, but now domain-specific:
class FloodExperimentProposer:
def propose(self, eval_surface, candidates, polb_config):
"""
Propose an experiment on flood-gate policies, subject to ETH and PoLB.
"""
# 1. Minimum detectable effect on safety
effect_size = self._min_detectable_effect(eval_surface, metric="safety")
# 2. Sample size and rollout duration
sample_size = self.sample_size_calc.calculate(eval_surface, effect_size)
# 3. Spatial PoLB: only one basin cluster as canary (portable-friendly: bp)
# 5% = 500 bp
canary_districts = self._select_canary_districts(max_population_share_bp=500)
assignment = self._design_assignment(
population=canary_districts,
candidates=candidates,
sample_size=sample_size,
polb_config=polb_config,
)
monitoring = MonitoringPlan(
metrics=eval_surface.objectives.primary + eval_surface.constraints.hard,
check_frequency="5min",
alert_thresholds=self._derive_alerts(eval_surface),
)
stop_rules = self._design_stop_rules(eval_surface, sample_size)
return FloodExperimentDesign(
assignment_scheme=assignment,
monitoring_plan=monitoring,
stop_rules=stop_rules,
# Keep naming consistent with later PoLB config blocks in this article.
mode="online", # envelope: online|shadow|sandbox|degraded
mode_name="ONLINE_EXPERIMENTAL_STRATIFIED",
risk_band="ONLINE_EXPERIMENTAL",
)
Here PoLB is used geographically:
- only a small district sees the new policy,
- the rest of the city remains on the baseline flood policy,
- EVAL defines both what we’re measuring and what counts as “harm”.
5.2 ETH-aware constraints on experiments
[ETH] adds extra constraints:
eth_constraints_for_flood_experiments:
hard:
- "No experiment may reduce hospital_access_score_bp below 9000 in any district"
- "No experiment on emergency routes during active flood warning"
soft:
- "Avoid assigning high-risk variants to historically under-served districts"
Every E-Jump (experiment design) is treated as a first-class action:
- has a principal (
role:city_ops), - is logged in [MEM] (EvalTrace + EthicsTrace),
- can be blocked by ETH if risk is too high (e.g., experiment in ER access during peak load).
6. Phase 3: the flood — emergency Jump+RML
The rain arrives. Water levels spike in several basins. We move into emergency mode.
6.1 Emergency observation bundles
[OBS] aggregates high-frequency semantics:
obs: city_state/v1
flood:
segments:
- id: "canal-12"
risk_score_bp: 9100
level_mm: 3400
trend: "rising_fast"
- id: "tunnel-north"
water_cm_on_road: 15
traffic:
roads:
- id: "artery-n1"
congestion_index_bp: 8000
incidents: 2
hospitals:
- id: "general-hospital"
er_load_bp: 9400
icu_load_bp: 9800
ambulance_queue: 7
Semantic compression policies auto-tighten (tighter loss budget, ε_loss) for critical streams, as described in the semantic compression article. (Terminology: ε_loss here is compression fidelity budget, not differential-privacy ε.)
6.2 Emergency Jump: flood + traffic + hospital in one SIR
We want a single SIR that coordinates all three:
sir city.emergency_flood_response {
obs_bundle: CityStateObservation
goals: GoalSurface<city.flood_response>
role_persona: persona:emergency_si_console
steps:
- jump: "city.compute_flood_options"
- jump: "city.compute_traffic_options"
- jump: "city.compute_hospital_load_sharing"
- jump: "city.compose_joint_plan"
- eth_check: "city_plan_eth"
- eval: "city_plan_risk_eval"
- rml: "apply_city_plan"
}
Each Jump is atomic but structurally chained in this SIR.
6.2.1 Example: compute flood options
jump city.compute_flood_options {
input obs: FloodObservation
input goals: GoalSurface<city.flood_response>
input constraints: ETHConstraints
candidates = generate_gate_schedules(obs, horizon=6*h)
ranked = rank_by_gcs(candidates, goals)
return JumpDraft(
primary_action = ranked[0],
alternatives = ranked[1:5],
gcs_vector = estimate_gcs(ranked[0], goals),
)
}
6.2.2 Example: compose joint plan
jump city.compose_joint_plan {
input flood_plan: FloodGatePlan
input traffic_plan: TrafficDiversionPlan
input hospital_plan: HospitalLoadPlan
input goals: GoalSurface<city.flood_response>
joint_candidates = combine_plans(
flood_plan.alternatives,
traffic_plan.alternatives,
hospital_plan.alternatives,
)
best = argmax(joint_candidates, p => goals.evaluate(p))
return JumpDraft(primary_action=best, gcs_vector=goals.evaluate(best))
}
6.3 RML: applying the emergency plan
RML splits the joint plan into concrete effect calls:
rml_call: apply_city_plan
effects:
- type: gate.actuate
id: "canal-12-gate-1"
target_position_bp: 8000
compensator: "restore_gates_baseline_profile"
- type: road.close
id: "tunnel-north"
compensator: "reopen_when_safe"
- type: signal.update_pattern
id: "artery-n1-east"
pattern_id: "emergency_corridor_1"
- type: notification.send
channel: "sms"
audience: "residents_in_zone_12"
template_id: "flood_warning_v3"
[ETH] verifies:
- no hospital is cut off,
- no disadvantaged district gets disproportionate closure burden,
- no RML call exceeds its role-capability budget (from the Role & Persona article).
[MEM] stores RMLTrace, JumpTrace, EthicsTrace, EvalTrace, SCover/SCI/CAS metadata.
7. Genius Replay in CityOS (GRP in context)
Now suppose we had a previous “near-perfect” response to a similar flood — a Genius Trace.
7.1 Genius Trace structure for CityOS
Non-normative:
genius_trace:
id: "GRP-2030-09-flood-north-basin"
domain: "city.flood_response"
context_signature:
basin_topology: "north"
hospital_layout: "v2"
population_profile: "weekday_evening"
goal_surface_snapshot: "city.flood_response@2030-09-01"
jump_sequence:
- "city.compute_flood_options@v3"
- "city.compute_traffic_options@v2"
- "city.compute_hospital_load_sharing@v1"
- "city.compose_joint_plan@v1"
eval_summary:
gcs_improvement_bp: 2300
fairness_score_bp: 9100
safety_incidents: 0
robustness:
success_rate_simulated_bp: 9200
worst_case_delta_gcs_bp: -300
status: "active"
7.2 Genius Replay engine in CityOS
When the new flood looks similar, GRP proposes a replay:
class CityGeniusReplayer:
def maybe_replay(self, current_context):
# 1. Candidate Genius traces for this domain
candidates = self.genius_matcher.find_candidates(
domain="city.flood_response",
context=current_context,
)
if not candidates:
return None # fall back to normal engine
best = max(candidates, key=lambda c: c.similarity)
# 2. Pre-replay safety checks (from GRP article)
ok, checks = self.safety_checker.pre_replay_check(
genius_trace=best,
current_context=current_context,
)
if not ok:
return None
# 3. Structural replay
return self.replay_sequence(best, current_context)
ETH / EVAL / PoLB still apply:
- we might only allow Genius replay for districts within some similarity / risk band,
- [EVAL] monitors that GCS is at least as good as baseline,
- PoLB can “canary” the replay to a subset of the city before global rollout.
8. Phase 4: recovery, postmortem, and learning
Once water recedes, CityOS enters a recovery phase:
- reopen roads,
- reset flood gates,
- normalize hospital flows,
- run a structured postmortem across SIRs.
8.1 Recovery Jumps
Recovery is still effectful, so it uses the same Jump/RML pattern:
jump city.recovery_plan {
input obs: PostFloodObservation
input goals: GoalSurface<city.recovery>
input constraints: ETHConstraints
actions = propose_reopening_actions(obs)
best = argmax(actions, a => goals.evaluate(a))
require constraints.check(best) else escalate
effect rml_level=1 {
execute_reopening(best)
compensator: "revert_to_previous_closure_state"
}
}
8.2 Postmortem using structural observability
We now use SCover / SCI / CAS in a post-incident SIR cluster:
- SCover: did we trace all the blocks we expected?
- SCI: where did ETH vs RML, Jump vs RML, Eval vs RML diverge?
- CAS: deterministic replay of Jumps — any drift due to non-determinism or missing inputs?
Non-normative queries:
sirs = trace_query_engine.query_by_domain_timerange(
domain="city.flood_response",
start="2031-05-10T00:00Z",
end="2031-05-11T00:00Z",
)
scover = scover_calculator.calculate_scover("city.flood_response", time_range=...)
sci_incidents = sci_detector.scan_domain("city.flood_response", time_range=...)
cas_sample = cas_calculator.sample_and_replay(sirs, sample_rate_bp=100) # 1% = 100 bp
[MEM] also runs Genius selection:
- some Jumps sequences might be candidates for new Genius traces (GRP),
- others might be anti-examples (things to never replay).
8.3 Memory governance and forgetting
Recovery is also when memory governance kicks in:
raw data for some streams can be aggressively forgotten, keeping only:
- semantic aggregates,
- structural traces,
- redacted Genius traces.
personal data in alerts / hospital records may be pseudonymized or erased per GDPR-compatible policies.
The Reconstruction Risk Score (RRS) and GCS-aware forgetting logic decide how to:
- reduce long-term privacy risk,
- without compromising future flood preparedness.
9. CI/CD, rollout, and PoLB in CityOS
So far, we have assumed a fixed set of algorithms. In reality, CityOS code changes:
- new flood models,
- new traffic routing heuristics,
- revised ETH policies,
- new GRP selection logic.
9.1 Release waves and PoLB modes
Non-normative PoLB configuration for CityOS:
polb_config:
domains:
city.flood_response:
waves:
- name: "shadow_prod"
mode: "shadow" # envelope: online|shadow|sandbox|degraded
mode_name: "SHADOW_PROD" # detailed PoLB profile (art-60-038)
risk_band: "SHADOW"
# shadow runs the logic for broad coverage but emits no live RML.
eval_coverage_share_bp: 10000
rml_effects: "none"
- name: "district_canary"
mode: "online"
mode_name: "ONLINE_EXPERIMENTAL_STRATIFIED"
risk_band: "ONLINE_EXPERIMENTAL"
population_share_bp: 500
constraints:
- "Only low-risk districts"
- "No hospitals fully dependent on canary districts"
- name: "citywide_gradual"
mode: "online"
mode_name: "ONLINE_SAFE_BASELINE"
risk_band: "ONLINE_SAFE"
population_share_bp: 10000
rollout_duration: "6h"
health_gates:
- "SCI == 0 in last 1h"
- "SCover_bp >= 9000"
- "CAS_bp >= 9700 for sampled SIRs (DET replay sample)"
PoLB is directly tied into:
- SCover / SCI / CAS metrics,
- Experiment risk assessment (EVAL),
- ETH (e.g. preventing high-risk treatment of vulnerable regions).
9.2 CI/CD workflow with SI-Core hooks
Every new flood policy version:
- gets a release ID (
city.flood_policy@v5.2), - is tested in shadow mode (no RML),
- is rolled out via PoLB waves,
- has post-release SCover/SCI/CAS dashboards for SRE / CityOps.
If a P0 incident occurs:
- auto-rollback can use RBL/RIR metrics and RML compensators,
- Genius traces attached to the previous “good version” remain reusable, with versioning.
10. Personas and explainability in CityOS
We reuse the Role & Persona Overlay to adapt explanations.
10.1 Resident view
For a resident in zone 12:
metrics are projected to simple concepts:
- “local flood risk: high / medium / low”,
- “recommended action: stay home / evacuate to shelter X”,
explanations avoid deep technical detail:
“We closed tunnel N-1 and diverted traffic to protect your evacuation route to Hospital A.”
10.2 CityOps / SRE view
For CityOps:
- dashboards show SCover, SCI, CAS, PoLB wave status,
- they can drill into SIR timelines and RML traces,
- they see which Jumps were Genius replays vs normal Jumps.
10.3 Auditor / regulator view
For an auditor:
- ETH traces and EvalTraces show how risk and fairness were evaluated,
- Memory governance reports show RRS, DP budgets, retention & forgetting,
- Genius trace lifecycle shows how high-impact playbooks were selected, validated, and versioned.
11. Implementation checklist: “Is our CityOS really under SI-Core?”
A non-normative checklist:
OBS
- All critical decisions read from typed semantic observations (not ad-hoc raw logs).
- Observation gaps are explicit (e.g., ObservationStatus == INCOMPLETE blocks effectful Jumps).
ID / roles / personas
- Every effectful Jump has a principal and role/persona overlay.
- Delegation chains (human → SI) are verifiable and revocable.
ETH
- Hard constraints for safety / access / fairness are explicit and enforced.
- High-risk experiments go through an approval workflow.
MEM
- SIR, JumpTrace, RMLTrace, EvalTrace, EthicsTrace are persisted with retention policies.
- Memory governance applies RRS + GCS-aware forgetting.
EVAL / experiments
- E-Jumps propose experiments with sample size, power, and stop rules.
- Multi-objective EvalSurfaces reflect real-world trade-offs (safety, access, cost).
Jump / RML
- All effectful operations cross a Jump+RML boundary with compensators where feasible.
- PoLB gates deployments based on SCover / SCI / CAS health signals.
GRP / Genius
- Genius traces are selected with robust criteria (GCS outliers, generalizability, robustness).
- Genius replay is guarded by pre-replay checks and post-replay validation.
Observability
- SCover, SCI, CAS are computed and surfaced for CityOS domains.
- Real-time alerts and postmortem tools exist for structural incidents.
If you can honestly check most of these for your CityOS, you’re close to what this article sketches: a flood-aware city that runs fully under SI-Core, where “clever” behavior is structured, replayable, and governable end-to-end.
12. Multi-domain coordination implementation
Challenge: CityOS must coordinate flood, traffic, and hospital subsystems, not just optimize each locally.
In §6 we sketched “compose joint plan” at a high level. This section makes the coordination fabric more explicit.
12.1 Coordination architecture
A simple pattern is a MultiDomainCoordinator that:
- lets each subsystem propose its own plan on its own GoalSurface,
- detects structural conflicts between those plans,
- resolves conflicts via a joint optimization over the global city GoalSurface.
class MultiDomainCoordinator:
def __init__(self, flood_subsystem, traffic_subsystem, hospital_subsystem):
self.flood_subsystem = flood_subsystem
self.traffic_subsystem = traffic_subsystem
self.hospital_subsystem = hospital_subsystem
def coordinate_emergency_response(self, obs_bundle, goal_surface):
"""Coordinate flood, traffic, and hospital subsystems in an emergency."""
# 1. Generate subsystem plans independently
flood_plan = self.flood_subsystem.propose(obs_bundle.flood)
traffic_plan = self.traffic_subsystem.propose(obs_bundle.traffic)
hospital_plan = self.hospital_subsystem.propose(obs_bundle.hospitals)
# 2. Identify conflicts across plans
conflicts = self._detect_conflicts(
flood_plan, traffic_plan, hospital_plan, obs_bundle
)
# 3. Resolve conflicts via joint optimization
if conflicts:
joint_plan = self._joint_optimization(
flood_plan, traffic_plan, hospital_plan,
conflicts, goal_surface
)
else:
joint_plan = self._compose_independent_plans(
flood_plan, traffic_plan, hospital_plan
)
return joint_plan
This coordinator is not a “super-brain” — it only:
- reads semantic plans from each domain,
- enforces hard invariants (hospital access, evacuation feasibility),
- aligns everything to the global city GoalSurface.
12.2 Conflict detection
Typical conflicts in CityOS:
- flood wants to close a tunnel,
- traffic wants to route cars through that tunnel,
- hospitals need that tunnel as a primary ambulance route.
We can capture those structurally:
class MultiDomainCoordinator:
...
def _detect_conflicts(self, flood_plan, traffic_plan, hospital_plan, obs_bundle):
"""Detect structural conflicts between subsystem plans."""
conflicts = []
# Conflict 1: Flood closes road that traffic wants to use
for flood_action in flood_plan.actions:
if flood_action.type == "road.close":
for traffic_action in traffic_plan.actions:
if getattr(traffic_action, "uses_road", None) and \
traffic_action.uses_road(flood_action.road_id):
conflicts.append(Conflict(
type="road_closure_vs_traffic_route",
subsystems=["flood", "traffic"],
actions=[flood_action, traffic_action],
))
# Conflict 2: Traffic diversion harms hospital access
for traffic_action in traffic_plan.actions:
if traffic_action.type == "route.divert":
for hospital in obs_bundle.hospitals:
if self._blocks_hospital_access(traffic_action, hospital):
conflicts.append(Conflict(
type="traffic_diversion_vs_hospital",
subsystems=["traffic", "hospital"],
hospital_id=hospital.id,
))
return conflicts
In practice this library grows with the city:
- more conflict types,
- better access checks (graph-based routing over road network),
- domain-specific “sensitivity” (e.g. oncology center vs clinic).
12.3 Joint optimization
When conflicts exist, we solve a small joint problem rather than acting on each plan independently.
Here’s a non-normative sketch:
class MultiDomainCoordinator:
...
def _joint_optimization(
self,
flood_plan,
traffic_plan,
hospital_plan,
conflicts,
goal_surface,
):
"""Jointly optimize to resolve cross-domain conflicts."""
variables = {
"flood_actions": flood_plan.alternatives,
"traffic_actions": traffic_plan.alternatives,
"hospital_actions": hospital_plan.alternatives,
}
# Hard constraints: expressed as predicates over (f, t, h)
constraints = [
# Maintain minimum hospital access
lambda f, t, h: self._hospital_access_ok(f, t, h, min_score=0.8),
# No complete blockage of evacuation routes
lambda f, t, h: self._evacuation_routes_ok(f, t),
# Conflict-specific handlers
*[self._resolve_conflict(c) for c in conflicts],
]
feasible_solutions = []
for flood_alt in variables["flood_actions"]:
for traffic_alt in variables["traffic_actions"]:
for hospital_alt in variables["hospital_actions"]:
candidate = JointPlan(
flood=flood_alt,
traffic=traffic_alt,
hospital=hospital_alt,
)
if all(c(flood_alt, traffic_alt, hospital_alt) for c in constraints):
gcs_vec = goal_surface.evaluate(candidate)
feasible_solutions.append((candidate, gcs_vec))
# Multi-objective: filter to Pareto frontier, then pick best composite
pareto_set = self._pareto_filter(feasible_solutions)
return max(pareto_set, key=lambda s: s[1].composite_score)
- The constraints encode the ETH hard rules (“no hospital cut off”).
- The GoalSurface encodes trade-offs (safety, access, traffic efficiency, fairness).
- The Pareto filter avoids “fake optima” that are strictly dominated.
12.4 Example coordination scenario
scenario: "Flood requires tunnel closure, but tunnel is critical hospital route"
independent_plans:
flood:
action: close_tunnel_north
reason: "water_ingress"
traffic:
action: route_via_tunnel_north
reason: "shortest_path"
hospital:
action: ambulance_route_tunnel_north
reason: "critical_patient"
conflict: "All three subsystems want contradictory tunnel state"
resolution:
flood:
action: close_tunnel_north
priority: high # Safety-critical
traffic:
action: route_via_bridge_west
added_travel_time_min: 5
hospital:
action: ambulance_via_bridge_west
alert: crews_about_delay
joint_gcs:
safety_bp: 9500 # Tunnel closed: good
hospital_access_bp: 8200
traffic_efficiency_bp: 6500
fairness_bp: 8800
This is exactly the kind of case where SI-Core earns its keep: the conflict is surfaced structurally and resolved globally, instead of being an emergent “oops” at 2am.
13. Failure scenarios and edge cases
Challenge: The neat “four phases” story hides messy real-world failure modes. CityOS must be explicit about how it fails and how it degrades.
13.1 Failure taxonomy
Non-normative taxonomy:
failure_scenarios:
observation_failures:
- "Sensor array offline"
- "SCE pipeline stalled"
- "Semantic compression produces contradictory state"
decision_failures:
- "Jump timeout (>5s for emergency Jumps)"
- "ETH deadlock (all candidate actions blocked)"
- "GCS evaluation failure (NaN / missing values)"
effect_failures:
- "Gate actuator jammed"
- "RML compensator fails"
- "Cascade: gate 1 fails → gate 2 overloads → gate 3 fails"
coordination_failures:
- "Multi-domain optimization timeout"
- "Conflict resolution finds no feasible solution"
Each class has its own mitigations and PoLB behavior.
13.2 Observation failure handling
When observations are incomplete or contradictory, [OBS] must degrade safely.
class ObservationFailureHandler:
def __init__(self, polb_controller):
self.polb = polb_controller
def handle_sensor_failure(self, obs_bundle):
"""Handle missing or corrupted sensors in CityOS observations."""
if obs_bundle.status != ObservationStatus.INCOMPLETE:
return obs_bundle
# Identify missing critical sensors
missing_critical = [
s for s in obs_bundle.sensors
if s.status == SensorStatus.OFFLINE and s.criticality == "critical"
]
if missing_critical:
# Fall back to degraded mode with explicit uncertainty
return self._degraded_mode_observations(obs_bundle, missing_critical)
else:
# Interpolate from nearby sensors / models
return self._interpolate_observations(obs_bundle)
def _degraded_mode_observations(self, obs_bundle, missing):
"""Use historical averages plus increased uncertainty for critical gaps."""
for sensor_id in missing:
obs_bundle.sensors[sensor_id] = ObservationValue(
value=historical_average(sensor_id),
uncertainty=Uncertainty.HIGH,
status=SensorStatus.DEGRADED,
source="historical_interpolation",
)
# Trigger PoLB mode change: operate in degraded-obs mode
self.polb.transition_to("DEGRADED_OBS_MODE")
return obs_bundle
The key invariant: we never silently pretend we “saw” something we didn’t — degraded modes are explicit, and PoLB reacts.
13.3 ETH deadlock handling
Sometimes ETH constraints block all candidate plans:
def handle_eth_deadlock(candidates, eth_constraints, escalation):
"""Handle the case where every candidate is blocked by ETH."""
feasible = [c for c in candidates if eth_constraints.check(c)]
if feasible:
return feasible # Not actually a deadlock
# True deadlock: no ETH-safe options
least_bad = compute_least_bad_option(candidates, eth_constraints)
return escalation.to_human(
reason="eth_deadlock",
context={
"candidates": candidates,
"constraints": eth_constraints.summary(),
"least_bad_option": least_bad,
},
)
SI-Core should never silently bypass ETH in a deadlock; it must escalate with clear context plus a “least bad” recommendation if one exists.
13.4 RML compensator failure and cascades
Compensators can fail too:
def handle_compensator_failure(rml_trace, sci_detector, escalation):
"""Handle failure of an RML compensator."""
# Log as SCI
sci_detector.log_incident(SCI(
type="compensator_failure",
rml_trace_id=rml_trace.id,
severity="high",
))
# Try alternative compensators if defined
for alt in rml_trace.alternative_compensators:
try:
result = alt.execute()
if result.ok:
return result
except Exception as e:
log_compensator_failure(alt, e)
# Fall back to manual intervention
return escalation.to_manual_ops(
reason="compensator_failure",
context={"rml_trace_id": rml_trace.id},
)
To prevent cascades:
cascade_prevention:
circuit_breakers:
- "If 2 gate operations fail within 5 minutes, freeze further gate operations"
- "If RBL p95 > 10s, halt new RML calls in that domain"
blast_radius_limits:
- "Max 3 concurrent RML failures before PoLB downgrades wave"
- "Single-domain failure must not automatically trigger multi-domain rollout"
This ties RBL/RIR and PoLB together: bad behavior shrinks the blast radius by design.
14. Cost analysis and resource planning
Challenge: CityOS under SI-Core is powerful—but not free. We should be explicit about the infrastructure bill and expected value.
14.1 Cost components (order-of-magnitude, non-normative)
infrastructure_costs:
compute:
jump_runtime: "$5k/month (distributed cluster)"
rml_engine: "$3k/month"
semantic_compression: "$10k/month (high-frequency sensors)"
genius_matcher: "$2k/month"
eval_engine: "$4k/month"
storage:
sim_hot: "$8k/month (90 days)"
sis_warm: "$15k/month (5 years)"
genius_library: "$2k/month"
audit_logs: "$10k/month (hash-chained, tamper-evident)"
observability:
trace_storage: "$12k/month"
scover_sci_cas_monitoring: "$3k/month"
dashboards: "$2k/month"
total_monthly: "$76k (illustrative)"
These are illustrative numbers, not guidance. The point is: SI-Core-style observability and governance are visible line items, not magical freebies.
14.2 Cost vs value (back-of-envelope)
For a flood-prone city:
incident_cost_comparison:
without_si_core:
major_flood_damage_avg: "$50M"
incidents_per_year: 2
annual_cost: "$100M"
with_si_core:
infrastructure: "$0.9M/year"
major_flood_damage_avg: "$35M"
incidents_per_year: 1.5
annual_cost: "$52.5M + $0.9M = $53.4M"
annual_savings: "$46.6M"
roi: "≈ 5100%"
Again, non-normative—but it frames SI-Core as “civic infrastructure”: like levees and power grids, not like a single SaaS subscription.
14.3 Resource planning for emergencies
CityOS must be sized for surge, not just normal days.
class ResourcePlanner:
def plan_for_emergency_capacity(self):
"""Plan compute/storage for emergency SIR surge."""
# Normal load (illustrative)
normal_sir_rate = 100 # SIRs per second
# Emergency surge (e.g. major flood)
emergency_sir_rate = 1000 # SIRs per second
compute = self._compute_requirements(emergency_sir_rate)
storage = self._storage_requirements(emergency_sir_rate)
return ResourcePlan(
compute=compute,
storage=storage,
autoscaling_rules=self._autoscaling_rules(
normal_sir_rate, emergency_sir_rate
),
)
The planner translates SIR/QPS targets into:
- cluster sizes,
- storage IO budgets,
- autoscaling policies (CPU, queue depth, or event rate based).
15. Simulation and testing strategies
Challenge: You do not want to “test in production” by waiting for real disasters. CityOS needs a simulation lab that exercises the full SI-Core stack.
15.1 CityOS simulation framework
A simulation harness runs the whole CityOS loop in silico:
class CitySimulator:
def __init__(self, obs_generator, cityos, world_simulator, evaluator):
self.obs_generator = obs_generator
self.cityos = cityos # The full SI-Core governed stack
self.world_simulator = world_simulator
self.evaluator = evaluator
def simulate_flood_scenario(self, scenario):
"""Simulate a flood scenario end-to-end."""
# 1. Generate synthetic observations over time
obs_timeline = self.obs_generator.generate(scenario)
# 2. Run CityOS decision loop
sir_timeline = []
actions = []
for obs in obs_timeline:
sir = self.cityos.run_decision_cycle(obs)
sir_timeline.append(sir)
actions.extend(sir.rml_actions)
# 3. Simulate physical world effects
world_state = self.world_simulator.simulate(
initial_state=scenario.initial_state,
actions=actions,
)
# 4. Evaluate outcomes on goal surface
evaluation = self.evaluator.evaluate(
goal_surface=scenario.goal_surface,
world_state=world_state,
)
return SimulationResult(
sir_timeline=sir_timeline,
world_state=world_state,
evaluation=evaluation,
)
This is where you:
- replay historical floods,
- generate synthetic extremes,
- test cascade and failure behaviors without touching real gates or roads.
15.2 Testing pyramid
Non-normative testing stack:
testing_strategies:
unit_tests:
- "Jump logic per domain (flood/traffic/hospital)"
- "RML effect execution and compensators"
- "ETH constraint checking"
- "GCS evaluation functions"
integration_tests:
- "Full SIR cycle (obs → jump → rml → effects)"
- "Multi-domain coordination path"
- "Genius replay end-to-end"
simulation_tests:
- "Historical flood replays"
- "Synthetic extreme scenarios"
- "Cascade failure scenarios"
chaos_tests:
- "Random sensor failures"
- "Random RML failures"
- "Network partitions / delayed observations"
15.3 Property-based & Genius-based regression tests
Property-based tests assert structural invariants, not just example cases:
from hypothesis import given, strategies as st
@given(flood_scenario=st.from_type(FloodScenario))
def test_hospital_access_always_maintained(flood_scenario):
"""Hospital access score must stay above 0.8 in all simulated floods."""
result = cityos_simulator.simulate_flood_scenario(flood_scenario)
for sir in result.sir_timeline:
assert sir.hospital_access_score_bp >= 8000
Genius traces (GRP) become regression tests:
def test_genius_trace_still_performs_well():
"""Genius trace from 2030 flood still yields strong GCS in simulation."""
genius_trace = genius_library.get("GRP-2030-09-flood-north-basin")
# Replay in simulation under slightly changed conditions
result = genius_replayer.replay_in_simulation(
genius_trace=genius_trace,
scenario=current_reference_scenario,
)
# It should still deliver a strong improvement vs baseline
assert result.gcs_improvement >= 0.20
This connects GRP back into CI/CD: a Genius trace is not just a “playbook,” it’s also a test of whether your latest CityOS changes preserved that playbook’s effectiveness.