Tyler King
89a054d656
Python SDK for shellbound Django applications. Provides ShellApp, ShardContext, ShellboundMiddleware. Emits Chronicle events to stdout in dev mode. Includes fix for IndexError in apps.py when DJANGO_SETTINGS_MODULE has no dots (e.g. instance_settings). Shard name now falls back safely without eager default argument parsing. Implements SHELLBOUND-APP-0001 §4 (dev mode). Wired into entropyopposition as of 2026-03-18.
110 lines
3.3 KiB
Python
110 lines
3.3 KiB
Python
"""Chronicle event emission via stdout (Phase A) or HFL direct binding (Phase B).
|
|
|
|
Phase A (current): Writes substrate_event JSON lines to stdout.
|
|
bascule-filter-python reads and translates to ChronicleEntry records.
|
|
|
|
Phase B (planned): When substrate-hfl-python (PyO3) is installed,
|
|
routes events through HFL kernel module directly:
|
|
chronicle_write(), accord_check(), session_get()
|
|
Tier 4 → Tier 1/2 upgrade with zero application code changes.
|
|
Detection: try import substrate_hfl; if available, use HFL path.
|
|
|
|
All writes are non-blocking. All failures are silently logged.
|
|
Chronicle emission MUST NOT affect application behavior.
|
|
"""
|
|
|
|
import json
|
|
import logging
|
|
import sys
|
|
import time
|
|
from typing import Optional
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# TODO Phase B: HFL direct binding detection
|
|
# When substrate-hfl-python (PyO3) is installed, use HFL path:
|
|
# try:
|
|
# import substrate_hfl
|
|
# _hfl_available = True
|
|
# except ImportError:
|
|
# _hfl_available = False
|
|
# Then in emit_event(): if _hfl_available, call substrate_hfl.chronicle_write()
|
|
# instead of stdout print(). Zero application code changes required.
|
|
_hfl_available = False
|
|
|
|
LAYER_TCB = 0
|
|
LAYER_IDENTITY = 1
|
|
LAYER_GOVERNANCE = 2
|
|
LAYER_APPLICATION = 3
|
|
|
|
|
|
def emit_event(
|
|
kind: str,
|
|
layer: int,
|
|
payload: dict,
|
|
actor_did: str = "",
|
|
accord_hash: str = "",
|
|
shard_id: str = "",
|
|
) -> bool:
|
|
"""Emit a Chronicle event to stdout.
|
|
|
|
Returns True if written, False on error (non-fatal).
|
|
The substrate_event marker tells bascule-filter this is a Chronicle event.
|
|
"""
|
|
event = {
|
|
"substrate_event": True,
|
|
"kind": kind,
|
|
"layer": layer,
|
|
"actor_did": actor_did,
|
|
"accord_hash": accord_hash,
|
|
"shard_id": shard_id,
|
|
"occurred_at": time.time_ns(),
|
|
"payload": payload,
|
|
}
|
|
try:
|
|
line = json.dumps(event, separators=(",", ":"), sort_keys=True)
|
|
print(line, flush=True)
|
|
return True
|
|
except Exception as e:
|
|
logger.warning("Chronicle emit failed (non-fatal): %s", e)
|
|
return False
|
|
|
|
|
|
class ShardEmitter:
|
|
"""Chronicle emitter scoped to a shard. Carries shard_id, actor_did, accord_hash."""
|
|
|
|
def __init__(self, context):
|
|
self._ctx = context
|
|
|
|
def emit(self, kind: str, layer: int, payload: dict, actor_did: Optional[str] = None) -> bool:
|
|
return emit_event(
|
|
kind=kind,
|
|
layer=layer,
|
|
payload=payload,
|
|
actor_did=actor_did or self._ctx.actor_did,
|
|
accord_hash=self._ctx.accord_hash,
|
|
shard_id=self._ctx.shard_id,
|
|
)
|
|
|
|
def request(
|
|
self,
|
|
method: str,
|
|
path: str,
|
|
status: int,
|
|
duration_ms: float,
|
|
actor_did: Optional[str] = None,
|
|
view_name: Optional[str] = None,
|
|
) -> bool:
|
|
return self.emit(
|
|
"APP_DJANGO_REQUEST",
|
|
LAYER_APPLICATION,
|
|
{"method": method, "path": path, "status": status, "duration_ms": duration_ms, "view_name": view_name},
|
|
actor_did=actor_did,
|
|
)
|
|
|
|
def signal(self, signal_name: str, model: str, instance_id: Optional[str] = None) -> bool:
|
|
return self.emit(
|
|
"APP_DJANGO_SIGNAL",
|
|
LAYER_APPLICATION,
|
|
{"signal": signal_name, "model": model, "instance_id": instance_id},
|
|
)
|